wikideck/server/market/canonical-inbox.ts

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<typeof pageSchema>;
export type CanonicalInboxResult = { streamId: string; cursor: number; head: number; height: number; tip: string;
  replayed: boolean; appendedEvents: z.infer<typeof eventSchema>[]; 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<typeof eventSchema> {
  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<string | undefined> {
  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<typeof eventSchema>, 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<typeof eventSchema>): 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<CanonicalInboxResult> {
  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<typeof eventSchema>[] = [];
  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<void> {
  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');
}