MEGAINFRA-WEBAPP/docs/n8n/purchase-order-ingest-implementation.md

Purchase Order Ingest — Implementation Notes

Current recovery behavior — 2026-10-04

The published workflow now selects both pending uploads and failed uploads whose nonempty next_retry_at has elapsed. The old pending-only filter stranded MEGCPO9390 after one attempt even though a retry was scheduled. Date cutoffs use PocketBase's UTC space-separated timestamp format.

PDF transcription includes the complete header and recipient block, in addition to the items table. Supplier means the recipient in To, not Mega Infrastructure in the heading. Missing supplier/project text no longer stops an otherwise readable PO before the app callback.

Before the callback, the workflow looks up the normalized supplier name and creates a missing, readable supplier automatically. Empty/unknown names are skipped. New supplier IDs are the first 15 hexadecimal characters of SHA-256 of po-supplier: plus the normalized name. Concurrent creation conflicts re-read the exact name before continuing. The app callback then resolves supplier_ref normally. The local callback source also implements automatic supplier resolution in src/lib/po-supplier.ts; the live workflow enables this behavior independently of an app deployment.

File-token, download, PDF transcription, structured extraction and image extraction failures now reach Escalate Failure, as do supplier lookup failures. Transient failures remain retryable after attempt three, with backoff of 1, 5, 15 and then 60 minutes. The recovery handler checks the current status, attempt number and claim timestamp before writing. It preserves completed records and app refusals requiring review. The 15-minute processing watchdog remains in place for interrupted runs.

Published version: 0cab4dcc-1147-453d-bb8d-7ac3ac33fb9e. Execution 317539 recovered MEGCPO9390 with seven line items and created KERBING CONSULTANT (bd70ceb64a1f7dc). See docs/audits/2026-10-04-purchase-order-recovery.md for verification. The operation record is scripts/n8n/po-ingest.patch.json; prior affected nodes/connections are retained in docs/n8n/po-ingest-before-2026-10-04.json. Both artifacts omit the callback secret and admin authentication body.

The sections below describe earlier versions; the recovery, supplier and escalation behavior above supersedes their old pending-only/dead-after-three policy.

Cron-driven async queue. Built via n8n MCP on 2026-08-19; repointed at purchase_orders on 2026-09-11 (ingestion restructure, Task 15 — docs/superpowers/plans/2026-09-11-ingestion-restructure.md).

What it does

/api/purchase-orders/upload creates the purchase_orders row itself, before OCR runs, with a placeholder number (PENDING-..., src/lib/po-placeholder.ts) and processing_status: pending. It owns the file and the site from the moment of upload — there is no separate batch row.

Every 3 minutes (the node is named "Every 5 minutes"; it has run at a 3-minute interval since 2026-08-19 — cosmetic mismatch, not fixed here) the workflow polls purchase_orders for processing_status=pending rows, atomically claims one by patching processing_status=processing && attempts+1 (conditional on the filter still matching pending), downloads the file from PocketBase, runs the AI Agent (MiniMax-M3 + Structured Output Parser) to extract the PO header and line items, and POSTs { purchase_order_id, extracted } to /api/purchase-orders/process-batch. That route fills in the existing row — it does not create a new one. See src/app/api/purchase-orders/ process-batch/route.ts for the three outcomes: fill in, merge onto an existing PO and delete this placeholder (real duplicate), or mark this row failed (invalid extraction — never deleted, so it stays visible and recoverable).

What changed on 2026-09-11 and why

data_import_batches is being retired across the whole app, not just this workflow — see the ingestion-restructure plan. For purchase orders specifically:

• The row is born at upload, not at process-batch. purchase_orders now carries the queue state (processing_status, attempts, last_error, next_retry_at, last_attempt_at, file_hash, date_uploaded) that data_import_batches used to carry. Every node that used to target data_import_batches now targets purchase_orders directly.
• Atomic Claim gained the race-guard it never actually had. The workflow's own name for this node, and this doc's prior description, both said "atomic claim" — but the deployed node had no ?filter=processing_status="pending" query parameter at all, just an unconditional PATCH. With one n8n instance and one row fetched per tick this had not caused visible harm, but it was not the race-safe design the docs claimed. Added as part of this repoint.
• The n8n-side duplicate short-circuit was removed. The old Build Callback Payload node queried purchase_orders for an existing PO number itself and, if found, PATCHed the batch row to processed/duplicate_short_circuited directly from n8n code — before ever calling the app. process-batch's merge branch now does the equivalent thing (delete the placeholder, point at the winner) entirely server-side, tested, versioned. Keeping both would have meant the same dedup decision made twice in two different places; n8n's job is narrowed back to "extract and call the API." The Is Duplicate? IF node is gone; Build Callback Payload connects straight to Send Callback.
• The callback field names changed: batch_id → purchase_order_id, and — this one matters — po_number → purchase_order_number. The route's validator (validateExtracted in process-batch/route.ts) requires the field named exactly purchase_order_number; the old n8n code sent po_number, which would have failed validation on every single real upload. po_number_extracted (the un-normalized original string, kept for the audit trail in raw_ocr_data) is unchanged.

The AI Agent's own system prompt still asks the model for a JSON key called po_number — that's the model's output schema, not the app's API contract, and is unaffected. Build Callback Payload is what translates between the two; that's exactly why the field rename lives there and nowhere else.

Workflow

• Workflow ID: 005iI4jX5mq9DsTR
• URL: https://wfdash.n8n.apphub.co.za/workflow/005iI4jX5mq9DsTR
• Trigger: Every 5 minutes (Schedule Trigger, actually 3-minute interval)
• Status: published
• Project: Personal (David Els david.els1@outlook.com)
• Credentials: MiniMax account is auto-bound to the four lmChatMinimax chat-model subnodes.

Node map

Every 5 minutes → Auth PB Admin
                    ├→ Watchdog: List Stuck → Watchdog: Reset Stuck Rows (terminal)
                    └→ Fetch Pending Batches → Loop Over Items
                                                  ├→ Skip Empty Items → Has ID? → Atomic Claim → Claim Successful?
                                                  │                                                            ├ (true)  → Get File Token → Download File → Route by Content Type → AI Agent / AI Agent1
                                                  │                                                            └ (false) → Loop Over Items
                                                  └──────────────────────────────────────────────────────────────────────────────── AI Agent / AI Agent1 → Build Callback Payload → Send Callback
                                                                                                                                                                                                                      ├ (ok)    → Loop Over Items
                                                                                                                                                                                                                      └ (error) → Escalate Failure → Loop Over Items

Node	Type	Purpose
Every 5 minutes	scheduleTrigger	Cron tick (actually every 3 min)
Auth PB Admin	httpRequest (executeOnce)	POST /api/collections/_superusers/auth-with-password → { token, record }. Downstream nodes read $('Auth PB Admin').item.json.token.
Watchdog: List Stuck	httpRequest (executeOnce)	GET purchase_orders where processing_status=processing and last_attempt_at < 15 min ago
Watchdog: Reset Stuck Rows	code (runOnceForAllItems)	PATCH each stuck purchase_orders row back to processing_status=pending with last_error="watchdog_reset"
Fetch Pending Batches	httpRequest (executeOnce)	GET purchase_orders where processing_status=pending and (next_retry_at="" or next_retry_at<now), sorted by date_uploaded, perPage=1
Loop Over Items	splitInBatches (batchSize=1)	Process one row at a time
Skip Empty Items	code (runOnceForAllItems)	Filter the empty wrapper splitInBatches emits on 0 items
Has ID?	if	Route items with id to Atomic Claim; terminate items without
Atomic Claim	httpRequest	PATCH purchase_orders/records/{id}?filter=processing_status="pending" — atomic transition to processing (race-safe; the query filter was added 2026-09-11, see above)
Claim Successful?	if	$json.processing_status === "processing" → Get File Token; otherwise loop back
Get File Token	httpRequest	POST /api/files/token with the PocketBase admin Authorization header. The purchase_order_file field is protected, so the admin header alone cannot download it.
Download File	httpRequest (responseFormat=file)	GET /files/purchase_orders/{id}/{purchase_order_file}?token={file_token} using the token from Get File Token and the id/filename from Claim Successful?
Route by Content Type	switch	PDF branch → AI Agent; image branch → AI Agent1
AI Agent / AI Agent1	langchain agent + parser + MiniMax chat model	Extract PO header + line items per the JSON schema (model-facing key: po_number)
Build Callback Payload	code (runOnceForEachItem)	Unwrap AI Agent output; normalize the PO number; build { purchase_order_id, extracted: { purchase_order_number, ... } }. No duplicate check here anymore — see above.
Send Callback	httpRequest (onError: continueErrorOutput)	POST /api/purchase-orders/process-batch with x-ingest-secret header
Escalate Failure	code (runOnceForEachItem)	On Send Callback error: PATCH the purchase_orders row to failed with next_retry_at (+1min / +5min), or dead after attempt 3. Loop continues.

Hardcoded values (no n8n-side env vars)

Self-hosted n8n on Coolify does not expose $env.* to workflows and has no Variables UI (paid feature only). Every value the workflow needs is baked into the workflow JSON itself.

Where	Value	Notes
Auth PB Admin, Watchdog, Fetch Pending, Atomic Claim, Download File	https://pb.mega.infra.apphub.co.za/...	PB base URL
Auth PB Admin	jsonBody: {"identity":"admin@mega.co.za","password":"<redacted>"}	PB superuser password, in plaintext, in the workflow JSON. Found during the 2026-09-11 repoint; not changed as part of it — rotating a superuser credential is a separate decision. Flagged here so it isn't missed.
Watchdog: List Stuck filter	processing_status="processing" && last_attempt_at<"{{ $now.toUTC().minus({minutes:15}).toFormat('yyyy-MM-dd HH:mm:ss.SSS') }}Z" on purchase_orders	Match PocketBase's space-separated timestamp format; an ISO T cutoff falsely matched fresh rows.
Fetch Pending Batches filter	processing_status="pending" && (next_retry_at="" || next_retry_at<"{{ $now.toISO() }}") on purchase_orders	Respects retry backoff
Atomic Claim filter	processing_status="pending"	The race-guard added 2026-09-11
Send Callback	url: https://mega.dash.apphub.co.za/api/purchase-orders/process-batch	APP_URL hardcoded
Send Callback	x-ingest-secret: b8dffe7b00fca75272bf9a090d15a3c59fa0ec20d6bb232f7bb792db0921ac3b	Mirrors the deployed N8N_INGEST_SECRET. Rotate both sides together.
Escalate Failure	Backoff constants 60000 (1 min) and 300000 (5 min) inline in JS	Adjust in the jsCode
AI Agent / AI Agent1	model: MiniMax-M3	Hardcoded model name in lmChatMinimax subnode

Atomic claim semantics

PATCH /api/collections/purchase_orders/records/{id}?filter=processing_status="pending"
Body: { "processing_status": "processing", "attempts": "={{ $json.attempts + 1 }}", ... }

If the row was already claimed (processing_status != pending), the filter rejects the PATCH and PB returns 0 records. Claim Successful? checks $json.processing_status === "processing" — if not, the loop skips this row. Two cron workers can never claim the same row.

Escalation semantics

Escalate Failure runs only on Send Callback error — not on a validation failure, which process-batch itself marks failed server-side without n8n ever seeing an error response to escalate.

• attempt 1 → processing_status=failed, next_retry_at=now+1min
• attempt 2 → processing_status=failed, next_retry_at=now+5min
• attempt 3 → processing_status=dead, next_retry_at=null
• attempt ≥ 4 → still dead (idempotent)

Fetch Pending Batches's filter respects next_retry_at, so a row with a future next_retry_at is skipped until the backoff elapses.

Watchdog semantics

Runs in parallel with Fetch Pending on every tick; resets any purchase_orders row stuck in processing_status=processing for more than 15 minutes (crash mid-run, n8n restart, killed execution) back to pending. Without this, a crashed row's attempts counter would never advance.

On 2026-09-29, a Cedar Gate upload reached 10 failed claims because Download File returned 404 on its protected PDF, and the watchdog compared the stored space-separated last_attempt_at against an ISO cutoff containing T. PocketBase treated even fresh rows as older than the cutoff. Get File Token and the timestamp format above fixed both issues; the upload then processed as MEGCPO9317 with two line items. A 404 at Download File now points first to a missing/expired file token or a genuinely missing file.

Why the file is not in the callback payload

The route accepts { purchase_order_id, extracted } — no binary. The file already lives on the purchase_order_file field of the row that purchase_order_id points at (it was uploaded there directly, not routed through n8n). This is stricter than the pre-2026-09-11 design, where the file lived on the batch row and process-batch had to move it onto the new PO row it created — there is no move step now, because there is no longer a second row.

PB filter syntax gotcha

PocketBase uses && and ||, not AND/OR (400 on those). The = prefix is required on any field containing {{ ... }} — n8n only evaluates fields that explicitly opt in to templating.

Manual smoke test

Step 1 — Confirm the cron is alive

Open the workflow. The last execution timestamp should advance every few minutes. If not, check Every 5 minutes is enabled and the workflow is Published.

Step 2 — Upload a real PO via the dashboard

1. Visit /dashboard/purchasing.
2. Click Upload Purchase Order, pick a site, and pick a real PDF PO.
3. Expect 202 { purchase_order_id, status: "queued" }. Copy the id.

Step 3 — Watch the cron pick it up

Within a few minutes, the n8n Executions tab should show a run with this trace:

Every 5 minutes → Auth PB Admin → Fetch Pending Batches → Loop Over Items
  → Skip Empty Items → Has ID? → Atomic Claim → Claim Successful?
  → Download File → Route by Content Type → AI Agent → Build Callback Payload
  → Send Callback

Step 4 — Verify the PO row transitioned

In the PB admin UI for purchase_orders, find the row by purchase_order_id. Expect:

Field	Value
purchase_order_number	the real number (no longer PENDING-...)
processing_status	processed
attempts	1
last_error	empty
line items	populated in purchase_order_items, purchase_order pointing at this row

Step 5 — Verify dedup

Upload the same PO file again. Expect a new placeholder row (different hash-dedup window on date_uploaded, or a genuinely different file — the upload route's own file_hash dedup only catches byte-identical re-uploads within 30 days). Once OCR extracts the same purchase_order_number, process-batch's merge branch deletes this second placeholder and the response points at the original row's id. Confirm in PB: only one purchase_orders row for that number, and the second placeholder id no longer exists.

Step 6 — Verify invalid-extraction handling

Upload a file OCR can't read cleanly (e.g. a very low-quality scan) or one that yields no PO number. After the cron runs, expect:

• processing_status = failed (not dead unless it's been retried 3 times)
• last_error mentions the validation failure
• The placeholder row still exists with its file attached — it is never deleted on this path, only on the merge-dedup path.

Step 7 — Cleanup

Delete any test purchase_orders rows (and their purchase_order_items) created during the smoke test.

---

If a step fails, check the n8n Executions tab first — the node trace shows exactly which step died.

• AI Agent timeout → row goes failed → next_retry_at=+1min; the next tick re-attempts. Check the model name in the lmChatMinimax subnode if it keeps timing out.
• Send Callback 4xx → row goes dead after 3 attempts. Check x-ingest-secret matches the deployed N8N_INGEST_SECRET, and that /api/purchase-orders/process-batch is reachable from n8n.
• Send Callback 5xx → row goes failed → re-attempted. Check the Next.js logs for the actual exception.