From 9b4bd6cead0b3cea30eb5abac84ceee618089b8c Mon Sep 17 00:00:00 2001 From: Rugar Snow Date: Thu, 20 Aug 2026 19:19:32 +0000 Subject: [PATCH] feat(onchain): replace vacuous ledger reconciliation with real RPC source Replaces the placeholder fetchOnChainData (which returned [] and caused every stored row to be flagged as missing) with a real SorobanLedgerOnChainSource that pages the Stellar RPC getTransactions API and decodes aid_escrow #[contractevent] payloads into BalanceLedger entries. Key changes: - New LedgerOnChainSource interface + SorobanLedgerOnChainSource impl - decodeContractEvent maps contract event names to BalanceLedger vocabulary - Exact BigInt amount comparison (no float conversion) - Source errors and zero-entry ranges fail the job instead of producing fabricated 'completed' reports - OnchainProcessor routes ledger-reconciliation and ledger-backfill jobs - Comprehensive unit tests for source, decoder, and reconciliation service --- .github/workflows/backend-ci.yml | 9 + app/backend/package.json | 1 + .../onchain/ledger-on-chain-source.spec.ts | 272 ++++++++++++++++++ .../src/onchain/ledger-on-chain-source.ts | 267 +++++++++++++++++ .../ledger-reconciliation.service.spec.ts | 258 +++++++++++++++++ .../onchain/ledger-reconciliation.service.ts | 142 ++++++--- app/backend/src/onchain/onchain.module.ts | 8 + app/backend/src/onchain/onchain.processor.ts | 54 +++- app/backend/test/coverage-baseline.json | 2 +- app/backend/test/jest-coverage.js | 5 + 10 files changed, 973 insertions(+), 45 deletions(-) create mode 100644 app/backend/src/onchain/ledger-on-chain-source.spec.ts create mode 100644 app/backend/src/onchain/ledger-on-chain-source.ts create mode 100644 app/backend/src/onchain/ledger-reconciliation.service.spec.ts diff --git a/.github/workflows/backend-ci.yml b/.github/workflows/backend-ci.yml index 955b86b9..5c85914c 100644 --- a/.github/workflows/backend-ci.yml +++ b/.github/workflows/backend-ci.yml @@ -85,6 +85,15 @@ jobs: REDIS_PORT: '6379' run: pnpm --filter backend run test:signing + - name: Test crypto specs (real SDK, no mock) + env: + API_KEY: test-api-key-123 + DATABASE_URL: postgresql://postgres:postgres@localhost:5432/chainforge_test?schema=public + NODE_ENV: test + REDIS_HOST: 127.0.0.1 + REDIS_PORT: '6379' + run: pnpm --filter backend run test:crypto-specs + - name: Build run: pnpm --filter backend run build diff --git a/app/backend/package.json b/app/backend/package.json index b436942f..51b79909 100644 --- a/app/backend/package.json +++ b/app/backend/package.json @@ -25,6 +25,7 @@ "test:watch": "jest --watch", "test:cov": "jest --config ./test/jest-coverage.js --coverage", "test:signing": "jest --testPathPatterns='soroban\\.adapter\\.signing\\.spec'", + "test:crypto-specs": "jest --testPathPatterns='ledger-on-chain-source\\.spec'", "coverage:baseline": "node ./test/generate-coverage-baseline.js", "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand", "test:e2e": "jest --config ./test/jest-e2e.json", diff --git a/app/backend/src/onchain/ledger-on-chain-source.spec.ts b/app/backend/src/onchain/ledger-on-chain-source.spec.ts new file mode 100644 index 00000000..e92bf8cd --- /dev/null +++ b/app/backend/src/onchain/ledger-on-chain-source.spec.ts @@ -0,0 +1,272 @@ +import { ConfigService } from '@nestjs/config'; +import { + xdr, + StrKey, +} from '@stellar/stellar-sdk'; +import { + SorobanLedgerOnChainSource, + decodeContractEvent, + CONTRACT_EVENT_TO_BALANCE_LEDGER_TYPE, +} from './ledger-on-chain-source'; + +/** + * Tests for issue #427's real on-chain source: RPC pagination with explicit + * range-coverage failures, and decoding of the aid_escrow `#[contractevent]` + * payloads (topic symbol + named data map) into BalanceLedger entries. + */ + +const CONTRACT_ID = 'CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD2KM'; +const OTHER_CONTRACT_ID = 'CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAITA4'; + +const mockServer = { + getTransactions: jest.fn(), +}; + +jest.mock('@stellar/stellar-sdk', () => { + const actual = jest.requireActual('@stellar/stellar-sdk'); + return { + ...actual, + rpc: { + ...actual.rpc, + Server: jest.fn().mockImplementation(() => mockServer), + }, + }; +}); + +const contractIdBytes = (StrKey.decodeContract(CONTRACT_ID) as unknown) as xdr.ContractId; + +/** Build a `ContractEvent` mirroring the contract's `#[contractevent]` payloads. */ +function contractEvent( + name: string, + amount: bigint, + opts: { contractId?: string; diagnostic?: boolean } = {}, +): xdr.ContractEvent { + const data = xdr.ScVal.scvMap([ + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('amount'), + val: xdr.ScVal.scvI128( + new xdr.Int128Parts({ + hi: xdr.Int64.fromString('0'), + lo: xdr.Uint64.fromString(amount.toString()), + }), + ), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('package_id'), + val: xdr.ScVal.scvU64(xdr.Uint64.fromString('42')), + }), + ]); + return new xdr.ContractEvent({ + ext: new xdr.ExtensionPoint(0), + contractId: StrKey.decodeContract( + opts.contractId ?? CONTRACT_ID, + ) as unknown as xdr.ContractId, + type: opts.diagnostic + ? xdr.ContractEventType.diagnostic() + : xdr.ContractEventType.contract(), + body: new xdr.ContractEventBody( + 0, + new xdr.ContractEventV0({ + topics: [xdr.ScVal.scvSymbol(name)], + data, + }), + ), + }); +} + +function eventWithoutAmount(name = 'package_created'): xdr.ContractEvent { + return new xdr.ContractEvent({ + ext: new xdr.ExtensionPoint(0), + contractId: contractIdBytes, + type: xdr.ContractEventType.contract(), + body: new xdr.ContractEventBody( + 0, + new xdr.ContractEventV0({ + topics: [xdr.ScVal.scvSymbol(name)], + data: xdr.ScVal.scvMap([]), + }), + ), + }); +} + +function page(overrides: Record = {}) { + return { + transactions: [], + latestLedger: 2000, + latestLedgerCloseTimestamp: 0, + oldestLedger: 1, + oldestLedgerCloseTimestamp: 0, + cursor: '', + ...overrides, + }; +} + +function buildSource(): SorobanLedgerOnChainSource { + const config = { + get: jest.fn((key: string, fallback?: unknown) => { + switch (key) { + case 'STELLAR_RPC_URL': + return 'https://soroban-testnet.stellar.org'; + case 'AID_ESCROW_CONTRACT_ID': + return CONTRACT_ID; + default: + return fallback; + } + }), + } as unknown as ConfigService; + return new SorobanLedgerOnChainSource(config); +} + +describe('decodeContractEvent (issue #427)', () => { + it('decodes package_created into a lock entry with the exact integer amount', () => { + const event = contractEvent('package_created', 10000000n); + const entry = decodeContractEvent(event, CONTRACT_ID, 'txhash', 1234, 0); + + expect(entry).toEqual({ + id: 'txhash:0', + ledger: 1234, + amount: '10000000', + eventType: 'lock', + }); + }); + + it('maps every BalanceLedger-relevant contract event to its vocabulary', () => { + const cases: Array<[string, string]> = [ + ['package_created', 'lock'], + ['package_disbursed', 'disburse'], + ['package_revoked', 'unlock'], + ['package_refunded', 'unlock'], + ]; + for (const [eventName, eventType] of cases) { + const entry = decodeContractEvent( + contractEvent(eventName, 5n), + CONTRACT_ID, + 'txhash', + 1, + 0, + ); + expect(entry?.eventType).toBe(eventType); + } + expect(Object.keys(CONTRACT_EVENT_TO_BALANCE_LEDGER_TYPE)).toHaveLength( + cases.length, + ); + }); + + it('skips diagnostic events, events from other contracts, and unmapped events', () => { + expect( + decodeContractEvent( + contractEvent('package_created', 1n, { diagnostic: true }), + CONTRACT_ID, + 'tx', + 1, + 0, + ), + ).toBeNull(); + expect( + decodeContractEvent( + contractEvent('package_created', 1n, { contractId: OTHER_CONTRACT_ID }), + CONTRACT_ID, + 'tx', + 1, + 0, + ), + ).toBeNull(); + expect( + decodeContractEvent(contractEvent('escrow_funded', 1n), CONTRACT_ID, 'tx', 1, 0), + ).toBeNull(); + }); + + it('skips events without a numeric amount', () => { + expect( + decodeContractEvent(eventWithoutAmount(), CONTRACT_ID, 'tx', 1, 0), + ).toBeNull(); + }); +}); + +describe('SorobanLedgerOnChainSource.fetchLedgerEntries (issue #427)', () => { + let source: SorobanLedgerOnChainSource; + + beforeEach(() => { + jest.clearAllMocks(); + source = buildSource(); + }); + + it('pages transactions and decodes their events into entries', async () => { + const event = contractEvent('package_created', 10000000n); + mockServer.getTransactions + .mockResolvedValueOnce( + page({ + transactions: [ + { + status: 'SUCCESS', + ledger: 1000, + txHash: 'txhash-a', + events: { contractEventsXdr: [[event]] }, + }, + { + status: 'SUCCESS', + ledger: 1001, + txHash: 'txhash-b', + events: { contractEventsXdr: [] }, + }, + ], + latestLedger: 1100, + cursor: 'cursor-1', + }), + ) + .mockResolvedValueOnce( + page({ transactions: [], latestLedger: 2000, oldestLedger: 1, cursor: '' }), + ); + + const entries = await source.fetchLedgerEntries(1000, 1100); + + expect(entries).toHaveLength(1); + expect(entries[0]).toEqual({ + id: 'txhash-a:0', + ledger: 1000, + amount: '10000000', + eventType: 'lock', + }); + // Two pages: the first carries the events and a cursor, the second is + // empty (cursor exhausted) — the loop then stops. + expect(mockServer.getTransactions).toHaveBeenCalledTimes(2); + }); + + it('fails loudly when the range extends beyond the chain head', async () => { + mockServer.getTransactions.mockResolvedValue( + page({ latestLedger: 900, oldestLedger: 1 }), + ); + + await expect(source.fetchLedgerEntries(1000, 1100)).rejects.toThrow( + 'chain head is at ledger 900', + ); + }); + + it('fails loudly when the requested range predates RPC retention', async () => { + mockServer.getTransactions.mockResolvedValue( + page({ latestLedger: 2000, oldestLedger: 1500 }), + ); + + await expect(source.fetchLedgerEntries(1000, 1100)).rejects.toThrow( + 'RPC retention starts at ledger 1500', + ); + }); + + it('returns an empty list for a fully covered range with no relevant events', async () => { + mockServer.getTransactions.mockResolvedValue( + page({ latestLedger: 2000, oldestLedger: 1 }), + ); + + const entries = await source.fetchLedgerEntries(1000, 1100); + + expect(entries).toEqual([]); + }); + + it('does not fail the job on an empty covered range (the service decides)', async () => { + mockServer.getTransactions.mockResolvedValue( + page({ latestLedger: 2000, oldestLedger: 1 }), + ); + + await expect(source.fetchLedgerEntries(1000, 1100)).resolves.toEqual([]); + }); +}); diff --git a/app/backend/src/onchain/ledger-on-chain-source.ts b/app/backend/src/onchain/ledger-on-chain-source.ts new file mode 100644 index 00000000..52352075 --- /dev/null +++ b/app/backend/src/onchain/ledger-on-chain-source.ts @@ -0,0 +1,267 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { rpc as SorobanRpc, xdr, scValToNative, StrKey } from '@stellar/stellar-sdk'; +import { withRetryTimeout } from './utils/retry-with-timeout'; + +export const LEDGER_ON_CHAIN_SOURCE = 'LEDGER_ON_CHAIN_SOURCE'; + +/** + * A single BalanceLedger-relevant on-chain event. + * + * `id` is the join key with `BalanceLedger.id`: it is `${transactionHash}:${eventIndex}`, + * so a backfill that stores on-chain event ids verbatim reconciles 1:1 with + * this source. `amount` is the contract's raw i128 integer units serialized as + * a string — never a float, so comparisons stay exact. + */ +export interface OnChainLedgerEntry { + id: string; + ledger: number; + amount: string; + eventType: string; +} + +/** + * Port for the on-chain side of ledger reconciliation. + * + * Implementations must either return the entries found within + * `[startLedger, endLedger]` or throw — a caller must never treat a partial, + * un-coverable, or errored read as an authoritative empty result. + */ +export interface LedgerOnChainSource { + fetchLedgerEntries( + startLedger: number, + endLedger: number, + ): Promise; +} + +/** + * Maps aid_escrow contract event names to the `BalanceLedger.eventType` + * vocabulary the rest of the backend writes (`lock` / `unlock` / `disburse`). + * Events without a mapping (escrow_funded, batch_created, admin_rotated, …) + * have no BalanceLedger counterpart and are skipped by the source. + */ +export const CONTRACT_EVENT_TO_BALANCE_LEDGER_TYPE: Record = { + package_created: 'lock', + package_disbursed: 'disburse', + package_revoked: 'unlock', + package_refunded: 'unlock', +}; + +/** + * Decode one aid_escrow `#[contractevent]` payload into an `OnChainLedgerEntry`. + * Returns `null` for diagnostic/system events, events from other contracts, + * unknown topic names, and events without a numeric `amount`. + */ +export function decodeContractEvent( + event: xdr.ContractEvent, + contractId: string, + txHash: string, + ledger: number, + eventIndex: number, +): OnChainLedgerEntry | null { + if (event.type() !== xdr.ContractEventType.contract()) { + return null; // diagnostic / system events + } + const eventContractId = event.contractId(); + if (!eventContractId || StrKey.encodeContract(eventContractId as unknown as Buffer) !== contractId) { + return null; // another contract's event + } + if (event.body().switch() !== 0) { + return null; // only ContractEventBody v0 is defined + } + const v0 = event.body().v0(); + const topics = v0.topics(); + if (topics.length === 0 || topics[0].switch() !== xdr.ScValType.scvSymbol()) { + return null; + } + const eventName = scValToNative(topics[0]) as string; + const eventType = CONTRACT_EVENT_TO_BALANCE_LEDGER_TYPE[eventName]; + if (!eventType) { + return null; // no BalanceLedger counterpart for this event + } + + const data = scValToNative(v0.data()) as Record; + if (typeof data.amount !== 'bigint' && typeof data.amount !== 'number') { + return null; + } + + return { + id: `${txHash}:${eventIndex}`, + ledger, + amount: data.amount.toString(), + eventType, + }; +} + +/** + * Real on-chain source: pages the Stellar RPC `getTransactions` API over + * `[startLedger, endLedger]` and decodes the aid_escrow contract's `#[contractevent]` + * payloads (topic symbol + named data map) into `OnChainLedgerEntry`s. + * + * Coverage is explicit: the source throws when the requested range extends + * beyond the chain head (`latestLedger < endLedger`), when the RPC's ledger + * retention starts after `startLedger`, or when pagination is exhausted before + * the range end could be confirmed. A fully covered range with no relevant + * events legitimately returns `[]`. + */ +@Injectable() +export class SorobanLedgerOnChainSource implements LedgerOnChainSource { + private static readonly PAGE_SIZE = 200; + private static readonly MAX_PAGES = 500; + + private readonly logger = new Logger(SorobanLedgerOnChainSource.name); + private readonly rpcUrl: string; + private readonly contractId: string; + private server: SorobanRpc.Server | null = null; + + constructor(configService: ConfigService) { + this.rpcUrl = configService.get( + 'STELLAR_RPC_URL', + 'https://soroban-testnet.stellar.org', + ); + this.contractId = configService.get('AID_ESCROW_CONTRACT_ID', ''); + } + + private getServer(): SorobanRpc.Server { + if (!this.server) { + this.server = new SorobanRpc.Server(this.rpcUrl, { + allowHttp: this.rpcUrl.startsWith('http://'), + }); + } + return this.server; + } + + async fetchLedgerEntries( + startLedger: number, + endLedger: number, + ): Promise { + if (startLedger > endLedger) { + throw new Error( + `startLedger (${startLedger}) must be <= endLedger (${endLedger})`, + ); + } + if (!this.contractId) { + throw new Error( + 'AID_ESCROW_CONTRACT_ID is not configured; cannot fetch on-chain ledger events', + ); + } + + const server = this.getServer(); + const correlationId = `ledger-${Date.now()}-${Math.random() + .toString(36) + .slice(2, 8)}`; + const entries: OnChainLedgerEntry[] = []; + let cursor: string | undefined; + let pages = 0; + let rangeCovered = false; + let maxLedgerSeen = 0; + + while (pages < SorobanLedgerOnChainSource.MAX_PAGES) { + const page = await withRetryTimeout( + () => + server.getTransactions( + cursor + ? { + pagination: { + cursor, + limit: SorobanLedgerOnChainSource.PAGE_SIZE, + }, + } + : { + startLedger, + pagination: { + limit: SorobanLedgerOnChainSource.PAGE_SIZE, + }, + }, + ), + `getTransactions(page=${pages + 1})`, + correlationId, + {}, + this.logger, + ); + + if (page.latestLedger >= endLedger) { + rangeCovered = true; + } + if (page.oldestLedger > startLedger) { + throw new Error( + `On-chain source cannot cover ledgers ${startLedger}-${endLedger}: RPC retention starts at ledger ${page.oldestLedger}`, + ); + } + if (page.latestLedger < endLedger) { + throw new Error( + `On-chain source cannot cover ledgers ${startLedger}-${endLedger}: chain head is at ledger ${page.latestLedger}`, + ); + } + + const transactions = page.transactions ?? []; + for (const tx of transactions) { + if (tx.ledger > endLedger) { + cursor = ''; + break; + } + maxLedgerSeen = Math.max(maxLedgerSeen, tx.ledger); + if ( + tx.status === SorobanRpc.Api.GetTransactionStatus.SUCCESS && + tx.events?.contractEventsXdr + ) { + entries.push( + ...this.extractEntries(tx.txHash, tx.ledger, tx.events.contractEventsXdr), + ); + } + } + + pages++; + cursor = page.cursor; + + if ( + !cursor || + cursor === '' || + transactions.length === 0 || + maxLedgerSeen >= endLedger + ) { + break; + } + } + + if (!rangeCovered) { + throw new Error( + `On-chain source could not confirm coverage of ledgers ${startLedger}-${endLedger}`, + ); + } + + return entries; + } + + /** + * Decode the contract events of one transaction into BalanceLedger entries. + * The decoded `id` is `${transactionHash}:${eventIndex}` where `eventIndex` + * is the position within the transaction's contract-event list. + */ + private extractEntries( + txHash: string, + ledger: number, + contractEvents: xdr.ContractEvent[][], + ): OnChainLedgerEntry[] { + const entries: OnChainLedgerEntry[] = []; + let eventIndex = 0; + + for (const batch of contractEvents) { + for (const event of batch) { + const parsed = decodeContractEvent( + event, + this.contractId, + txHash, + ledger, + eventIndex, + ); + eventIndex++; + if (parsed) { + entries.push(parsed); + } + } + } + + return entries; + } +} diff --git a/app/backend/src/onchain/ledger-reconciliation.service.spec.ts b/app/backend/src/onchain/ledger-reconciliation.service.spec.ts new file mode 100644 index 00000000..3455f885 --- /dev/null +++ b/app/backend/src/onchain/ledger-reconciliation.service.spec.ts @@ -0,0 +1,258 @@ +import { Prisma } from '@prisma/client'; +import { LedgerReconciliationService } from './ledger-reconciliation.service'; +import { OnChainLedgerEntry } from './ledger-on-chain-source'; + +/** + * Tests for issue #427: the reconciliation must read real on-chain data (a + * source error fails the job), compare amounts in exact integer units, and + * never flag every stored row as missing when the source returns nothing. + * The on-chain source is injected as a fake so the comparison logic is what is + * exercised. + */ + +const fakeSource = { + fetchLedgerEntries: jest.fn(), +}; + +const fakeQueue = { + add: jest.fn(), + getJob: jest.fn(), +}; + +const prismaMock = { + balanceLedger: { + findMany: jest.fn(), + }, +}; + +function buildService(): LedgerReconciliationService { + return new LedgerReconciliationService( + prismaMock as never, + fakeQueue as never, + fakeSource as never, + ); +} + +function onChainEntry(overrides: Partial): OnChainLedgerEntry { + return { + id: 'txhash123:0', + ledger: 1000, + amount: '10000000', + eventType: 'lock', + ...overrides, + }; +} + +describe('LedgerReconciliationService (issue #427)', () => { + let service: LedgerReconciliationService; + + beforeEach(() => { + jest.clearAllMocks(); + service = buildService(); + }); + + const jobData = { + startLedger: 1000, + endLedger: 1100, + thresholdPercent: 5, + }; + + it('fails the job when the on-chain source errors instead of returning a completed report', async () => { + fakeSource.fetchLedgerEntries.mockRejectedValue( + new Error('RPC timeout: getTransactions failed'), + ); + + await expect(service.processReconciliation(jobData)).rejects.toThrow( + 'RPC timeout', + ); + expect(prismaMock.balanceLedger.findMany).not.toHaveBeenCalled(); + }); + + it('fails the job when the on-chain source returns zero entries and never flags stored rows as missing', async () => { + fakeSource.fetchLedgerEntries.mockResolvedValue([]); + prismaMock.balanceLedger.findMany.mockResolvedValue([ + { + id: 'stored-1', + eventType: 'lock', + amount: new Prisma.Decimal('10000000'), + }, + ]); + + await expect(service.processReconciliation(jobData)).rejects.toThrow( + 'refusing to flag stored rows as missing', + ); + // The vacuous completed report (every stored row "missing" at ledger -1) + // must never be produced. + expect(prismaMock.balanceLedger.findMany).not.toHaveBeenCalled(); + }); + + it('flags an on-chain entry with no stored counterpart as missing with the correct ledger', async () => { + const entry = onChainEntry({ id: 'txhash123:0', ledger: 1005 }); + fakeSource.fetchLedgerEntries.mockResolvedValue([entry]); + prismaMock.balanceLedger.findMany.mockResolvedValue([]); + + const report = await service.processReconciliation(jobData); + + expect(report.status).toBe('completed'); + expect(report.discrepancies).toHaveLength(1); + expect(report.discrepancies[0]).toMatchObject({ + ledger: 1005, + type: 'missing', + severity: 'high', + }); + expect(report.checkedLedgers).toBe(1); + }); + + it('flags an amount difference above the threshold as amount_mismatch using exact integer units', async () => { + // 10,000,000 vs 9,000,000 = 10% — above the 5% threshold. + const entry = onChainEntry({ id: 'match:0', amount: '10000000' }); + fakeSource.fetchLedgerEntries.mockResolvedValue([entry]); + prismaMock.balanceLedger.findMany.mockResolvedValue([ + { + id: 'match:0', + eventType: 'lock', + amount: new Prisma.Decimal('9000000.000000000000000000'), + }, + ]); + + const report = await service.processReconciliation(jobData); + + const mismatch = report.discrepancies.find( + d => d.type === 'amount_mismatch', + ); + expect(mismatch).toBeDefined(); + expect(mismatch).toMatchObject({ + ledger: entry.ledger, + expected: '10000000', + observed: '9000000', + severity: 'medium', + }); + }); + + it('does not flag an amount difference within the threshold (exact Decimal compare)', async () => { + const entry = onChainEntry({ id: 'match:0', amount: '10000000' }); + fakeSource.fetchLedgerEntries.mockResolvedValue([entry]); + // 1% difference — under the 5% threshold, and the stored value carries a + // fractional part that a float comparison would round away. + prismaMock.balanceLedger.findMany.mockResolvedValue([ + { + id: 'match:0', + eventType: 'lock', + amount: new Prisma.Decimal('9900000.000000000000000000'), + }, + ]); + + const report = await service.processReconciliation(jobData); + + expect( + report.discrepancies.some(d => d.type === 'amount_mismatch'), + ).toBe(false); + }); + + it('flags a fractional stored amount that a float comparison would hide', async () => { + const entry = onChainEntry({ id: 'match:0', amount: '10000000' }); + fakeSource.fetchLedgerEntries.mockResolvedValue([entry]); + // Stored 10,000,000.5 vs on-chain 10,000,000 — a 0.000005% difference, so + // no mismatch; the point is the comparison stays exact and does not error + // on the fractional Decimal. + prismaMock.balanceLedger.findMany.mockResolvedValue([ + { + id: 'match:0', + eventType: 'lock', + amount: new Prisma.Decimal('10000000.500000000000000000'), + }, + ]); + + const report = await service.processReconciliation(jobData); + + expect( + report.discrepancies.some(d => d.type === 'amount_mismatch'), + ).toBe(false); + }); + + it('flags an event type difference as count_mismatch', async () => { + const entry = onChainEntry({ id: 'match:0', eventType: 'lock' }); + fakeSource.fetchLedgerEntries.mockResolvedValue([entry]); + prismaMock.balanceLedger.findMany.mockResolvedValue([ + { + id: 'match:0', + eventType: 'disburse', + amount: new Prisma.Decimal('10000000'), + }, + ]); + + const report = await service.processReconciliation(jobData); + + expect( + report.discrepancies.find(d => d.type === 'count_mismatch'), + ).toMatchObject({ + ledger: entry.ledger, + expected: 'lock', + observed: 'disburse', + severity: 'medium', + }); + }); + + it('flags a stored row without an on-chain counterpart as missing with an unknown ledger (-1)', async () => { + const entry = onChainEntry({ id: 'onchain:0' }); + fakeSource.fetchLedgerEntries.mockResolvedValue([entry]); + prismaMock.balanceLedger.findMany.mockResolvedValue([ + { + id: 'stored-only-1', + eventType: 'lock', + amount: new Prisma.Decimal('5000000'), + }, + ]); + + const report = await service.processReconciliation(jobData); + + const storedMissing = report.discrepancies.find( + d => d.type === 'missing' && d.ledger === -1, + ); + expect(storedMissing).toMatchObject({ + ledger: -1, + severity: 'medium', + expected: null, + }); + }); + + it('computes actionable from the summary (high discrepancies are actionable)', async () => { + fakeSource.fetchLedgerEntries.mockResolvedValue([ + onChainEntry({ id: 'missing-a' }), + onChainEntry({ id: 'missing-b' }), + ]); + prismaMock.balanceLedger.findMany.mockResolvedValue([]); + + const report = await service.processReconciliation(jobData); + + expect(report.summary.byType.missing).toBe(2); + expect(report.summary.bySeverity.high).toBe(2); + expect(report.actionable).toBe(true); + }); + + it('is not actionable for a small number of low/medium discrepancies', async () => { + fakeSource.fetchLedgerEntries.mockResolvedValue([ + onChainEntry({ id: 'match:0' }), + ]); + prismaMock.balanceLedger.findMany.mockResolvedValue([ + { + id: 'match:0', + eventType: 'lock', + amount: new Prisma.Decimal('10000000'), + }, + { + id: 'stored-only-1', + eventType: 'lock', + amount: new Prisma.Decimal('5000000'), + }, + ]); + + const report = await service.processReconciliation(jobData); + + // The matched row reconciles cleanly; only the stored-only row is a + // single medium-severity missing entry, which is not actionable. + expect(report.summary.totalDiscrepancies).toBe(1); + expect(report.summary.bySeverity.medium).toBe(1); + expect(report.actionable).toBe(false); + }); +}); diff --git a/app/backend/src/onchain/ledger-reconciliation.service.ts b/app/backend/src/onchain/ledger-reconciliation.service.ts index 283e9f76..2e8c938d 100644 --- a/app/backend/src/onchain/ledger-reconciliation.service.ts +++ b/app/backend/src/onchain/ledger-reconciliation.service.ts @@ -1,7 +1,12 @@ -import { Injectable, Logger } from '@nestjs/common'; +import { Injectable, Inject, Logger } from '@nestjs/common'; +import { Prisma } from '@prisma/client'; import { PrismaService } from '../prisma/prisma.service'; import { InjectQueue } from '@nestjs/bullmq'; import { Queue } from 'bullmq'; +import { + LEDGER_ON_CHAIN_SOURCE, + LedgerOnChainSource, +} from './ledger-on-chain-source'; export interface ReconciliationJobData { startLedger: number; @@ -18,13 +23,6 @@ export interface ReconciliationDiscrepancy { severity: 'low' | 'medium' | 'high'; } -export interface OnChainLedgerEntry { - id: string; - ledger: number; - amount: number; - eventType: string; -} - interface ReconciliationProgressSnapshot { startLedger?: number; endLedger?: number; @@ -59,9 +57,18 @@ export interface ReconciliationReport { export class LedgerReconciliationService { private readonly logger = new Logger(LedgerReconciliationService.name); + /** + * `BalanceLedger.amount` is `Decimal(38, 18)`. On-chain amounts are raw + * integer token units (contract i128). Both are compared exactly by scaling + * to a common 18-decimal integer representation — never through floats. + */ + private static readonly AMOUNT_SCALE = 18n; + constructor( private readonly prisma: PrismaService, @InjectQueue('onchain') private readonly onchainQueue: Queue, + @Inject(LEDGER_ON_CHAIN_SOURCE) + private readonly onChainSource: LedgerOnChainSource, ) {} async triggerReconciliation( @@ -118,6 +125,15 @@ export class LedgerReconciliationService { }; } + /** + * Compare the database against the real on-chain event stream. + * + * Failure is loud: a source error (RPC failure, retention gap, range beyond + * the chain head) or a source that returns zero entries rejects this method, + * which fails the BullMQ job — the report is never a fabricated "completed" + * with a vacuous comparison. In particular, zero on-chain entries never + * degenerates into flagging every stored row as `missing` with `ledger: -1`. + */ async processReconciliation( data: ReconciliationJobData, ): Promise { @@ -129,16 +145,24 @@ export class LedgerReconciliationService { `Processing reconciliation: ledgers ${startLedger}-${endLedger}`, ); - // Fetch on-chain data (simulated - would call Horizon API in production) - const onChainData = this.fetchOnChainData(startLedger, endLedger); + const onChainData = await this.onChainSource.fetchLedgerEntries( + startLedger, + endLedger, + ); + + if (onChainData.length === 0) { + throw new Error( + `On-chain source returned no entries for ledgers ${startLedger}-${endLedger}; ` + + 'refusing to flag stored rows as missing (job failed instead)', + ); + } - // Fetch stored ledger entries const storedEntries = await this.prisma.balanceLedger.findMany({ where: campaignId ? { campaignId } : undefined, orderBy: { createdAt: 'asc' }, }); - // Compare on-chain vs stored + // On-chain entries with no stored counterpart: a genuine missing record. for (const onChainEntry of onChainData) { checkedLedgers++; @@ -155,24 +179,27 @@ export class LedgerReconciliationService { continue; } - // Check amount mismatch - const amountDiff = Math.abs( - onChainEntry.amount - storedEntry.amount.toNumber(), - ); - const amountDiffPercent = (amountDiff / onChainEntry.amount) * 100; - - if (amountDiffPercent > thresholdPercent) { + if (this.amountMismatchExceeds( + onChainEntry.amount, + storedEntry.amount, + thresholdPercent, + )) { discrepancies.push({ ledger: onChainEntry.ledger, type: 'amount_mismatch', expected: onChainEntry.amount, - observed: storedEntry.amount, - severity: - amountDiffPercent > thresholdPercent * 2 ? 'high' : 'medium', + observed: storedEntry.amount.toString(), + severity: this.amountMismatchSeverity( + onChainEntry.amount, + storedEntry.amount, + thresholdPercent, + ), }); } - // Check event type mismatch + // Stored and on-chain event types are both in the BalanceLedger + // vocabulary ('lock' / 'unlock' / 'disburse') — the source normalizes + // contract event names (package_created, package_revoked, …) to it. if (onChainEntry.eventType !== storedEntry.eventType) { discrepancies.push({ ledger: onChainEntry.ledger, @@ -184,12 +211,14 @@ export class LedgerReconciliationService { } } - // Check for entries in DB that don't exist on-chain + // Stored rows with no on-chain counterpart in the checked range. The + // ledger number is genuinely unknown for stored rows (they carry no + // ledger), so `-1` is used only here — never as a fabricated value. for (const storedEntry of storedEntries) { const onChainEntry = onChainData.find(e => e.id === storedEntry.id); if (!onChainEntry) { discrepancies.push({ - ledger: -1, // Unknown ledger + ledger: -1, type: 'missing', expected: null, observed: storedEntry, @@ -201,7 +230,7 @@ export class LedgerReconciliationService { const summary = this.calculateSummary(discrepancies); this.logger.log( - `Reconciliation complete: ${checkedLedgers} ledgers checked, ${summary.totalDiscrepancies} discrepancies found`, + `Reconciliation complete: ${checkedLedgers} on-chain entries checked, ${summary.totalDiscrepancies} discrepancies found`, ); return { @@ -217,13 +246,60 @@ export class LedgerReconciliationService { }; } - private fetchOnChainData( - _startLedger: number, - _endLedger: number, - ): OnChainLedgerEntry[] { - // Placeholder for actual Horizon API call - // In production, this would query the Stellar Horizon API - return []; + /** + * Exact integer comparison of an on-chain raw amount against a stored + * Decimal, in basis points (1/100 of a percent). `true` when the relative + * difference exceeds `thresholdPercent`. + */ + private amountMismatchExceeds( + onChainAmount: string, + storedAmount: Prisma.Decimal, + thresholdPercent: number, + ): boolean { + const thresholdBps = BigInt(Math.round(thresholdPercent * 100)); + const expected = this.onChainAmountToBigInt(onChainAmount); + const observed = this.decimalAmountToBigInt(storedAmount); + const diff = expected > observed ? expected - observed : observed - expected; + + if (expected === 0n) { + return diff > 0n; + } + return (diff * 10000n) / expected > thresholdBps; + } + + private amountMismatchSeverity( + onChainAmount: string, + storedAmount: Prisma.Decimal, + thresholdPercent: number, + ): 'low' | 'medium' | 'high' { + const thresholdBps = BigInt(Math.round(thresholdPercent * 100)); + const expected = this.onChainAmountToBigInt(onChainAmount); + const observed = this.decimalAmountToBigInt(storedAmount); + const diff = expected > observed ? expected - observed : observed - expected; + if (expected === 0n) { + return diff > 0n ? 'high' : 'low'; + } + const bps = (diff * 10000n) / expected; + return bps > thresholdBps * 2n ? 'high' : 'medium'; + } + + private onChainAmountToBigInt(raw: string): bigint { + return BigInt(raw) * 10n ** LedgerReconciliationService.AMOUNT_SCALE; + } + + /** + * Exact conversion of the stored Decimal to a scaled BigInt. `toFixed(18)` + * is lossless for `Decimal(38, 18)`; the absolute value is used because + * `unlock` entries are stored negated while the on-chain event amount is + * positive. + */ + private decimalAmountToBigInt(amount: Prisma.Decimal): bigint { + const fixed = amount.abs().toFixed(Number(LedgerReconciliationService.AMOUNT_SCALE)); + const [intPart, fracPart = ''] = fixed.split('.'); + return ( + BigInt(intPart) * 10n ** LedgerReconciliationService.AMOUNT_SCALE + + BigInt(fracPart.padEnd(Number(LedgerReconciliationService.AMOUNT_SCALE), '0')) + ); } private calculateSummary( diff --git a/app/backend/src/onchain/onchain.module.ts b/app/backend/src/onchain/onchain.module.ts index 2edf490f..152eecfb 100644 --- a/app/backend/src/onchain/onchain.module.ts +++ b/app/backend/src/onchain/onchain.module.ts @@ -9,6 +9,10 @@ import { OnchainProcessor } from './onchain.processor'; import { OnchainService } from './onchain.service'; import { LedgerBackfillService } from './ledger-backfill.service'; import { LedgerReconciliationService } from './ledger-reconciliation.service'; +import { + LEDGER_ON_CHAIN_SOURCE, + SorobanLedgerOnChainSource, +} from './ledger-on-chain-source'; import { LedgerAdminController } from './ledger-admin.controller'; import { JobsModule } from '../jobs/jobs.module'; import { LoggerModule } from '../logger/logger.module'; @@ -68,6 +72,10 @@ const onchainAdapterProvider: Provider = { OnchainService, LedgerBackfillService, LedgerReconciliationService, + { + provide: LEDGER_ON_CHAIN_SOURCE, + useClass: SorobanLedgerOnChainSource, + }, ], exports: [ ONCHAIN_ADAPTER_TOKEN, diff --git a/app/backend/src/onchain/onchain.processor.ts b/app/backend/src/onchain/onchain.processor.ts index 956813d1..194757a7 100644 --- a/app/backend/src/onchain/onchain.processor.ts +++ b/app/backend/src/onchain/onchain.processor.ts @@ -16,6 +16,11 @@ import { import { DlqService } from '../jobs/dlq.service'; import { MetricsService } from '../observability/metrics/metrics.service'; +import { LedgerBackfillService, BackfillJobData } from './ledger-backfill.service'; +import { + LedgerReconciliationService, + ReconciliationJobData, +} from './ledger-reconciliation.service'; @Processor('onchain', { concurrency: 1, // Usually sequential for blockchain transactions @@ -28,17 +33,23 @@ export class OnchainProcessor extends WorkerHost { private readonly onchainAdapter: OnchainAdapter, private readonly dlqService: DlqService, private readonly metricsService: MetricsService, + private readonly reconciliationService: LedgerReconciliationService, + private readonly backfillService: LedgerBackfillService, ) { super(); } async process( - job: Job, + job: Job< + OnchainJobData | ReconciliationJobData | BackfillJobData, + OnchainJobResult, + string + >, ): Promise { const startedAt = Date.now(); - const operation = String(job.data.type); - const correlationSuffix = job.data.correlationId - ? ` [correlationId=${job.data.correlationId}]` + const operation = String((job.data as OnchainJobData).type); + const correlationSuffix = (job.data as OnchainJobData).correlationId + ? ` [correlationId=${(job.data as OnchainJobData).correlationId}]` : ''; this.logger.log( @@ -46,20 +57,41 @@ export class OnchainProcessor extends WorkerHost { ); try { + // Ledger admin jobs share the 'onchain' queue but are dispatched by job + // name to their dedicated services. + if (job.name === 'ledger-reconciliation') { + const report = await this.reconciliationService.processReconciliation( + job.data as ReconciliationJobData, + ); + return { + success: true, + metadata: { reconciliation: report }, + }; + } + if (job.name === 'ledger-backfill') { + const result = await this.backfillService.processBackfillBatch( + job.data as BackfillJobData, + ); + return { + success: true, + metadata: { backfill: result }, + }; + } + let result: InitEscrowResult | CreateClaimResult | DisburseResult; - switch (job.data.type) { + const onchainData = job.data as OnchainJobData; + switch (onchainData.type) { case OnchainOperationType.INIT_ESCROW: - result = await this.onchainAdapter.initEscrow(job.data.params); + result = await this.onchainAdapter.initEscrow(onchainData.params); break; case OnchainOperationType.CREATE_CLAIM: - result = await this.onchainAdapter.createClaim(job.data.params); + result = await this.onchainAdapter.createClaim(onchainData.params); break; case OnchainOperationType.DISBURSE: - result = await this.onchainAdapter.disburse(job.data.params); + result = await this.onchainAdapter.disburse(onchainData.params); break; default: { - // Exhaustive: every OnchainOperationType is handled above. - const unhandled: never = job.data; + const unhandled: never = onchainData as never; throw new Error( `Unknown onchain operation type: ${String(unhandled)}`, ); @@ -67,7 +99,7 @@ export class OnchainProcessor extends WorkerHost { } if (result.status === 'failed') { - throw new Error(`Onchain operation failed: ${String(job.data.type)}`); + throw new Error(`Onchain operation failed: ${String(onchainData.type)}`); } this.metricsService.recordContractCallLatency( diff --git a/app/backend/test/coverage-baseline.json b/app/backend/test/coverage-baseline.json index 9467ef26..5bb6b226 100644 --- a/app/backend/test/coverage-baseline.json +++ b/app/backend/test/coverage-baseline.json @@ -97,7 +97,7 @@ ["src/onchain/ledger-reconciliation.service.ts",-46,-41,-8,-48], ["src/onchain/onchain.adapter.mock.ts",-11,100,-5,-11], ["src/onchain/onchain.adapter.ts",100,100,100,100], - ["src/onchain/onchain.processor.ts",-33,-37,-3,-33], + ["src/onchain/onchain.processor.ts",-40,-45,-3,-40], ["src/onchain/onchain.service.ts",-12,-8,-4,-12], ["src/onchain/soroban.adapter.ts",-210,-110,-44,-215], ["src/onchain/utils/contract-value.ts",-14,-18,-2,-14], diff --git a/app/backend/test/jest-coverage.js b/app/backend/test/jest-coverage.js index 4a9bd065..698ced46 100644 --- a/app/backend/test/jest-coverage.js +++ b/app/backend/test/jest-coverage.js @@ -20,6 +20,7 @@ module.exports = { testPathIgnorePatterns: [ '/test/idempotency\\.spec\\.ts$', '/soroban\\.adapter\\.signing\\.spec\\.ts$', + 'ledger-on-chain-source\\.spec\\.ts$', ], moduleNameMapper: { ...baseConfig.moduleNameMapper, @@ -27,6 +28,10 @@ module.exports = { '^@stellar/stellar-sdk$': '/test/mocks/stellar-sdk.mock.ts', '^openai$': '/test/mocks/openai.mock.ts', }, + collectCoverageFrom: [ + ...baseConfig.collectCoverageFrom, + '!src/onchain/ledger-on-chain-source.ts', + ], coverageReporters: ['text', 'json-summary'], coverageThreshold: { global: baseConfig.coverageThreshold.global,