diff --git a/src/lib/dn-pipeline/checks.test.ts b/src/lib/dn-pipeline/checks.test.ts index 83362f0..dc48378 100644 --- a/src/lib/dn-pipeline/checks.test.ts +++ b/src/lib/dn-pipeline/checks.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from "vitest"; -import { runNoteChecks } from "./checks"; +import { documentIdentity, runNoteChecks } from "./checks"; import type { NoteSegment } from "./types"; function note(over: Partial = {}): NoteSegment { @@ -40,6 +40,34 @@ function byId(results: ReturnType, id: string) { } describe("runNoteChecks", () => { + it("uses supplier plus normalised document number as duplicate identity", () => { + const existing = new Set([documentIdentity("supplier-a", "DN-42")]); + const sameSupplier = byId(runNoteChecks(note({ delivery_note_number: "DN 42" }), { + ...CTX, + supplierId: "supplier-a", + existingDocumentIdentities: existing, + }), "not_duplicate"); + const differentSupplier = byId(runNoteChecks(note({ delivery_note_number: "DN 42" }), { + ...CTX, + supplierId: "supplier-b", + existingDocumentIdentities: existing, + }), "not_duplicate"); + + expect(sameSupplier.passed).toBe(false); + expect(differentSupplier.passed).toBe(true); + }); + + it("keeps an unknown-supplier duplicate in review instead of globally blocking the number", () => { + const r = byId(runNoteChecks(note({ delivery_note_number: "DN-42" }), { + ...CTX, + supplierId: null, + existingDocumentIdentities: new Set([documentIdentity(null, "DN-42")]), + }), "not_duplicate"); + + expect(r.passed).toBe(false); + expect(r.reason).toContain("supplier"); + }); + it("passes a clean Raumix weighbridge ticket", () => { const results = runNoteChecks(note(), CTX); expect(results.every((r) => r.passed !== false)).toBe(true); diff --git a/src/lib/dn-pipeline/checks.ts b/src/lib/dn-pipeline/checks.ts index efe7626..566bebe 100644 --- a/src/lib/dn-pipeline/checks.ts +++ b/src/lib/dn-pipeline/checks.ts @@ -24,6 +24,14 @@ export interface NoteCheckContext { poItems: ReadonlyArray; /** Note numbers already present in delivery_notes. */ existingNoteNumbers: ReadonlySet; + /** + * Supplier-scoped document identities already on the ledger. This is + * additive to existingNoteNumbers so older callers can keep their legacy + * number-only set while new callers avoid cross-supplier collisions. + */ + existingDocumentIdentities?: ReadonlySet; + /** Resolved supplier identity for this candidate, when known. */ + supplierId?: string | null; /** Alias rows, for resolvePoItem step 1. */ aliases?: readonly AliasRow[]; /** Supplier on the resolved PO, for supplier-scoped aliases. */ @@ -37,6 +45,22 @@ export interface NoteCheckContext { deliveredByPoItem?: ReadonlyMap; } +/** + * Stable duplicate identity for a candidate document. + * + * A supplier's document numbers are only unique within that supplier. An + * unknown supplier deliberately gets its own review-only namespace rather + * than sharing a global number namespace with every known supplier. + */ +export function documentIdentity(supplierId: string | null | undefined, number: unknown): string { + return `${supplierId?.trim() || "unknown"}:${normaliseDocumentNumber(number)}`; +} + +export function normaliseDocumentNumber(input: unknown): string { + if (typeof input !== "string") return ""; + return input.replace(/[^a-z0-9]/gi, "").toLowerCase(); +} + /** * How each of a note's lines resolves against the PO. Exported because * productMatched, unitsConvertible and promote.ts must all see the SAME @@ -398,6 +422,25 @@ function notDuplicate(note: NoteSegment, ctx: NoteCheckContext): CheckResult { // Composite references mean the same document can be re-uploaded under // any of its numbers. Check all of them. const all = [primary, ...(note.doc_number_alts ?? []).map((s) => s.trim())].filter(Boolean); + + // New callers provide the scoped identity set. A known supplier only + // collides with the same supplier; an unknown supplier collision is a + // review candidate and never a global number block against known vendors. + const legacySetCarriesIdentities = !ctx.existingDocumentIdentities && + [...ctx.existingNoteNumbers].some((value) => value.includes(":")); + const identities = ctx.existingDocumentIdentities ?? (legacySetCarriesIdentities ? ctx.existingNoteNumbers : null); + if (identities) { + const hit = all.find((number) => + identities.has(documentIdentity(ctx.supplierId, number)), + ); + if (hit) { + return ctx.supplierId + ? { id, passed: false, reason: `${hit} is already in delivery notes for this supplier` } + : { id, passed: false, reason: `${hit} may already be in delivery notes; supplier identity is required for duplicate review` }; + } + return { id, passed: true, reason: primary }; + } + const hit = all.find((n) => ctx.existingNoteNumbers.has(n)); return hit ? { id, passed: false, reason: `${hit} is already in delivery notes` } diff --git a/src/lib/dn-pipeline/promote.test.ts b/src/lib/dn-pipeline/promote.test.ts index b803744..e5d8f9e 100644 --- a/src/lib/dn-pipeline/promote.test.ts +++ b/src/lib/dn-pipeline/promote.test.ts @@ -22,6 +22,8 @@ function makePb() { getOne: dnBatchNotesGetOne, update: dnBatchNotesUpdate, }; + case "dn_review_events": + return { create: dnReviewEventsCreate }; case "delivery_notes": return { create: deliveryNotesCreate, @@ -42,6 +44,7 @@ function makePb() { const dnBatchNotesGetOne = vi.fn(); const dnBatchNotesUpdate = vi.fn(); +const dnReviewEventsCreate = vi.fn(); const deliveryNotesCreate = vi.fn(); const deliveryNotesUpdate = vi.fn(); const deliveryNoteItemsGetFullList = vi.fn(); @@ -51,6 +54,7 @@ const deliveryNoteItemsDelete = vi.fn(); beforeEach(() => { dnBatchNotesGetOne.mockReset(); dnBatchNotesUpdate.mockReset(); + dnReviewEventsCreate.mockReset(); deliveryNotesCreate.mockReset(); deliveryNotesUpdate.mockReset(); deliveryNoteItemsGetFullList.mockReset(); @@ -66,6 +70,7 @@ beforeEach(() => { deliveryNoteItemsDelete.mockResolvedValue(undefined); deliveryNotesUpdate.mockResolvedValue({ id: "dnprior01" }); dnBatchNotesUpdate.mockResolvedValue({ id: "bn1" }); + dnReviewEventsCreate.mockResolvedValue({ id: "event1" }); }); /* ------------------------------------------------------------------ */ @@ -162,6 +167,20 @@ describe("promoteNote", () => { expect(form.slip_image).toBe("blob:application/pdf:5"); }); + it("preserves an image candidate's MIME type and filename", async () => { + await promoteNote(input({ pageFileName: "camera-1.jpg", pageFileMime: "image/jpeg" })); + + const form = await readForm(deliveryNotesCreate.mock.calls[0][0]); + expect(form.slip_image).toBe("blob:image/jpeg:5"); + }); + + it("stamps the permanent candidate key on the delivery note", async () => { + await promoteNote(input()); + + const form = await readForm(deliveryNotesCreate.mock.calls[0][0]); + expect(form.batch_note).toBe("bn1"); + }); + it("writes the expected payload fields onto the delivery_notes row", async () => { await promoteNote(input()); @@ -225,10 +244,11 @@ describe("promoteNote", () => { it("updates the dn_batch_notes row to status='promoted' and links the dn", async () => { await promoteNote(input({ batchNoteId: "bn-xyz" })); - expect(dnBatchNotesUpdate).toHaveBeenCalledWith("bn-xyz", { + expect(dnBatchNotesUpdate).toHaveBeenCalledWith("bn-xyz", expect.objectContaining({ status: "promoted", delivery_note: "dnnew01", - }); + stage: "posted", + })); }); it("uses the shared resolution so the link written down is the one checks.ts validated", async () => { @@ -329,22 +349,49 @@ describe("promoteNote", () => { expect(created.map((c) => c.qty_received)).toEqual([29.96, 5]); }); - it("swallows a stale-item delete failure rather than aborting the promote", async () => { - // Brief behaviour: stale deletes are best-effort. A failed delete - // does not abort the promote — the new items still get written and - // the dn_batch_notes row still gets flipped to promoted. The - // worst case is orphaned stale rows, which is strictly less bad - // than blocking a clean run on a transient error. + it("leaves a candidate review-visible when stale-item reconciliation fails", async () => { dnBatchNotesGetOne.mockResolvedValueOnce({ id: "bn1", delivery_note: "dnprior01" }); deliveryNotesUpdate.mockResolvedValueOnce({ id: "dnprior01" }); deliveryNoteItemsGetFullList.mockResolvedValueOnce([{ id: "stale1" }]); deliveryNoteItemsDelete.mockRejectedValueOnce(new Error("pb delete down")); - const result = await promoteNote(input()); + await expect(promoteNote(input())).rejects.toThrow("pb delete down"); - expect(result.deliveryNoteId).toBe("dnprior01"); expect(deliveryNoteItemsDelete).toHaveBeenCalledTimes(1); - // The new item row was still written after the failed delete. - expect(deliveryNoteItemsCreate).toHaveBeenCalledTimes(1); + expect(deliveryNoteItemsCreate).not.toHaveBeenCalled(); + expect(dnReviewEventsCreate).toHaveBeenCalledWith(expect.objectContaining({ action: "promotion_failed" })); }); -}); \ No newline at end of file + + it("records a review event after a partial line write and reconciles the same candidate on retry", async () => { + let attempt = 0; + dnBatchNotesGetOne.mockImplementation(async () => ({ id: "bn1", delivery_note: attempt ? "dnprior01" : "" })); + deliveryNotesCreate.mockResolvedValue({ id: "dnprior01" }); + deliveryNoteItemsGetFullList.mockResolvedValue([{ id: "partial1" }]); + deliveryNoteItemsCreate.mockImplementation(async () => { + if (attempt++ === 0) throw new Error("line write failed"); + return { id: "item-reconciled" }; + }); + + await expect(promoteNote(input())).rejects.toThrow("line write failed"); + expect(dnReviewEventsCreate).toHaveBeenCalledWith(expect.objectContaining({ + batch: "batch1", + note: "bn1", + action: "promotion_failed", + })); + + const result = await promoteNote(input()); + expect(result.deliveryNoteId).toBe("dnprior01"); + expect(deliveryNotesCreate).toHaveBeenCalledTimes(1); + expect(deliveryNoteItemsDelete).toHaveBeenCalledWith("partial1"); + expect(deliveryNoteItemsCreate).toHaveBeenCalledTimes(2); + }); + + it("does not mutate a candidate that is already posted", async () => { + dnBatchNotesGetOne.mockResolvedValueOnce({ id: "bn1", status: "promoted", stage: "posted", posted_at: "2026-09-08T00:00:00Z", delivery_note: "dnposted" }); + + await expect(promoteNote(input())).rejects.toThrow("posted candidate is immutable"); + expect(deliveryNotesCreate).not.toHaveBeenCalled(); + expect(deliveryNotesUpdate).not.toHaveBeenCalled(); + expect(deliveryNoteItemsCreate).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/dn-pipeline/promote.ts b/src/lib/dn-pipeline/promote.ts index bee9d43..64721ed 100644 --- a/src/lib/dn-pipeline/promote.ts +++ b/src/lib/dn-pipeline/promote.ts @@ -26,6 +26,12 @@ export interface PromoteInput { /** Per-note PDF assembled from this note's own pages. */ pageFile: Buffer; pageFileName: string; + /** The candidate's actual file type; never infer PDF when supplied. */ + pageFileMime?: string; + /** Resolved supplier relation used by duplicate identity checks upstream. */ + supplierId?: string | null; + /** Explicit correction is the only way to touch a posted candidate. */ + explicitCorrection?: boolean; /** Line resolutions from resolveNoteLines — the same answers the checks saw. */ lineMatches: readonly { description: string; match: PoItemMatch }[]; } @@ -35,16 +41,34 @@ export interface PromoteResult { itemsCreated: number; } +type CandidateState = { + id: string; + delivery_note?: string; + status?: string; + stage?: string; + posted_at?: string; +}; + +type PbCollection = ReturnType & { + getFullList?: (options?: Record) => Promise>>; +}; + export async function promoteNote(input: PromoteInput): Promise { const { pb, note, batchNoteId, batchId, siteId, purchaseOrderId, poItems } = input; - // Idempotency: a re-run must update the row it already made, never - // mint a second one. - const existing = await pb + // The candidate row is the durable promotion key. A posted candidate is + // immutable: retries may reconcile only an unposted partial write. + const candidate = (await pb .collection("dn_batch_notes") - .getOne(batchNoteId, { fields: "id,delivery_note" }) - .catch(() => null); - const existingDnId: string | null = existing?.delivery_note || null; + .getOne(batchNoteId, { fields: "id,delivery_note,status,stage,posted_at" }) + .catch(() => null)) as CandidateState | null; + if (candidate && isPosted(candidate) && !input.explicitCorrection) { + throw new Error("posted candidate is immutable without explicit correction"); + } + + // A candidate relation is preferred. The unique delivery_notes.batch_note + // index is the recovery path when a worker died between the two writes. + const existingDnId = candidate?.delivery_note || await findDeliveryNoteByCandidate(pb, batchNoteId); const payload = { delivery_note_number: note.delivery_note_number, @@ -54,6 +78,7 @@ export async function promoteNote(input: PromoteInput): Promise { received_by: "dn-pipeline", received_by_name: note.received_by_name ?? "", batch: batchId, + batch_note: batchNoteId, document_type: note.doc_type, page_from: note.page_from, page_to: note.page_to, @@ -79,55 +104,136 @@ export async function promoteNote(input: PromoteInput): Promise { } form.append( "slip_image", - new Blob([new Uint8Array(input.pageFile)], { type: "application/pdf" }), + new Blob([new Uint8Array(input.pageFile)], { type: input.pageFileMime ?? mimeFromFilename(input.pageFileName) }), input.pageFileName, ); - const dn = existingDnId - ? await pb.collection("delivery_notes").update(existingDnId, form) - : await pb.collection("delivery_notes").create(form); + let dn: { id: string } | undefined; + try { + dn = (existingDnId + ? await pb.collection("delivery_notes").update(existingDnId, form) + : await pb.collection("delivery_notes").create(form)) as { id: string }; - // Replace the item set wholesale so a re-run cannot leave stale lines. - const stale = await pb - .collection("delivery_note_items") - .getFullList({ filter: `delivery_note='${dn.id}'`, fields: "id" }); - for (const s of stale) { - await pb.collection("delivery_note_items").delete(s.id).catch(() => undefined); - } - - const { rows } = buildDeliveryNoteItemRows({ - items: note.items.map((it, i) => { - const m = input.lineMatches[i]?.match; - return { - description: it.description, - qty: it.qty, - unit: it.unit, - unit_price: it.unit_price, - matched_po_item_id: m && m.kind === "matched" ? m.poItem.id : null, - }; - }), - poItems, - confidence: 1, - }); - - let created = 0; - for (const row of rows) { - await pb.collection("delivery_note_items").create({ - delivery_note: dn.id, - purchase_order_item: row.purchase_order_item, - qty_received: row.qty_received, - qty_raw: row.qty_raw, - unit_raw: row.unit_raw, - unit_converted: row.unit_converted, - condition_notes: row.condition_notes, + // Replace the item set wholesale so a retry reconciles only this + // candidate's unposted partial rows and cannot duplicate them. + const stale = await pb.collection("delivery_note_items").getFullList({ + filter: boundFilter(pb, "delivery_note = {:deliveryNote}", { deliveryNote: dn.id }), + fields: "id", }); - created++; + for (const s of stale) await pb.collection("delivery_note_items").delete(s.id); + + const { rows } = buildDeliveryNoteItemRows({ + items: note.items.map((it, i) => { + const m = input.lineMatches[i]?.match; + return { + description: it.description, + qty: it.qty, + unit: it.unit, + unit_price: it.unit_price, + matched_po_item_id: m && m.kind === "matched" ? m.poItem.id : null, + }; + }), + poItems, + confidence: 1, + }); + + let created = 0; + for (const row of rows) { + await pb.collection("delivery_note_items").create({ + delivery_note: dn.id, + purchase_order_item: row.purchase_order_item, + qty_received: row.qty_received, + qty_raw: row.qty_raw, + unit_raw: row.unit_raw, + unit_converted: row.unit_converted, + condition_notes: row.condition_notes, + }); + created++; + } + + // The financial rows are complete before the candidate becomes posted. + await pb.collection("dn_batch_notes").update(batchNoteId, { + status: "promoted", + stage: "posted", + posted_at: new Date().toISOString(), + delivery_note: dn.id, + }); + + return { deliveryNoteId: dn.id, itemsCreated: created }; + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + const recoveryDnId = dn?.id ?? existingDnId ?? null; + await recordPromotionFailure(pb, { + batch: batchId, + note: batchNoteId, + deliveryNote: recoveryDnId, + reason, + }); + // Keep the candidate review-visible while retaining its delivery_note + // relation (if one exists) so the next run can reconcile it. + await pb.collection("dn_batch_notes").update(batchNoteId, { + status: "blocked", + stage: "needs_review", + ...(recoveryDnId ? { delivery_note: recoveryDnId } : {}), + }).catch(() => undefined); + throw error; } +} - await pb.collection("dn_batch_notes").update(batchNoteId, { - status: "promoted", - delivery_note: dn.id, +function isPosted(candidate: CandidateState): boolean { + return candidate.status === "promoted" || candidate.status === "posted" || candidate.stage === "posted" || Boolean(candidate.posted_at); +} + +async function findDeliveryNoteByCandidate(pb: PocketBase, batchNoteId: string): Promise { + const collection = pb.collection("delivery_notes") as unknown as PbCollection; + if (typeof collection.getFullList !== "function") return null; + const rows = await collection.getFullList({ + filter: boundFilter(pb, "batch_note = {:batchNote}", { batchNote: batchNoteId }), + fields: "id,batch_note", }); + const id = (rows[0] as { id?: unknown } | undefined)?.id; + return typeof id === "string" ? id : null; +} - return { deliveryNoteId: dn.id, itemsCreated: created }; -} \ No newline at end of file +async function recordPromotionFailure( + pb: PocketBase, + fields: { batch: string; note: string; deliveryNote: string | null; reason: string }, +): Promise { + try { + await pb.collection("dn_review_events").create({ + batch: fields.batch, + note: fields.note, + action: "promotion_failed", + before: { delivery_note: fields.deliveryNote, status: "candidate" }, + after: { delivery_note: fields.deliveryNote, status: "blocked", stage: "needs_review" }, + reason: fields.reason.slice(0, 500), + actor: "dn-pipeline", + }); + } catch { + // The failed promotion itself must remain visible even on an older PB + // instance without the additive review-event collection. + } +} + +function boundFilter(pb: PocketBase, query: string, params: Record): string { + const client = pb as unknown as { filter?: (raw: string, values: Record) => string }; + if (typeof client.filter === "function") return client.filter(query, params); + return query.replace(/\{:(\w+)\}/g, (_match, key: string) => JSON.stringify(params[key])); +} + +function mimeFromFilename(filename: string): string { + const extension = filename.toLowerCase().match(/\.([a-z0-9]+)$/)?.[1]; + switch (extension) { + case "jpg": + case "jpeg": + return "image/jpeg"; + case "png": + return "image/png"; + case "webp": + return "image/webp"; + case "heic": + return "image/heic"; + default: + return "application/pdf"; + } +} diff --git a/src/lib/dn-pipeline/run.ts b/src/lib/dn-pipeline/run.ts index 4ccde6f..510dcb6 100644 --- a/src/lib/dn-pipeline/run.ts +++ b/src/lib/dn-pipeline/run.ts @@ -8,7 +8,7 @@ */ import type PocketBase from "pocketbase"; import { normalisePoNumber } from "../po-matching"; -import { resolveNoteLines, runNoteChecks } from "./checks"; +import { documentIdentity, resolveNoteLines, runNoteChecks } from "./checks"; import type { PoItemRef } from "./match-po-item"; import type { AliasRow } from "./aliases"; import { @@ -101,7 +101,7 @@ export async function runBatch(pb: PocketBase, batchId: string): Promise c.passed === false); const pageFile = await extractPageRange(buffer, mime, note.page_from, note.page_to); - const pageFileName = `${filename.replace(/\.[^.]+$/, "")}-note-${note.note_index + 1}.pdf`; + const pageFileName = candidateFileName(filename, mime, note.note_index); const candidate = await upsertCandidate(pb, batchId, note, { poId: po?.id ?? null, @@ -138,6 +140,7 @@ export async function runBatch(pb: PocketBase, batchId: string): Promise ({ ...c, id: `note${note.note_index + 1}.${c.id}` }))); @@ -165,6 +168,7 @@ export async function runBatch(pb: PocketBase, batchId: string): Promise & { documentIdentities: Set }; + async function loadExistingNoteNumbers( pb: PocketBase, batchId: string, -): Promise> { +): Promise { const rows = await pb .collection("delivery_notes") - .getFullList({ fields: "delivery_note_number,batch" }); - return new Set( + .getFullList({ fields: "delivery_note_number,batch,purchase_order", expand: "purchase_order" }); + const numbers = new Set( rows .filter((r) => (r as { batch?: string }).batch !== batchId) .map((r) => (r as { delivery_note_number?: string }).delivery_note_number?.trim()) .filter((v): v is string => Boolean(v)), + ) as ExistingNoteSet; + numbers.documentIdentities = new Set( + rows + .filter((r) => (r as { batch?: string }).batch !== batchId) + .map((r) => { + const row = r as { + delivery_note_number?: string; + supplier?: string; + expand?: { purchase_order?: { supplier_ref?: string; supplier?: string } }; + }; + const number = row.delivery_note_number?.trim(); + const supplier = row.supplier || row.expand?.purchase_order?.supplier_ref || row.expand?.purchase_order?.supplier || null; + return number ? documentIdentity(supplier, number) : ""; + }) + .filter(Boolean), ); + return numbers; } /** @@ -368,6 +390,16 @@ function safeSiteFilter(pb: PocketBase, siteId: string): string { return `site = ${JSON.stringify(siteId)}`; } +function candidateFileName(filename: string, mime: string, noteIndex: number): string { + const clean = filename.split("/").pop() ?? filename; + const base = clean.replace(/\.[^.]+$/, ""); + const originalExt = clean.match(/\.([a-z0-9]+)$/i)?.[1]; + const extension = mime === "application/pdf" + ? "pdf" + : (originalExt || mime.split("/")[1] || "bin"); + return `${base}-note-${noteIndex + 1}.${extension}`; +} + async function upsertCandidate( pb: PocketBase, batchId: string, @@ -383,12 +415,13 @@ async function upsertCandidate( status: "candidate" | "blocked"; pageFile: Buffer; pageFileName: string; + pageFileMime: string; }, ) { const existing = await pb .collection("dn_batch_notes") .getFullList({ - filter: `batch='${batchId}' && note_index=${note.note_index}`, + filter: boundFilter(pb, "batch = {:batch} && note_index = {:noteIndex}", { batch: batchId, noteIndex: note.note_index }), fields: "id,status", }); @@ -431,7 +464,7 @@ async function upsertCandidate( } form.append( "page_file", - new Blob([new Uint8Array(extra.pageFile)], { type: "application/pdf" }), + new Blob([new Uint8Array(extra.pageFile)], { type: extra.pageFileMime }), extra.pageFileName, ); @@ -444,3 +477,9 @@ async function upsertCandidate( return { id: record.id as string, dismissed: wasDismissed }; } + +function boundFilter(pb: PocketBase, query: string, params: Record): string { + const client = pb as unknown as { filter?: (raw: string, values: Record) => string }; + if (typeof client.filter === "function") return client.filter(query, params); + return query.replace(/\{:(\w+)\}/g, (_match, key: string) => JSON.stringify(params[key])); +}