diff --git a/tests/p0/aud0-outbox.test.ts b/tests/p0/aud0-outbox.test.ts new file mode 100644 index 0000000..c584194 --- /dev/null +++ b/tests/p0/aud0-outbox.test.ts @@ -0,0 +1,155 @@ +import { describe, expect, it } from "vitest"; + +import type { Aud0Client, Aud0Event } from "@/src/clients/aud0-client"; +import { + Aud0OutboxDispatcher, + openLineageIdempotencyKey, + outboxId, + type Aud0OutboxRecord, + type Aud0OutboxRepository, + type Aud0OutboxStats, +} from "@/src/evidence/aud0-outbox"; +import type { OpenLineageRunEvent } from "@/src/lineage/openlineage"; +import { SiaxError } from "@/src/platform/errors"; + +const event: Aud0Event = { + event_id: "evt-1", + event_type: "0pt.test", + tenant_id: "tenant-a", + project_id: "project-a", + actor_id: "tester", + app_id: "0pt", + capability_id: "0pt.test.v1", + resource_type: "test", + resource_id: "r1", + risk_level: "L1", + created_at: "2026-09-07T12:00:00.000Z", + idempotency_key: "idem-1", +}; + +const ol: OpenLineageRunEvent = { + eventType: "COMPLETE", + eventTime: "2026-09-07T12:00:00.000Z", + producer: "https://git.siax.io/sax3l/0pt", + run: { runId: "run-1" }, + job: { namespace: "0pt", name: "0pt.test.v1" }, + inputs: [], + outputs: [], +}; + +class FakeRepo implements Aud0OutboxRepository { + claimed: Aud0OutboxRecord[] = []; + delivered: string[] = []; + dead: Array<{ id: string; error: string }> = []; + retried: Array<{ id: string; error: string; delayMs: number }> = []; + + async enqueueEvent(): Promise { throw new Error("unused"); } + async enqueueOpenLineage(): Promise { throw new Error("unused"); } + async claimPending(): Promise { return this.claimed; } + async markDelivered(id: string): Promise { this.delivered.push(id); } + async markRetry(id: string, _worker: string, error: string, delayMs: number): Promise { + this.retried.push({ id, error, delayMs }); + } + async markDead(id: string, _worker: string, error: string): Promise { + this.dead.push({ id, error }); + } + async stats(): Promise { return { pending: this.claimed.length, delivered: 0, dead: 0 }; } +} + +function row( + kind: "event" | "openlineage", + payload: Aud0Event | OpenLineageRunEvent, + overrides: Partial = {}, +): Aud0OutboxRecord { + return { + outbox_id: "obx-1", + tenant_id: "tenant-a", + app_id: "0pt", + evidence_kind: kind, + idempotency_key: "idem-1", + payload, + status: "pending", + attempts: 1, + next_attempt_at: "2026-09-07T12:00:00.000Z", + created_at: "2026-09-07T12:00:00.000Z", + updated_at: "2026-09-07T12:00:00.000Z", + ...overrides, + }; +} + +function client(impl?: Partial): Aud0Client { + return { + async ingestEvent(input) { return input; }, + async ingestOpenLineageEvent(tenantId, _appId, input) { + return { tenant_id: tenantId, run_id: input.run.runId, job: input.job.name, event_id: "aud0-1" }; + }, + async ingestLineageEvent(input) { return input; }, + ...impl, + }; +} + +describe("AUD0 durable outbox", () => { + it("derives stable tenant-bound outbox ids", () => { + expect(outboxId("tenant-a", "event", "k1")).toBe(outboxId("tenant-a", "event", "k1")); + expect(outboxId("tenant-a", "event", "k1")).not.toBe(outboxId("tenant-b", "event", "k1")); + expect(outboxId("tenant-a", "event", "k1")).not.toBe(outboxId("tenant-a", "openlineage", "k1")); + }); + + it("uses logical OpenLineage run+eventType identity rather than delivery time", () => { + expect(openLineageIdempotencyKey(ol)).toBe("ol:run-1:COMPLETE"); + expect(openLineageIdempotencyKey({ ...ol, eventTime: "2030-01-01T00:00:00Z" })).toBe("ol:run-1:COMPLETE"); + }); + + it("marks a successfully delivered generic event delivered", async () => { + const repo = new FakeRepo(); + repo.claimed = [row("event", event)]; + let calls = 0; + const dispatcher = new Aud0OutboxDispatcher(repo, client({ + async ingestEvent(input) { calls += 1; return input; }, + }), "worker-a"); + + expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 1, dead: 0, retried: 0 }); + expect(calls).toBe(1); + expect(repo.delivered).toEqual(["obx-1"]); + }); + + it("retries network/dependency failures with bounded exponential backoff", async () => { + const repo = new FakeRepo(); + repo.claimed = [row("openlineage", ol, { attempts: 4 })]; + const dispatcher = new Aud0OutboxDispatcher(repo, client({ + async ingestOpenLineageEvent() { + throw new SiaxError("DEPENDENCY_UNAVAILABLE", "aud0", "offline"); + }, + }), "worker-a"); + + expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 0, dead: 0, retried: 1 }); + expect(repo.retried[0].delayMs).toBe(8_000); + expect(repo.retried[0].error).toBe("DEPENDENCY_UNAVAILABLE:aud0"); + }); + + it("parks permanent 4xx-style failures dead instead of poisoning the queue", async () => { + const repo = new FakeRepo(); + repo.claimed = [row("event", event)]; + const dispatcher = new Aud0OutboxDispatcher(repo, client({ + async ingestEvent() { + throw new SiaxError("INVALID_REQUEST", "aud0", "bad event"); + }, + }), "worker-a"); + + expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 0, dead: 1, retried: 0 }); + expect(repo.dead).toEqual([{ id: "obx-1", error: "INVALID_REQUEST:aud0" }]); + }); + + it("fails closed on an immutable tenant/app envelope mismatch before calling AUD0", async () => { + const repo = new FakeRepo(); + repo.claimed = [row("event", { ...event, tenant_id: "tenant-b" })]; + let calls = 0; + const dispatcher = new Aud0OutboxDispatcher(repo, client({ + async ingestEvent(input) { calls += 1; return input; }, + }), "worker-a"); + + expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 0, dead: 1, retried: 0 }); + expect(calls).toBe(0); + expect(repo.dead[0].error).toBe("PERMANENT_FAILURE:0pt.aud0_outbox"); + }); +});