David/MEGAINFRA-WEBAPP
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

docs: add PO import queue spec/plan, ignore cooler.env, rewrite n8n ingest impl

Browse source 
Captures the async-queue redesign of the PO/DN import pipeline (see
spec + plan) and rewrites the n8n implementation doc to match the
new cron-driven architecture (workflow 005iI4jX5mq9DsTR). Also
gitignores cooler.env so the local-only deployment reference is no
longer tracked.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DEls 2026-08-19 22:03:39 +02:00
parent 34f272f520
commit 84ff4c25b2
4 changed files with 2325 additions and 125 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -8,6 +8,8 @@ next-env.d.ts
.env
.env.local
.env.*.local
# Local-only deployment/secret reference files (do not commit)
cooler.env
public/sw.js
public/sw.js.map
public/workbox-*.js

313
docs/n8n/purchase-order-ingest-implementation.md
View file

@ -1,160 +1,223 @@
# ⚠️ Superseded by the cron-driven queue (2026-08-19)
This document describes the OLD webhook-sync flow, where the browser's
upload forwarded through Next.js to an n8n webhook which then called
back to a synchronous /api/purchase-orders/ingest route. That flow is
no longer the canonical path.
The current path is documented in
[`docs/superpowers/specs/2026-08-19-po-import-queue-redesign.md`](../superpowers/specs/2026-08-19-po-import-queue-redesign.md)
and implemented as a Schedule Trigger + cron-polled
`data_import_batches` queue. The `/api/purchase-orders/ingest` route is
deprecated; the new callback target is
`/api/purchase-orders/process-batch`.
---
# Purchase Order Ingest — Implementation Notes
Built via n8n MCP on 2026-08-19.
Cron-driven async queue. Built via n8n MCP on 2026-08-19.
## What was built
## What it does
A 15-node n8n workflow (`MEGA - Purchase Order Ingest`) that:
Every 5 minutes the workflow polls `data_import_batches` for `batch_type=purchase_orders && status=pending` rows, atomically claims them by patching `status=processing && attempts+1`, downloads the original file from PocketBase, runs the AI Agent (MiniMax-M3 + Structured Output Parser) to extract the PO header and line items, and POSTs the structured payload to `/api/purchase-orders/process-batch` on the dashboard. The dashboard route owns the PB inserts and updates the batch row to `processed`/`failed`/`dead`.
1. Receives a PO file (PDF or image) via webhook (multipart form).
2. Normalizes the multipart form fields into top-level JSON fields and preserves the binary file.
3. Routes by `contentType`:
- `application/pdf` → AI Agent (LangChain) with `lmChatMinimax` + Structured Output Parser — parses PDF via `passthroughBinaryPdfs`
- `image/(jpeg|png|webp)` → AI Agent with vision (`passthroughBinaryImages`) + parser
4. Asks the model for a strict JSON object matching the [Next.js `PurchaseOrderIngestPayload` schema](../superpowers/specs/2026-08-19-purchase-order-upload-design.md) (minus `file`, which is added back from the original binary).
5. Base64-encodes the binary file and POSTs the assembled payload back to Next.js with the shared secret header.
The previous webhook-based synchronous flow is retired — the browser upload no longer forwards to n8n at all.
## Workflow
- **Workflow ID:** `005iI4jX5mq9DsTR`
- **URL:** https://wfdash.n8n.apphub.co.za/workflow/005iI4jX5mq9DsTR
- **Webhook path:** `purchase-order-upload`
- **Production URL:** `https://wfdash.n8n.apphub.co.za/webhook/purchase-order-upload`
- **Test URL:** `https://wfdash.n8n.apphub.co.za/webhook-test/purchase-order-upload`
- **Status:** draft (not yet published — see [Activate](#activate-the-workflow))
- **Trigger:** `Every 5 minutes` (Schedule Trigger)
- **Status:** published (active version is the cron flow)
- **Project:** Personal (David Els <david.els1@outlook.com>)
- **Credentials:** `MiniMax account` is auto-bound to all four `lmChatMinimax` chat-model subnodes (PDF Agent, Image Agent, Parser Model PDF, Parser Model Image).
## Design choices that differ from the handoff doc
## Node map
| Handoff doc said | Built with | Why |
```
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) → 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 |
|---|---|---|
| Tesseract for images, `pdf-parse` for PDFs | MiniMax-M3 via LangChain AI Agent for both branches | n8n has no first-class Tesseract node; the AI Agent reads PDFs natively via `passthroughBinaryPdfs` and images via `passthroughBinaryImages`. One model family, one prompt, simpler pipeline. |
| Two OCR nodes + one parser | AI Agent + Structured Output Parser with `autoFix=true` (model-driven self-correction) | Less moving parts, fewer round-trips. The parser's `autoFix` lets the model retry when it returns malformed JSON. |
| 10s/30s/90s retry on 500/408 | No retry layer yet | v1 assumes the Next.js side responds promptly; dead-letter handling can be added with a second Error Trigger workflow later. |
| Tesseract node / pdf-parse via Function | LangChain `lmChatMinimax` chat-model subnode | Lighter config; structured JSON output enforced by the Structured Output Parser. |
| Every 5 minutes | scheduleTrigger | Cron tick |
| Auth PB Admin | httpRequest (executeOnce) | POST `/api/collections/_superusers/auth-with-password` → returns `{ token, record }`. Downstream nodes reference `$('Auth PB Admin').item.json.token`. |
| Watchdog: List Stuck | httpRequest (executeOnce) | GET batches where `status=processing` and `last_attempt_at < 15 min ago` |
| Watchdog: Reset Stuck Rows | code (runOnceForAllItems) | PATCH each stuck row back to `pending` with `last_error="watchdog_reset"` |
| Fetch Pending Batches | httpRequest (executeOnce) | GET batches where `status=pending` and (`next_retry_at=""` or `next_retry_at<now`), sorted by `date_uploaded`, max 10 |
| Loop Over Items | splitInBatches (batchSize=1) | Process one batch at a time |
| Skip Empty Items | code (runOnceForAllItems) | Filter the empty wrapper that splitInBatches emits when input has 0 items; replace with `{ skipped: true }` |
| Has ID? | if | Route items with `id` to Atomic Claim; terminate items without |
| Atomic Claim | httpRequest | PATCH `/records/{id}?filter=status="pending"` — atomic state transition to `processing` (race-safe) |
| Claim Successful? | if | `status=processing` → Download File; otherwise loop back |
| Download File | httpRequest (responseFormat=file, outputPropertyName=data) | GET `/files/data_import_batches/{id}/{original_file}` — file appears in `$binary.data` |
| 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 |
| Build Callback Payload | code (runOnceForEachItem) | Unwrap AI Agent `output` wrapper; build `{ batch_id, extracted: {...} }`. The file is NOT in the payload — PB has it on the batch row. |
| 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 batch to `failed` with `next_retry_at=+1min` (or `+5min` after attempt 2); 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). The two values the Send Callback nodes need are therefore baked into the workflow JSON itself:
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. Search for these literals if you ever rotate them:
- **`APP_URL`** — hardcoded as the literal `https://mega.dash.apphub.co.za` in the `url` field of both Send PDF Callback and Send Image Callback. If the dashboard base URL ever changes, edit both nodes in the n8n UI (or ask me to update them via MCP).
- **`N8N_INGEST_SECRET`** — currently set to the dev value matching `cooler.env` in the `x-ingest-secret` header value of both Send Callback nodes. Before going live, regenerate both sides (Next.js env + n8n workflow) and keep them in sync. Two options for the n8n side:
- **Inline (simplest):** open the workflow in n8n, click each Send Callback node, paste the new secret into the Headers section. Re-publish.
- **HTTP Header Auth credential (cleaner long-term):** Credentials → New → HTTP Header Auth, name it `x-ingest-secret`, value = the secret. Then ask me to bind it to both Send nodes via `setNodeCredential` so the secret lives in the credential store instead of the workflow JSON.
| 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":"12345678"}` | PB superuser creds — change both before going live |
| Watchdog: List Stuck filter | `batch_type="purchase_orders" && status="processing" && last_attempt_at<"{{ $now.minus({minutes:15}).toISO() }}"` | Stuck-row threshold is 15 min |
| Fetch Pending Batches filter | `batch_type="purchase_orders" && status="pending" && (next_retry_at="" \|\| next_retry_at<"{{ $now.toISO() }}")` | Respects retry backoff |
| Send Callback | `url: https://mega.dash.apphub.co.za/api/purchase-orders/process-batch` | APP_URL hardcoded |
| Send Callback | `x-ingest-secret: b8dffe7b00fca75272bf9a090d15a3c59fa0ec20d6bb232f7bb792db0921ac3b` | Mirrors `cooler.env 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 |
## Next.js side
## Atomic claim semantics
Set `N8N_PO_INGEST_URL` in `.env.local` (or wherever Next.js reads env) to the **Production URL** above. The existing `/api/purchase-orders/upload` route already reads it via the `N8N_PO_INGEST_URL` variable.
Atomic Claim uses PB's filter-on-PATCH trick:
## Activate the workflow
The workflow is currently a draft. To make the production webhook URL live:
1. Open https://wfdash.n8n.apphub.co.za/workflow/005iI4jX5mq9DsTR in n8n.
2. Click **Publish** (top right). The production webhook URL becomes reachable.
For quick iteration during testing, use the **Test URL** instead — it works without publishing. (For local dev, the **Test URL** is fine and matches `N8N_PO_INGEST_URL` in `cooler.env` if you swap to the `/webhook-test/...` variant.)
## End-to-end smoke test (Task 7)
Once activated:
```bash
curl -X POST "$N8N_PO_INGEST_URL" \
-F "file=@/path/to/test-po.pdf" \
-F "filename=test-po.pdf" \
-F "contentType=application/pdf" \
-F "size=$(stat -f%z /path/to/test-po.pdf)"
```
PATCH /api/collections/data_import_batches/records/{id}?filter=status="pending"
Body: { "status": "processing", "attempts": "={{ $json.attempts + 1 }}", ... }
```
Then in n8n's Executions tab, confirm:
If the row was already claimed by another worker (status != pending), the filter rejects the PATCH and PB returns 0 records. Claim Successful? checks `$json.status === "processing"` — if not, the loop skips this batch. Two cron workers can never claim the same row.
1. The webhook receives the file.
2. The Switch routes the item into the PDF branch (`Route by Content Type` → `AI Agent`) or Image branch (`AI Agent1`).
3. The AI Agent returns valid JSON with all required fields (via Structured Output Parser).
4. The Send Callback returns `200` from `/api/purchase-orders/ingest`.
5. The new PO appears in PocketBase `purchase_orders` with matching line items.
## Escalation semantics
Test cases to cover (from handoff doc):
Escalate Failure runs only on Send Callback error. It does **not** run if the AI Agent returned no items (that's a "nothing to do" condition that the process-batch route handles by transitioning the batch to `dead` itself).
- Clean PDF (expect 200)
- Scanned JPG (expect 200)
- PDF with no extractable text (PDF branch should still work — OpenAI handles it)
- Duplicate `po_number` (expect 409 from Next.js, no retry)
- Unknown `project_code` (expect 200 with `warnings: ['site_unmatched']`)
- attempt 1 → status=`failed`, `next_retry_at=now+1min`
- attempt 2 → status=`failed`, `next_retry_at=now+5min`
- attempt 3 → status=`dead`, `next_retry_at=null`
- attempt ≥ 4 → still `dead` (idempotent)
## Binary access in 'separate' binary mode (resolved)
The cron tick's `Fetch Pending Batches` filter respects `next_retry_at`, so a row with `next_retry_at=...future` is skipped until the backoff elapses.
The workflow runs with `binaryMode: 'separate'`, which means n8n stores uploaded files in its own database and passes only metadata objects between nodes. The `item.binary.data` field on items from `Normalize Upload Fields` is the string `"database"` — it is not a Buffer.
## Watchdog semantics
```json
{
"mimeType": "image/jpeg",
"fileType": "image",
"fileExtension": "jpg",
"data": "database",
"id": "database:00843d08-f67e-48f3-8c15-d9d29dfbe599",
"fileSize": "62.4 kB",
"bytes": 62428
}
Watchdog runs in parallel with Fetch Pending on every tick. It picks up rows that have been stuck in `processing` for more than 15 minutes (e.g. crash mid-process, n8n restart, killed execution) and resets them to `pending`. Without this, a crashed row would block its `attempts` counter from advancing forever.
## Why the file is NOT in the callback payload
The new `/api/purchase-orders/process-batch` route accepts just `{ batch_id, extracted: {...} }`. The original file lives on the batch row in PB; the route fetches it from PB on success, copies it to the new `purchase_orders` row, and clears `original_file` from the batch. This keeps payloads small and avoids the binary-passthrough bugs that plagued the old `/api/purchase-orders/ingest` route.
## PB filter syntax gotcha
PocketBase uses `&&` and `||` for logical operators — `AND` and `OR` return 400. Both filters in this workflow use the `&&`/`||` form. The `=` prefix is required on expression fields containing `{{ ... }}` (n8n only evaluates fields that explicitly opt in to templating).
## Known cosmetic warnings
Two `INVALID_PARAMETER` warnings persist:
```
Field "parameters.queryParameters.parameters.1.value" has wrong type. Expected string, but got number.
Field "parameters.queryParameters.parameters.2.value" has wrong type. Expected string, but got number.
```
Calling `Buffer.from(item.binary.data0)` on that object throws `TypeError [ERR_INVALID_ARG_TYPE]: Received an instance of Object`.
The AI Agent also strips the binary from its own output items (even with `passthroughBinaryImages: true`), so `$binary` is empty by the time control reaches `Build * Callback Payload`.
**Workaround used here:** both Build nodes fetch the actual buffer via the Code helper, which loads from the binary store:
```js
const fileBuffer = await this.helpers.getBinaryDataBuffer(
0, // item index on Normalize Upload Fields
'Normalize Upload Fields', // upstream node that preserved the binary
'data0' // the binary property name on that node
);
const base64 = fileBuffer.toString('base64');
```
The webhook fires once per upload (single item), so item index `0` is correct. The Code node's `mode` is `runOnceForEachItem`, but we look up by node name + binary property so we don't depend on the current item index.
If you ever refactor the binary property name on the webhook (`binaryPropertyName`) or the Set node, this key (`'data0'`) needs to be updated to match. The first binary key on `Normalize Upload Fields` is the one to use.
## JSON Schema nullable fields — MCP server pitfall (resolved)
n8n's MCP server has a serialization quirk: when a JSON Schema declares a nullable field as `type: ["string", "null"]` or `oneOf: [{"type":"string"},{"type":null}]`, the MCP server strips the bare `null` and stores an empty string instead. The result is invalid JSON Schema, and the Structured Output Parser throws `"[object Object]" is not valid JSON` at runtime.
**Workaround used here:** the `outputParserStructured` node uses Ajv under the hood, which honors OpenAPI's `nullable: true` extension. For the PDF branch we therefore declare each optional field as:
```json
{
"description": "...",
"type": "string",
"nullable": true
}
```
The image branch still uses `oneOf:[{"type":"string"},{"type":null}]` (set in an earlier MCP round-trip before the mangling pattern was identified) and that pattern survives the wire today, but if a future MCP update strips it too, switch that branch to `nullable:true` for symmetry.
If you need to fully reset the PDF parser's schema, pass `replace: true` on `updateNodeParameters` — otherwise the MCP server does a deep merge and keeps the old broken `oneOf` alongside your new fields.
These fire on the `perPage` query parameter (which the schema expects as a string). n8n auto-coerces numeric values into the URL query string at HTTP time, so the warnings are harmless. Suppressing them would require switching to a string literal; not worth the loss of readability.
## Future work
- **Retry on 500/504/timeout** — wrap the Send Callback nodes in `retryOnFail` settings (per-node `retryOnFail: true, maxTries: 3, waitBetweenTries: ...`) or chain three HTTP nodes with `Wait` nodes between them. The handoff doc's 10s/30s/90s schedule would be added via the wait expression.
- **Dead-letter workflow** — set `workflow.errorWorkflow` on this workflow to a separate "PO Ingest Failures" workflow that posts to Slack.
- **Email ingestion** — see the "Future: email ingestion" section of the handoff doc; the callback contract is identical so the existing `/api/purchase-orders/ingest` route does not need to change.
- **Per-node retry** — Send Callback currently uses `onError: continueErrorOutput` so the error branch can run Escalate Failure. The HTTP node itself doesn't retry. If you want to retry transient network errors within the same execution, set `retryOnFail: true, maxTries: 3, waitBetweenTries: 5000` on Send Callback.
- **Email ingestion** — the callback contract is identical, so adding email attachments to the same queue requires no changes here.
- **Webhook fallback** — if cron latency (up to 5 min) is too long for some flows, add a webhook node that writes the batch row directly to PB and lets the cron pick it up.
## Manual smoke test
The cron fires every 5 minutes — the empty-input path is verified (execution
`55206` succeeded). For the happy path with a real file, run these steps
from the deployed stack. Each step has an expected outcome you can check.
### Step 1 — Confirm the cron is alive
Open [the workflow](https://wfdash.n8n.apphub.co.za/workflow/005iI4jX5mq9DsTR).
The last execution timestamp at top-right should advance within 5 min of
the previous one. If it doesn't, check `Every 5 minutes` is enabled and
the workflow is Published.
### Step 2 — Upload a real PO via the dashboard
1. Visit `/dashboard/purchasing` on the deployed app.
2. Click **Upload Purchase Order** and pick a real PDF PO that has at least
one matching site in the `sites` collection.
3. Expect a 202 + `{ batch_id, status: "pending" }` toast within a couple
seconds. Copy the `batch_id`.
### Step 3 — Watch the cron pick it up
Wait up to 5 min. In the n8n workflow's **Executions** tab, a new run
should appear with the following node 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 batch row transitioned
In the PB admin UI for `data_import_batches`, find the row by `batch_id`.
Expect:
| Field | Value |
|---|---|
| `status` | `processed` |
| `attempts` | `1` |
| `processed_at` | ISO timestamp from the run |
| `processed_record_id` | new `purchase_orders` row id |
| `last_error` | empty |
| `original_file` | empty (file moved off the batch) |
A matching `purchase_orders` row should exist with `original_file` pointing
to the same filename, and the PO header + line items populated.
### Step 5 — Verify dedup
Upload the **same** PO file again. Expect either:
- The upload route returns 409 with the existing `purchase_orders.id`, or
- A new batch row is created but the AI Agent extracts the same
`po_number`, and `process-batch` either links to the existing PO or
rejects with `duplicate`.
(Exact behavior is set in Task 7 — check the route's source if unsure.)
### Step 6 — Verify dead-state on unmatched site
Upload a PO whose site name doesn't match any row in the `sites`
collection. After the cron runs, expect:
- `status = dead`
- `attempts = 3`
- `next_retry_at = null`
- `last_error` mentions "site not matched" (or similar)
- No `purchase_orders` row created
### Step 7 — Verify retry UI
In `/dashboard/purchasing` or wherever the batch list lives, find the
`dead` row from Step 6. Expect a **Retry** button. Clicking it sets
`status=pending` and `attempts=0` (or whatever the route contract is —
see the retry route source).
### Step 8 — Verify manual retry runs the cron
Within 5 min of clicking Retry, the row should re-enter `processing` and
either transition to `processed` or back to `failed`/`dead` per the
outcome. Check the **Executions** tab for a new run that includes the
batch id.
### Step 9 — Cleanup
Delete any test `purchase_orders` rows created during the smoke test.
Delete any test `data_import_batches` rows that don't have
`status=processed` and a real `processed_record_id`.
---
If any step fails, check the **Executions** tab in n8n first — the node
trace shows exactly which step died. The most common failure modes:
- **AI Agent timeout** → row goes `failed` → `next_retry_at=+1min`; the
next cron tick re-attempts. If it keeps timing out, check the model
name in the `lmChatMinimax` subnode.
- **Send Callback 4xx** → row goes `dead` after 3 attempts. Check
`x-ingest-secret` matches `cooler.env 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.

1682
docs/superpowers/plans/2026-08-19-po-import-queue-implementation.md Normal file
View file

File diff suppressed because it is too large Load diff

453
docs/superpowers/specs/2026-08-19-po-import-queue-redesign.md Normal file
View file

@ -0,0 +1,453 @@
# PO/DN Import Queue Redesign — Design
**Date:** 2026-08-19
**Status:** Approved (pending implementation)
**Owner:** DELs
**Supersedes:** [2026-08-19-purchase-order-upload-design.md](2026-08-19-purchase-order-upload-design.md) for the PO upload path. The old webhook-sync flow stays in git history but the canonical PO ingest path becomes the queue documented here.
## Goal
Replace the synchronous browser → Next.js → n8n → Next.js → PocketBase upload path with an async queue built on the existing `data_import_batches` collection. The browser returns immediately, n8n processes on a 5-minute cron, and retries happen by re-processing the same row instead of asking the user to re-upload.
The same queue serves both `purchase_orders` and (future) `delivery_notes` uploads. Timesheets and fuel-logs CSV imports keep their synchronous path — fixed-format CSV needs no AI and no retries.
## Non-goals (v1)
- Email ingestion (existing callback can be re-pointed at this endpoint later).
- Realtime/SSE UI updates — 2 s polling is fine.
- Slack/email alerting on `dead` status — dashboard visibility only for now.
- Migrating timesheets/fuel-logs to the async path — they stay sync. Their batch rows are still written but `status` is set to `processed` on create.
- Bulk multi-file uploads — one file per upload.
## Architecture
```
Browser ──multipart──► Next.js POST /api/purchase-orders/upload
│
├─ validate (MIME, size)
├─ write data_import_batches row (status=pending)
│ batch_type='purchase_orders' ← set by button
│ content_type='image/jpeg' ← set by file.type
│ original_file=<bytes>
└─ return 202 { batch_id, status: "queued" }
Browser stores batch_id, polls /api/data-import-batches/[id]/status every 2s
┌──────────── 5-min n8n Schedule Trigger ────────────┐
│ │
▼ │
1. watchdog: reset stuck 'processing' (>15min) → 'pending' │
│ │
▼ │
2. GET pending batches (filter status IN pending │
│ AND batch_type='purchase_orders') │
│ │
└─► for each batch: │
│ │
├─► atomic claim: │
│ PATCH status='processing' │
│ attempts=attempts+1 │
│ filter: id=X AND status IN (pending, failed)│
│ (skip if 0 rows matched = race lost) │
│ │
├─► download file: GET /api/files/... │
│ (admin-token auth) │
│ │
├─► AI Agent route by batch.content_type │
│ → extracts PO JSON │
│ │
└─► POST /api/purchase-orders/process-batch │
{ batch_id, extracted } │
│ │
▼ │
┌──────────────────────────────────┐ │
│ Next.js /process-batch │ │
│ │ │
│ • verify secret │ │
│ • validate extracted │ │
│ • dedup po_number (idempotent) │ │
│ • matchSite(project_code) │ │
│ • create PO + items │ │
│ • move file: batch → PO │ │
│ • PATCH batch: │ │
│ status='processed' │ │
│ processed_record_id │ │
│ processed_at │ │
│ original_file='' │ │
└──────────────────────────────────┘ │
│
────────────────────────────────────────────────────────────┘
```
### Why this is better than what we have
- **No file round-trip.** The file lives in PocketBase once. No browser→n8n base64, no `Buffer.from` of binary metadata, no `getBinaryDataBuffer` 3-arg misroute. The whole class of bugs we just debugged (executions 54311, 54392) is gone.
- **Retry by row, not by user.** Failed extraction does not lose the file. The same batch can be retried (auto by cron, manual by button) without the user re-uploading anything.
- **Failure observable.** `data_import_batches` is filterable by `status` and `last_error`. Open PocketBase admin and see exactly what's stuck.
- **Reusable.** Same queue hosts delivery_notes later; same n8n cron hosts more doc types later.
## Step-by-step
### 1. Upload (`POST /api/purchase-orders/upload`)
- MIME check on `file.type` (PDF/JPG/PNG/WEBP).
- Size check on `file.size` (≤ 20 MB).
- Create `data_import_batches` row:
```ts
{
batch_type: "purchase_orders",
content_type: file.type,
original_file: file,
date_uploaded: new Date().toISOString(),
uploaded_by: "web-upload",
status: "pending",
attempts: 0,
}
```
- Response: `202 { batch_id, status: "queued" }` — no waiting on n8n.
### 2. UI
- `UploadPurchaseOrderButton` stores `batch_id` in component state on success.
- New light polling component (or in-dialog progress): `GET /api/data-import-batches/[id]/status` every 2 s.
- Status transitions surface as `queued → processing → imported (with PO id) | failed (with error) | dead (with "retry" button)`.
- Each terminal state has its own panel copy; clicks navigate to the new PO.
### 3. Cron (n8n Schedule Trigger, every 5 min)
**Step A — Watchdog first.** PATCH any batch where `status='processing' AND last_attempt_at < now() - 15min` to `status='pending'`. Recovers from n8n crashes mid-run.
**Step B — Fetch pending.** `GET /api/collections/data_import_batches/records` with filter:
```
batch_type="purchase_orders" AND status="pending" AND (next_retry_at="" OR next_retry_at<=NOW())
sort=date_uploaded asc, perPage=10
```
**Step C — Per batch.** For each row:
1. **Atomic claim.** PATCH the batch, then filter by the current status to detect races. If 0 rows match → skip.
2. **Download the file.** `GET /api/files/data_import_batches/{id}/{filename}` with the admin token header.
3. **AI Agent.** Route by `content_type`:
- `application/pdf` → AI Agent PDF branch (existing passthrough binary).
- `image/(jpeg|png|webp)` → AI Agent image branch.
Same LangChain Structured Output Parser as today, autoFix on for retry within the model.
4. **Callback.** `POST https://mega.dash.apphub.co.za/api/purchase-orders/process-batch` with `{ batch_id, extracted }`. Same `x-ingest-secret` header convention.
5. **Mark outcome.** Based on the HTTP response:
- `200` → success terminal; n8n does nothing else (Next.js updated the batch).
- `4xx/5xx` → n8n PATCHes batch: `status='failed'`, `last_error=<truncated message>`, `next_retry_at=now() + backoff(attempts)`. If `attempts >= 3` → `status='dead'`.
### 4. Process route (`POST /api/purchase-orders/process-batch`)
- Verify `x-ingest-secret` matches `process.env.N8N_INGEST_SECRET`.
- Validate `{ batch_id, extracted }`:
- `batch_id` is a PB record id.
- `extracted` matches `PurchaseOrderIngestPayload` minus `file` (file comes from the batch).
- Fetch batch from PB; if `status !== 'processing'` → 409 conflict (race lost). n8n will log and move on.
- **Dedup.** If `purchase_orders.purchase_order_number = extracted.po_number` exists → mark batch `processed` with `processed_record_id=<existing>` (idempotent — "already imported"), return 200.
- Resolve site via existing `matchSite(project_code)`. Track warnings.
- **Move the file.** Read the batch's `original_file` URL with admin auth, wrap as `File`, attach to the new PO row, then clear `original_file` from the batch on success.
- Create `purchase_orders` row (file attached, denormalized fields set, `is_verified=false`).
- Sequential `for` loop to create `purchase_order_items` (same as today).
- PATCH batch: `status='processed'`, `processed_record_id=<po.id>`, `processed_at=now()`, `original_file=""`.
- Return `{ ok: true, po_id, warnings? }` to n8n.
## Data model
### Schema changes (PocketBase)
`data_import_batches` (existing collection, see [src/lib/pb-types.ts:68-74](../../src/lib/pb-types.ts#L68-L74)) gets the following fields added:
| Field | Type | Notes |
|---|---|---|
| `batch_type` | select enum | existing — extend to include `purchase_orders` and `delivery_notes` |
| `content_type` | text (nullable) | NEW — set by upload route; null for legacy T/F rows |
| `original_file` | file | existing — cleared on `processed` |
| `date_uploaded` | datetime | existing |
| `uploaded_by` | text | existing |
| `status` | select enum | NEW — `pending` \| `processing` \| `processed` \| `failed` \| `dead` |
| `attempts` | number | NEW — int, default 0 |
| `last_error` | text (nullable) | NEW — last error message, truncate to 2000 chars |
| `last_attempt_at` | datetime (nullable) | NEW |
| `next_retry_at` | datetime (nullable) | NEW — earliest next eligible attempt |
| `processed_at` | datetime (nullable) | NEW — set on transition to `processed` |
| `processed_record_id` | text (nullable) | NEW — id of the PO/DN row created |
New index (PB side, via admin UI or `pb_bootstrap`):
```
CREATE INDEX idx_batches_pending
ON data_import_batches (status, batch_type, next_retry_at)
```
`purchase_orders` — see [src/lib/pb-types.ts:76-90](../../src/lib/pb-types.ts#L76-L90). No schema change; we keep writing `purchase_order_file`, `raw_ocr_data`, `is_verified`, `site` as today.
### TypeScript types
[src/lib/pb-types.ts](../../src/lib/pb-types.ts) grows:
```typescript
export type BatchType =
| "timesheets"
| "fuel_logs"
| "purchase_orders"
| "delivery_notes";
export type BatchStatus =
| "pending"
| "processing"
| "processed"
| "failed"
| "dead";
export interface DataImportBatchRecord extends RecordModel {
collectionName: "data_import_batches";
batch_type: BatchType;
content_type?: string;
original_file: string;
date_uploaded: string;
uploaded_by: string;
status: BatchStatus;
attempts: number;
last_error?: string;
last_attempt_at?: string;
next_retry_at?: string;
processed_at?: string;
processed_record_id?: string;
}
```
### State machine
```
upload
│
▼
┌────────────┐
│ pending │ ◄────────────┐
└──────┬──────┘ │
│ cron picks up │
▼ │
┌──────────────┐ │
│ processing │ │
└─┬────────┬───┘ │
│ │ │
ok fail │
│ │ attempts<3 │
│ └────────────────►┤ status=failed
▼ │ next_retry_at=+backoff(attempts)
┌────────────┐ │
│ processed │ │
└────────────┘ │
│ attempts==3
▼
┌────────────┐
│ dead │ (manual fix + retry button)
└────────────┘
```
### Retry policy
| Attempts left | Wait on failure |
|---|---|
| 2 (after attempt 1 fails) | 1 min |
| 1 (after attempt 2 fails) | 5 min |
| 0 (after attempt 3 fails) | terminal → `dead` |
Worst-case time from upload to `dead` for a poison-pill file: **~6 min** + 3 × AI runtime.
### Backwards compatibility
- Existing timesheets/fuel-logs routes get a one-line change: add `status='processed'` and `processed_at=<now>` on row creation, so all batch rows look uniform in the future. Old rows without `status` are read with `?? 'processed'` fallback so legacy data still renders.
- Existing `purchase_orders` rows created by the old webhook-sync flow are unaffected.
## Files
### New
| File | Purpose |
|---|---|
| `src/app/api/data-import-batches/[id]/status/route.ts` | `GET` returns `{status, last_error, attempts, processed_record_id}`. Used by the upload button for polling. |
| `src/app/api/data-import-batches/[id]/retry/route.ts` | `POST` resets `failed/dead` batch to `pending`, `attempts=0`, `next_retry_at=null`. Used by the manual retry button. |
| `src/app/api/purchase-orders/process-batch/route.ts` | Replaces `/api/purchase-orders/ingest`. Takes `{batch_id, extracted}`, performs the create + status update. |
| `src/lib/batch-state.ts` | Pure functions: `nextBatchState(status, lastResult, attempts) → status` and `nextRetryAt(attempts) → Date`. Unit-tested in isolation. |
| `scripts/migrations/2026-08-19-extend-data-import-batches.py` | One-shot migration applied via PocketBase admin / `pb_bootstrap_schema.py`. Adds new fields and the index to `data_import_batches`. |
### Modified
| File | Change |
|---|---|
| `src/app/api/purchase-orders/upload/route.ts` | Stops forwarding to n8n. Writes the batch row, returns `{batch_id}`. |
| `src/app/api/timesheets/import/route.ts` | Add `status='processed'`, `processed_at=<now>` on `data_import_batches` create. Behavior unchanged. |
| `src/app/api/fuel-logs/import/route.ts` | Same — add `status='processed'` on batch create. |
| `src/lib/pb-types.ts` | Extend `BatchType`, add `BatchStatus`, extend `DataImportBatchRecord`. |
| `src/components/upload-purchase-order-button.tsx` | On `202` response, store `batch_id`, show queued panel, poll status every 2 s until terminal. Render `processed` → link to PO; `failed` → show error; `dead` → show error + retry button. |
| `docs/n8n/purchase-order-ingest-implementation.md` | Rewrite the n8n handoff doc to describe the cron + download-from-PB + process-batch flow. Marks the old webhook-sync flow as superseded. |
| `cooler.env` (deployment reference) | Add `POCKETBASE_ADMIN_TOKEN` (server-only, used by the new process-batch and the n8n cron to call PB). Keep existing `N8N_INGEST_SECRET` (now shared between upload route, process-batch, and the n8n callback). |
### Unchanged
- `src/app/api/purchase-orders/ingest/route.ts` — **deprecated**, kept around for one release in case anything still hits it; document the deprecation in the file header. Remove in a follow-up once we confirm nothing hits it.
- `src/components/upload-delivery-note-button.tsx` — stays sync for now; future work migrates it to the same queue.
## UX
### Upload flow (`UploadPurchaseOrderButton`)
After the user picks a file and submits:
| Phase | What user sees |
|---|---|
| Submitting (≤ 1 s) | Spinner over the upload button: "Uploading…" |
| Queued | Toast/banner: "Uploaded — processing." Stays open. Shows a small live status pill: `Queued` (yellow) → `Processing` (blue) → `Imported as PO #MEGCP09199` (green, link to PO) — or — `Failed: <last_error>` (red, retry button) — or — `Dead, needs manual retry` (red, retry button). |
| Terminal | Status pill becomes the only state. User clicks "Done" or navigates away. |
Polling stops on any terminal state. Each new upload starts fresh.
### Dashboard
Existing `/dashboard/purchasing` view is `force-dynamic`; on next navigation the new PO renders. No realtime needed.
## Error handling
### Failure modes and recovery
| Where it fails | What happens | Batch state | Recovery |
|---|---|---|---|
| n8n AI Agent throws parsing the file | n8n catches, PATCHes batch to `failed` | `failed`, `last_error` populated | Cron re-runs after backoff |
| `process-batch` validation rejects `extracted` | 400 → n8n marks batch failed | `failed`, `last_error` = validation detail | Cron re-runs; AI autoFix tightens output |
| `po_number` already exists | Mark `processed` with `processed_record_id=<existing>` — not an error | `processed` (idempotent) | None — UI shows "already imported" |
| PocketBase unreachable from n8n during claim | n8n throws; no batch state change | `pending` | Next cron tick |
| `process-batch` throws mid-write (e.g., PB timeout) | n8n sees connection error, marks failed | `processing` → watchdog resets → `pending` | Watchdog + cron re-run. Acceptable: if items table missed some rows, a retry will discover and recreate (idempotent on dedup). |
| n8n crashes mid-processing (workflow dies) | Cron watchdog finds `processing > 15min` and resets | reset → `pending` | Next cron tick |
| `attempts == 3` failure | n8n PATCHes `status='dead'` | `dead` | Manual button only |
### Manual retry
- `POST /api/data-import-batches/[id]/retry` resets `status='pending'`, `next_retry_at=null`, `attempts=0`, clears `last_error`. Cron picks it up next tick.
- The dashboard's "Retry" button is shown for `status IN ('failed', 'dead')`.
### Notifications
- **v1**: dashboard visibility only — failed/dead batches surface in PocketBase admin and on the import panel. No external alerting.
- **v2** (later): Slack/email on `dead` transitions. Few per week when something is genuinely broken.
### Observability
Console logs (existing pattern):
- `[po-upload]` success: `{ batch_id, filename, size, contentType, duration_ms }`.
- `[po-process-batch]` success: `{ batch_id, po_number, matched_site_id, items_count, total_value, duration_ms, warnings }`.
- `[po-process-batch]` failure: `{ batch_id, step, error }`.
- `[batches-watchdog]` if any rows were reset this tick: `{ count }`.
## Environment variables
Existing:
```
N8N_INGEST_SECRET=<32-byte-hex>
```
New (server-only, no `NEXT_PUBLIC_` prefix):
```
# PocketBase admin token used by Next.js (server-side) to read/write the
# data_import_batches collection and to download the batch's file when
# moving it onto the purchase_orders record.
POCKETBASE_ADMIN_TOKEN=<token-from-pocketbase-admin>
# PocketBase public URL — already exists via NEXT_PUBLIC_POCKETBASE_URL,
# but server-side code needs to read it as PB_URL too (no NEXT_PUBLIC on
# the import process-batch route). Re-use the same value.
PB_URL=https://pb.mega.infra.apphub.co.za
# Shared secret between n8n and the upload route / process-batch route.
# Already exists as N8N_INGEST_SECRET; no new value added.
```
n8n-side (hardcoded into the workflow nodes, since self-hosted Coolify n8n has no `$env.*`):
- `PB_URL` (matches above) — hardcoded in HTTP nodes that talk to PocketBase.
- `POCKETBASE_ADMIN_TOKEN` — hardcoded in the `Authorization: <token>` header on n8n's PB HTTP nodes.
- `APP_URL` (`https://mega.dash.apphub.co.za`) — already hardcoded in Send Callback nodes, no change.
- `N8N_INGEST_SECRET` (`x-ingest-secret` header value) — already hardcoded in Send Callback nodes, no change.
## Testing
### Unit / route tests (Vitest)
- **`batch-state.ts`** — `nextBatchState`:
- `pending + ok + attempts=1` → `processed`.
- `pending + fail + attempts=1` → `failed`, nextRetryAt = now+1min.
- `failed + fail + attempts=2` → `failed`, nextRetryAt = now+5min.
- `failed + fail + attempts=3` → `dead`.
- `processing + ok` → `processed`.
- Covers the whole state machine in one focused table.
- **`batch-state.ts`** — `nextRetryAt`:
- attempts=1 → +60 000 ms.
- attempts=2 → +300 000 ms.
- attempts=3 → undefined (terminal).
- **`/api/data-import-batches/[id]/status/route.ts`**:
- Existing batch → returns status, last_error, attempts, processed_record_id.
- Missing batch → 404.
- Bad batch id format → 400.
- **`/api/data-import-batches/[id]/retry/route.ts`**:
- `failed` batch → reset to `pending`, `attempts=0`.
- `dead` batch → same.
- `processed` batch → 409 (already done).
- `pending` batch → 409 (nothing to retry).
- **`/api/purchase-orders/process-batch/route.ts`**:
- Bad secret → 401.
- Bad extracted payload → 400 with details.
- Bad batch_id → 404.
- Status not `processing` → 409.
- Duplicate `po_number` → 200, marks batch `processed` with `processed_record_id` of existing PO.
- Unknown `project_code` → 200 with `warnings: ['site_unmatched']`.
- **`/api/purchase-orders/upload/route.ts`** (rewritten):
- Bad MIME → 400.
- Oversized → 400.
- Happy path → creates batch with `status='pending'`, returns 202 + `batch_id`.
- PB write failure → 500.
### Integration / smoke tests (manual, post-deploy)
```bash
# 1. Upload a real PO PDF
curl -X POST "$APP_URL/api/purchase-orders/upload" \
-F "file=@test-fixtures/test-po.pdf" \
-F "contentType=application/pdf"
# 2. Confirm batch row exists with status=pending
batch_id="<from response>"
curl "$APP_URL/api/data-import-batches/$batch_id/status"
# 3. Trigger an out-of-cron run (manual n8n run)
# (click "Execute workflow" in n8n UI for the process-import-batches workflow)
# 4. Watch status transition: pending → processing → processed
# 5. Verify PO created, file attached, batch has processed_record_id,
# original_file on batch cleared
```
### Failure-case smoke (one curl per row in the table above)
| Scenario | Expected |
|---|---|
| Upload a known-bad PDF (random non-PO file) | After 3 cron ticks → `dead`, `last_error` populated, retry button enabled |
| Upload a PDF twice | First goes `processed`; second goes `processed` with `processed_record_id` pointing to the existing PO |
| Upload with unknown `project_code` | PO created, batch `processed`, response includes `warnings: ['site_unmatched']` |
| Upload then leave n8n off for 30 min, then turn on | Watchdog finds no `processing` rows; pending batch is picked up on next 5-min tick |
| Simulate n8n crash mid-AI-Agent: kill workflow between claim and callback | After 15 min, watchdog resets batch to `pending`; next cron tick re-processes |
### Out of test scope (v1)
- AI extraction accuracy (n8n's territory; autoFix covers most cases).
- Load test (cron processes 10 at a time, uploads << 1/min — thousand-x headroom).
- Realtime UI (not building).
- Slack/email alerting (v2).