import type { PoolClient } from 'pg'; import { z } from 'zod'; import { GENESIS_HASH, sha256 } from '../testnet/consensus.ts'; const hex = z.string().regex(/^[a-f0-9]{64}$/); const MAX_REHEARSAL_EVENTS = 4096; const eventSchema = z.object({ sequence: z.number().int().positive().safe(), type: z.enum(['connect', 'disconnect']), height: z.number().int().positive().safe(), hash: hex, previousHash: hex, transactionIds: z.array(hex).max(64), eventHash: hex }).strict(); const pageSchema = z.object({ streamId: z.string().uuid(), previousAnchor: hex, anchor: hex, events: z.array(eventSchema).max(64), cursor: z.number().int().nonnegative().safe(), head: z.number().int().nonnegative().safe(), caughtUp: z.boolean() }).strict(); export type CanonicalInboxPage = z.infer; export type CanonicalInboxResult = { streamId: string; cursor: number; head: number; height: number; tip: string; replayed: boolean; appendedEvents: z.infer[]; reviewRequired?: boolean }; export class CanonicalInboxError extends Error { readonly code: string; constructor(code: string) { super(`Canonical event inbox rejected input: ${code}`); this.code = code; } } const streamKey = 'canonical-inbox:v1:stream'; const eventKey = (sequence: number) => `canonical-inbox:v1:event:${sequence.toString().padStart(16, '0')}`; function storedEvent(text: string): z.infer { let decoded: unknown; try { decoded = JSON.parse(text); } catch { throw new CanonicalInboxError('stored-event-json-invalid'); } const parsed = eventSchema.safeParse(decoded); if (!parsed.success) throw new CanonicalInboxError('stored-event-invalid'); return parsed.data; } async function anchorAt(client: PoolClient, sequence: number): Promise { if (sequence === 0) return GENESIS_HASH; const row = await client.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1', [eventKey(sequence)]); return row.rows[0] ? storedEvent(row.rows[0].value).eventHash : undefined; } function eventDigest(streamId: string, e: z.infer, previousAnchor: string): string { return sha256(JSON.stringify(['wikideck-canonical-event-v1', streamId, e.sequence, e.type, e.height, e.hash, e.previousHash, e.transactionIds, previousAnchor])); } function applyTip(stack: string[], e: z.infer): void { if (e.type === 'connect') { if (e.height !== stack.length + 1 || e.previousHash !== (stack.at(-1) ?? GENESIS_HASH)) throw new CanonicalInboxError('connect-parent-mismatch'); stack.push(e.hash); return; } if (e.height !== stack.length || e.hash !== stack.at(-1) || e.height < 1 || e.previousHash !== (stack.at(-2) ?? GENESIS_HASH)) throw new CanonicalInboxError('disconnect-ancestry-mismatch'); stack.pop(); } async function readCanonicalHistory(client: PoolClient) { const rows = await client.query(`SELECT key,value FROM wk_custody.custody_meta WHERE key=$1 OR key LIKE 'canonical-inbox:v1:event:%' ORDER BY key LIMIT $2`, [streamKey, MAX_REHEARSAL_EVENTS + 2]); if (rows.rows.length > MAX_REHEARSAL_EVENTS + 1) throw new CanonicalInboxError('rehearsal-event-budget-exceeded'); const stream = rows.rows.find(r => r.key === streamKey)?.value as string | undefined; if (rows.rows.length && (!stream || !z.string().uuid().safeParse(stream).success)) throw new CanonicalInboxError('stream-record-invalid'); const stack: string[] = []; let cursor = 0, anchor = GENESIS_HASH; for (const row of rows.rows) { if (row.key === streamKey) continue; const match = /^canonical-inbox:v1:event:(\d{16})$/.exec(row.key); const e = storedEvent(row.value); if (!match || Number(match[1]) !== cursor + 1 || e.sequence !== cursor + 1) throw new CanonicalInboxError('history-gap-or-invalid-record'); if (e.eventHash !== eventDigest(stream!, e, anchor)) throw new CanonicalInboxError('event-anchor-invalid'); applyTip(stack, e); cursor++; anchor = e.eventHash; } return { streamId: stream, cursor, anchor, stack }; } function assertPageShape(page: CanonicalInboxPage): void { const finalCursor = page.events.at(-1)?.sequence ?? page.cursor; if ((page.events.length && page.events[0].sequence < 1) || page.cursor !== finalCursor || page.head < page.cursor || page.caughtUp !== (page.cursor === page.head) || (!page.events.length && !page.caughtUp) || page.anchor !== (page.events.at(-1)?.eventHash ?? page.previousAnchor) || page.events.some((e, i) => i > 0 && e.sequence !== page.events[i - 1].sequence + 1)) throw new CanonicalInboxError('page-shape-invalid'); } /** Persist only events returned by the local ledger's validated canonical-event API. This verifies * continuity/integrity, not PoW or provenance; callers must not pass peer-supplied, unvalidated data. */ export async function ingestCanonicalPage(client: PoolClient, raw: unknown): Promise { const parsed = pageSchema.safeParse(raw); if (!parsed.success) throw new CanonicalInboxError('page-schema-invalid'); const page = parsed.data; assertPageShape(page); if (page.head > MAX_REHEARSAL_EVENTS || page.cursor > MAX_REHEARSAL_EVENTS) throw new CanonicalInboxError('rehearsal-event-budget-exceeded'); const history = await readCanonicalHistory(client); const storedStream = history.streamId; if (storedStream && storedStream !== page.streamId) throw new CanonicalInboxError('stream-changed'); let { cursor, anchor } = history; const stack = history.stack; if (!storedStream) { if (cursor !== 0 || page.previousAnchor !== GENESIS_HASH || (page.events.length ? page.events[0].sequence !== 1 : page.cursor !== 0)) { throw new CanonicalInboxError('initial-history-required'); } await client.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2)', [streamKey, page.streamId]); } if (!page.events.length) { const savedAnchor = await anchorAt(client, page.cursor); if (page.cursor > cursor || savedAnchor !== page.anchor || page.previousAnchor !== page.anchor) throw new CanonicalInboxError('empty-cursor-mismatch'); return { streamId: page.streamId, cursor, head: Math.max(page.head, cursor), height: stack.length, tip: stack.at(-1) ?? GENESIS_HASH, replayed: true, appendedEvents: [] }; } const firstSequence = page.events[0].sequence; if (firstSequence > cursor + 1) throw new CanonicalInboxError('cursor-gap'); const predecessor = await anchorAt(client, firstSequence - 1); if (!predecessor || page.previousAnchor !== predecessor) throw new CanonicalInboxError('cursor-anchor-mismatch'); let priorAnchor = predecessor; let appended = false; const appendedEvents: z.infer[] = []; for (const e of page.events) { if (e.eventHash !== eventDigest(page.streamId, e, priorAnchor)) throw new CanonicalInboxError('event-hash-invalid'); if (e.sequence <= cursor) { const saved = await client.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1', [eventKey(e.sequence)]); if (!saved.rowCount || JSON.stringify(storedEvent(saved.rows[0].value)) !== JSON.stringify(e)) throw new CanonicalInboxError('replayed-event-mismatch'); } else { if (e.sequence !== cursor + 1 || priorAnchor !== anchor) throw new CanonicalInboxError('cursor-anchor-mismatch'); applyTip(stack, e); await client.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2)', [eventKey(e.sequence), JSON.stringify(e)]); cursor = e.sequence; anchor = e.eventHash; appended = true; appendedEvents.push(e); } priorAnchor = e.eventHash; } if (page.cursor !== page.events.at(-1)!.sequence || page.anchor !== priorAnchor) throw new CanonicalInboxError('cursor-anchor-mismatch'); return { streamId: page.streamId, cursor, head: Math.max(page.head, cursor), height: stack.length, tip: stack.at(-1) ?? GENESIS_HASH, replayed: !appended, appendedEvents }; } /** Full append-only inbox verification. Expected to run under the caller's coherent audit snapshot. */ export async function auditCanonicalInbox(client: PoolClient): Promise { await readCanonicalHistory(client); const reviews = await client.query(`SELECT key,value FROM wk_custody.custody_meta WHERE key LIKE 'sale-reorg-review:%' ORDER BY key`); for (const row of reviews.rows) { let record: unknown; try { record = JSON.parse(row.value); } catch { throw new CanonicalInboxError('sale-review-record-invalid'); } const schema = z.object({ version: z.literal(1), streamId: z.string().uuid(), orderId: z.string().uuid(), terminalTransactionId: hex, stateRoot: hex, eventSequence: z.number().int().positive().safe(), eventHash: hex }).strict(); const parsed = schema.safeParse(record), keyMatch = /^sale-reorg-review:([a-f0-9-]{36}):(\d+)$/.exec(row.key); if (!parsed.success || !keyMatch || keyMatch[1] !== parsed.data.orderId || Number(keyMatch[2]) !== parsed.data.eventSequence) { throw new CanonicalInboxError('sale-review-record-invalid'); } const event = await client.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1', [eventKey(parsed.data.eventSequence)]); if (!event.rowCount) throw new CanonicalInboxError('sale-review-event-missing'); const disconnected = storedEvent(event.rows[0].value); if (disconnected.type !== 'disconnect' || disconnected.eventHash !== parsed.data.eventHash || !disconnected.transactionIds.includes(parsed.data.terminalTransactionId)) { throw new CanonicalInboxError('sale-review-event-mismatch'); } } if (reviews.rowCount) throw new CanonicalInboxError('posted-sale-terminal-disconnected'); }