diff --git a/tests/p0/aud0-outbox-postgres.realinfra.test.ts b/tests/p0/aud0-outbox-postgres.realinfra.test.ts new file mode 100644 index 0000000..b3f0855 --- /dev/null +++ b/tests/p0/aud0-outbox-postgres.realinfra.test.ts @@ -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; +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); + }); +});