From 79942a2431ed1069a95ae575c03613046672ec28 Mon Sep 17 00:00:00 2001 From: Simon Axelsson Date: Mon, 7 Sep 2026 17:13:02 +0200 Subject: [PATCH] feat(evidence): implement durable AUD0 outbox and dispatcher --- src/evidence/aud0-outbox.ts | 377 ++++++++++++++++++++++++++++++++++++ 1 file changed, 377 insertions(+) create mode 100644 src/evidence/aud0-outbox.ts diff --git a/src/evidence/aud0-outbox.ts b/src/evidence/aud0-outbox.ts new file mode 100644 index 0000000..c2bec81 --- /dev/null +++ b/src/evidence/aud0-outbox.ts @@ -0,0 +1,377 @@ +import crypto from "node:crypto"; + +import type { Aud0Client, Aud0Event } from "../clients/aud0-client"; +import type { OpenLineageRunEvent } from "../lineage/openlineage"; +import { SiaxError } from "../platform/errors"; +import { + getRepositorySql, + repositoryDatabaseUrl, + type RepositorySql, +} from "../repositories/db"; + +export type Aud0EvidenceKind = "event" | "openlineage"; +export type Aud0OutboxStatus = "pending" | "delivered" | "dead"; + +export interface Aud0OutboxRecord { + outbox_id: string; + tenant_id: string; + app_id: string; + evidence_kind: Aud0EvidenceKind; + idempotency_key: string; + payload: Aud0Event | OpenLineageRunEvent; + status: Aud0OutboxStatus; + attempts: number; + next_attempt_at: string; + leased_by?: string; + leased_until?: string; + last_error?: string; + created_at: string; + updated_at: string; + delivered_at?: string; + dead_at?: string; +} + +export interface Aud0OutboxStats { + pending: number; + delivered: number; + dead: number; + oldest_pending_at?: string; +} + +export interface Aud0OutboxRepository { + enqueueEvent(event: Aud0Event): Promise; + enqueueOpenLineage( + tenantId: string, + appId: string, + event: OpenLineageRunEvent, + ): Promise; + claimPending(input: { + workerId: string; + limit: number; + leaseMs: number; + }): Promise; + markDelivered(outboxId: string, workerId: string): Promise; + markRetry(outboxId: string, workerId: string, error: string, delayMs: number): Promise; + markDead(outboxId: string, workerId: string, error: string): Promise; + stats(): Promise; +} + +interface OutboxRow { + outbox_id: string; + tenant_id: string; + app_id: string; + evidence_kind: Aud0EvidenceKind; + idempotency_key: string; + payload: Aud0Event | OpenLineageRunEvent; + status: Aud0OutboxStatus; + attempts: number; + next_attempt_at: string | Date; + leased_by: string | null; + leased_until: string | Date | null; + last_error: string | null; + created_at: string | Date; + updated_at: string | Date; + delivered_at: string | Date | null; + dead_at: string | Date | null; +} + +type JsonValue = Parameters[0]; + +function iso(value: string | Date): string { + return value instanceof Date ? value.toISOString() : new Date(value).toISOString(); +} + +function optionalIso(value: string | Date | null): string | undefined { + return value == null ? undefined : iso(value); +} + +function mapRow(row: OutboxRow): Aud0OutboxRecord { + return { + outbox_id: row.outbox_id, + tenant_id: row.tenant_id, + app_id: row.app_id, + evidence_kind: row.evidence_kind, + idempotency_key: row.idempotency_key, + payload: row.payload, + status: row.status, + attempts: Number(row.attempts), + next_attempt_at: iso(row.next_attempt_at), + ...(row.leased_by ? { leased_by: row.leased_by } : {}), + ...(row.leased_until ? { leased_until: optionalIso(row.leased_until) } : {}), + ...(row.last_error ? { last_error: row.last_error } : {}), + created_at: iso(row.created_at), + updated_at: iso(row.updated_at), + ...(row.delivered_at ? { delivered_at: optionalIso(row.delivered_at) } : {}), + ...(row.dead_at ? { dead_at: optionalIso(row.dead_at) } : {}), + }; +} + +function required(value: string, field: string): string { + const normalized = value.trim(); + if (!normalized) { + throw new SiaxError("INVALID_REQUEST", "0pt.aud0_outbox", `${field} is required`); + } + return normalized; +} + +function outboxId(tenantId: string, kind: Aud0EvidenceKind, idempotencyKey: string): string { + const digest = crypto + .createHash("sha256") + .update(tenantId) + .update("\0") + .update(kind) + .update("\0") + .update(idempotencyKey) + .digest("hex") + .slice(0, 40); + return `obx_${digest}`; +} + +function openLineageIdempotencyKey(event: OpenLineageRunEvent): string { + return `ol:${required(event.run.runId, "run.runId")}:${required(event.eventType, "eventType")}`; +} + +function safeError(error: unknown): string { + if (error instanceof SiaxError) return `${error.code}:${error.source}`.slice(0, 500); + if (error instanceof Error) return error.name.slice(0, 500); + return "UnknownError"; +} + +export class PostgresAud0OutboxRepository implements Aud0OutboxRepository { + constructor(private readonly sql: RepositorySql) {} + + async enqueueEvent(event: Aud0Event): Promise { + const key = required(event.idempotency_key ?? event.event_id, "event idempotency key"); + return this.enqueue({ + tenantId: required(event.tenant_id, "tenant_id"), + appId: required(event.app_id, "app_id"), + kind: "event", + idempotencyKey: key, + payload: event, + }); + } + + async enqueueOpenLineage( + tenantId: string, + appId: string, + event: OpenLineageRunEvent, + ): Promise { + return this.enqueue({ + tenantId: required(tenantId, "tenant_id"), + appId: required(appId, "app_id"), + kind: "openlineage", + idempotencyKey: openLineageIdempotencyKey(event), + payload: event, + }); + } + + private async enqueue(input: { + tenantId: string; + appId: string; + kind: Aud0EvidenceKind; + idempotencyKey: string; + payload: Aud0Event | OpenLineageRunEvent; + }): Promise { + const id = outboxId(input.tenantId, input.kind, input.idempotencyKey); + const rows = await this.sql` + INSERT INTO opt_aud0_evidence_outbox + (outbox_id, tenant_id, app_id, evidence_kind, idempotency_key, payload) + VALUES + (${id}, ${input.tenantId}, ${input.appId}, ${input.kind}, ${input.idempotencyKey}, + ${this.sql.json(input.payload as unknown as JsonValue)}) + ON CONFLICT (tenant_id, evidence_kind, idempotency_key) + DO UPDATE SET updated_at = now() + WHERE opt_aud0_evidence_outbox.payload = EXCLUDED.payload + RETURNING *`; + + if (rows.length === 0) { + throw new SiaxError( + "CONFLICT", + "0pt.aud0_outbox", + "the same evidence idempotency key was reused with a different payload", + ); + } + return mapRow(rows[0]); + } + + async claimPending(input: { + workerId: string; + limit: number; + leaseMs: number; + }): Promise { + const workerId = required(input.workerId, "workerId"); + if (!Number.isSafeInteger(input.limit) || input.limit < 1 || input.limit > 500) { + throw new SiaxError("INVALID_REQUEST", "0pt.aud0_outbox", "limit must be between 1 and 500"); + } + if (!Number.isSafeInteger(input.leaseMs) || input.leaseMs < 1_000 || input.leaseMs > 15 * 60_000) { + throw new SiaxError("INVALID_REQUEST", "0pt.aud0_outbox", "leaseMs must be between 1000 and 900000"); + } + + const rows = await this.sql` + WITH candidates AS ( + SELECT outbox_id + FROM opt_aud0_evidence_outbox + WHERE status = 'pending' + AND next_attempt_at <= now() + AND (leased_until IS NULL OR leased_until < now()) + ORDER BY next_attempt_at ASC, created_at ASC + LIMIT ${input.limit} + FOR UPDATE SKIP LOCKED + ) + UPDATE opt_aud0_evidence_outbox AS o + SET leased_by = ${workerId}, + leased_until = now() + (${input.leaseMs} * interval '1 millisecond'), + attempts = o.attempts + 1, + updated_at = now() + FROM candidates + WHERE o.outbox_id = candidates.outbox_id + RETURNING o.*`; + return rows.map(mapRow); + } + + async markDelivered(outboxId: string, workerId: string): Promise { + await this.finish(outboxId, workerId, "delivered"); + } + + async markDead(outboxId: string, workerId: string, error: string): Promise { + await this.finish(outboxId, workerId, "dead", error); + } + + private async finish( + outboxId: string, + workerId: string, + status: "delivered" | "dead", + error?: string, + ): Promise { + const rows = await this.sql<{ outbox_id: string }[]>` + UPDATE opt_aud0_evidence_outbox + SET status = ${status}, + delivered_at = CASE WHEN ${status} = 'delivered' THEN now() ELSE NULL END, + dead_at = CASE WHEN ${status} = 'dead' THEN now() ELSE NULL END, + last_error = ${error?.slice(0, 500) ?? null}, + leased_by = NULL, + leased_until = NULL, + updated_at = now() + WHERE outbox_id = ${required(outboxId, "outboxId")} + AND status = 'pending' + AND leased_by = ${required(workerId, "workerId")} + RETURNING outbox_id`; + if (rows.length !== 1) { + throw new SiaxError("CONFLICT", "0pt.aud0_outbox", "outbox lease is no longer owned by this worker"); + } + } + + async markRetry(outboxId: string, workerId: string, error: string, delayMs: number): Promise { + if (!Number.isSafeInteger(delayMs) || delayMs < 0 || delayMs > 24 * 60 * 60_000) { + throw new SiaxError("INVALID_REQUEST", "0pt.aud0_outbox", "delayMs must be between 0 and 86400000"); + } + const rows = await this.sql<{ outbox_id: string }[]>` + UPDATE opt_aud0_evidence_outbox + SET next_attempt_at = now() + (${delayMs} * interval '1 millisecond'), + last_error = ${error.slice(0, 500)}, + leased_by = NULL, + leased_until = NULL, + updated_at = now() + WHERE outbox_id = ${required(outboxId, "outboxId")} + AND status = 'pending' + AND leased_by = ${required(workerId, "workerId")} + RETURNING outbox_id`; + if (rows.length !== 1) { + throw new SiaxError("CONFLICT", "0pt.aud0_outbox", "outbox lease is no longer owned by this worker"); + } + } + + async stats(): Promise { + const rows = await this.sql<{ + pending: string; + delivered: string; + dead: string; + oldest_pending_at: string | Date | null; + }[]>` + SELECT + count(*) FILTER (WHERE status = 'pending')::text AS pending, + count(*) FILTER (WHERE status = 'delivered')::text AS delivered, + count(*) FILTER (WHERE status = 'dead')::text AS dead, + min(created_at) FILTER (WHERE status = 'pending') AS oldest_pending_at + FROM opt_aud0_evidence_outbox`; + const row = rows[0] ?? { pending: "0", delivered: "0", dead: "0", oldest_pending_at: null }; + return { + pending: Number(row.pending), + delivered: Number(row.delivered), + dead: Number(row.dead), + ...(row.oldest_pending_at ? { oldest_pending_at: iso(row.oldest_pending_at) } : {}), + }; + } +} + +export function createAud0OutboxRepositoryFromEnv( + env: NodeJS.ProcessEnv = process.env, +): PostgresAud0OutboxRepository | undefined { + const url = repositoryDatabaseUrl(env); + return url ? new PostgresAud0OutboxRepository(getRepositorySql(url)) : undefined; +} + +export interface Aud0OutboxSweepResult { + claimed: number; + delivered: number; + dead: number; + retried: number; +} + +export class Aud0OutboxDispatcher { + constructor( + private readonly repo: Aud0OutboxRepository, + private readonly aud0: Aud0Client, + private readonly workerId = `0pt-${process.pid}-${crypto.randomUUID()}`, + ) {} + + async sweep(limit = 100): Promise { + const items = await this.repo.claimPending({ + workerId: this.workerId, + limit, + leaseMs: 60_000, + }); + let delivered = 0; + let dead = 0; + let retried = 0; + + for (const item of items) { + try { + if (item.evidence_kind === "event") { + const event = item.payload as Aud0Event; + if (event.tenant_id !== item.tenant_id || event.app_id !== item.app_id) { + throw new SiaxError( + "PERMANENT_FAILURE", + "0pt.aud0_outbox", + "queued AUD0 event tenant/app binding does not match its immutable outbox envelope", + ); + } + await this.aud0.ingestEvent(event); + } else { + await this.aud0.ingestOpenLineageEvent( + item.tenant_id, + item.app_id, + item.payload as OpenLineageRunEvent, + ); + } + await this.repo.markDelivered(item.outbox_id, this.workerId); + delivered += 1; + } catch (error) { + const description = safeError(error); + if (error instanceof SiaxError && !error.retryable) { + await this.repo.markDead(item.outbox_id, this.workerId, description); + dead += 1; + } else { + const exponent = Math.min(Math.max(item.attempts - 1, 0), 10); + const delayMs = Math.min(60 * 60_000, 1_000 * 2 ** exponent); + await this.repo.markRetry(item.outbox_id, this.workerId, description, delayMs); + retried += 1; + } + } + } + + return { claimed: items.length, delivered, dead, retried }; + } +} + +export { openLineageIdempotencyKey, outboxId, safeError };