import { Ledger } from '../testnet/ledger.ts'; import type { PostgresCustody } from './postgres-custody.ts'; export type CanonicalObserverState = 'stopped' | 'running' | 'failed' | 'review-required'; export type CanonicalObserverStatus = { state: CanonicalObserverState; cursor: number; head: number; height: number; tip: string; caughtUp: boolean; failureCode: string | null }; /** Bounded polling loop for the explicitly disposable PostgreSQL rehearsal only. * It reads local Ledger events; it does not verify candidate consensus or trigger custody payouts. */ export class CanonicalObserver { private readonly custody: PostgresCustody; private readonly ledger: Ledger; private readonly intervalMs: number; private readonly pagesPerCycle: number; private active = false; private timer: ReturnType | undefined; private cycle: Promise | undefined; private current: CanonicalObserverStatus = { state: 'stopped', cursor: 0, head: 0, height: 0, tip: '', caughtUp: false, failureCode: null }; constructor(custody: PostgresCustody, ledger: Ledger, options: { intervalMs?: number; pagesPerCycle?: number } = {}) { if (!custody.rehearsalObserverEnabled) throw new Error('Canonical observer requires disposable rehearsal custody'); if (process.env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL !== '1' || process.env.WIKIDECK_PG_DISPOSABLE_TEST !== '1' || process.env.NODE_ENV === 'production') throw new Error('Canonical observer requires explicit non-production disposable runtime opt-in'); if (!(ledger instanceof Ledger)) throw new Error('Canonical observer requires a local Ledger instance'); this.custody = custody; this.ledger = ledger; this.intervalMs = options.intervalMs ?? 1000; this.pagesPerCycle = options.pagesPerCycle ?? 4; if (!Number.isSafeInteger(this.intervalMs) || this.intervalMs < 25 || this.intervalMs > 60000 || !Number.isSafeInteger(this.pagesPerCycle) || this.pagesPerCycle < 1 || this.pagesPerCycle > 16) { throw new Error('Invalid canonical observer bounds'); } } status(): CanonicalObserverStatus { return { ...this.current }; } async start(): Promise { if (this.active) { await this.cycle; return this.status(); } if (this.current.state === 'failed' || this.current.state === 'review-required') return this.status(); this.active = true; this.current = { ...this.current, state: 'running', failureCode: null }; await this.runCycle(); return this.status(); } /** Deterministic wake-up hook; the scheduled poll and manual wake are serialized. */ async pollNow(): Promise { if (!this.active) throw new Error('Canonical observer is not running'); if (this.timer) { clearTimeout(this.timer); this.timer = undefined; } await this.runCycle(); return this.status(); } async close(): Promise { this.active = false; if (this.timer) { clearTimeout(this.timer); this.timer = undefined; } await this.cycle; if (this.current.state === 'running') this.current = { ...this.current, state: 'stopped' }; } private runCycle(): Promise { if (this.cycle) return this.cycle; const work = (async () => { for (let page = 0; page < this.pagesPerCycle && this.active; page++) { try { const result = await this.custody.observeLedgerCanonicalEvents(this.ledger); this.current = { state: result.reviewRequired ? 'review-required' : 'running', cursor: result.cursor, head: result.head, height: result.height, tip: result.tip, caughtUp: result.cursor >= result.head, failureCode: result.reviewRequired ? 'posted-sale-terminal-disconnected' : null }; if (result.reviewRequired) { this.active = false; break; } if (result.cursor >= result.head) break; } catch { this.current = { ...this.current, state: 'failed', failureCode: 'canonical-observer-failed' }; this.active = false; break; } } })(); this.cycle = work.finally(() => { this.cycle = undefined; if (this.active && !this.timer) { this.timer = setTimeout(() => { this.timer = undefined; void this.runCycle(); }, this.intervalMs); this.timer.unref(); } }); return this.cycle; } }