feat(backend): scheduled reconciliation, manifest persistence, resuma…

backend/prisma/migrations/20260627120000_add_manifest_reconciliation_backfill/migration.sql

@@ -0,0 +1,60 @@
+-- CreateTable
+CREATE TABLE "ExportManifest" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "requester" TEXT NOT NULL,
+    "reportType" TEXT NOT NULL,
+    "filters" TEXT NOT NULL,
+    "checksum" TEXT NOT NULL,
+    "generatedAt" DATETIME NOT NULL,
+    "fileName" TEXT NOT NULL,
+    "rowCount" INTEGER NOT NULL,
+    "bulkExportJobId" TEXT,
+    "artifactId" TEXT
+);
+
+-- CreateTable
+CREATE TABLE "ReconciliationSnapshot" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "generatedAt" DATETIME NOT NULL,
+    "traceId" TEXT,
+    "status" TEXT NOT NULL,
+    "windowFrom" DATETIME NOT NULL,
+    "windowTo" DATETIME NOT NULL,
+    "summaryJson" TEXT NOT NULL,
+    "driftCount" INTEGER NOT NULL
+);
+
+-- CreateTable
+CREATE TABLE "TransactionBackfillJob" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "jobKey" TEXT NOT NULL,
+    "startLedger" INTEGER NOT NULL,
+    "endLedger" INTEGER NOT NULL,
+    "batchSize" INTEGER NOT NULL,
+    "dryRun" BOOLEAN NOT NULL,
+    "status" TEXT NOT NULL,
+    "rpcUrl" TEXT NOT NULL,
+    "contractId" TEXT NOT NULL,
+    "lastProcessedLedger" INTEGER,
+    "progressJson" TEXT NOT NULL,
+    "errorMessage" TEXT,
+    "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" DATETIME NOT NULL,
+    "completedAt" DATETIME
+);
+
+-- CreateIndex
+CREATE INDEX "ExportManifest_generatedAt_idx" ON "ExportManifest"("generatedAt");
+CREATE INDEX "ExportManifest_requester_idx" ON "ExportManifest"("requester");
+CREATE INDEX "ExportManifest_reportType_idx" ON "ExportManifest"("reportType");
+CREATE INDEX "ExportManifest_checksum_idx" ON "ExportManifest"("checksum");
+
+-- CreateIndex
+CREATE INDEX "ReconciliationSnapshot_generatedAt_idx" ON "ReconciliationSnapshot"("generatedAt");
+CREATE INDEX "ReconciliationSnapshot_status_idx" ON "ReconciliationSnapshot"("status");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "TransactionBackfillJob_jobKey_key" ON "TransactionBackfillJob"("jobKey");
+CREATE INDEX "TransactionBackfillJob_status_idx" ON "TransactionBackfillJob"("status");
+CREATE INDEX "TransactionBackfillJob_createdAt_idx" ON "TransactionBackfillJob"("createdAt");
+CREATE INDEX "TransactionBackfillJob_dryRun_idx" ON "TransactionBackfillJob"("dryRun");

backend/prisma/schema.prisma

@@ -315,3 +315,57 @@ model FeatureFlagOverride {
   @@index([expiresAt])
   @@index([actor])
 }
+
+model ExportManifest {
+  id              String   @id
+  requester       String
+  reportType      String
+  filters         String
+  checksum        String
+  generatedAt     DateTime
+  fileName        String
+  rowCount        Int
+  bulkExportJobId String?
+  artifactId      String?
+
+  @@index([generatedAt])
+  @@index([requester])
+  @@index([reportType])
+  @@index([checksum])
+}
+
+model ReconciliationSnapshot {
+  id          String   @id
+  generatedAt DateTime
+  traceId     String?
+  status      String
+  windowFrom  DateTime
+  windowTo    DateTime
+  summaryJson String
+  driftCount  Int
+
+  @@index([generatedAt])
+  @@index([status])
+}
+
+model TransactionBackfillJob {
+  id                  String    @id
+  jobKey              String    @unique
+  startLedger         Int
+  endLedger           Int
+  batchSize           Int
+  dryRun              Boolean
+  status              String
+  rpcUrl              String
+  contractId          String
+  lastProcessedLedger Int?
+  progressJson        String
+  errorMessage        String?
+  createdAt           DateTime  @default(now())
+  updatedAt           DateTime  @updatedAt
+  completedAt         DateTime?
+
+  @@index([status])
+  @@index([createdAt])
+  @@index([dryRun])
+}

backend/src/__tests__/exportManifest.test.ts

@@ -0,0 +1,99 @@
+import crypto from 'crypto';
+import {
+  createExportManifest,
+  verifyExportManifestChecksum,
+  listExportManifests,
+  pruneExportManifests,
+  resetExportManifestsForTests,
+} from '../exportManifest';
+
+jest.mock('../prismaClient', () => ({
+  getPrismaClient: () => ({
+    exportManifest: {
+      create: jest.fn().mockResolvedValue({}),
+      findMany: jest.fn().mockResolvedValue([]),
+      findUnique: jest.fn().mockResolvedValue(null),
+      count: jest.fn().mockResolvedValue(0),
+      deleteMany: jest.fn().mockResolvedValue({ count: 0 }),
+    },
+  }),
+}));
+
+jest.mock('../middleware/structuredLogging', () => ({
+  logger: { log: jest.fn(), configure: jest.fn() },
+}));
+
+describe('exportManifest persistence', () => {
+  beforeEach(() => {
+    resetExportManifestsForTests();
+    process.env.EXPORT_MANIFEST_STORAGE = 'memory';
+  });
+
+  it('creates a manifest with deterministic checksum', async () => {
+    const manifest = await createExportManifest({
+      requester: 'admin@test',
+      reportType: 'transactions',
+      filters: { status: 'completed' },
+      rows: [{ id: '1', amount: '10' }],
+    });
+
+    expect(manifest.id).toMatch(/^exp-/);
+    expect(manifest.checksum).toHaveLength(64);
+    expect(manifest.rowCount).toBe(1);
+  });
+
+  it('verifies checksum match and mismatch without exposing rows', async () => {
+    const manifest = await createExportManifest({
+      requester: 'admin@test',
+      reportType: 'transactions',
+      filters: {},
+      rows: [{ id: '1' }],
+    });
+
+    const match = await verifyExportManifestChecksum(manifest.id, manifest.checksum);
+    expect(match.match).toBe(true);
+    expect(match.manifest?.rowCount).toBe(1);
+    expect((match.manifest as any).rows).toBeUndefined();
+
+    const mismatch = await verifyExportManifestChecksum(manifest.id, crypto.randomBytes(32).toString('hex'));
+    expect(mismatch.match).toBe(false);
+  });
+
+  it('lists manifests with pagination from memory store', async () => {
+    await createExportManifest({
+      requester: 'a',
+      reportType: 'transactions',
+      filters: {},
+      rows: [{ id: '1' }],
+    });
+    await createExportManifest({
+      requester: 'b',
+      reportType: 'transactions',
+      filters: {},
+      rows: [{ id: '2' }],
+    });
+
+    const page = await listExportManifests(1, 0);
+    expect(page.total).toBe(2);
+    expect(page.data).toHaveLength(1);
+  });
+
+  it('prunes manifests beyond retention limit', async () => {
+    process.env.EXPORT_MANIFEST_RETENTION = '2';
+
+    for (let i = 0; i < 4; i += 1) {
+      await createExportManifest({
+        requester: 'admin',
+        reportType: 'transactions',
+        filters: { i },
+        rows: [{ id: String(i) }],
+      });
+    }
+
+    const pruned = await pruneExportManifests();
+    expect(pruned).toBeGreaterThan(0);
+
+    const remaining = await listExportManifests(10, 0);
+    expect(remaining.total).toBeLessThanOrEqual(2);
+  });
+});

backend/src/__tests__/ledgerReconciliationJob.test.ts

@@ -0,0 +1,163 @@
+/**
+ * Tests for scheduled ledger reconciliation drift detection.
+ */
+
+import {
+  runReconciliationReport,
+  reconcile,
+  resetReconciliationStateForTests,
+  type LedgerRecord,
+} from '../reconciliationReport';
+import { runLedgerReconciliationJob, resetLedgerReconciliationSchedulerForTests } from '../positionReconciliationJob';
+import {
+  reconciliationDriftTotal,
+  reconciliationStatus,
+  reconciliationLastRunTimestamp,
+  register,
+} from '../metrics';
+import { resetJobGovernance } from '../jobGovernance';
+
+const mockFindMany = jest.fn().mockResolvedValue([]);
+const mockSnapshotCreate = jest.fn().mockResolvedValue({});
+
+jest.mock('../prismaClient', () => ({
+  getPrismaClient: () => ({
+    transaction: {
+      findMany: (...args: unknown[]) => mockFindMany(...args),
+    },
+    reconciliationSnapshot: {
+      create: (...args: unknown[]) => mockSnapshotCreate(...args),
+    },
+  }),
+}));
+
+jest.mock('../middleware/structuredLogging', () => ({
+  logger: { log: jest.fn(), configure: jest.fn() },
+}));
+
+describe('runReconciliationReport', () => {
+  beforeEach(() => {
+    resetReconciliationStateForTests();
+    resetLedgerReconciliationSchedulerForTests();
+    resetJobGovernance();
+    mockFindMany.mockReset();
+    mockFindMany.mockResolvedValue([]);
+  });
+
+  it('reports CLEAN when ledger and database records match', async () => {
+    const records: LedgerRecord[] = [
+      {
+        transactionHash: 'tx-1',
+        type: 'deposit',
+        amount: '100',
+        walletAddress: 'GABC',
+        timestamp: '2026-06-20T10:00:00Z',
+      },
+    ];
+
+    mockFindMany.mockResolvedValueOnce([
+      {
+        id: 'tx-1',
+        type: 'deposit',
+        amount: '100',
+        user: 'GABC',
+        timestamp: new Date('2026-06-20T10:00:00Z'),
+      },
+    ]);
+
+    const report = await runReconciliationReport({
+      from: '2026-06-20T00:00:00Z',
+      to: '2026-06-21T00:00:00Z',
+      ledgerFetcher: async () => records,
+      storeAsAutomated: true,
+      persistSnapshot: false,
+    });
+
+    expect(report.status).toBe('CLEAN');
+    expect(report.counts.matched).toBe(1);
+    expect(report.counts.drifted).toBe(0);
+  });
+
+  it('detects drift when ledger record is missing in DB', async () => {
+    const ledgerRecords: LedgerRecord[] = [
+      {
+        transactionHash: 'tx-missing',
+        type: 'deposit',
+        amount: '50',
+        walletAddress: 'GXYZ',
+        timestamp: '2026-06-20T10:00:00Z',
+      },
+    ];
+
+    const report = await runReconciliationReport({
+      from: '2026-06-20T00:00:00Z',
+      to: '2026-06-21T00:00:00Z',
+      ledgerFetcher: async () => ledgerRecords,
+      storeAsAutomated: true,
+      persistSnapshot: false,
+    });
+
+    expect(report.status).toBe('DRIFT_DETECTED');
+    expect(report.driftEntries[0].issue).toBe('MISSING_IN_DB');
+  });
+});
+
+describe('runLedgerReconciliationJob metrics', () => {
+  beforeEach(async () => {
+    resetReconciliationStateForTests();
+    resetJobGovernance();
+    reconciliationDriftTotal.reset();
+    reconciliationStatus.reset();
+    reconciliationLastRunTimestamp.reset();
+  });
+
+  it('increments drift counters and sets status gauge on drift', async () => {
+    const ledgerRecords: LedgerRecord[] = [
+      {
+        transactionHash: 'tx-drift',
+        type: 'withdrawal',
+        amount: '10',
+        walletAddress: 'GDRIFT',
+        timestamp: '2026-06-20T10:00:00Z',
+      },
+    ];
+
+    jest.spyOn(require('../reconciliationReport'), 'runReconciliationReport').mockResolvedValue({
+      generatedAt: new Date().toISOString(),
+      traceId: 'trace-1',
+      window: { from: '2026-06-20T00:00:00Z', to: '2026-06-21T00:00:00Z' },
+      counts: { ledgerRecords: 1, databaseRecords: 0, matched: 0, drifted: 1 },
+      driftEntries: [{ transactionHash: 'tx-drift', issue: 'MISSING_IN_DB', details: {} }],
+      status: 'DRIFT_DETECTED',
+    });
+
+    await runLedgerReconciliationJob();
+
+    const metrics = await register.metrics();
+    expect(metrics).toContain('reconciliation_drift_total');
+    expect(metrics).toContain('reconciliation_status');
+    expect(metrics).toContain('reconciliation_last_run_timestamp');
+  });
+});
+
+describe('reconcile', () => {
+  it('flags amount mismatches', () => {
+    const ledger: LedgerRecord[] = [{
+      transactionHash: 'tx-1',
+      type: 'deposit',
+      amount: '100',
+      walletAddress: 'GABC',
+      timestamp: '2026-06-20T10:00:00Z',
+    }];
+    const db: LedgerRecord[] = [{
+      transactionHash: 'tx-1',
+      type: 'deposit',
+      amount: '99',
+      walletAddress: 'GABC',
+      timestamp: '2026-06-20T10:00:00Z',
+    }];
+
+    const result = reconcile(ledger, db);
+    expect(result.driftEntries[0].issue).toBe('AMOUNT_MISMATCH');
+  });
+});