feat(indexer): decode Soroban event XDR and upsert contributions & loans (closes #2)

jest.config.js

@@ -0,0 +1,8 @@
+/** @type {import('jest').Config} */
+module.exports = {
+  preset: "ts-jest",
+  testEnvironment: "node",
+  rootDir: "src",
+  testRegex: ".*\\.spec\\.ts$",
+  moduleFileExtensions: ["js", "json", "ts"],
+};

src/common/stellar-indexer.service.spec.ts

@@ -0,0 +1,137 @@
+import { Test, TestingModule } from "@nestjs/testing";
+import { ConfigService } from "@nestjs/config";
+import { nativeToScVal, Address, Keypair } from "@stellar/stellar-sdk";
+import { StellarIndexerService } from "./stellar-indexer.service";
+import { PrismaService } from "./prisma.service";
+
+/**
+ * Build a base64 XDR ScVal the same way the Soroban contracts emit their
+ * event topics/values, so the decoder is exercised end-to-end.
+ */
+const sym = (s: string) => nativeToScVal(s, { type: "symbol" }).toXDR("base64");
+const addr = (a: string) => nativeToScVal(new Address(a), { type: "address" });
+
+// A throwaway but valid Stellar public key for the borrower/member fields.
+const MEMBER = Keypair.random().publicKey();
+
+function contributionEvent() {
+  const value = nativeToScVal([
+    addr(MEMBER),
+    nativeToScVal(1000n, { type: "i128" }),
+    nativeToScVal(3, { type: "u32" }),
+  ]).toXDR("base64");
+  return {
+    id: "ev-contribution-1",
+    type: "contract",
+    ledger: 42,
+    transaction_hash: "tx-abc",
+    in_successful_contract_call: true,
+    topic: [sym("contribution")],
+    value,
+  };
+}
+
+function loanRequestedEvent() {
+  const value = nativeToScVal([
+    nativeToScVal(7, { type: "u32" }),
+    addr(MEMBER),
+    nativeToScVal(5000n, { type: "i128" }),
+  ]).toXDR("base64");
+  return {
+    id: "ev-loan-1",
+    type: "contract",
+    ledger: 43,
+    transaction_hash: "tx-loan",
+    in_successful_contract_call: true,
+    topic: [sym("loan_requested")],
+    value,
+  };
+}
+
+describe("StellarIndexerService", () => {
+  let service: StellarIndexerService;
+  let prisma: {
+    contribution: { upsert: jest.Mock };
+    loan: { findFirst: jest.Mock; create: jest.Mock; update: jest.Mock };
+  };
+
+  beforeEach(async () => {
+    prisma = {
+      contribution: { upsert: jest.fn() },
+      loan: { findFirst: jest.fn().mockResolvedValue(null), create: jest.fn(), update: jest.fn() },
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        StellarIndexerService,
+        { provide: PrismaService, useValue: prisma },
+        { provide: ConfigService, useValue: { get: (_k: string, d: string) => d } },
+      ],
+    }).compile();
+
+    service = module.get(StellarIndexerService);
+  });
+
+  it("is defined", () => {
+    expect(service).toBeDefined();
+  });
+
+  it("decodes a contribution event and upserts it on txHash", async () => {
+    await (service as any).processEvent("group-1", contributionEvent());
+
+    expect(prisma.contribution.upsert).toHaveBeenCalledTimes(1);
+    const arg = prisma.contribution.upsert.mock.calls[0][0];
+    expect(arg.where).toEqual({ txHash: "tx-abc" });
+    expect(arg.create).toMatchObject({
+      groupId: "group-1",
+      memberAddress: MEMBER,
+      amount: 1000n,
+      period: 3,
+      txHash: "tx-abc",
+      ledger: 42,
+    });
+  });
+
+  it("decodes a loan_requested event and creates a Pending loan", async () => {
+    await (service as any).processEvent("group-1", loanRequestedEvent());
+
+    expect(prisma.loan.findFirst).toHaveBeenCalledWith({
+      where: { groupId: "group-1", onChainId: 7 },
+      select: { id: true },
+    });
+    expect(prisma.loan.create).toHaveBeenCalledTimes(1);
+    expect(prisma.loan.create.mock.calls[0][0].data).toMatchObject({
+      groupId: "group-1",
+      onChainId: 7,
+      borrower: MEMBER,
+      amount: 5000n,
+      status: "Pending",
+      txHash: "tx-loan",
+    });
+  });
+
+  it("updates an existing loan instead of duplicating it", async () => {
+    prisma.loan.findFirst.mockResolvedValueOnce({ id: "loan-row-1" });
+
+    const approved = loanRequestedEvent();
+    approved.topic = [sym("loan_approved")];
+
+    await (service as any).processEvent("group-1", approved);
+
+    expect(prisma.loan.create).not.toHaveBeenCalled();
+    expect(prisma.loan.update).toHaveBeenCalledTimes(1);
+    const data = prisma.loan.update.mock.calls[0][0].data;
+    expect(data.status).toBe("Approved");
+    expect(data.approvedAt).toBeInstanceOf(Date);
+  });
+
+  it("ignores unrelated events (e.g. withdrawal)", async () => {
+    const withdrawal = contributionEvent();
+    withdrawal.topic = [sym("withdrawal")];
+
+    await (service as any).processEvent("group-1", withdrawal);
+
+    expect(prisma.contribution.upsert).not.toHaveBeenCalled();
+    expect(prisma.loan.create).not.toHaveBeenCalled();
+  });
+});

src/common/stellar-indexer.service.ts

@@ -1,12 +1,24 @@
 import { Injectable, Logger } from "@nestjs/common";
 import { Cron, CronExpression } from "@nestjs/schedule";
 import { ConfigService } from "@nestjs/config";
+import { xdr, scValToNative } from "@stellar/stellar-sdk";
 import { PrismaService } from "./prisma.service";
 
 /**
- * Polls Stellar Horizon for contract events and syncs them to the database.
- * This bridges on-chain state with the off-chain PostgreSQL store.
+ * A single contract event as returned by Horizon's
+ * `/contracts/{id}/events` endpoint. Only the fields we consume are typed;
+ * `topic`/`value` carry the base64 XDR ScVal payloads emitted on-chain.
  */
+interface ContractEventRecord {
+  id: string;
+  type: string;
+  ledger?: number;
+  transaction_hash?: string;
+  in_successful_contract_call: boolean;
+  topic: Array<string | { xdr: string }>;
+  value: string | { xdr: string };
+}
+
 @Injectable()
 export class StellarIndexerService {
   private readonly logger = new Logger(StellarIndexerService.name);
@@ -67,11 +79,159 @@ export class StellarIndexerService {
     if (!res.ok) return;
 
     const { _embedded: { records } } = await res.json() as {
-      _embedded: { records: Array<{ id: string; type: string; in_successful_contract_call: boolean }> }
+      _embedded: { records: ContractEventRecord[] };
     };
 
     this.logger.debug(`Found ${records.length} ${eventType} events for contract ${contractId}`);
-    // TODO: Parse and upsert into contributions / loans tables
-    // This is where you'd decode the XDR event payloads from Soroban
+
+    for (const record of records) {
+      // Only index events emitted by a successful contract invocation.
+      if (record.in_successful_contract_call === false) continue;
+
+      try {
+        await this.processEvent(groupId, record);
+      } catch (err) {
+        // A single malformed record must not abort the whole batch.
+        this.logger.warn(
+          `Skipping unparseable event ${record.id} on ${contractId}: ${
+            err instanceof Error ? err.message : err
+          }`,
+        );
+      }
+    }
+  }
+
+  /**
+   * Decode a single Soroban event's XDR payload and upsert it into the
+   * corresponding table. Routing is driven by the event name carried in the
+   * first topic (e.g. `contribution`, `loan_requested`, ...).
+   */
+  private async processEvent(groupId: string, record: ContractEventRecord) {
+    const name = this.decodeTopicName(record.topic);
+    if (!name) return;
+
+    const value = scValToNative(this.toScVal(record.value));
+
+    switch (name) {
+      case "contribution":
+        // Emitted as `(member: Address, amount: i128, period: u32)`.
+        await this.upsertContribution(groupId, record, value as [string, bigint, number]);
+        break;
+
+      case "loan_requested":
+      case "loan_approved":
+      case "loan_repaid":
+        // `(id/loan_id: u32, borrower: Address, amount: i128, [status])`.
+        await this.upsertLoan(groupId, record, name, value as unknown[]);
+        break;
+
+      // `withdrawal`, `member_added`, etc. carry no row to sync here.
+      default:
+        break;
+    }
+  }
+
+  private async upsertContribution(
+    groupId: string,
+    record: ContractEventRecord,
+    [memberAddress, amount, period]: [string, bigint, number],
+  ) {
+    const txHash = record.transaction_hash ?? record.id;
+    const ledger = record.ledger ?? null;
+
+    await this.prisma.contribution.upsert({
+      where: { txHash },
+      update: {
+        amount: BigInt(amount),
+        period: Number(period),
+        ledger,
+      },
+      create: {
+        groupId,
+        memberAddress,
+        amount: BigInt(amount),
+        period: Number(period),
+        txHash,
+        ledger,
+      },
+    });
+  }
+
+  private async upsertLoan(
+    groupId: string,
+    record: ContractEventRecord,
+    name: string,
+    value: unknown[],
+  ) {
+    const [onChainIdRaw, borrower, amount, status] = value as [
+      number | bigint,
+      string,
+      bigint,
+      unknown?,
+    ];
+    const onChainId = Number(onChainIdRaw);
+    const txHash = record.transaction_hash ?? record.id;
+
+    const statusByEvent: Record<string, string> = {
+      loan_requested: "Pending",
+      loan_approved: "Approved",
+      loan_repaid: "Repaid",
+    };
+    // `loan_repaid` also carries the on-chain status; prefer it when present
+    // (a partial repayment keeps the loan `Approved`).
+    const derivedStatus = this.normalizeStatus(status) ?? statusByEvent[name];
+
+    const existing = await this.prisma.loan.findFirst({
+      where: { groupId, onChainId },
+      select: { id: true },
+    });
+
+    if (existing) {
+      await this.prisma.loan.update({
+        where: { id: existing.id },
+        data: {
+          amount: BigInt(amount),
+          status: derivedStatus,
+          txHash,
+          ...(name === "loan_approved" ? { approvedAt: new Date() } : {}),
+          ...(name === "loan_repaid" && derivedStatus === "Repaid"
+            ? { repaidAt: new Date() }
+            : {}),
+        },
+      });
+    } else {
+      await this.prisma.loan.create({
+        data: {
+          groupId,
+          onChainId,
+          borrower,
+          amount: BigInt(amount),
+          interestBps: 0,
+          purpose: "",
+          status: derivedStatus,
+          txHash,
+        },
+      });
+    }
+  }
+
+  /** Decode the event name from the first topic ScVal (a Symbol). */
+  private decodeTopicName(topic: ContractEventRecord["topic"]): string | null {
+    if (!topic || topic.length === 0) return null;
+    const decoded = scValToNative(this.toScVal(topic[0]));
+    return typeof decoded === "string" ? decoded : String(decoded);
+  }
+
+  /** Accept either a raw base64 XDR string or a `{ xdr }` wrapper. */
+  private toScVal(payload: string | { xdr: string }): xdr.ScVal {
+    const base64 = typeof payload === "string" ? payload : payload.xdr;
+    return xdr.ScVal.fromXDR(base64, "base64");
+  }
+
+  /** Contract enums decode to their variant name (as a string or `[name]`). */
+  private normalizeStatus(status: unknown): string | undefined {
+    if (typeof status === "string") return status;
+    if (Array.isArray(status) && typeof status[0] === "string") return status[0];
+    return undefined;
   }
 }