test(evidence): verify AUD0 outbox durability against Postgres

This commit is contained in:
2026-09-07 17:13:57 +02:00
parent 76a3cae776
commit 2a79ae6e60
@@ -0,0 +1,94 @@
import fs from "node:fs/promises";
import path from "node:path";
import postgres from "postgres";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { PostgresAud0OutboxRepository } from "@/src/evidence/aud0-outbox";
const TEST_DATABASE_URL = process.env.TEST_DATABASE_URL;
const describeReal = TEST_DATABASE_URL ? describe : describe.skip;
const schema = `opt_outbox_test_${process.pid}_${Date.now()}`.replace(/[^a-zA-Z0-9_]/g, "_");
let sql: ReturnType<typeof postgres>;
let repo: PostgresAud0OutboxRepository;
const baseEvent = {
event_id: "evt-real-1",
event_type: "0pt.test",
tenant_id: "tenant-real-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-real-1",
};
describeReal("AUD0 outbox Postgres integration", () => {
beforeAll(async () => {
const admin = postgres(TEST_DATABASE_URL!, { max: 1, onnotice: () => undefined });
await admin.unsafe(`CREATE SCHEMA ${schema}`);
await admin.end();
sql = postgres(TEST_DATABASE_URL!, {
max: 3,
prepare: false,
onnotice: () => undefined,
connection: { search_path: schema },
});
const migration = await fs.readFile(path.join(process.cwd(), "sql/011_aud0_evidence_outbox.sql"), "utf8");
await sql.unsafe(migration);
repo = new PostgresAud0OutboxRepository(sql);
});
afterAll(async () => {
if (sql) await sql.end({ timeout: 5 });
if (TEST_DATABASE_URL) {
const admin = postgres(TEST_DATABASE_URL, { max: 1, onnotice: () => undefined });
await admin.unsafe(`DROP SCHEMA IF EXISTS ${schema} CASCADE`);
await admin.end();
}
});
it("deduplicates identical producer replays and rejects same-key payload drift", async () => {
const first = await repo.enqueueEvent(baseEvent);
const replay = await repo.enqueueEvent({ ...baseEvent });
expect(replay.outbox_id).toBe(first.outbox_id);
await expect(
repo.enqueueEvent({ ...baseEvent, resource_id: "different" }),
).rejects.toMatchObject({ code: "CONFLICT" });
});
it("leases rows to only one worker and recovers retry state", async () => {
await repo.enqueueEvent({ ...baseEvent, event_id: "evt-real-2", idempotency_key: "idem-real-2" });
const a = await repo.claimPending({ workerId: "worker-a", limit: 100, leaseMs: 60_000 });
expect(a.length).toBeGreaterThanOrEqual(2);
const b = await repo.claimPending({ workerId: "worker-b", limit: 100, leaseMs: 60_000 });
expect(b).toHaveLength(0);
const target = a.find((r) => r.idempotency_key === "idem-real-2")!;
await repo.markRetry(target.outbox_id, "worker-a", "DEPENDENCY_UNAVAILABLE:aud0", 0);
const reclaimed = await repo.claimPending({ workerId: "worker-b", limit: 100, leaseMs: 60_000 });
expect(reclaimed.some((r) => r.outbox_id === target.outbox_id)).toBe(true);
});
it("keeps delivered/dead rows as visible terminal evidence-delivery state", async () => {
const delivered = await repo.enqueueEvent({ ...baseEvent, event_id: "evt-real-3", idempotency_key: "idem-real-3" });
const dead = await repo.enqueueEvent({ ...baseEvent, event_id: "evt-real-4", idempotency_key: "idem-real-4" });
const claimed = await repo.claimPending({ workerId: "terminal-worker", limit: 100, leaseMs: 60_000 });
if (claimed.some((r) => r.outbox_id === delivered.outbox_id)) {
await repo.markDelivered(delivered.outbox_id, "terminal-worker");
}
if (claimed.some((r) => r.outbox_id === dead.outbox_id)) {
await repo.markDead(dead.outbox_id, "terminal-worker", "INVALID_REQUEST:aud0");
}
const stats = await repo.stats();
expect(stats.delivered).toBeGreaterThanOrEqual(1);
expect(stats.dead).toBeGreaterThanOrEqual(1);
});
});