feat(dn-batches): add the tick route with compare-and-set claiming
This commit is contained in:
DEls
parent
540bea8b5d
commit
ebede76c40
3 changed files with 764 additions and 0 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
640
src/app/api/dn-batches/tick/route.test.ts
Normal file
640
src/app/api/dn-batches/tick/route.test.ts
Normal file
|
|
@ -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<string, string> } = {}): 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<string, unknown> }> {
|
||||
return dnBatchesUpdate.mock.calls.map((c) => ({
|
||||
id: c[0] as string,
|
||||
data: c[1] as Record<string, unknown>,
|
||||
}));
|
||||
}
|
||||
|
||||
/** 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<MAX → back in queue
|
||||
});
|
||||
|
||||
it("parks a failing batch at needs_attention after MAX_ATTEMPTS", async () => {
|
||||
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 },
|
||||
]);
|
||||
});
|
||||
});
|
||||
114
src/app/api/dn-batches/tick/route.ts
Normal file
114
src/app/api/dn-batches/tick/route.ts
Normal file
|
|
@ -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<string, unknown>[] = [];
|
||||
|
||||
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<ReturnType<typeof pbAdmin>>, id: string): Promise<boolean> {
|
||||
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<ReturnType<typeof pbAdmin>>): Promise<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue