feat(indexer): reorg-aware event ingestion with ledger rollback (#724)

prisma/schema/indexer.prisma

@@ -1,5 +1,60 @@
 // prisma/schema/indexer.prisma
 
+/// Finality status of an ingested chain event with respect to the canonical
+/// ledger chain (#724).
+enum IngestedEventStatus {
+  /// Event has been ingested and its ledger is still considered canonical.
+  CONFIRMED
+  /// The event's ledger was rolled back by a reorg; its database side
+  /// effects have been reversed by the reorg-guard rollback.
+  REORGED
+}
+
+/// One row per ingested chain event, stamped with the ledger sequence and
+/// ledger hash it was observed in (#724).
+///
+/// Written by the reorg-aware ingestion entry point (`ingestLedger`) so a
+/// ledger rollback can identify exactly which events to mark REORGED and
+/// reverse, and so re-ingestion after a rollback is auditable.
+model IngestedEvent {
+  id             String              @id @default(cuid())
+
+  /// Ledger sequence the event was ingested in.
+  ledgerSequence Int
+  /// Hash of the ledger the event was ingested in.
+  ledgerHash     String
+
+  /// On-chain transaction hash (dedup key with eventIndex).
+  txHash         String
+  /// Index of the event within its transaction.
+  eventIndex     Int
+  /// Domain event type (e.g. KEY_BOUGHT, KEY_SOLD).
+  eventType      String
+
+  /// Wallet that performed the action, when applicable.
+  actor          String?
+  /// Creator/key the event applies to, when applicable.
+  creatorId      String?
+  /// Key quantity the event moved, as a decimal string, when applicable.
+  amount         String?
+  /// Per-key trade price in stroops, as a decimal string, when applicable.
+  priceStroops   String?
+  /// Trade timestamp carried by the event, when applicable.
+  tradeAt        DateTime?
+
+  status         IngestedEventStatus @default(CONFIRMED)
+  /// Set when a reorg rolled this event back.
+  reorgedAt      DateTime?
+
+  createdAt      DateTime             @default(now())
+
+  @@unique([txHash, eventIndex])
+  @@index([ledgerSequence])
+  @@index([status])
+  @@index([creatorId])
+  @@map("ingested_events")
+}
+
 /**
  * Dead-letter queue (DLQ) for failed indexing jobs.
  * 

prisma/schema/ledger.prisma

@@ -10,6 +10,13 @@ model IndexedLedger {
 model LedgerCheckpoint {
   id                    Int      @id @default(autoincrement())
   ledger                Int
+  /// Hash of the ledger at `ledger`, as reported by the network (#724).
+  /// Stored for every ledger ingested by the reorg-aware pipeline so the
+  /// next ledger's `previousLedgerHash` can be verified against it.
+  /// Nullable only for checkpoints written before hash tracking existed.
+  ledgerHash            String?
+  /// The `previousLedgerHash` the ingested ledger carried, when available.
+  previousLedgerHash    String?
   lastProcessedEventIndex Int
   batchHash             String
   completedAt           DateTime @default(now())

prisma/schema/migrations/20261020000000_add_reorg_aware_ingestion/migration.sql

@@ -0,0 +1,47 @@
+-- Reorg-aware Stellar event ingestion (#724).
+--
+-- 1. ledger_checkpoints gains ledgerHash/previousLedgerHash so the pipeline
+--    can verify that each incoming ledger chains onto the last ingested one
+--    and detect ledger rollbacks.
+-- 2. New ingested_events table records every ingested event with the ledger
+--    sequence and hash it was observed in, so a reorg can mark exactly the
+--    affected events as REORGED and reverse their side effects.
+
+-- AlterTable
+ALTER TABLE "ledger_checkpoints" ADD COLUMN     "ledgerHash" TEXT,
+ADD COLUMN     "previousLedgerHash" TEXT;
+
+-- CreateEnum
+CREATE TYPE "IngestedEventStatus" AS ENUM ('CONFIRMED', 'REORGED');
+
+-- CreateTable
+CREATE TABLE "ingested_events" (
+    "id" TEXT NOT NULL,
+    "ledgerSequence" INTEGER NOT NULL,
+    "ledgerHash" TEXT NOT NULL,
+    "txHash" TEXT NOT NULL,
+    "eventIndex" INTEGER NOT NULL,
+    "eventType" TEXT NOT NULL,
+    "actor" TEXT,
+    "creatorId" TEXT,
+    "amount" TEXT,
+    "priceStroops" TEXT,
+    "tradeAt" TIMESTAMP(3),
+    "status" "IngestedEventStatus" NOT NULL DEFAULT 'CONFIRMED',
+    "reorgedAt" TIMESTAMP(3),
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "ingested_events_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "ingested_events_txHash_eventIndex_key" ON "ingested_events"("txHash", "eventIndex");
+
+-- CreateIndex
+CREATE INDEX "ingested_events_ledgerSequence_idx" ON "ingested_events"("ledgerSequence");
+
+-- CreateIndex
+CREATE INDEX "ingested_events_status_idx" ON "ingested_events"("status");
+
+-- CreateIndex
+CREATE INDEX "ingested_events_creatorId_idx" ON "ingested_events"("creatorId");

src/__tests__/integration/invoice-protocol.integration.test.ts

@@ -266,7 +266,10 @@ describeDatabase('invoice protocol against PostgreSQL', () => {
             return { result: { status: 'PENDING', hash } };
          }
          if (body.method === 'getTransaction')
-            return { result: { status: 'SUCCESS', ledger: 901 } };
+            // `latestLedger` is part of a real getTransaction response and is
+            // what the #724 ledger-depth gate reads: report the network head
+            // two ledgers past inclusion so buys/sells confirm immediately.
+            return { result: { status: 'SUCCESS', ledger: 901, latestLedger: 903 } };
          throw new Error(`Unexpected RPC method ${body.method}`);
       });
    });

src/modules/contracts/contract.service.test.ts

@@ -40,9 +40,15 @@ function makeDeps(
 ) {
    let currentTime = 0;
    const pollCalls: number[] = [];
+   // latestLedger: 102 satisfies the #724 ledger-depth gate (tx ledger 100
+   // + depth 2) on the first poll so gate polling adds no extra delays.
    const getTransaction =
       overrides.getTransaction ??
-      jest.fn().mockResolvedValue({ status: 'SUCCESS', ledger: 100 });
+      jest.fn().mockResolvedValue({
+         status: 'SUCCESS',
+         ledger: 100,
+         latestLedger: 102,
+      });
 
    return {
       deps: {

src/modules/contracts/contract.service.ts

@@ -32,6 +32,7 @@ import {
    pollForConfirmation,
    type TransactionStatusClient,
 } from './confirmation-poller.utils';
+import { waitForLedgerDepth } from './ledger-depth-gate.utils';
 import {
    classifyContractError,
    ContractSubmitError,
@@ -287,6 +288,20 @@ export async function submitContractCall(
 
    // ── 4. Persist resolution + emit events ─────────────────────────────
    if (outcome?.status === 'SUCCESS') {
+      // Ledger-depth gate (#724): buy/sell responses are withheld until the
+      // ledger that included the transaction is 2 ledgers deep, to reduce
+      // reorg exposure. Budget exhaustion releases with a warning (logged
+      // inside the gate) so a stalled RPC head cannot block confirmations
+      // forever.
+      await waitForLedgerDepth(outcome.ledger, {
+         statusClient,
+         txHash,
+         intervalMs: pollIntervalMs,
+         timeoutMs: pollTimeoutMs,
+         delay,
+         now,
+      });
+
       try {
          await prisma.contractTransaction.update({
             where: { txHash },

src/modules/contracts/ledger-depth-gate.utils.test.ts

@@ -0,0 +1,179 @@
+// src/modules/contracts/ledger-depth-gate.utils.test.ts
+// Unit tests for #724 — buy/sell confirmations are gated on confirmed
+// ledger depth ≥ 2.
+
+import {
+   CONFIRMED_LEDGER_DEPTH,
+   waitForLedgerDepth,
+} from './ledger-depth-gate.utils';
+import { logger } from '../../utils/logger.utils';
+import { submitContractCall } from './contract.service';
+
+jest.mock('../../utils/prisma.utils', () => ({
+   prisma: {
+      contractTransaction: {
+         create: jest.fn().mockResolvedValue({}),
+         update: jest.fn().mockResolvedValue({}),
+      },
+   },
+}));
+
+function makeGateDeps(
+   getTransaction: jest.Mock,
+   overrides: { intervalMs?: number; timeoutMs?: number } = {}
+) {
+   let currentTime = 0;
+   const delays: number[] = [];
+   return {
+      deps: {
+         statusClient: { getTransaction },
+         txHash: 'tx-hash-1',
+         intervalMs: overrides.intervalMs ?? 5,
+         timeoutMs: overrides.timeoutMs ?? 500,
+         delay: async (ms: number) => {
+            delays.push(ms);
+            currentTime += ms;
+         },
+         now: () => currentTime,
+      },
+      delays,
+   };
+}
+
+describe('waitForLedgerDepth (#724)', () => {
+   beforeEach(() => {
+      jest.clearAllMocks();
+   });
+
+   it('requires a depth of 2 ledgers', () => {
+      expect(CONFIRMED_LEDGER_DEPTH).toBe(2);
+   });
+
+   it('resolves immediately when the ledger is already 2 deep', async () => {
+      const getTransaction = jest
+         .fn()
+         .mockResolvedValue({ status: 'SUCCESS', ledger: 100, latestLedger: 102 });
+      const { deps } = makeGateDeps(getTransaction);
+
+      const result = await waitForLedgerDepth(100, deps);
+
+      expect(result).toEqual({ depthConfirmed: true, latestLedger: 102 });
+      expect(getTransaction).toHaveBeenCalledTimes(1);
+   });
+
+   it('keeps polling until the head advances to the required depth', async () => {
+      const heads = [100, 101, 102];
+      let call = 0;
+      const getTransaction = jest.fn().mockImplementation(async () => ({
+         status: 'SUCCESS',
+         ledger: 100,
+         latestLedger: heads[Math.min(call++, heads.length - 1)],
+      }));
+      const { deps, delays } = makeGateDeps(getTransaction);
+
+      const result = await waitForLedgerDepth(100, deps);
+
+      expect(result.depthConfirmed).toBe(true);
+      expect(result.latestLedger).toBe(102);
+      // Head was polled at least twice and waited between polls.
+      expect(getTransaction.mock.calls.length).toBeGreaterThanOrEqual(3);
+      expect(delays.length).toBeGreaterThanOrEqual(2);
+   });
+
+   it('reports depthConfirmed=false when the budget is exhausted', async () => {
+      const warnSpy = jest.spyOn(logger, 'warn');
+      const getTransaction = jest
+         .fn()
+         .mockResolvedValue({ status: 'SUCCESS', ledger: 100, latestLedger: 100 });
+      const { deps } = makeGateDeps(getTransaction, { timeoutMs: 30 });
+
+      const result = await waitForLedgerDepth(100, deps);
+
+      expect(result.depthConfirmed).toBe(false);
+      expect(result.latestLedger).toBe(100);
+      expect(warnSpy).toHaveBeenCalledWith(
+         expect.objectContaining({
+            tx_ledger: 100,
+            latest_ledger: 100,
+            required_depth: 2,
+         }),
+         'Ledger depth gate timed out before the required confirmation depth'
+      );
+      warnSpy.mockRestore();
+   });
+
+   it('retries transient RPC failures instead of throwing', async () => {
+      let call = 0;
+      const getTransaction = jest.fn().mockImplementation(async () => {
+         call += 1;
+         if (call === 1) {
+            throw new Error('rpc unavailable');
+         }
+         return { status: 'SUCCESS', ledger: 100, latestLedger: 103 };
+      });
+      const { deps } = makeGateDeps(getTransaction);
+
+      const result = await waitForLedgerDepth(100, deps);
+
+      expect(result.depthConfirmed).toBe(true);
+      expect(result.latestLedger).toBe(103);
+   });
+
+   it('treats a missing latestLedger as "not deep enough yet"', async () => {
+      const getTransaction = jest
+         .fn()
+         .mockResolvedValue({ status: 'SUCCESS', ledger: 100 });
+      const { deps } = makeGateDeps(getTransaction, { timeoutMs: 20 });
+
+      const result = await waitForLedgerDepth(100, deps);
+
+      expect(result.depthConfirmed).toBe(false);
+   });
+});
+
+describe('submitContractCall ledger-depth gate (#724)', () => {
+   beforeEach(() => {
+      jest.clearAllMocks();
+   });
+
+   it('withholds the CONFIRMED result until the including ledger is 2 deep', async () => {
+      const sendTransaction = jest
+         .fn()
+         .mockResolvedValue({ hash: 'hash-gated', status: 'PENDING' });
+
+      // Confirmation lands at ledger 100; the head lags at 100/101 and only
+      // reaches 102 after several polls.
+      const heads = [100, 100, 101, 102];
+      let call = 0;
+      const getTransaction = jest.fn().mockImplementation(async () => ({
+         status: 'SUCCESS',
+         ledger: 100,
+         latestLedger: heads[Math.min(call++, heads.length - 1)],
+      }));
+
+      let currentTime = 0;
+      const deps = {
+         submitClient: { sendTransaction },
+         statusClient: { getTransaction },
+         delay: async (ms: number) => {
+            currentTime += ms;
+         },
+         now: () => currentTime,
+         pollIntervalMs: 5,
+         pollTimeoutMs: 500,
+      };
+
+      const result = await submitContractCall(
+         {
+            operation: 'buy' as const,
+            submitterWallet: 'GWALLET',
+            transactionXdr: 'AAAAbase64xdr==',
+         },
+         deps
+      );
+
+      expect(result.status).toBe('CONFIRMED');
+      // At least one confirmation poll plus the depth-gate polls.
+      expect(getTransaction.mock.calls.length).toBeGreaterThan(1);
+   });
+});