test(backend): verify snapshot totals reconcile with derived tables

backend/src/services/derivedTableStore.ts

@@ -153,6 +153,22 @@ export class InMemoryDerivedTableStore implements DerivedTableStore {
     return Array.from(this.tables.invoices.values());
   }
 
+  async listBids(): Promise<any[]> {
+    return Array.from(this.tables.bids.values());
+  }
+
+  async listSettlements(): Promise<any[]> {
+    return Array.from(this.tables.settlements.values());
+  }
+
+  async listDisputes(): Promise<any[]> {
+    return Array.from(this.tables.disputes.values());
+  }
+
+  async listNotifications(): Promise<any[]> {
+    return Array.from(this.tables.notifications.values());
+  }
+
   async getBid(id: string): Promise<any | null> {
     return this.tables.bids.get(id) || null;
   }

backend/src/services/snapshotService.ts

@@ -46,9 +46,18 @@ interface SnapshotClient extends SnapshotQueryable {
   release(): void;
 }
 
-interface SnapshotPoolLike extends SnapshotQueryable {
-  connect(): Promise<SnapshotClient>;
-}
+interface SnapshotPoolLike extends SnapshotQueryable {
+  connect(): Promise<SnapshotClient>;
+}
+
+type SnapshotReadableDerivedStore = DerivedTableStore & {
+  listInvoices?: () => Promise<any[]>;
+  listBids?: () => Promise<any[]>;
+  listSettlements?: () => Promise<any[]>;
+  listDisputes?: () => Promise<any[]>;
+  listNotifications?: () => Promise<any[]>;
+  getTableCounts?: () => TableCounts;
+};
 
 // ── PII redaction helpers ─────────────────────────────────────────────────────
 
@@ -129,26 +138,23 @@ export class SnapshotScheduler {
   private timer: NodeJS.Timeout | null = null;
   private isRunning = false;
   private activeBatchCount = 0;
-  private snapshotCounter = 0;
-
-  private constructor(
-    private readonly derivedStore: DerivedTableStore & {
-      listInvoices?: () => Promise<any[]>;
-      getTableCounts?: () => { invoices: number; bids: number; settlements: number; disputes: number; notifications: number };
-    },
-    config: Partial<SnapshotScheduleConfig> = {},
-  ) {
+  private snapshotCounter = 0;
+
+  private constructor(
+    private readonly derivedStore: SnapshotReadableDerivedStore,
+    config: Partial<SnapshotScheduleConfig> = {},
+  ) {
     this.config = {
       intervalMs: config.intervalMs ?? DEFAULT_INTERVAL_MS,
       maxRetained: config.maxRetained ?? DEFAULT_MAX_RETAINED,
       hmacSecret: config.hmacSecret ?? process.env.SNAPSHOT_HMAC_SECRET ?? "changeme-use-secrets-manager",
     };
   }
 
-  static getInstance(
-    derivedStore: DerivedTableStore & { listInvoices?: () => Promise<any[]>; getTableCounts?: () => any },
-    config: Partial<SnapshotScheduleConfig> = {},
-  ): SnapshotScheduler {
+  static getInstance(
+    derivedStore: SnapshotReadableDerivedStore,
+    config: Partial<SnapshotScheduleConfig> = {},
+  ): SnapshotScheduler {
     if (!SnapshotScheduler.instance) {
       SnapshotScheduler.instance = new SnapshotScheduler(derivedStore, config);
     }
@@ -201,21 +207,39 @@ export class SnapshotScheduler {
     const stateHash = await this.derivedStore.getStateHash();
 
     // Collect raw rows from the store
-    const rawInvoices: any[] = this.derivedStore.listInvoices
-      ? await this.derivedStore.listInvoices()
-      : [];
-
-    const counts: TableCounts = this.derivedStore.getTableCounts
-      ? this.derivedStore.getTableCounts()
-      : { invoices: rawInvoices.length, bids: 0, settlements: 0, disputes: 0, notifications: 0 };
-
-    const tables: RedactedTablePayload = {
-      invoices: redactRows(rawInvoices, this.config.hmacSecret),
-      bids: [],
-      settlements: [],
-      disputes: [],
-      notifications: [],
-    };
+    const rawInvoices: any[] = this.derivedStore.listInvoices
+      ? await this.derivedStore.listInvoices()
+      : [];
+    const rawBids: any[] = this.derivedStore.listBids
+      ? await this.derivedStore.listBids()
+      : [];
+    const rawSettlements: any[] = this.derivedStore.listSettlements
+      ? await this.derivedStore.listSettlements()
+      : [];
+    const rawDisputes: any[] = this.derivedStore.listDisputes
+      ? await this.derivedStore.listDisputes()
+      : [];
+    const rawNotifications: any[] = this.derivedStore.listNotifications
+      ? await this.derivedStore.listNotifications()
+      : [];
+
+    const counts: TableCounts = this.derivedStore.getTableCounts
+      ? this.derivedStore.getTableCounts()
+      : {
+          invoices: rawInvoices.length,
+          bids: rawBids.length,
+          settlements: rawSettlements.length,
+          disputes: rawDisputes.length,
+          notifications: rawNotifications.length,
+        };
+
+    const tables: RedactedTablePayload = {
+      invoices: redactRows(rawInvoices, this.config.hmacSecret),
+      bids: redactRows(rawBids, this.config.hmacSecret),
+      settlements: redactRows(rawSettlements, this.config.hmacSecret),
+      disputes: redactRows(rawDisputes, this.config.hmacSecret),
+      notifications: redactRows(rawNotifications, this.config.hmacSecret),
+    };
 
     const snapshotId = `snap_${++this.snapshotCounter}_${Date.now()}`;
     const snapshot: DerivedStateSnapshot = {

backend/src/tests/snapshot-integrity.test.ts

@@ -0,0 +1,183 @@
+import Database from "better-sqlite3";
+import { InMemoryDerivedTableStore } from "../services/derivedTableStore";
+import { SnapshotScheduler } from "../services/snapshotService";
+import type { RedactedRow } from "../types/snapshot";
+
+type LedgerTable = "invoices" | "bids" | "settlements";
+type InvoiceRow = { id: string; amount: number };
+type BidRow = { id: string; invoice_id: string; bid_amount: number };
+type SettlementRow = { id: string; invoice_id: string; amount: number };
+type ScalarRow = { value: number };
+
+function createLedgerDb(): Database.Database {
+  const db = new Database(":memory:");
+  db.exec(`
+    CREATE TABLE invoices (
+      id TEXT PRIMARY KEY,
+      amount INTEGER NOT NULL
+    );
+    CREATE TABLE bids (
+      id TEXT PRIMARY KEY,
+      invoice_id TEXT NOT NULL,
+      bid_amount INTEGER NOT NULL
+    );
+    CREATE TABLE settlements (
+      id TEXT PRIMARY KEY,
+      invoice_id TEXT NOT NULL,
+      amount INTEGER NOT NULL
+    );
+  `);
+  return db;
+}
+
+function queryCount(db: Database.Database, table: LedgerTable): number {
+  const row = db.prepare(`SELECT COUNT(*) AS value FROM ${table}`).get() as ScalarRow;
+  return Number(row.value);
+}
+
+function querySum(db: Database.Database, table: "bids", column: "bid_amount"): number;
+function querySum(db: Database.Database, table: "settlements", column: "amount"): number;
+function querySum(db: Database.Database, table: "bids" | "settlements", column: "bid_amount" | "amount"): number {
+  const row = db.prepare(`SELECT COALESCE(SUM(${column}), 0) AS value FROM ${table}`).get() as ScalarRow;
+  return Number(row.value);
+}
+
+function sumRows(rows: RedactedRow[], column: string): number {
+  return rows.reduce((total, row) => total + Number(row[column] ?? 0), 0);
+}
+
+async function hydrateStoreFromDb(
+  db: Database.Database,
+  store: InMemoryDerivedTableStore,
+): Promise<void> {
+  const invoices = db.prepare("SELECT id, amount FROM invoices ORDER BY id").all() as InvoiceRow[];
+  const bids = db.prepare("SELECT id, invoice_id, bid_amount FROM bids ORDER BY id").all() as BidRow[];
+  const settlements = db.prepare("SELECT id, invoice_id, amount FROM settlements ORDER BY id").all() as SettlementRow[];
+
+  for (const invoice of invoices) {
+    await store.upsertInvoice(invoice);
+  }
+  for (const bid of bids) {
+    await store.upsertBid({
+      bid_id: bid.id,
+      invoice_id: bid.invoice_id,
+      bid_amount: bid.bid_amount,
+    });
+  }
+  for (const settlement of settlements) {
+    await store.upsertSettlement(settlement);
+  }
+}
+
+describe("SnapshotScheduler integrity", () => {
+  let db: Database.Database;
+  let store: InMemoryDerivedTableStore;
+  let scheduler: SnapshotScheduler;
+
+  beforeEach(() => {
+    db = createLedgerDb();
+    store = new InMemoryDerivedTableStore();
+    scheduler = SnapshotScheduler.getInstance(store, {
+      intervalMs: 60_000,
+      maxRetained: 10,
+      hmacSecret: "snapshot-integrity-test-secret",
+    });
+  });
+
+  afterEach(() => {
+    scheduler.clearForTests();
+    db.close();
+  });
+
+  it("captures empty derived tables with valid zero counts and totals", async () => {
+    const snapshot = await scheduler.captureNow(0);
+
+    expect(snapshot.tableCounts).toEqual({
+      invoices: 0,
+      bids: 0,
+      settlements: 0,
+      disputes: 0,
+      notifications: 0,
+    });
+    expect(sumRows(snapshot.tables.bids, "bid_amount")).toBe(0);
+    expect(sumRows(snapshot.tables.settlements, "amount")).toBe(0);
+  });
+
+  it("reconciles snapshot counts and totals with derived-table sums", async () => {
+    db.prepare("INSERT INTO invoices (id, amount) VALUES (?, ?)").run("inv_1", 1000);
+    db.prepare("INSERT INTO invoices (id, amount) VALUES (?, ?)").run("inv_2", 2000);
+    db.prepare("INSERT INTO bids (id, invoice_id, bid_amount) VALUES (?, ?, ?)").run("bid_1", "inv_1", 400);
+    db.prepare("INSERT INTO bids (id, invoice_id, bid_amount) VALUES (?, ?, ?)").run("bid_2", "inv_1", 250);
+    db.prepare("INSERT INTO bids (id, invoice_id, bid_amount) VALUES (?, ?, ?)").run("bid_3", "inv_2", 900);
+    db.prepare("INSERT INTO settlements (id, invoice_id, amount) VALUES (?, ?, ?)").run("set_1", "inv_1", 300);
+    db.prepare("INSERT INTO settlements (id, invoice_id, amount) VALUES (?, ?, ?)").run("set_2", "inv_2", 700);
+    await hydrateStoreFromDb(db, store);
+
+    const snapshot = await scheduler.captureNow(42);
+
+    expect(snapshot.tableCounts.invoices).toBe(queryCount(db, "invoices"));
+    expect(snapshot.tableCounts.bids).toBe(queryCount(db, "bids"));
+    expect(snapshot.tableCounts.settlements).toBe(queryCount(db, "settlements"));
+    expect(sumRows(snapshot.tables.bids, "bid_amount")).toBe(querySum(db, "bids", "bid_amount"));
+    expect(sumRows(snapshot.tables.settlements, "amount")).toBe(querySum(db, "settlements", "amount"));
+  });
+
+  it("redacts PII while preserving aggregate reconciliation fields", async () => {
+    await store.upsertInvoice({ id: "inv_pii", amount: 1000, business: "GRAWBUSINESSWALLET" });
+    await store.upsertBid({
+      bid_id: "bid_pii",
+      invoice_id: "inv_pii",
+      investor: "GINVESTORWALLET",
+      bid_amount: 1000,
+    });
+
+    const first = await scheduler.captureNow(7);
+    const second = await scheduler.captureNow(8);
+
+    expect(first.tables.invoices[0].business).not.toBe("GRAWBUSINESSWALLET");
+    expect(first.tables.bids[0].investor).not.toBe("GINVESTORWALLET");
+    expect(first.tables.bids[0].bid_amount).toBe(1000);
+    expect(first.tables.bids[0].investor).toBe(second.tables.bids[0].investor);
+  });
+
+  it("keeps snapshots point-in-time when derived tables mutate", async () => {
+    db.prepare("INSERT INTO invoices (id, amount) VALUES (?, ?)").run("inv_1", 1000);
+    db.prepare("INSERT INTO bids (id, invoice_id, bid_amount) VALUES (?, ?, ?)").run("bid_1", "inv_1", 500);
+    await hydrateStoreFromDb(db, store);
+    const before = await scheduler.captureNow(100);
+
+    db.prepare("INSERT INTO settlements (id, invoice_id, amount) VALUES (?, ?, ?)").run("set_1", "inv_1", 500);
+    await store.upsertSettlement({ id: "set_1", invoice_id: "inv_1", amount: 500 });
+    const after = await scheduler.captureNow(101);
+
+    expect(after.snapshotId).not.toBe(before.snapshotId);
+    expect(after.stateHash).not.toBe(before.stateHash);
+    expect(before.tableCounts.settlements).toBe(0);
+    expect(after.tableCounts.settlements).toBe(1);
+    expect(sumRows(before.tables.settlements, "amount")).toBe(0);
+    expect(sumRows(after.tables.settlements, "amount")).toBe(500);
+  });
+
+  it("marks mid-batch snapshots and prunes retained snapshots deterministically", async () => {
+    scheduler.clearForTests();
+    scheduler = SnapshotScheduler.getInstance(store, {
+      intervalMs: 60_000,
+      maxRetained: 2,
+      hmacSecret: "snapshot-integrity-test-secret",
+    });
+
+    scheduler.markBatchStart();
+    const midBatch = await scheduler.captureNow(1);
+    scheduler.markBatchEnd();
+    const second = await scheduler.captureNow(2);
+    const third = await scheduler.captureNow(3);
+
+    expect(midBatch.midBatch).toBe(true);
+    expect(second.midBatch).toBe(false);
+    expect(scheduler.getSnapshot(midBatch.snapshotId)).toBeUndefined();
+    expect(scheduler.listSnapshots().map((snapshot) => snapshot.snapshotId)).toEqual([
+      second.snapshotId,
+      third.snapshotId,
+    ]);
+  });
+});

backend/src/types/replay.ts

@@ -142,6 +142,18 @@ export interface DerivedTableStore {
 
   // List invoices currently persisted by the indexer
   listInvoices?(): Promise<any[]>;
+
+  // List bids currently persisted by the indexer
+  listBids?(): Promise<any[]>;
+
+  // List settlements currently persisted by the indexer
+  listSettlements?(): Promise<any[]>;
+
+  // List disputes currently persisted by the indexer
+  listDisputes?(): Promise<any[]>;
+
+  // List notifications currently persisted by the indexer
+  listNotifications?(): Promise<any[]>;
 }
 
 // Security validation interface