diff --git a/scripts/pb-add-dn-staged-ingestion.py b/scripts/pb-add-dn-staged-ingestion.py index be14276..406cfc7 100644 --- a/scripts/pb-add-dn-staged-ingestion.py +++ b/scripts/pb-add-dn-staged-ingestion.py @@ -164,6 +164,17 @@ def self_test(): assert merged == [old[0], replacement] assert changes == [("replace", replacement)] assert merge_indexes(merged, [replacement])[1] == [] + candidate_key = field("candidate_key") + assert candidate_key == {"name": "candidate_key", "type": "text", "required": False} + candidate_index = ( + "CREATE UNIQUE INDEX `idx_dn_batch_notes_candidate_key` " + "ON `dn_batch_notes` (`batch`, `candidate_key`) " + "WHERE `candidate_key` IS NOT NULL AND `candidate_key` != ''" + ) + candidate_indexes, candidate_changes = merge_indexes([], [candidate_index]) + assert candidate_indexes == [candidate_index] + assert candidate_changes == [("add", candidate_index)] + assert merge_indexes(candidate_indexes, [candidate_index])[1] == [] print("self-test: index replacement and idempotence passed") return 0 @@ -187,10 +198,10 @@ def main(): print(f"offline preview: {error}") print("would create: dn_batch_sources, dn_batch_pages, " "supplier_document_aliases, dn_review_events, dn_pipeline_locks") - print("would extend: dn_batches, dn_batch_notes, delivery_notes") + print("would extend: dn_batches, dn_batch_notes(candidate_key), delivery_notes") print("would add unique indexes: source(batch,file_hash), " "page(source,source_page,region), alias(normalised_key), " - "lock(batch), delivery_notes(batch_note)") + "lock(batch), candidate(batch,candidate_key), delivery_notes(batch_note)") print("would add partial unique capture-session index on dn_batches") print("\nDRY RUN. No schema changes written.") return 0 @@ -281,9 +292,14 @@ def main(): values=["matching", "needs_review", "ready_to_post", "posted", "dismissed"]), field("assembly_revision", "number"), field("pages", "json", maxSize=2000000), relation("supplier", supplier_id), field("supplier_name_raw"), - field("match_method"), field("match_evidence", "json", maxSize=2000000), + field("match_method"), field("match_evidence", "json", maxSize=2000000), field("candidate_key"), field("posted_at", "date"), ]) + add_indexes(token, "dn_batch_notes", [ + "CREATE UNIQUE INDEX `idx_dn_batch_notes_candidate_key` " + "ON `dn_batch_notes` (`batch`, `candidate_key`) " + "WHERE `candidate_key` IS NOT NULL AND `candidate_key` != ''", + ]) add_fields(token, "delivery_notes", [relation("batch_note", notes_id)]) add_indexes(token, "delivery_notes", [ diff --git a/src/app/api/dn-batches/[id]/assembly/route.test.ts b/src/app/api/dn-batches/[id]/assembly/route.test.ts index 697ab94..96dfb60 100644 --- a/src/app/api/dn-batches/[id]/assembly/route.test.ts +++ b/src/app/api/dn-batches/[id]/assembly/route.test.ts @@ -89,9 +89,8 @@ describe("assembly revision route", () => { it("reclaims an expired lease and releases it after a failed revision", async () => { lockCreate.mockRejectedValueOnce(new Error("duplicate lock")).mockResolvedValueOnce({ id: "new-lock" }); - lockGet - .mockResolvedValueOnce({ id: "expired-lock", batch: id, worker: "dead-worker", expires_at: new Date(Date.now() - 60_000).toISOString() }) - .mockImplementationOnce(async () => lockCreate.mock.calls[1][0]); + lockList.mockResolvedValueOnce([{ id: "expired-lock", batch: id, worker: "dead-worker", expires_at: new Date(Date.now() - 60_000).toISOString() }]); + lockGet.mockImplementationOnce(async () => lockCreate.mock.calls[1][0]); notesUpdate.mockRejectedValueOnce(new Error("write failed")); const response = await POST(request({ revision: 1, groups: [{ pages: [1] }, { pages: [2] }], dismissedPages: [], reason: "try" }), { params: Promise.resolve({ id }) }); expect(response.status).toBe(500); diff --git a/src/lib/dn-pipeline/lock.test.ts b/src/lib/dn-pipeline/lock.test.ts index a9db1fd..0f2089b 100644 --- a/src/lib/dn-pipeline/lock.test.ts +++ b/src/lib/dn-pipeline/lock.test.ts @@ -3,96 +3,109 @@ import { withBatchLease } from "./lock"; const BATCH_ID = "abcdefghijklmno"; -function lockPb(options: { - create?: (data: Record) => Promise>; - getOne?: (id: string) => Promise>; - remove?: (id: string) => Promise; -}) { - const create = vi.fn(options.create ?? (async (data) => ({ ...data }))); - const getOne = vi.fn(options.getOne ?? (async () => { throw new Error("not found"); })); - const remove = vi.fn(options.remove ?? (async () => undefined)); +type Lock = Record & { id: string; batch: string; worker: string; expires_at: string }; + +function lockPb(initial: Lock[] = []) { + const records = new Map(initial.map((record) => [record.id, { ...record }])); + const create = vi.fn(async (data: Record) => { + if ([...records.values()].some((record) => record.batch === data.batch)) { + throw new Error("unique batch lock"); + } + const record = data as Lock; + records.set(record.id, { ...record }); + return { ...record }; + }); + const getOne = vi.fn(async (id: string) => { + const record = records.get(id); + if (!record) throw new Error("not found"); + return { ...record }; + }); + const getFullList = vi.fn(async () => [...records.values()].map((record) => ({ ...record }))); + const remove = vi.fn(async (id: string) => { records.delete(id); }); return { + records, create, getOne, + getFullList, remove, - pb: { - collection: vi.fn(() => ({ create, getOne, delete: remove })), - }, + pb: { collection: vi.fn(() => ({ create, getOne, getFullList, delete: remove })) }, }; } describe("withBatchLease", () => { - it("uses one deterministic PocketBase record id so only the create winner runs", async () => { - const records = new Map>(); - const create = async (data: Record) => { - const id = String(data.id); - if (records.has(id)) throw new Error("duplicate record id"); - records.set(id, { ...data }); - return { ...data }; - }; - const { pb, create: createMock, remove } = lockPb({ - create, - getOne: async (id) => records.get(id) ?? Promise.reject(new Error("not found")), - remove: async (id) => { records.delete(id); }, - }); - let finishWinner: ((value: string) => void) | undefined; - const winner = vi.fn(async () => new Promise((resolve) => { finishWinner = resolve; })); - const loser = vi.fn(async () => "must not run"); + it("uses a distinct valid record id for each contender while UNIQUE(batch) admits only one", async () => { + const { pb, create } = lockPb(); + let releaseFirst: (() => void) | undefined; + const first = withBatchLease(pb as never, BATCH_ID, "worker-a", async () => new Promise((resolve) => { releaseFirst = resolve; })); + await vi.waitFor(() => expect(create).toHaveBeenCalledOnce()); + const secondFn = vi.fn(async () => "must not run"); + const second = await withBatchLease(pb as never, BATCH_ID, "worker-b", secondFn); + releaseFirst!(); - const first = withBatchLease(pb as never, BATCH_ID, "worker-a", winner); - await vi.waitFor(() => expect(winner).toHaveBeenCalledOnce()); - const second = await withBatchLease(pb as never, BATCH_ID, "worker-b", loser); - finishWinner!("ran"); - - expect(await first).toEqual({ acquired: true, value: "ran" }); + expect(await first).toEqual({ acquired: true, value: undefined }); expect(second).toEqual({ acquired: false }); - expect(winner).toHaveBeenCalledOnce(); - expect(loser).not.toHaveBeenCalled(); - const claims = createMock.mock.calls.map((call) => call[0]); - expect(claims[0].id).toMatch(/^[a-z0-9]{15}$/); - expect(claims[1].id).toBe(claims[0].id); - expect(remove).toHaveBeenCalledOnce(); + expect(secondFn).not.toHaveBeenCalled(); + expect(create.mock.calls[0][0].id).toMatch(/^[a-z0-9]{15}$/); + expect(create.mock.calls[1][0].id).toMatch(/^[a-z0-9]{15}$/); + expect(create.mock.calls[1][0].id).not.toBe(create.mock.calls[0][0].id); }); - it("reclaims one expired deterministic lock then releases its own replacement", async () => { - const expired = new Date(Date.now() - 1_000).toISOString(); - const { pb, create, getOne, remove } = lockPb({}); - create - .mockRejectedValueOnce(new Error("duplicate record id")) - .mockImplementationOnce(async (data: Record) => ({ ...data })); - getOne - .mockResolvedValueOnce({ id: "expired-lock", worker: "dead-worker", expires_at: expired }) - .mockImplementationOnce(async () => create.mock.calls[1][0]); + it("reclaims one expired lock into a new acquisition id", async () => { + const old: Lock = { + id: "oldacquisition1", batch: BATCH_ID, worker: "dead", expires_at: new Date(Date.now() - 1_000).toISOString(), + }; + const { pb, create, remove } = lockPb([old]); const result = await withBatchLease(pb as never, BATCH_ID, "worker-a", async () => 42); expect(result).toEqual({ acquired: true, value: 42 }); expect(create).toHaveBeenCalledTimes(2); - expect(getOne).toHaveBeenCalledTimes(2); - expect(remove).toHaveBeenNthCalledWith(1, "expired-lock"); + expect(create.mock.calls[0][0].id).toBe(create.mock.calls[1][0].id); + expect(create.mock.calls[1][0].id).not.toBe(old.id); + expect(remove).toHaveBeenNthCalledWith(1, old.id); expect(remove).toHaveBeenNthCalledWith(2, create.mock.calls[1][0].id); }); - it("does not release a lock reclaimed by another owner before finally", async () => { - let live: Record | undefined; - const { pb, create, getOne, remove } = lockPb({ - create: async (data) => { - live = { ...data }; - return live; - }, - getOne: async () => live ?? Promise.reject(new Error("not found")), - remove: async () => { live = undefined; }, + it("does not delete a replacement created after expired-lock lookup but before old deletion", async () => { + const old: Lock = { + id: "oldacquisition1", batch: BATCH_ID, worker: "dead", expires_at: new Date(Date.now() - 1_000).toISOString(), + }; + const replacement: Lock = { + id: "replacementlock", batch: BATCH_ID, worker: "worker-c", expires_at: new Date(Date.now() + 60_000).toISOString(), + }; + const fixture = lockPb([old]); + fixture.remove.mockImplementationOnce(async (id: string) => { + fixture.records.delete(old.id); + fixture.records.set(replacement.id, { ...replacement }); + fixture.records.delete(id); + }); + const fn = vi.fn(async () => "must not run"); + + const result = await withBatchLease(fixture.pb as never, BATCH_ID, "worker-b", fn); + + expect(result).toEqual({ acquired: false }); + expect(fn).not.toHaveBeenCalled(); + expect(fixture.remove).toHaveBeenCalledWith(old.id); + expect(fixture.records.get(replacement.id)).toEqual(replacement); + }); + + it("does not delete a replacement created after final ownership read", async () => { + const fixture = lockPb(); + const replacement: Lock = { + id: "replacementlock", batch: BATCH_ID, worker: "worker-b", expires_at: new Date(Date.now() + 60_000).toISOString(), + }; + fixture.getOne.mockImplementationOnce(async (id: string) => { + const old = fixture.records.get(id)!; + const snapshot = { ...old }; + fixture.records.delete(id); + fixture.records.set(replacement.id, { ...replacement }); + return snapshot; }); - const result = await withBatchLease(pb as never, BATCH_ID, "worker-a", async () => { - live = { id: live!.id, worker: "worker-b:replacement", expires_at: new Date(Date.now() + 60_000).toISOString() }; - return "done"; - }); + const result = await withBatchLease(fixture.pb as never, BATCH_ID, "worker-a", async () => "done"); expect(result).toEqual({ acquired: true, value: "done" }); - expect(create).toHaveBeenCalledOnce(); - expect(getOne).toHaveBeenCalledOnce(); - expect(remove).not.toHaveBeenCalled(); - expect(live?.worker).toBe("worker-b:replacement"); + expect(fixture.remove).toHaveBeenCalledWith(fixture.create.mock.calls[0][0].id); + expect(fixture.records.get(replacement.id)).toEqual(replacement); }); }); diff --git a/src/lib/dn-pipeline/lock.ts b/src/lib/dn-pipeline/lock.ts index 45d1cf7..290f5d8 100644 --- a/src/lib/dn-pipeline/lock.ts +++ b/src/lib/dn-pipeline/lock.ts @@ -6,9 +6,9 @@ const LEASE_MS = 5 * 60 * 1000; type LockRecord = { id?: string; worker?: string; expires_at?: string }; type LeaseResult = { acquired: true; value: T } | { acquired: false }; -/** A valid, deterministic PocketBase record id for one batch's unique lease. */ -export function batchLockId(batchId: string): string { - return createHash("sha256").update(`dn-pipeline-lock:${batchId}`).digest("hex").slice(0, 15); +/** A valid record id deterministic for one fresh acquisition, never the batch alone. */ +export function batchLockId(batchId: string, owner: string): string { + return createHash("sha256").update(`dn-pipeline-lock:${batchId}:${owner}`).digest("hex").slice(0, 15); } function isExpired(lock: LockRecord, now: number): boolean { @@ -19,10 +19,9 @@ function isExpired(lock: LockRecord, now: number): boolean { /** * Run `fn` only while this batch's create-based lease is held. * - * The initial create is the atomic claim. Reading happens only after a failed - * create, to distinguish a live owner from one expired row that may be - * reclaimed once. The per-acquisition worker token prevents a worker that has - * timed out from releasing another worker's replacement lease. + * The initial create is the atomic claim, with UNIQUE(batch) as the exclusion + * gate. Every acquisition receives a different id derived from its fresh owner + * token, so deleting a previous acquisition id cannot delete a replacement. */ export async function withBatchLease( pb: PocketBase, @@ -30,8 +29,8 @@ export async function withBatchLease( worker: string, fn: () => Promise, ): Promise> { - const id = batchLockId(batchId); const owner = `${worker}:${randomUUID()}`; + const id = batchLockId(batchId, owner); const locks = pb.collection("dn_pipeline_locks"); const payload = { id, @@ -44,13 +43,17 @@ export async function withBatchLease( try { claimed = (await locks.create(payload)) as unknown as LockRecord; } catch { - let existing: LockRecord; + let existing: LockRecord | undefined; try { - existing = (await locks.getOne(id)) as unknown as LockRecord; + const rows = (await locks.getFullList({ + filter: `batch = ${JSON.stringify(batchId)}`, + fields: "id,batch,worker,expires_at", + })) as unknown as LockRecord[]; + existing = rows[0]; } catch { return { acquired: false }; } - if (!isExpired(existing, Date.now())) return { acquired: false }; + if (!existing || !isExpired(existing, Date.now())) return { acquired: false }; try { await locks.delete(existing.id ?? id); claimed = (await locks.create(payload)) as unknown as LockRecord; @@ -63,9 +66,8 @@ export async function withBatchLease( try { return { acquired: true, value: await fn() }; } finally { - // An expiry/reclaim can replace the deterministic row id before this - // worker reaches finally. Check the unique owner token first so this - // loser never deletes the replacement lease. + // A replacement acquisition has another record id. Retain the owner check + // as a belt-and-braces guard before deleting this acquisition's own row. const current = await locks.getOne(claimedId).catch(() => null) as LockRecord | null; if (current?.worker === owner) { await locks.delete(claimedId).catch(() => undefined); diff --git a/src/lib/dn-pipeline/run-staged.test.ts b/src/lib/dn-pipeline/run-staged.test.ts index 23095a0..48776fc 100644 --- a/src/lib/dn-pipeline/run-staged.test.ts +++ b/src/lib/dn-pipeline/run-staged.test.ts @@ -12,7 +12,7 @@ vi.mock("./extract", async (importOriginal) => { return { ...actual, extractPage: extractPageMock }; }); -import { runStagedBatch } from "./run-staged"; +import { candidateKey, runStagedBatch } from "./run-staged"; const BATCH_ID = "abcdefghijklmno"; const SOURCE_ID = "source-00000001"; @@ -32,7 +32,7 @@ function read(page: number, number: string, po: string | null): PageRead { function stagedPb() { const batch: { id: string; status: string; stage: string; site: string; source_count: number; - page_count?: number; open_pages?: unknown; + page_count?: number; open_pages?: unknown; claimed_at?: string; } = { id: BATCH_ID, status: "inbox", stage: "queued", site: "site-1", source_count: 1 }; const pages = new Map>(); const candidates = new Map>(); @@ -58,7 +58,7 @@ function stagedPb() { const row: Record = { id, batch: data.get("batch"), source: data.get("source"), source_page: Number(data.get("source_page")), - region: Number(data.get("region")), image: data.get("image"), image_hash: data.get("image_hash"), + region: Number(data.get("region")), image: `${id}.png`, image_hash: data.get("image_hash"), mime: data.get("mime"), width: Number(data.get("width")), height: Number(data.get("height")), rotation: Number(data.get("rotation")), read_status: data.get("read_status"), }; @@ -90,7 +90,7 @@ function stagedPb() { const locks = new Map>(); const lockCollection = { create: vi.fn(async (data: Record) => { - if (locks.has(String(data.id))) throw new Error("duplicate id"); + if ([...locks.values()].some((lock) => lock.batch === data.batch)) throw new Error("duplicate batch"); locks.set(String(data.id), { ...data }); return data; }), @@ -99,10 +99,13 @@ function stagedPb() { if (!row) throw new Error("not found"); return row; }), + getFullList: vi.fn(async () => [...locks.values()]), delete: vi.fn(async (id: string) => { locks.delete(id); }), }; + const filesGetURL = vi.fn((record: { id: string }, filename: string) => `http://files.test/${record.id}/${filename}`); const pb = { authStore: { token: "token" }, + files: { getURL: filesGetURL }, filter: (raw: string, params: Record) => raw.replace(/\{:(\w+)\}/g, (_match, key: string) => JSON.stringify(params[key])), collection: (name: string) => { if (name === "dn_batches") return { @@ -119,7 +122,7 @@ function stagedPb() { throw new Error(`unexpected collection ${name}`); }, }; - return { pb, batch, pages, candidates, updates, pageCollection, candidateCollection }; + return { pb, batch, pages, candidates, updates, pageCollection, candidateCollection, filesGetURL }; } beforeEach(() => { @@ -132,6 +135,34 @@ beforeEach(() => { afterEach(() => vi.unstubAllEnvs()); describe("runStagedBatch", () => { + it("keeps unknown document candidates page-specific while normalising known printed numbers", () => { + const page = { id: "page-1", source: "source-1", source_page: 1, region: 0 }; + expect(candidateKey(BATCH_ID, page, "DN-001")).toBe(candidateKey(BATCH_ID, page, "dn 001")); + expect(candidateKey(BATCH_ID, page, "")).not.toBe(candidateKey(BATCH_ID, { ...page, source_page: 2 }, "")); + }); + + it("stamps claimed_at before reading and leaves an overlapping worker not_claimed", async () => { + const fixture = stagedPb(); + let releaseNormalisation: (() => void) | undefined; + const normalisationGate = new Promise((resolve) => { releaseNormalisation = resolve; }); + normaliseSourceMock.mockImplementationOnce(async () => { + await normalisationGate; + return [{ sourcePage: 1, png: Buffer.from("one"), mime: "image/png", width: 100, height: 200, rotation: 0 }]; + }); + extractPageMock.mockResolvedValue(read(1, "DN-1", "PO-1")); + + const first = runStagedBatch(fixture.pb as never, BATCH_ID); + await vi.waitFor(() => expect(fixture.batch.claimed_at).toEqual(expect.any(String))); + const firstClaimedAt = fixture.batch.claimed_at; + const second = await runStagedBatch(fixture.pb as never, BATCH_ID); + releaseNormalisation!(); + + expect(second).toMatchObject({ notClaimed: true }); + expect(fixture.batch.claimed_at).toBe(firstClaimedAt); + await expect(first).resolves.toMatchObject({ status: "processed" }); + expect(fixture.batch.claimed_at).toBe(""); + }); + it("persists canonical pages, reuses successful reads, and leaves a clear sibling ready when another needs review", async () => { const fixture = stagedPb(); normaliseSourceMock.mockResolvedValue([ @@ -156,6 +187,9 @@ describe("runStagedBatch", () => { ])); expect(fixture.batch.stage).toBe("split_review"); expect(fixture.batch.page_count).toBe(2); + expect(typeof fixture.updates[0].data.claimed_at).toBe("string"); + expect(Date.parse(String(fixture.updates[0].data.claimed_at))).toBeGreaterThan(Date.now() - 5_000); + expect(fixture.updates.at(-1)?.data.claimed_at).toBe(""); expect(fixture.updates.map((update) => update.data.stage).filter(Boolean)).toEqual([ "preparing", "reading", "split_review", "preparing", "reading", "split_review", ]); @@ -179,4 +213,48 @@ describe("runStagedBatch", () => { expect(fixture.batch.open_pages).toEqual(expect.arrayContaining([expect.objectContaining({ page: 2, reason: "read_failed" })])); expect([...fixture.pages.values()]).toEqual(expect.arrayContaining([expect.objectContaining({ source_page: 2, read_status: "failed", read_error: "vision unavailable" })])); }); + + it("re-extracts from the persisted canonical page file when normalisation changes on rerun", async () => { + const fixture = stagedPb(); + const persistedPng = Buffer.from("persisted-canonical-png"); + vi.stubGlobal("fetch", vi.fn(async (url: string) => new Response( + String(url).includes("page-1.png") ? persistedPng : Buffer.from("original-source"), + { headers: { "content-type": "image/png" } }, + ))); + normaliseSourceMock + .mockResolvedValueOnce([{ sourcePage: 1, png: Buffer.from("first-normalisation"), mime: "image/png", width: 100, height: 200, rotation: 0 }]) + .mockResolvedValueOnce([{ sourcePage: 1, png: Buffer.from("changed-normalisation"), mime: "image/png", width: 100, height: 200, rotation: 0 }]); + extractPageMock.mockResolvedValue(read(1, "DN-1", "PO-1")); + + await runStagedBatch(fixture.pb as never, BATCH_ID); + fixture.pages.get("page-1")!.prompt_version = "old"; + await runStagedBatch(fixture.pb as never, BATCH_ID); + + expect(extractPageMock).toHaveBeenCalledTimes(2); + expect(extractPageMock.mock.calls[1][0]).toEqual(persistedPng); + expect(fixture.filesGetURL).toHaveBeenCalledWith(expect.objectContaining({ id: SOURCE_ID }), "original.jpg"); + expect(fixture.filesGetURL).toHaveBeenCalledWith(expect.objectContaining({ id: "page-1" }), "page-1.png"); + }); + + it("moves the existing DN-2 candidate by stable key when an earlier failed page later recovers as DN-1", async () => { + const fixture = stagedPb(); + normaliseSourceMock.mockResolvedValue([ + { sourcePage: 1, png: Buffer.from("one"), mime: "image/png", width: 100, height: 200, rotation: 0 }, + { sourcePage: 2, png: Buffer.from("two"), mime: "image/png", width: 100, height: 200, rotation: 0 }, + ]); + extractPageMock.mockImplementation(async (_image: Buffer, page: number) => { + if (page === 1) throw new Error("temporary vision failure"); + return read(page, "DN-2", "PO-1"); + }); + + await runStagedBatch(fixture.pb as never, BATCH_ID); + extractPageMock.mockImplementation(async (_image: Buffer, page: number) => read(page, "DN-1", "PO-1")); + await runStagedBatch(fixture.pb as never, BATCH_ID); + + expect([...fixture.candidates.values()]).toEqual(expect.arrayContaining([ + expect.objectContaining({ id: "candidate-1", delivery_note_number: "DN-2", note_index: 1, candidate_key: expect.any(String) }), + expect.objectContaining({ id: "candidate-2", delivery_note_number: "DN-1", note_index: 0, candidate_key: expect.any(String) }), + ])); + expect(fixture.candidates.size).toBe(2); + }); }); diff --git a/src/lib/dn-pipeline/run-staged.ts b/src/lib/dn-pipeline/run-staged.ts index 7bf19b6..9ed2878 100644 --- a/src/lib/dn-pipeline/run-staged.ts +++ b/src/lib/dn-pipeline/run-staged.ts @@ -26,8 +26,8 @@ export interface StagedRunResult { type Batch = { id: string; site: string; status?: string; file?: string }; type Source = { id: string; file: string; mime: string; ordinal?: number; delivery_session_id?: string }; -type PersistedPage = { id: string; source?: string; source_page?: number; region?: number }; -type ExistingCandidate = { id: string; note_index?: number; status?: string; stage?: string; posted_at?: string }; +type PersistedPage = { id: string; source?: string; source_page?: number; region?: number; image?: string }; +type ExistingCandidate = { id: string; candidate_key?: string; note_index?: number; status?: string; stage?: string; posted_at?: string }; /** * Run the non-financial stages for one batch. A successful run only produces @@ -42,7 +42,9 @@ export async function runStagedBatch(pb: PocketBase, batchId: string): Promise { const batch = (await pb.collection("dn_batches").getOne(batchId)) as unknown as Batch; - await pb.collection("dn_batches").update(batchId, { status: "analysing", stage: "preparing", last_error: "" }); + await pb.collection("dn_batches").update(batchId, { + status: "analysing", stage: "preparing", claimed_at: new Date().toISOString(), last_error: "", + }); const sources = ((await pb.collection("dn_batch_sources").getFullList({ filter: batchFilter(pb, batchId), @@ -62,6 +64,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string): Promise = []; const pageDeliverySessionIds: Record = {}; + const persistedPages = new Map(); let pageCount = 0; let pageOffset = 0; @@ -85,7 +88,9 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string): Promise { - const base = process.env.NEXT_PUBLIC_POCKETBASE_URL; - if (!base) throw new Error("NEXT_PUBLIC_POCKETBASE_URL is not set"); - const response = await fetch(`${base}/api/files/dn_batch_sources/${source.id}/${source.file}`, { + const response = await fetch(pb.files.getURL(source as never, source.file), { headers: { Authorization: pb.authStore.token }, }); if (!response.ok) throw new Error(`source file fetch failed: ${response.status}`); return { bytes: Buffer.from(await response.arrayBuffer()), mime: response.headers.get("content-type") ?? source.mime }; } +async function fetchPersistedPageImage( + pb: PocketBase, + page: PersistedPage, + fallback: Buffer, +): Promise { + if (!page.image) return fallback; + const response = await fetch(pb.files.getURL(page as never, page.image), { + headers: { Authorization: pb.authStore.token }, + }); + if (!response.ok) throw new Error(`persisted page fetch failed: ${response.status}`); + return Buffer.from(await response.arrayBuffer()); +} + async function createOrReusePage( pb: PocketBase, batchId: string, @@ -160,7 +179,7 @@ async function createOrReusePage( ): Promise { const pages = pb.collection("dn_batch_pages"); const filter = pageFilter(pb, sourceId, image.sourcePage); - const found = (await pages.getFullList({ filter, fields: "id,source,source_page,region" })) as unknown as PersistedPage[]; + const found = (await pages.getFullList({ filter, fields: "id,source,source_page,region,image" })) as unknown as PersistedPage[]; if (found.length > 0) return found[0]; const form = new FormData(); @@ -183,7 +202,7 @@ async function createOrReusePage( } catch (error) { // The unique source/page/region index is the durable idempotency key. A // concurrent writer may have inserted the same page after our lookup. - const raced = (await pages.getFullList({ filter, fields: "id,source,source_page,region" })) as unknown as PersistedPage[]; + const raced = (await pages.getFullList({ filter, fields: "id,source,source_page,region,image" })) as unknown as PersistedPage[]; if (raced.length > 0) return raced[0]; throw error; } @@ -193,6 +212,7 @@ async function upsertCandidate( pb: PocketBase, batchId: string, existing: ExistingCandidate | undefined, + candidateKey: string, noteIndex: number, group: PageGroup, first: PageRead, @@ -204,6 +224,7 @@ async function upsertCandidate( if (existing && (existing.status === "dismissed" || existing.stage === "dismissed" || existing.stage === "posted" || Boolean(existing.posted_at))) return; const fields = { batch: batchId, + candidate_key: candidateKey, note_index: noteIndex, pages: group.pages, page_from: Math.min(...group.pages), @@ -226,12 +247,25 @@ async function upsertCandidate( else await pb.collection("dn_batch_notes").create(fields); } -async function loadCandidates(pb: PocketBase, batchId: string): Promise> { +async function loadCandidates(pb: PocketBase, batchId: string): Promise> { const records = (await pb.collection("dn_batch_notes").getFullList({ filter: batchFilter(pb, batchId), - fields: "id,note_index,status,stage,posted_at", + fields: "id,candidate_key,note_index,status,stage,posted_at", })) as unknown as ExistingCandidate[]; - return new Map(records.filter((record) => Number.isInteger(record.note_index)).map((record) => [record.note_index!, record])); + return new Map(records.filter((record) => Boolean(record.candidate_key)).map((record) => [record.candidate_key!, record])); +} + +/** + * The first persisted page is stable when an earlier failed sibling recovers, + * while the normalised printed number distinguishes documents sharing a scan. + * An unknown number remains safely page-specific rather than colliding with + * another unknown page. + */ +export function candidateKey(batchId: string, page: PersistedPage, documentNumber: string): string { + const pageIdentity = `${page.source ?? "unknown-source"}:${page.source_page ?? 0}:${page.region ?? 0}`; + const documentIdentity = normalisePoNumber(documentNumber); + const identity = documentIdentity ? `document:${documentIdentity}` : `unknown-page:${pageIdentity}`; + return createHash("sha256").update(`${batchId}|${pageIdentity}|${identity}`).digest("hex"); } async function loadPurchaseOrders(pb: PocketBase, siteId: string): Promise { diff --git a/src/lib/pb-types.ts b/src/lib/pb-types.ts index 097714b..478d6a6 100644 --- a/src/lib/pb-types.ts +++ b/src/lib/pb-types.ts @@ -452,6 +452,8 @@ export interface DnBatchNoteRecord extends RecordModel { /** Back-reference to the DELIVERY_NOTES row this note was promoted into. Empty while a candidate or blocked. */ delivery_note: string; stage?: CandidateStage; + /** Stable staged identity derived from first persisted page plus document identity. */ + candidate_key?: string; assembly_revision?: number; pages?: unknown; supplier?: string;