fix: verify audit collisions and purchase orders

scripts/compare-dn-pipelines.ts

@@ -1,5 +1,5 @@
 import { pbAdmin } from "../src/lib/pb";
-import { classifyUnit } from "../src/lib/dn-pipeline/compare";
+import { classifyUnit, poMatchCounts, verifyPurchaseOrderIds } from "../src/lib/dn-pipeline/compare";
 
 type Row = Record<string, any>;
 
@@ -21,7 +21,11 @@ for (const batch of batches) {
     pb.collection("dn_batch_notes").getFullList({ filter: `batch = ${JSON.stringify(batch.id)}`, sort: "note_index", fields: "id,stage,status,pages,purchase_order,items,checks,doc_number,delivery_note_number" }),
   ]);
   const poIds = [...new Set((notes as Row[]).map((note) => typeof note.purchase_order === "string" ? note.purchase_order : "").filter(Boolean))];
-  const poItems = (await Promise.all(poIds.map((po) => pb.collection("purchase_order_items").getFullList({ filter: `purchase_order = ${JSON.stringify(po)}`, fields: "id,purchase_order,unit,qty,rate,description" })))).flat() as Row[];
+  const verifiedPoIds = await verifyPurchaseOrderIds(poIds, async (po) => {
+    try { const row = await pb.collection("purchase_orders").getOne(po, { fields: "id" }) as Row; return { id: String(row.id) }; }
+    catch (error) { throw new Error(`purchase order verification failed for ${po}: ${error instanceof Error ? error.message : String(error)}`, { cause: error }); }
+  });
+  const poItems = (await Promise.all([...verifiedPoIds].map((po) => pb.collection("purchase_order_items").getFullList({ filter: `purchase_order = ${JSON.stringify(po)}`, fields: "id,purchase_order,unit,qty,rate,description" })))).flat() as Row[];
   const itemById = new Map(poItems.map((item) => [item.id, item]));
   let matchedItems = 0;
   let itemCount = 0;
@@ -56,7 +60,7 @@ for (const batch of batches) {
     pageCoverage: { total: pages.length, assigned: coveredPages.size, open: Math.max(0, pages.length - coveredPages.size), read: (pages as Row[]).filter((page) => page.read_status === "read").length },
     candidateCount: notes.length,
     stages: Object.fromEntries([...new Set((notes as Row[]).map((note) => String(note.stage ?? note.status ?? "unknown")))].map((stage) => [stage, (notes as Row[]).filter((note) => String(note.stage ?? note.status ?? "unknown") === stage).length])),
-    poMatch: { candidatesWithPo: (notes as Row[]).filter((note) => typeof note.purchase_order === "string" && note.purchase_order).length, total: notes.length },
+    poMatch: poMatchCounts(notes as Row[], verifiedPoIds),
     itemMatch: { matched: matchedItems, total: itemCount },
     units: { compatible: unitCompatible, incompatible: unitIncompatible, unknown: unitUnknown, matchedItems },
     quantity,

src/app/api/dn-batches/[id]/notes/[noteId]/post/route.test.ts

@@ -16,6 +16,7 @@ const state = vi.hoisted(() => ({
   poItems: [] as Record<string, unknown>[],
   pages: [] as Record<string, unknown>[],
   auditFails: false,
+  auditCollision: false,
 }));
 
 vi.mock("@/lib/pb", () => ({ pbAdmin: vi.fn(async () => ({
@@ -32,7 +33,10 @@ vi.mock("@/lib/pb", () => ({ pbAdmin: vi.fn(async () => ({
     if (name === "delivery_note_items") return { getFullList: vi.fn(async () => [{ id: "item-row" }]) };
     if (name === "dn_review_events") return {
       getFullList: vi.fn(async () => state.events),
-      create: vi.fn(async (event: Record<string, unknown>) => { if (state.auditFails) throw new Error("audit unavailable"); state.events.push(event); return { id: "event" }; }),
+      create: vi.fn(async (event: Record<string, unknown>) => {
+        if (state.auditCollision) { state.events.push({ id: "winner", event_key: event.event_key, batch: BATCH, note: NOTE, action: "post" }); throw new Error("unique event key"); }
+        if (state.auditFails) throw new Error("audit unavailable"); state.events.push(event); return { id: "event" };
+      }),
     };
     throw new Error(`unexpected collection ${name}`);
   },
@@ -55,6 +59,7 @@ beforeEach(() => {
   state.delivery = null;
   state.events = [];
   state.auditFails = false;
+  state.auditCollision = false;
   state.poItems = [{ id: ITEM, purchase_order: PO, description: "Aggregate", unit: "ton", qty: 10, rate: 5 }];
   state.pages = [{ id: "page00000000001", image: "page.png", mime: "image/png", source_page: 1, region: 0, expand: { source: { ordinal: 0 } } }];
   promoteNote.mockClear();
@@ -116,6 +121,22 @@ describe("post delivery note route", () => {
     expect(body).not.toHaveProperty("error", "promotion_failed");
   });
 
+  it("confirms a concurrent deterministic audit winner after create collision", async () => {
+    state.auditCollision = true;
+    const { POST } = await import("./route");
+    const response = await POST(request(), context());
+    expect(response.status).toBe(200);
+    expect(await response.json()).toEqual({ deliveryNoteId: DELIVERY, itemsCreated: 1, stage: "posted" });
+  });
+
+  it("does not accept a deterministic audit key owned by another note", async () => {
+    state.events = [{ id: "wrong", event_key: `dn-post:${NOTE}`, batch: BATCH, note: "note00000000002", action: "post" }];
+    const { POST } = await import("./route");
+    const response = await POST(request(), context());
+    expect(response.status).toBe(200);
+    expect(await response.json()).toMatchObject({ stage: "posted", auditPending: true, auditError: "audit_key_collision" });
+  });
+
   it("fails closed when a posted delivery note no longer matches its PO/site association", async () => {
     state.note = { ...state.note, status: "promoted", stage: "posted", delivery_note: DELIVERY };
     state.delivery = { id: DELIVERY, batch_note: NOTE, batch: BATCH, site: "site00000000002", purchase_order: PO };

src/app/api/dn-batches/[id]/notes/[noteId]/post/route.ts

@@ -125,8 +125,8 @@ export async function POST(_request: Request, { params }: { params: Promise<{ id
       return NextResponse.json({ error: "posted_delivery_note_relation_mismatch" }, { status: 409 });
     }
     const items = await pb.collection("delivery_note_items").getFullList({ filter: filter(pb, "delivery_note = {:deliveryNote}", { deliveryNote: deliveryNoteId }), fields: "id" });
-    const auditPending = !(await ensurePostedAudit(pb, id, noteId, deliveryNoteId));
-    return NextResponse.json({ deliveryNoteId, itemsCreated: items.length, stage: "posted", ...(auditPending ? { auditPending: true } : {}) });
+    const audit = await ensurePostedAudit(pb, id, noteId, deliveryNoteId);
+    return NextResponse.json({ deliveryNoteId, itemsCreated: items.length, stage: "posted", ...(audit.ok ? {} : { auditPending: true, auditError: audit.error }) });
   }
   if (note.status === "dismissed" || note.stage === "dismissed") return NextResponse.json({ error: "dismissed_candidate" }, { status: 409 });
 
@@ -184,23 +184,34 @@ export async function POST(_request: Request, { params }: { params: Promise<{ id
       })),
       pageFile, pageFileName: `${normalizedNote.delivery_note_number || noteId}.pdf`, pageFileMime: "application/pdf", lineMatches,
     });
-    const auditPending = !(await ensurePostedAudit(pb, id, noteId, result.deliveryNoteId));
-    return NextResponse.json({ ...result, stage: "posted", ...(auditPending ? { auditPending: true } : {}) });
+    const audit = await ensurePostedAudit(pb, id, noteId, result.deliveryNoteId);
+    return NextResponse.json({ ...result, stage: "posted", ...(audit.ok ? {} : { auditPending: true, auditError: audit.error }) });
   } catch (error) {
     return NextResponse.json({ error: "promotion_failed", detail: error instanceof Error ? error.message : String(error) }, { status: 500 });
   }
 }
 
-async function ensurePostedAudit(pb: Awaited<ReturnType<typeof pbAdmin>>, batchId: string, noteId: string, deliveryNoteId: string): Promise<boolean> {
+type AuditResult = { ok: true } | { ok: false; error: "audit_unavailable" | "audit_key_collision" };
+
+async function ensurePostedAudit(pb: Awaited<ReturnType<typeof pbAdmin>>, batchId: string, noteId: string, deliveryNoteId: string): Promise<AuditResult> {
   const eventKey = `dn-post:${noteId}`;
+  const identity = (event: Record<string, unknown>) => relationId(event.batch) === batchId && relationId(event.note) === noteId && event.action === "post";
   try {
-    const events = await pb.collection("dn_review_events").getFullList({ filter: filter(pb, "event_key = {:eventKey}", { eventKey }), fields: "id" });
-    if (events.length > 0) return true;
-    await pb.collection("dn_review_events").create({ event_key: eventKey, batch: batchId, note: noteId, action: "post", before: { stage: "ready_to_post" }, after: { stage: "posted", delivery_note: deliveryNoteId }, reason: "candidate posted", actor: "web" });
-    return true;
+    const events = await pb.collection("dn_review_events").getFullList({ filter: filter(pb, "event_key = {:eventKey}", { eventKey }), fields: "id,event_key,batch,note,action" }) as Record<string, unknown>[];
+    if (events.length > 0) return identity(events[0]) ? { ok: true } : { ok: false, error: "audit_key_collision" };
+    try {
+      await pb.collection("dn_review_events").create({ event_key: eventKey, batch: batchId, note: noteId, action: "post", before: { stage: "ready_to_post" }, after: { stage: "posted", delivery_note: deliveryNoteId }, reason: "candidate posted", actor: "web" });
+      return { ok: true };
+    } catch {
+      // A concurrent writer may have won the deterministic key. Re-read and
+      // only confirm success when the winner belongs to this batch/note.
+      const winner = await pb.collection("dn_review_events").getFullList({ filter: filter(pb, "event_key = {:eventKey}", { eventKey }), fields: "id,event_key,batch,note,action" }) as Record<string, unknown>[];
+      if (winner.length > 0) return identity(winner[0]) ? { ok: true } : { ok: false, error: "audit_key_collision" };
+      return { ok: false, error: "audit_unavailable" };
+    }
   } catch {
     // Financial promotion is already durable. Never report promotion_failed;
     // the retry path will converge once the audit collection is available.
-    return false;
+    return { ok: false, error: "audit_unavailable" };
   }
 }

src/lib/dn-pipeline/compare.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { classifyUnit } from "./compare";
+import { classifyUnit, poMatchCounts, verifyPurchaseOrderIds } from "./compare";
 
 describe("delivery-note comparison units", () => {
   it("distinguishes compatible, incompatible, and unknown units", () => {
@@ -9,3 +9,17 @@ describe("delivery-note comparison units", () => {
     expect(classifyUnit("kg", null)).toBe("unknown");
   });
 });
+
+describe("delivery-note comparison PO verification", () => {
+  it("separates linked references from verified existing POs", () => {
+    expect(poMatchCounts([{ purchase_order: "po-1" }, { purchase_order: "missing" }, {}], new Set(["po-1"]))).toEqual({ linked: 2, verified: 1, total: 3 });
+  });
+
+  it("keeps genuine missing POs unresolved but fails on read errors", async () => {
+    await expect(verifyPurchaseOrderIds(["present", "missing"], async (id) => {
+      if (id === "missing") throw Object.assign(new Error("not found"), { status: 404 });
+      return { id };
+    })).resolves.toEqual(new Set(["present"]));
+    await expect(verifyPurchaseOrderIds(["broken"], async () => { throw new Error("PB unavailable"); })).rejects.toThrow("PB unavailable");
+  });
+});

src/lib/dn-pipeline/compare.ts

@@ -4,3 +4,24 @@ export function classifyUnit(lineUnit: unknown, poUnit: unknown): UnitCompatibil
   if (typeof lineUnit !== "string" || !lineUnit.trim() || typeof poUnit !== "string" || !poUnit.trim()) return "unknown";
   return lineUnit.trim().toLowerCase() === poUnit.trim().toLowerCase() ? "compatible" : "incompatible";
 }
+
+export function poMatchCounts(notes: readonly Record<string, unknown>[], existingPoIds: ReadonlySet<string>) {
+  const linked = notes.filter((note) => typeof note.purchase_order === "string" && note.purchase_order.trim()).length;
+  const verified = notes.filter((note) => typeof note.purchase_order === "string" && existingPoIds.has(note.purchase_order)).length;
+  return { linked, verified, total: notes.length };
+}
+
+export async function verifyPurchaseOrderIds(
+  ids: readonly string[],
+  read: (id: string) => Promise<{ id: string }>,
+): Promise<Set<string>> {
+  const rows = await Promise.all(ids.map(async (id) => {
+    try { return await read(id); }
+    catch (error) {
+      const status = (error as { status?: unknown }).status;
+      if (status === 404 || /\b404\b|not found/i.test(error instanceof Error ? error.message : String(error))) return null;
+      throw error;
+    }
+  }));
+  return new Set(rows.filter((row): row is { id: string } => Boolean(row?.id)).map((row) => row.id));
+}