From 933eefd6c9ad2fc4db2eb635d4babd182e519a59 Mon Sep 17 00:00:00 2001 From: Vyacheslav-Tomashevskiy Date: Sat, 4 Jul 2026 22:50:33 +0200 Subject: [PATCH] feat(indexer): decode Soroban event XDR and upsert contributions & loans (closes #2) fetchAndStoreEvents had a TODO where decoded events should be persisted, so the Horizon sync loop was a no-op. Decode each event's topic/value XDR with scValToNative and route by event name onto the Prisma models: - contribution (member, amount, period) -> Contribution, idempotent on txHash - loan_requested/approved/repaid (id, borrower, amount[, status]) -> Loan, matched by (groupId, onChainId) so lifecycle updates don't duplicate rows Unsuccessful calls are skipped; a single unparseable record is logged and skipped instead of aborting the batch. Adds a ts-jest config and a spec that builds real event XDR the same way the contracts emit it. --- jest.config.js | 8 + src/common/stellar-indexer.service.spec.ts | 137 +++++++++++++++++ src/common/stellar-indexer.service.ts | 170 ++++++++++++++++++++- 3 files changed, 310 insertions(+), 5 deletions(-) create mode 100644 jest.config.js create mode 100644 src/common/stellar-indexer.service.spec.ts diff --git a/jest.config.js b/jest.config.js new file mode 100644 index 0000000..cbd58bb --- /dev/null +++ b/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"], +}; diff --git a/src/common/stellar-indexer.service.spec.ts b/src/common/stellar-indexer.service.spec.ts new file mode 100644 index 0000000..19a8dfa --- /dev/null +++ b/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(); + }); +}); diff --git a/src/common/stellar-indexer.service.ts b/src/common/stellar-indexer.service.ts index 5951b57..d6f40c6 100644 --- a/src/common/stellar-indexer.service.ts +++ b/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; + 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 = { + 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; } }