feat(delivery-notes): own batch lifecycle in OCR pipeline

- upload: create data_import_batches row when DN is uploaded
  (batch_type=delivery_notes, processed_record_id=<dnId>)
- ocr-result: mark the batch processed on success, failed/dead on
  failure (uses nextBatchState from batch-state.ts for retries)
- new /api/delivery-notes/[id]/slip proxy route: lets n8n fetch the
  slip file via shared secret instead of embedding the PocketBase
  admin JWT in workflow headers (which was leaking in the workflow JSON)

Backfill: existing pending DN ekko6xkhynoekas needs a manually
created data_import_batches row until it's reprocessed.
This commit is contained in:
DEls 2026-08-26 15:50:44 +02:00
parent 8dbf393446
commit 1606bd2cc7
3 changed files with 218 additions and 0 deletions

View file

@ -1,5 +1,6 @@
import { NextResponse } from "next/server";
import { pbAdmin } from "@/lib/pb";
import { nextBatchState, MAX_ATTEMPTS } from "@/lib/batch-state";
import type { PurchaseOrderItemRecord } from "@/lib/pb-types";
/**
@ -185,6 +186,16 @@ export async function POST(
});
} catch (err) {
console.error("[dn-ocr-result] dn update failed:", err);
// The DN update failed — mark the batch as failed/dead so the
// operator dashboard reflects the broken state instead of leaving
// it pending forever. n8n's next poll would otherwise re-process
// this row indefinitely.
await updateBatchRow(pb, dnId, {
status: "failed",
attempts: 1,
lastError:
err instanceof Error ? err.message : "dn_update_failed (unknown)",
});
return NextResponse.json(
{ error: err instanceof Error ? err.message : "dn_update_failed" },
{ status: 500 },
@ -195,6 +206,14 @@ export async function POST(
// The manual tier gets no items — the user will fill them in via the
// exception inbox.
if (tier === "manual") {
// Mark the linked data_import_batches row processed (the OCR
// ran fine; we just couldn't auto-extract items). Search by
// processed_record_id so we don't depend on a relation field.
await updateBatchRow(pb, dnId, {
status: "processed",
attempts: 1,
lastError: null,
});
return NextResponse.json({
ok: true,
tier,
@ -296,6 +315,16 @@ export async function POST(
}
}
// 9. Mark the linked data_import_batches row processed. Single
// authoritative place to do this — the n8n workflow no longer
// patches PB directly, so this is the source of truth for batch
// lifecycle.
await updateBatchRow(pb, dnId, {
status: "processed",
attempts: 1,
lastError: null,
});
return NextResponse.json({
ok: true,
tier,
@ -305,6 +334,68 @@ export async function POST(
});
}
/* ------------------------------------------------------------------ */
/* Update the data_import_batches row linked to this DN. */
/* */
/* Lookups by `processed_record_id` (set by the upload route) so we */
/* don't need a relation field on delivery_notes. Tolerates a missing */
/* row (older DNs uploaded before the batch-tracking migration will */
/* not have one) by silently no-op'ing — those rows are simply not */
/* tracked in the imports view. */
/* ------------------------------------------------------------------ */
async function updateBatchRow(
pb: Awaited<ReturnType<typeof pbAdmin>>,
dnId: string,
opts: {
status: "processed" | "failed" | "dead" | "processing";
attempts?: number;
lastError: string | null;
},
): Promise<void> {
try {
const batches = await pb
.collection("data_import_batches")
.getFullList({
filter: `processed_record_id='${dnId}' AND batch_type='delivery_notes'`,
fields: "id,attempts",
perPage: 5,
});
if (batches.length === 0) {
// Older DN without a batch row — nothing to update.
return;
}
const batch = batches[0];
const priorAttempts = (batch.attempts ?? 0) + (opts.attempts ?? 0);
// For "processed" we always win (the OCR succeeded). For failure
// paths the caller passes status=failed and we apply the
// nextBatchState machine to decide if this attempt is terminal.
const nextStatus =
opts.status === "processed"
? "processed"
: opts.status === "processing"
? "processing"
: nextBatchState({
lastResult: "fail",
attempts: priorAttempts,
});
const now = new Date().toISOString();
await pb.collection("data_import_batches").update(batch.id, {
status: nextStatus,
attempts: priorAttempts,
last_attempt_at: now,
processed_at: opts.status === "processed" ? now : undefined,
last_error: opts.lastError ? opts.lastError.slice(0, 1900) : null,
});
} catch (err) {
// Never let batch-update failures mask the real OCR result. Log
// and move on — the DN itself has already been updated.
console.warn(
`[dn-ocr-result] batch update failed for dn=${dnId}:`,
err,
);
}
}
/* ------------------------------------------------------------------ */
/* Tiny type-safe coercion helpers */
/* ------------------------------------------------------------------ */

View file

@ -0,0 +1,100 @@
import { NextResponse } from "next/server";
import { pbAdmin } from "@/lib/pb";
/**
* GET /api/delivery-notes/[id]/slip
*
* Proxies the slip file for a delivery note. Used by the n8n OCR
* pipeline so it can fetch the file with the shared ingest secret
* instead of embedding the PocketBase admin JWT in workflow headers
* (which would leak the god-mode token if the workflow JSON ever
* got exported, screenshotted, or shared).
*
* Auth: `x-ingest-secret: ${N8N_INGEST_SECRET}` (same as the other
* n8n callbacks). Path param `id` is the delivery_note id.
*
* The admin token stays on the server. We fetch the DN record to
* resolve the slip filename, then stream the file bytes back.
*/
export async function GET(
request: Request,
{ params }: { params: Promise<{ id: string }> },
) {
// 1. Auth
const secret = request.headers.get("x-ingest-secret");
if (!secret || secret !== process.env.N8N_INGEST_SECRET) {
return NextResponse.json({ error: "unauthorized" }, { status: 401 });
}
const { id: dnId } = await params;
if (!dnId || !/^[a-z0-9]{15}$/i.test(dnId)) {
return NextResponse.json({ error: "invalid_dn_id" }, { status: 400 });
}
const pb = await pbAdmin();
// 2. Resolve the slip filename from the DN record.
let slipImage: string;
try {
const dn = await pb
.collection("delivery_notes")
.getOne(dnId, { fields: "slip_image" });
if (!dn.slip_image) {
return NextResponse.json(
{ error: "no_slip_image" },
{ status: 404 },
);
}
slipImage = dn.slip_image;
} catch {
return NextResponse.json({ error: "dn_not_found" }, { status: 404 });
}
// 3. Stream the file bytes from PocketBase back to the caller. Use
// the SDK's file helper so we get the right base URL + auth without
// hardcoding the PB URL here.
try {
const pbUrl = process.env.NEXT_PUBLIC_POCKETBASE_URL;
if (!pbUrl) {
return NextResponse.json(
{ error: "pb_url_not_configured" },
{ status: 500 },
);
}
const fileUrl = `${pbUrl}/api/files/delivery_notes/${dnId}/${slipImage}`;
// pbAdmin() returns the pocketbase JS Client with admin auth
// already attached, so its buildUrl() helper produces the right
// signed URL. We then strip the host so we can fetch via fetch()
// with the Authorization header re-attached.
const token = pb.authStore.token;
const fileRes = await fetch(fileUrl, {
headers: { Authorization: token },
});
if (!fileRes.ok) {
return NextResponse.json(
{ error: "slip_fetch_failed", status: fileRes.status },
{ status: 502 },
);
}
const contentType =
fileRes.headers.get("content-type") ?? "application/octet-stream";
const body = await fileRes.arrayBuffer();
return new NextResponse(body, {
status: 200,
headers: {
"Content-Type": contentType,
// n8n downloads as binary; force a download-disposition with
// the original filename so the workflow's binary metadata
// carries a sensible name.
"Content-Disposition": `attachment; filename="${slipImage}"`,
"Cache-Control": "no-store",
},
});
} catch (err) {
console.error("[dn-slip] proxy error:", err);
return NextResponse.json(
{ error: err instanceof Error ? err.message : "slip_proxy_failed" },
{ status: 500 },
);
}
}

View file

@ -64,6 +64,33 @@ export async function POST(request: Request) {
processing_status: "pending",
});
// Create a data_import_batches row so the n8n OCR pipeline (and
// the dashboard's "imports" view) can track this upload through
// pending → processed/failed/dead. The row is linked back to the
// delivery_note via `processed_record_id` so the OCR result route
// can find it without a relation field on delivery_notes.
try {
await pb.collection("data_import_batches").create({
batch_type: "delivery_notes",
content_type: file.type || "application/octet-stream",
status: "pending",
attempts: 0,
date_uploaded: new Date().toISOString(),
uploaded_by: "web-upload",
original_file: file,
processed_record_id: dn.id,
});
} catch (batchErr) {
// Don't fail the upload if the batch row can't be created — the
// DN is already persisted and the OCR pipeline can still pick
// it up via processing_status='pending'. The batch row is just
// a tracking convenience.
console.warn(
"[dn-upload] failed to create data_import_batches row:",
batchErr,
);
}
return NextResponse.json({
id: dn.id,
deliveryNoteNumber: deliveryNoteNumber || "",