Disposable canonical-event inbox rehearsal

server/market/postgres-custody.ts

@@ -2,6 +2,7 @@ import { Pool, type PoolClient, type PoolConfig } from 'pg';
 import { createHash, randomUUID } from 'node:crypto';
 import { IS_MAINNET, NETWORK } from '../../shared/network.ts';
 import { CHAIN_ID, verifySignature } from '../testnet/consensus.ts';
+import { Ledger } from '../testnet/ledger.ts';
 import { ADDITIONAL_MARKET_FEE } from '../../shared/market.ts';
 import { allocateXmrSale } from '../../shared/xmr-pricing.ts';
 import { parseAtomicXmr } from '../../shared/xmr-amount.ts';
@@ -13,6 +14,7 @@ import { CUSTODY_WRITE_LOCK, verifyCustodySchema } from './postgres-schema.ts';
 import { resolvedMarketPolicy, validateMarketPolicy, permitsMarketAction, type MarketPolicy } from './policy.ts';
 import { CustodyIntegrityError, auditBusinessLiabilities, auditOrder, auditWithdrawal } from './custody-integrity.ts';
 import { replayReservedSale } from './sale-rehearsal.ts';
+import { CanonicalInboxError, auditCanonicalInbox, ingestCanonicalPage } from './canonical-inbox.ts';
 
 export type CustodyOptions = { feeBps: number; card(hash: string, owner: string, sequence: number): { title: string } | undefined;
   now?: () => number; disposableSaleRehearsal?: boolean } & MarketPolicy;
@@ -337,6 +339,38 @@ export class PostgresCustody {
       return { order: o, action: replay.decision.action, reservationDigest: replay.reservationDigest, stateRoot: replay.stateRoot };
     });
   }
+  /** Disposable observer rehearsal only; input must come from Ledger.canonicalEvents(), never peers/HTTP. */
+  observeCanonicalEvents(page: unknown) {
+    check(this.options.disposableSaleRehearsal === true, 'Canonical event inbox rehearsal is unavailable on this custody instance');
+    return this.transaction(true, async c => {
+      check(!this.diagnostics().mutationHold, 'Custody reconciliation hold');
+      check(!(await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'")).rowCount, 'Custody integrity incident unresolved');
+      try { return await ingestCanonicalPage(c, page); }
+      catch (error) {
+        if (error instanceof CanonicalInboxError) throw new CustodyIntegrityError(`canonical-inbox-${error.code}`);
+        throw error;
+      }
+    });
+  }
+  /** Consume one bounded page directly from this process's locally validated ledger; no HTTP/peer feed. */
+  async observeLedgerCanonicalEvents(ledger: Ledger) {
+    check(this.options.disposableSaleRehearsal === true, 'Canonical event inbox rehearsal is unavailable on this custody instance');
+    check(ledger instanceof Ledger, 'Canonical inbox requires a local Ledger instance');
+    const summary = ledger.summary();
+    check(summary.network === NETWORK && summary.chainId === CHAIN_ID, 'Canonical ledger identity mismatch');
+    const cursor = await this.transaction(false, async c => {
+      const row = await c.query(`SELECT key FROM wk_custody.custody_meta
+        WHERE key LIKE 'canonical-inbox:v1:event:%' ORDER BY key DESC LIMIT 1`);
+      if (!row.rowCount) return 0;
+      const match = /^canonical-inbox:v1:event:(\d{16})$/.exec(row.rows[0].key);
+      if (!match) throw new CustodyIntegrityError('canonical-inbox-cursor-invalid');
+      return Number(match[1]);
+    });
+    let page: unknown;
+    try { page = ledger.canonicalEvents(cursor, 64); }
+    catch { throw new CustodyIntegrityError('canonical-ledger-cursor-mismatch'); }
+    return this.observeCanonicalEvents(page);
+  }
   audit(): Promise<CustodyAudit> {
     if (this.auditInFlight) return this.auditInFlight;
     this.auditState.lastAttemptAt = Date.now();
@@ -359,6 +393,11 @@ export class PostgresCustody {
 export async function auditPostgres(c: PoolClient): Promise<CustodyAudit> {
   const incident = await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'");
   if (incident.rowCount) throw new CustodyIntegrityError('unresolved-integrity-incident');
+  try { await auditCanonicalInbox(c); }
+  catch (error) {
+    if (error instanceof CanonicalInboxError) throw new CustodyIntegrityError(`canonical-inbox-${error.code}`);
+    throw error;
+  }
   const bad = await c.query(`SELECT 1 FROM wk_custody.journals j LEFT JOIN wk_custody.entries e ON e.journal=j.id
     GROUP BY j.id HAVING count(e.position)=0 OR sum(e.debit-e.credit)<>0 LIMIT 1`);
   if (bad.rowCount) throw new CustodyIntegrityError('unbalanced-journal');