feat(reconciliation): design + canonical comparison foundation

docs/reconciliation-control-plane.md

@@ -0,0 +1,152 @@
+# Reconciliation, drift forensics & zero-downtime projection repair
+
+> Design document for issue #8. This is the architecture; it is delivered with a
+> foundational implementation slice (the run/discrepancy model + the canonical
+> comparison core) and is followed by the implementation PRs listed at the end.
+
+## Problem
+
+The OrbitPay indexer is a **derived-state pipeline, not a source of truth**.
+Even with resumable, idempotent ingestion, the PostgreSQL projections can drift
+from chain reality: missed ledger ranges, decoder regressions, contract
+upgrades, RPC inconsistencies, partial migrations, or manual DB edits all corrupt
+projections silently.
+
+We need an **independent control plane** that can:
+
+1. Prove what was checked, at a reproducible ledger boundary.
+2. Explain every discrepancy with enough forensic evidence to attribute the
+   cause (ingestion vs decode vs schema vs RPC vs operator).
+3. Repair or shadow-rebuild projections **without** mutating chain state or
+   interrupting reads.
+
+A reconciliation engine is a correctness amplifier or a correctness destroyer:
+an over-eager repair masks real loss, an under-eager one floods ops. So the
+design is **report-only by default**, exact-integer throughout, and every
+mutation is gated, planned, checksummed, and reversible in audit.
+
+## Components
+
+```
+                ┌──────────────────────────────────────────────┐
+   chain RPC ──▶│  evidence collector  (pinned ledger, XDR)     │
+                └───────────────┬──────────────────────────────┘
+                                ▼ canonical snapshot (base units, fingerprint)
+   projection ─▶  canonical projector ─▶  comparator  ─▶  discrepancies
+                                                │
+                          ┌─────────────────────┼─────────────────────┐
+                          ▼                      ▼                     ▼
+                    evidence store        repair planner        shadow rebuild
+                    (immutable)        (report→plan→apply)     (build + cutover)
+```
+
+This PR implements the **canonical projector ↔ comparator** core and the
+**run/discrepancy data model**. The rest is specified below and lands in
+follow-ups.
+
+## 1. Reconciliation run model
+
+Every run persists an immutable **pin** of all inputs that could change its
+verdict (`RunPin` in `src/reconciliation/types.ts`):
+
+- network passphrase, RPC endpoint identity,
+- contract registry version, contract IDs → deployed WASM hashes,
+- the **pinned ledger** every comparison is taken at,
+- decoder version, DB schema version,
+- start/end time, terminal status (`RunStatus`).
+
+Pinning makes a run **exactly reproducible** and lets us detect a contract
+upgrade (WASM hash change) or a schema migration mid-flight.
+
+- **Distributed lease.** A run (or a cutover) acquires a Redis lease keyed by
+  `domain + projection`, so only one conflicting operation runs at a time.
+  Leases are fenced (monotonic token) so a paused run cannot resume over a newer
+  one.
+- **Resumable shards.** Work is partitioned into bounded shards with explicit
+  high-water marks, persisted as the run progresses, so cancellation, timeout,
+  or a crash resumes without losing completed comparison work.
+
+## 2. Canonical comparison  *(implemented in this PR)*
+
+`src/reconciliation/canonical.ts` + `compare.ts`:
+
+- **Exact base units only.** `toBaseUnits` coerces amounts to `bigint` and
+  **rejects** floats, non-integers, NaN/Infinity, and unsafe-integer numbers.
+  `compareAmounts` / `amountsEqual` compare exactly. No JavaScript or SQL
+  floating point ever participates in a money comparison.
+
+  > Note: the current projection schema stores amounts as `Float` (e.g.
+  > `Stream.totalAmount`). The canonical layer treats those as a **migration
+  > target** — a follow-up moves them to base-unit integer columns. Until then
+  > the projector converts at the boundary and the float-ban guard flags any
+  > value that cannot be represented exactly.
+
+- **Normalization.** `normalize` sorts object keys, drops optional/default-absent
+  fields, renders bigints as decimal strings, and normalizes dates — so two
+  semantically-equal snapshots produce identical output regardless of key order
+  or absent optionals. Order-insensitive collections are sorted via `sortBy`
+  before hashing.
+
+- **Deterministic fingerprints.** `fingerprint` is the sha256 of the canonical
+  string; two entities are equal iff their fingerprints match.
+
+- **Classification.** `compareDomain` (pure) diffs chain vs projection per entity
+  and classifies each difference (`DiscrepancyKind`): `missing`, `unexpected`,
+  `stale`, `value_mismatch`, `invalid_transition`, `aggregate_mismatch`,
+  `decode_failure`, `unavailable_historical_state`, `unsupported_contract_version`.
+
+Canonical snapshots are defined per domain: treasury config + withdrawals,
+payroll streams + claims, vesting schedules + claims, and governance proposals,
+membership, weights, and votes.
+
+## 3. Evidence & forensics
+
+Each discrepancy persists the forensic chain so the cause is attributable
+**without re-running**:
+
+- source ledger, transaction/event identifiers, raw XDR reference or checksum,
+  decoded payload, projection row + version, expected vs observed value, decoder
+  provenance.
+- Discrepancy records are **immutable**; a resolution writes a linked,
+  append-only **action** record (never an in-place edit).
+- Secrets and auth material are **redacted** from all logs and evidence.
+- Retention + export are defined for incident review.
+
+## 4. Repair safety
+
+- **Report-only by default** (`RepairMode = 'report_only'`).
+- A repair requires: an authorized operator, explicit network + contract scope,
+  ledger range, the specific discrepancy set, an idempotency key, a reason, and
+  dry-run approval.
+- A deterministic **repair plan** is generated; its checksum is approved before
+  execution.
+- Application uses transactional **compare-and-swap** guards (on row version) so
+  any row changed after planning is never blindly overwritten.
+- The control plane **never submits Soroban transactions** — it only repairs
+  derived state.
+
+## 5. Shadow rebuild & zero-downtime cutover
+
+- A projection can be rebuilt in a **shadow** schema from the pinned ledger.
+- Readers are proven never to observe partially-repaired or mixed-version state:
+  cutover is an atomic pointer swap behind a fenced lease, with the schema
+  version stamped so a reader can assert it read a single consistent version.
+
+## 6. Observability
+
+- Per-run metrics: entities checked, discrepancies by kind/severity, drift
+  amount per domain, shard high-water marks, lease contention.
+- A run emits a terminal summary; criticals page, warnings aggregate.
+
+## Delivery plan (per the issue: design + several reviewable PRs)
+
+1. **This PR** — design doc + run/discrepancy data model + canonical comparison
+   core (`toBaseUnits`, `normalize`, `fingerprint`, `compareDomain`) with tests.
+2. Evidence store + Prisma persistence of runs/discrepancies/actions wired to a
+   first read-only domain (treasury).
+3. Pinned-ledger evidence collector (Soroban historical state + XDR) and the
+   canonical projectors for all four domains.
+4. Distributed lease + resumable shards + run lifecycle.
+5. Report-only reconciliation worker + metrics.
+6. Gated repair planner/executor (compare-and-swap, idempotency, approval).
+7. Shadow rebuild + zero-downtime cutover.

package.json

@@ -11,7 +11,8 @@
     "start:indexer": "node dist/indexer.js",
     "prisma:generate": "prisma generate",
     "prisma:migrate:deploy": "prisma migrate deploy",
-    "test": "npm run build"
+    "test": "npm run build",
+    "test:unit": "node -r ts-node/register/transpile-only --test src/reconciliation/canonical.test.ts src/reconciliation/compare.test.ts"
   },
   "keywords": [],
   "author": "",

src/reconciliation/canonical.test.ts

@@ -0,0 +1,66 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+
+import {
+  toBaseUnits,
+  compareAmounts,
+  amountsEqual,
+  normalize,
+  canonicalString,
+  fingerprint,
+  sortBy,
+} from './canonical';
+
+test('toBaseUnits accepts exact integer representations', () => {
+  assert.equal(toBaseUnits(10n), 10n);
+  assert.equal(toBaseUnits(42), 42n);
+  assert.equal(toBaseUnits('123456789012345678901234567890'), 123456789012345678901234567890n);
+  assert.equal(toBaseUnits('-5'), -5n);
+});
+
+test('toBaseUnits rejects anything that could lose precision', () => {
+  assert.throws(() => toBaseUnits(1.5), TypeError);
+  assert.throws(() => toBaseUnits(Number.NaN), TypeError);
+  assert.throws(() => toBaseUnits(Number.MAX_SAFE_INTEGER + 1), TypeError);
+  assert.throws(() => toBaseUnits('1.0'), TypeError);
+  assert.throws(() => toBaseUnits('abc'), TypeError);
+});
+
+test('amount comparison is exact, even past 2^53', () => {
+  assert.equal(compareAmounts('9007199254740993', '9007199254740992'), 1);
+  assert.equal(compareAmounts(100, 100n), 0);
+  assert.equal(compareAmounts('5', 9), -1);
+  assert.ok(amountsEqual('1000', 1000n));
+  assert.ok(!amountsEqual('1000', '1001'));
+});
+
+test('normalize is key-order independent and drops nullish fields', () => {
+  const a = normalize({ b: 2, a: 1, c: null, d: undefined });
+  const b = normalize({ a: 1, b: 2 });
+  assert.deepEqual(a, b);
+});
+
+test('normalize renders bigints as exact decimal strings and dates as ISO', () => {
+  assert.equal(
+    canonicalString({ amount: 10n, at: new Date('2026-01-01T00:00:00.000Z') }),
+    '{"amount":"10","at":"2026-01-01T00:00:00.000Z"}',
+  );
+});
+
+test('normalize refuses non-integer numbers (no float money)', () => {
+  assert.throws(() => normalize({ amount: 1.23 }), TypeError);
+  assert.throws(() => normalize(Number.POSITIVE_INFINITY), TypeError);
+});
+
+test('fingerprint is stable across key order and sensitive to value changes', () => {
+  assert.equal(fingerprint({ a: 1, b: '2' }), fingerprint({ b: '2', a: 1 }));
+  assert.notEqual(fingerprint({ a: 1 }), fingerprint({ a: 2 }));
+});
+
+test('sortBy gives a deterministic order for order-insensitive collections', () => {
+  const items = [{ id: 'c' }, { id: 'a' }, { id: 'b' }];
+  assert.deepEqual(
+    sortBy(items, (i) => i.id).map((i) => i.id),
+    ['a', 'b', 'c'],
+  );
+});

src/reconciliation/canonical.ts

@@ -0,0 +1,136 @@
+/**
+ * Canonical comparison primitives for reconciliation (issue #8).
+ *
+ * Money is compared as exact base-unit integers — never JavaScript or SQL
+ * floating point. Snapshots from chain and from the projection are normalized to
+ * a canonical form (sorted keys, base units as decimal strings, ordering and
+ * optional/default fields normalized) and reduced to a deterministic fingerprint
+ * so two snapshots are equal iff their fingerprints match.
+ *
+ * Pure and dependency-free (only `node:crypto`), so it is trivially testable and
+ * safe to run anywhere in the pipeline.
+ */
+
+import { createHash } from 'node:crypto';
+
+export type Canonical =
+  | string
+  | boolean
+  | null
+  | Canonical[]
+  | { [key: string]: Canonical };
+
+/**
+ * Coerce a base-unit amount to an exact bigint, rejecting anything that could
+ * silently lose precision. Accepts a bigint, an integer-valued number that is
+ * still safe, or a decimal string of an integer. Floats, NaN, Infinity, and
+ * non-integers throw — this is the guard against floating-point money.
+ */
+export function toBaseUnits(value: bigint | number | string): bigint {
+  if (typeof value === 'bigint') return value;
+
+  if (typeof value === 'number') {
+    if (!Number.isInteger(value)) {
+      throw new TypeError(`base-unit amount must be an integer, got ${value}`);
+    }
+    if (!Number.isSafeInteger(value)) {
+      throw new TypeError(
+        `base-unit amount ${value} exceeds Number.MAX_SAFE_INTEGER; pass a string or bigint`,
+      );
+    }
+    return BigInt(value);
+  }
+
+  if (typeof value === 'string') {
+    if (!/^-?\d+$/.test(value.trim())) {
+      throw new TypeError(`base-unit amount must be an integer string, got "${value}"`);
+    }
+    return BigInt(value.trim());
+  }
+
+  throw new TypeError(`unsupported base-unit amount type: ${typeof value}`);
+}
+
+/** Exact three-way comparison of two base-unit amounts. */
+export function compareAmounts(
+  a: bigint | number | string,
+  b: bigint | number | string,
+): -1 | 0 | 1 {
+  const x = toBaseUnits(a);
+  const y = toBaseUnits(b);
+  if (x < y) return -1;
+  if (x > y) return 1;
+  return 0;
+}
+
+/** True iff two base-unit amounts are exactly equal. */
+export function amountsEqual(
+  a: bigint | number | string,
+  b: bigint | number | string,
+): boolean {
+  return compareAmounts(a, b) === 0;
+}
+
+/**
+ * Normalize an arbitrary value into canonical form:
+ *  - object keys sorted, `undefined` / nullish-default fields dropped;
+ *  - bigints rendered as decimal strings (exact);
+ *  - numbers required to be integers (floats are a bug here) and stringified;
+ *  - arrays normalized element-wise (order preserved — callers sort
+ *    order-insensitive collections via {@link sortBy} before normalizing).
+ */
+export function normalize(value: unknown): Canonical {
+  if (value === null || value === undefined) return null;
+
+  if (typeof value === 'bigint') return value.toString();
+
+  if (typeof value === 'number') {
+    if (!Number.isFinite(value)) {
+      throw new TypeError(`cannot canonicalize non-finite number ${value}`);
+    }
+    if (!Number.isInteger(value)) {
+      throw new TypeError(
+        `refusing to canonicalize a non-integer number (${value}); convert money to base-unit strings first`,
+      );
+    }
+    return value.toString();
+  }
+
+  if (typeof value === 'boolean') return value;
+  if (typeof value === 'string') return value;
+
+  if (value instanceof Date) return value.toISOString();
+
+  if (Array.isArray(value)) return value.map((v) => normalize(v));
+
+  if (typeof value === 'object') {
+    const out: { [key: string]: Canonical } = {};
+    for (const key of Object.keys(value as Record<string, unknown>).sort()) {
+      const v = (value as Record<string, unknown>)[key];
+      if (v === undefined || v === null) continue; // drop optional/default-absent fields
+      out[key] = normalize(v);
+    }
+    return out;
+  }
+
+  throw new TypeError(`cannot canonicalize value of type ${typeof value}`);
+}
+
+/** Deterministic, key-order-independent string for a canonical value. */
+export function canonicalString(value: unknown): string {
+  return JSON.stringify(normalize(value));
+}
+
+/** Deterministic fingerprint (sha256 hex) of a value's canonical form. */
+export function fingerprint(value: unknown): string {
+  return createHash('sha256').update(canonicalString(value)).digest('hex');
+}
+
+/** Stable sort helper for order-insensitive collections, by a derived key. */
+export function sortBy<T>(items: readonly T[], key: (item: T) => string): T[] {
+  return [...items].sort((a, b) => {
+    const ka = key(a);
+    const kb = key(b);
+    return ka < kb ? -1 : ka > kb ? 1 : 0;
+  });
+}