diff --git a/.env.local.example b/.env.local.example index aaee842..26d2648 100644 --- a/.env.local.example +++ b/.env.local.example @@ -40,6 +40,16 @@ N8N_INGEST_SECRET=replace-with-output-of-openssl-rand-hex-32 # The Delivery Note OCR workflow in n8n imports docs/n8n/delivery-note-ocr-workflow.json # and calls POST {MEGA_NEXT_URL}/api/delivery-notes/{id}/ocr-result with this header. +# ── Delivery-note batch pipeline (Task 10+) ──────────────────────── +# Customer VAT number the pipeline uses to check the printed +# identity block on every slip. Suppliers print it on every weighbridge +# ticket; mismatches get marked needs_attention. +MEGA_CUSTOMER_VAT_NO=4280307796 + +# How many dn_batches rows the /api/dn-batches/tick route claims per +# cron hit. Tune for your batch size / model latency. +DN_TICK_BATCH_SIZE=3 + # Public URL of the n8n instance (used by your local Codex MCP config). N8N_MCP_URL=https://wfdash.n8n.apphub.co.za/mcp-server/http diff --git a/src/app/api/dn-batches/tick/route.test.ts b/src/app/api/dn-batches/tick/route.test.ts new file mode 100644 index 0000000..fc2365f --- /dev/null +++ b/src/app/api/dn-batches/tick/route.test.ts @@ -0,0 +1,640 @@ +/** + * Tests for POST /api/dn-batches/tick. + * + * The orchestrator (runBatch) is mocked at the module boundary so we + * can assert on the route's behaviour without standing up any of the + * pipeline. PocketBase is mocked in the same shape every other route + * test uses: a name-keyed collection dispatch + recording mocks. + * + * What the route is responsible for (and what we test here): + * 1. Auth: rejects without x-ingest-secret, accepts the right one. + * 2. Reclaim: at the start, finds analysing batches with stale + * claimed_at and either resets them to inbox (attempts < MAX) or + * parks them at needs_attention (attempts >= MAX). + * 3. List: pulls every dn_batches row with status='inbox', in + * created order, capped by DN_TICK_BATCH_SIZE. + * 4. Claim: compare-and-set on status — only the winner of the race + * runs the batch, the loser is recorded as skipped:'not_claimed'. + * 5. Run: delegates to the orchestrator and merges its result. + * 6. Failure: catches throws from runBatch, increments attempts, + * writes last_error (truncated) and claimed_at="", parks as + * needs_attention after MAX_ATTEMPTS, otherwise puts it back in + * inbox for the next tick. + * 7. Response: { ok, claimed, results } with one entry per candidate + * (skipped, success, or error). + * + * Defaults used by every test unless overridden: + * - one inbox candidate with id=ID, status='inbox' + * - runBatch resolves to { status:'processed', notesDetected:1, + * notesPromoted:1, checks:[{ id:'pages_accounted', passed:true }] } + * - no stale analysing batches (the reclaim query returns []) + * + * Helper structure mirrors purchase-orders/process-batch/route.test.ts. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +/* ------------------------------------------------------------------ */ +/* Hoisted mocks — vi.mock factories run before module imports, so */ +/* every constant the factory touches must be hoisted too. */ +/* ------------------------------------------------------------------ */ + +const { dnBatchesGetFullList, dnBatchesGetOne, dnBatchesUpdate, runBatchMock, viPbAdmin } = + vi.hoisted(() => ({ + dnBatchesGetFullList: vi.fn(), + dnBatchesGetOne: vi.fn(), + dnBatchesUpdate: vi.fn(), + runBatchMock: vi.fn(), + viPbAdmin: vi.fn(), + })); + +/** + * The route calls getFullList twice: first for the reclaim query + * (filter has status='analysing' && claimed_at<...), then for the + * inbox list (filter has status='inbox'). We route them by inspecting + * the filter string — keeps individual tests focused on the shape + * they care about. + */ +function makePb() { + return { + authStore: { token: "test-admin-token" }, + collection: (name: string) => { + if (name !== "dn_batches") { + throw new Error(`unexpected collection in test: ${name}`); + } + return { + getFullList: dnBatchesGetFullList, + getOne: dnBatchesGetOne, + update: dnBatchesUpdate, + }; + }, + }; +} + +vi.mock("@/lib/pb", () => ({ + pbAdmin: viPbAdmin, +})); + +vi.mock("@/lib/dn-pipeline/run", () => ({ + runBatch: runBatchMock, +})); + +import { pbAdmin } from "@/lib/pb"; +import { POST } from "./route"; + +const ID = "abcdefghijklmno"; + +function makeRequest(opts: { headers?: Record } = {}): Request { + return new Request("http://x/api/dn-batches/tick", { + method: "POST", + headers: { ...opts.headers }, + }); +} + +function postWithValidAuth(): Request { + return makeRequest({ headers: { "x-ingest-secret": "test-secret" } }); +} + +/** Read every dn_batches.update payload, in order. */ +function allUpdates(): Array<{ id: string; data: Record }> { + return dnBatchesUpdate.mock.calls.map((c) => ({ + id: c[0] as string, + data: c[1] as Record, + })); +} + +/** Read the getFullList call index by filter prefix. */ +function callIndex(prefix: string): number { + return dnBatchesGetFullList.mock.calls.findIndex( + (c) => (c[0] as { filter?: string }).filter?.startsWith(prefix), + ); +} + +/* ------------------------------------------------------------------ */ +/* Lifecycle */ +/* ------------------------------------------------------------------ */ + +beforeEach(() => { + vi.stubEnv("N8N_INGEST_SECRET", "test-secret"); + vi.stubEnv("POCKETBASE_ADMIN_EMAIL", "test@example.com"); + vi.stubEnv("POCKETBASE_ADMIN_PASSWORD", "test-password"); + vi.stubEnv("DN_TICK_BATCH_SIZE", "3"); + + dnBatchesGetFullList.mockReset(); + dnBatchesGetOne.mockReset(); + dnBatchesUpdate.mockReset(); + runBatchMock.mockReset(); + + viPbAdmin.mockReset(); + viPbAdmin.mockResolvedValue(makePb() as never); + + // Default: no stale claims, one inbox candidate. + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='analysing'")) return []; + if (opts.filter?.includes("status='inbox'")) return [{ id: ID }]; + throw new Error(`unexpected getFullList filter in test: ${opts.filter}`); + }, + ); + dnBatchesGetOne.mockImplementation(async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields?.startsWith("status,claimed_at")) return { id, status: "analysing" }; + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields === "attempts") return { id, attempts: 0 }; + throw new Error(`unexpected getOne in test: id=${id} fields=${opts.fields}`); + }); + dnBatchesUpdate.mockResolvedValue({ id: ID }); + + runBatchMock.mockResolvedValue({ + status: "processed", + notesDetected: 1, + notesPromoted: 1, + checks: [{ id: "pages_accounted", passed: true, reason: "ok" }], + }); +}); + +afterEach(() => { + vi.unstubAllEnvs(); +}); + +/* ------------------------------------------------------------------ */ +/* Tests */ +/* ------------------------------------------------------------------ */ + +describe("POST /api/dn-batches/tick — auth", () => { + it("returns 401 when x-ingest-secret is missing", async () => { + const res = await POST(makeRequest()); + expect(res.status).toBe(401); + expect(await res.json()).toEqual({ error: "unauthorized" }); + expect(dnBatchesGetFullList).not.toHaveBeenCalled(); + expect(runBatchMock).not.toHaveBeenCalled(); + }); + + it("returns 401 when x-ingest-secret is wrong", async () => { + const res = await POST(makeRequest({ headers: { "x-ingest-secret": "wrong" } })); + expect(res.status).toBe(401); + expect(await res.json()).toEqual({ error: "unauthorized" }); + expect(dnBatchesGetFullList).not.toHaveBeenCalled(); + }); +}); + +describe("POST /api/dn-batches/tick — reclaim", () => { + it("runs the stale-claim query before listing the inbox", async () => { + await POST(postWithValidAuth()); + expect(callIndex("status='analysing' && claimed_at <")).toBeGreaterThanOrEqual(0); + // The reclaim call comes before the inbox list. + const reclaim = callIndex("status='analysing' && claimed_at <"); + const inbox = callIndex("status='inbox'"); + expect(reclaim).toBeGreaterThanOrEqual(0); + expect(inbox).toBeGreaterThan(reclaim); + }); + + it("uses a 10-minute stale cutoff", async () => { + const before = Date.now(); + await POST(postWithValidAuth()); + const after = Date.now(); + const reclaimCall = dnBatchesGetFullList.mock.calls.find( + (c) => (c[0] as { filter?: string }).filter?.startsWith("status='analysing'"), + ); + expect(reclaimCall).toBeDefined(); + const filter = (reclaimCall![0] as { filter: string }).filter; + const m = /claimed_at < "([^"]+)"/.exec(filter); + expect(m).not.toBeNull(); + const cutoff = Date.parse(m![1]); + // cutoff is `now - 10min`, so it should be between `before - 10min` + // and `after - 10min`. + expect(cutoff).toBeGreaterThanOrEqual(before - 10 * 60 * 1000 - 50); + expect(cutoff).toBeLessThanOrEqual(after - 10 * 60 * 1000 + 50); + }); + + it("resets a stale analysing batch to inbox and bumps attempts (attempts<3)", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='analysing'")) + return [{ id: "stale1", attempts: 1 }]; + if (opts.filter?.includes("status='inbox'")) return [{ id: ID }]; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + expect(res.status).toBe(200); + + const stale = allUpdates().find((u) => u.id === "stale1"); + expect(stale).toBeDefined(); + expect(stale!.data.attempts).toBe(2); + expect(stale!.data.status).toBe("inbox"); + expect(stale!.data.claimed_at).toBe(""); + expect(stale!.data.last_error).toMatch(/reclaimed/); + }); + + it("parks a stale analysing batch at needs_attention when attempts hit MAX_ATTEMPTS", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='analysing'")) + return [{ id: "stale3", attempts: 2 }]; // +1 = 3 → MAX_ATTEMPTS + if (opts.filter?.includes("status='inbox'")) return []; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + expect(res.status).toBe(200); + + const stale = allUpdates().find((u) => u.id === "stale3"); + expect(stale).toBeDefined(); + expect(stale!.data.attempts).toBe(3); + expect(stale!.data.status).toBe("needs_attention"); + }); + + it("does not 500 when the reclaim query itself fails", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='analysing'")) + throw new Error("PB down"); + if (opts.filter?.includes("status='inbox'")) return [{ id: ID }]; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + expect(res.status).toBe(200); + // The inbox tick still runs. + expect(runBatchMock).toHaveBeenCalledTimes(1); + }); +}); + +describe("POST /api/dn-batches/tick — list failure", () => { + it("returns 500 with error=list_failed when the inbox list throws", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='analysing'")) return []; + if (opts.filter?.includes("status='inbox'")) + throw new Error("PB outage"); + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + expect(res.status).toBe(500); + expect(await res.json()).toEqual({ error: "list_failed" }); + expect(runBatchMock).not.toHaveBeenCalled(); + }); +}); + +describe("POST /api/dn-batches/tick — happy path", () => { + it("claims and runs every candidate up to DN_TICK_BATCH_SIZE", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='inbox'")) + return [{ id: "a1" }, { id: "a2" }, { id: "a3" }]; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + expect(res.status).toBe(200); + + const body = await res.json(); + expect(body.ok).toBe(true); + expect(body.claimed).toBe(3); + expect(body.results).toHaveLength(3); + expect(runBatchMock).toHaveBeenCalledTimes(3); + expect(runBatchMock.mock.calls[0][1]).toBe("a1"); + expect(runBatchMock.mock.calls[1][1]).toBe("a2"); + expect(runBatchMock.mock.calls[2][1]).toBe("a3"); + }); + + it("uses the DN_TICK_BATCH_SIZE env var as the cap", async () => { + vi.stubEnv("DN_TICK_BATCH_SIZE", "2"); + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='inbox'")) + return [{ id: "a1" }, { id: "a2" }, { id: "a3" }]; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body.claimed).toBe(2); + expect(runBatchMock).toHaveBeenCalledTimes(2); + expect(body.results.map((r: { id: string }) => r.id)).toEqual(["a1", "a2"]); + }); + + it("defaults DN_TICK_BATCH_SIZE to 3 when env is unset", async () => { + vi.stubEnv("DN_TICK_BATCH_SIZE", ""); + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='inbox'")) + return [ + { id: "a1" }, + { id: "a2" }, + { id: "a3" }, + { id: "a4" }, + { id: "a5" }, + ]; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body.claimed).toBe(3); + expect(runBatchMock).toHaveBeenCalledTimes(3); + }); + + it("does nothing when there are no inbox candidates", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='inbox'")) return []; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(res.status).toBe(200); + expect(body).toEqual({ ok: true, claimed: 0, results: [] }); + expect(runBatchMock).not.toHaveBeenCalled(); + }); + + it("strips checks from the success results", async () => { + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body.results[0]).toEqual({ + id: ID, + status: "processed", + notesDetected: 1, + notesPromoted: 1, + // checks is intentionally removed — operators see the checks on + // the batch record itself, not in the tick log. + }); + expect(body.results[0]).not.toHaveProperty("checks"); + }); +}); + +describe("POST /api/dn-batches/tick — compare-and-set claim", () => { + it("skips a candidate whose re-read after PATCH shows a status other than 'analysing'", async () => { + // claim() reads status='inbox' (passes pre-check), PATCHes to + // 'analysing', then re-reads — if the re-read returns anything + // other than 'analysing', we lost the compare-and-set. We model + // that here by returning a stale 'inbox' on the post-PATCH read. + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields?.startsWith("status,claimed_at")) + return { id, status: "inbox" }; // CAS lost — not analysing + if (opts.fields === "attempts") return { id, attempts: 0 }; + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(res.status).toBe(200); + expect(body.results).toEqual([{ id: ID, skipped: "not_claimed" }]); + expect(runBatchMock).not.toHaveBeenCalled(); + }); + + it("skips a candidate whose status was already moved off 'inbox' before we even PATCHed it", async () => { + // The first getOne for status=inbox returns something else (e.g. + // a duplicate tick raced us). claim() short-circuits to false. + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields === "status") return { id, status: "analysing" }; + if (opts.fields === "attempts") return { id, attempts: 0 }; + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(res.status).toBe(200); + expect(body.results).toEqual([{ id: ID, skipped: "not_claimed" }]); + // The claim PATCH must never have fired. + expect(dnBatchesUpdate).not.toHaveBeenCalled(); + }); + + it("runs the batch when the claim succeeds (re-read shows analysing)", async () => { + // Default mocks already give us: status='inbox' on the pre-check, + // status='analysing' on the post-PATCH re-read. This test just + // asserts that the default path produces a real run. + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body.results[0].id).toBe(ID); + expect(body.results[0].status).toBe("processed"); + expect(runBatchMock).toHaveBeenCalledTimes(1); + }); + + it("skips when the PATCH itself throws (claim failure short-circuits)", async () => { + dnBatchesUpdate.mockRejectedValueOnce(new Error("PB patch failed")); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(res.status).toBe(200); + expect(body.results).toEqual([{ id: ID, skipped: "not_claimed" }]); + expect(runBatchMock).not.toHaveBeenCalled(); + }); +}); + +describe("POST /api/dn-batches/tick — run failures", () => { + it("increments attempts, writes last_error and clears claimed_at on a thrown error", async () => { + runBatchMock.mockRejectedValueOnce(new Error("model exploded")); + // After the throw, the route re-reads attempts on the same id. + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields?.startsWith("status,claimed_at")) + return { id, status: "analysing" }; + if (opts.fields === "attempts") return { id, attempts: 1 }; + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(res.status).toBe(200); + expect(body.results).toEqual([ + { id: ID, error: "model exploded", attempts: 2 }, + ]); + + const failure = allUpdates().find( + (u) => + u.id === ID && + typeof u.data.last_error === "string" && + !u.data.last_error.includes("reclaimed"), + ); + expect(failure).toBeDefined(); + expect(failure!.data.attempts).toBe(2); + expect(failure!.data.last_error).toBe("model exploded"); + expect(failure!.data.claimed_at).toBe(""); + expect(failure!.data.status).toBe("inbox"); // attempts { + runBatchMock.mockRejectedValueOnce(new Error("permanent failure")); + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields?.startsWith("status,claimed_at")) + return { id, status: "analysing" }; + if (opts.fields === "attempts") return { id, attempts: 2 }; + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body.results[0].attempts).toBe(3); + + const failure = allUpdates().find( + (u) => + u.id === ID && + typeof u.data.last_error === "string" && + !u.data.last_error.includes("reclaimed"), + ); + expect(failure!.data.status).toBe("needs_attention"); + expect(failure!.data.attempts).toBe(3); + }); + + it("truncates last_error to 2000 characters", async () => { + const longMsg = "x".repeat(3000); + runBatchMock.mockRejectedValueOnce(new Error(longMsg)); + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields?.startsWith("status,claimed_at")) + return { id, status: "analysing" }; + if (opts.fields === "attempts") return { id, attempts: 0 }; + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + + const res = await POST(postWithValidAuth()); + expect(res.status).toBe(200); + + const failure = allUpdates().find( + (u) => + u.id === ID && + typeof u.data.last_error === "string" && + !u.data.last_error.includes("reclaimed"), + ); + expect(failure!.data.last_error).toHaveLength(2000); + }); + + it("falls back to 0 attempts and still parks the batch if the post-throw attempts read fails", async () => { + runBatchMock.mockRejectedValueOnce(new Error("oops")); + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields?.startsWith("status,claimed_at")) + return { id, status: "analysing" }; + // attempts read throws, the route's catch returns { attempts: 0 }. + if (opts.fields === "attempts") throw new Error("PB read failed"); + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + + const res = await POST(postWithValidAuth()); + expect(res.status).toBe(200); + + const failure = allUpdates().find( + (u) => + u.id === ID && + typeof u.data.last_error === "string" && + !u.data.last_error.includes("reclaimed"), + ); + expect(failure!.data.attempts).toBe(1); + // 1 < 3 → back to inbox; we still know the row exists. + expect(failure!.data.status).toBe("inbox"); + }); + + it("coerces non-Error throws into a string message", async () => { + runBatchMock.mockRejectedValueOnce("just a string"); + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields?.startsWith("status,claimed_at")) + return { id, status: "analysing" }; + if (opts.fields === "attempts") return { id, attempts: 0 }; + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body.results[0].error).toBe("just a string"); + }); +}); + +describe("POST /api/dn-batches/tick — admin auth is awaited", () => { + it("awaits pbAdmin before doing any PocketBase calls", async () => { + let resolveAdmin!: (pb: unknown) => void; + viPbAdmin.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveAdmin = resolve; + }), + ); + + const inflight = POST(postWithValidAuth()); + // No PB calls have happened yet — the route is awaiting pbAdmin. + expect(dnBatchesGetFullList).not.toHaveBeenCalled(); + resolveAdmin(makePb()); + const res = await inflight; + expect(res.status).toBe(200); + }); +}); + +describe("POST /api/dn-batches/tick — response shape", () => { + it("returns ok=true, claimed and a results array on success", async () => { + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body).toMatchObject({ + ok: true, + claimed: 1, + }); + expect(Array.isArray(body.results)).toBe(true); + }); + + it("mixes skipped + success + error entries in the results array", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='inbox'")) + return [{ id: "a1" }, { id: "a2" }, { id: "a3" }]; + return []; + }, + ); + // a1: claim succeeds → success + // a2: pre-claim getOne sees 'analysing' → skipped + // a3: claim succeeds but runBatch throws → error + dnBatchesGetOne.mockImplementation( + async (id: string, opts: { fields?: string } = {}) => { + if (id === "a2" && opts.fields === "status") + return { id, status: "analysing" }; + if (opts.fields === "status") return { id, status: "inbox" }; + if (opts.fields?.startsWith("status,claimed_at")) + return { id, status: "analysing" }; + if (opts.fields === "attempts") return { id, attempts: 0 }; + throw new Error(`unexpected getOne: ${id} ${opts.fields}`); + }, + ); + runBatchMock.mockImplementation(async (pb: unknown, id: string) => { + if (id === "a3") throw new Error("model timeout"); + return { + status: "processed", + notesDetected: 1, + notesPromoted: 1, + checks: [], + }; + }); + + const res = await POST(postWithValidAuth()); + const body = await res.json(); + expect(body.claimed).toBe(3); + expect(body.results).toEqual([ + { id: "a1", status: "processed", notesDetected: 1, notesPromoted: 1 }, + { id: "a2", skipped: "not_claimed" }, + { id: "a3", error: "model timeout", attempts: 1 }, + ]); + }); +}); \ No newline at end of file diff --git a/src/app/api/dn-batches/tick/route.ts b/src/app/api/dn-batches/tick/route.ts new file mode 100644 index 0000000..897ca2b --- /dev/null +++ b/src/app/api/dn-batches/tick/route.ts @@ -0,0 +1,114 @@ +import { NextResponse } from "next/server"; +import { pbAdmin } from "@/lib/pb"; +import { runBatch } from "@/lib/dn-pipeline/run"; + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; +export const maxDuration = 300; + +/** A batch claimed longer ago than this is assumed dead and reclaimed. */ +const STALE_CLAIM_MS = 10 * 60 * 1000; +const MAX_ATTEMPTS = 3; + +/** + * POST /api/dn-batches/tick + * + * The only thing n8n calls. Claims up to DN_TICK_BATCH_SIZE inbox + * batches and runs each through the pipeline. + * + * Claiming is compare-and-set on `status`: we PATCH inbox → analysing + * and re-read. PocketBase has no transactions, so a losing racer sees a + * status that is no longer `inbox` and skips the batch rather than + * double-processing it. + */ +export async function POST(request: Request) { + const secret = request.headers.get("x-ingest-secret"); + if (!secret || secret !== process.env.N8N_INGEST_SECRET) { + return NextResponse.json({ error: "unauthorized" }, { status: 401 }); + } + + const pb = await pbAdmin(); + const limit = Number(process.env.DN_TICK_BATCH_SIZE || 3); + const results: Record[] = []; + + await reclaimStale(pb); + + let candidates: { id: string }[] = []; + try { + candidates = await pb.collection("dn_batches").getFullList({ + filter: `status='inbox'`, + fields: "id", + sort: "created", + }); + } catch (err) { + console.error("[dn-tick] list failed", err); + return NextResponse.json({ error: "list_failed" }, { status: 500 }); + } + + for (const c of candidates.slice(0, limit)) { + const claimed = await claim(pb, c.id); + if (!claimed) { + results.push({ id: c.id, skipped: "not_claimed" }); + continue; + } + try { + const r = await runBatch(pb, c.id); + results.push({ id: c.id, ...r, checks: undefined }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.error("[dn-tick] run failed", { id: c.id, message }); + const current = await pb + .collection("dn_batches") + .getOne(c.id, { fields: "attempts" }) + .catch(() => ({ attempts: 0 })); + const attempts = ((current as { attempts?: number }).attempts ?? 0) + 1; + await pb.collection("dn_batches").update(c.id, { + attempts, + last_error: message.slice(0, 2000), + claimed_at: "", + // Give up after MAX_ATTEMPTS rather than looping forever and + // spending model calls on a document that will never parse. + status: attempts >= MAX_ATTEMPTS ? "needs_attention" : "inbox", + }); + results.push({ id: c.id, error: message, attempts }); + } + } + + return NextResponse.json({ ok: true, claimed: results.length, results }); +} + +async function claim(pb: Awaited>, id: string): Promise { + try { + const before = await pb.collection("dn_batches").getOne(id, { fields: "status" }); + if ((before as { status?: string }).status !== "inbox") return false; + await pb.collection("dn_batches").update(id, { + status: "analysing", + claimed_at: new Date().toISOString(), + }); + const after = await pb.collection("dn_batches").getOne(id, { fields: "status,claimed_at" }); + return (after as { status?: string }).status === "analysing"; + } catch { + return false; + } +} + +async function reclaimStale(pb: Awaited>): Promise { + const cutoff = new Date(Date.now() - STALE_CLAIM_MS).toISOString(); + try { + const stuck = await pb.collection("dn_batches").getFullList({ + filter: `status='analysing' && claimed_at < "${cutoff}"`, + fields: "id,attempts", + }); + for (const s of stuck) { + const attempts = ((s as { attempts?: number }).attempts ?? 0) + 1; + await pb.collection("dn_batches").update(s.id, { + attempts, + status: attempts >= MAX_ATTEMPTS ? "needs_attention" : "inbox", + claimed_at: "", + last_error: "reclaimed after a stalled run", + }); + } + } catch (err) { + console.warn("[dn-tick] reclaim failed", err); + } +} \ No newline at end of file