feat(evidence): implement durable AUD0 outbox and dispatcher
This commit is contained in:
@@ -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<Aud0OutboxRecord>;
|
||||
enqueueOpenLineage(
|
||||
tenantId: string,
|
||||
appId: string,
|
||||
event: OpenLineageRunEvent,
|
||||
): Promise<Aud0OutboxRecord>;
|
||||
claimPending(input: {
|
||||
workerId: string;
|
||||
limit: number;
|
||||
leaseMs: number;
|
||||
}): Promise<Aud0OutboxRecord[]>;
|
||||
markDelivered(outboxId: string, workerId: string): Promise<void>;
|
||||
markRetry(outboxId: string, workerId: string, error: string, delayMs: number): Promise<void>;
|
||||
markDead(outboxId: string, workerId: string, error: string): Promise<void>;
|
||||
stats(): Promise<Aud0OutboxStats>;
|
||||
}
|
||||
|
||||
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<RepositorySql["json"]>[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<Aud0OutboxRecord> {
|
||||
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<Aud0OutboxRecord> {
|
||||
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<Aud0OutboxRecord> {
|
||||
const id = outboxId(input.tenantId, input.kind, input.idempotencyKey);
|
||||
const rows = await this.sql<OutboxRow[]>`
|
||||
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<Aud0OutboxRecord[]> {
|
||||
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<OutboxRow[]>`
|
||||
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<void> {
|
||||
await this.finish(outboxId, workerId, "delivered");
|
||||
}
|
||||
|
||||
async markDead(outboxId: string, workerId: string, error: string): Promise<void> {
|
||||
await this.finish(outboxId, workerId, "dead", error);
|
||||
}
|
||||
|
||||
private async finish(
|
||||
outboxId: string,
|
||||
workerId: string,
|
||||
status: "delivered" | "dead",
|
||||
error?: string,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
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<Aud0OutboxStats> {
|
||||
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<Aud0OutboxSweepResult> {
|
||||
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 };
|
||||
Reference in New Issue
Block a user