diff --git a/backend/src/services/derivedTableStore.ts b/backend/src/services/derivedTableStore.ts index 0f3260c7..fb9db7cb 100644 --- a/backend/src/services/derivedTableStore.ts +++ b/backend/src/services/derivedTableStore.ts @@ -153,6 +153,22 @@ export class InMemoryDerivedTableStore implements DerivedTableStore { return Array.from(this.tables.invoices.values()); } + async listBids(): Promise { + return Array.from(this.tables.bids.values()); + } + + async listSettlements(): Promise { + return Array.from(this.tables.settlements.values()); + } + + async listDisputes(): Promise { + return Array.from(this.tables.disputes.values()); + } + + async listNotifications(): Promise { + return Array.from(this.tables.notifications.values()); + } + async getBid(id: string): Promise { return this.tables.bids.get(id) || null; } diff --git a/backend/src/services/snapshotService.ts b/backend/src/services/snapshotService.ts index e861dfc6..c6ba1a6e 100644 --- a/backend/src/services/snapshotService.ts +++ b/backend/src/services/snapshotService.ts @@ -46,9 +46,18 @@ interface SnapshotClient extends SnapshotQueryable { release(): void; } -interface SnapshotPoolLike extends SnapshotQueryable { - connect(): Promise; -} +interface SnapshotPoolLike extends SnapshotQueryable { + connect(): Promise; +} + +type SnapshotReadableDerivedStore = DerivedTableStore & { + listInvoices?: () => Promise; + listBids?: () => Promise; + listSettlements?: () => Promise; + listDisputes?: () => Promise; + listNotifications?: () => Promise; + getTableCounts?: () => TableCounts; +}; // ── PII redaction helpers ───────────────────────────────────────────────────── @@ -129,15 +138,12 @@ 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; - getTableCounts?: () => { invoices: number; bids: number; settlements: number; disputes: number; notifications: number }; - }, - config: Partial = {}, - ) { + private snapshotCounter = 0; + + private constructor( + private readonly derivedStore: SnapshotReadableDerivedStore, + config: Partial = {}, + ) { this.config = { intervalMs: config.intervalMs ?? DEFAULT_INTERVAL_MS, maxRetained: config.maxRetained ?? DEFAULT_MAX_RETAINED, @@ -145,10 +151,10 @@ export class SnapshotScheduler { }; } - static getInstance( - derivedStore: DerivedTableStore & { listInvoices?: () => Promise; getTableCounts?: () => any }, - config: Partial = {}, - ): SnapshotScheduler { + static getInstance( + derivedStore: SnapshotReadableDerivedStore, + config: Partial = {}, + ): 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 = { diff --git a/backend/src/tests/snapshot-integrity.test.ts b/backend/src/tests/snapshot-integrity.test.ts new file mode 100644 index 00000000..25e9fd30 --- /dev/null +++ b/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 { + 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, + ]); + }); +}); diff --git a/backend/src/types/replay.ts b/backend/src/types/replay.ts index 11f289dd..4f93128d 100644 --- a/backend/src/types/replay.ts +++ b/backend/src/types/replay.ts @@ -142,6 +142,18 @@ export interface DerivedTableStore { // List invoices currently persisted by the indexer listInvoices?(): Promise; + + // List bids currently persisted by the indexer + listBids?(): Promise; + + // List settlements currently persisted by the indexer + listSettlements?(): Promise; + + // List disputes currently persisted by the indexer + listDisputes?(): Promise; + + // List notifications currently persisted by the indexer + listNotifications?(): Promise; } // Security validation interface