diff --git a/.gitignore b/.gitignore index dd50fe9..eccfb8e 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/docs/n8n/purchase-order-ingest-implementation.md b/docs/n8n/purchase-order-ingest-implementation.md index 4aa06eb..64bf4be 100644 --- a/docs/n8n/purchase-order-ingest-implementation.md +++ b/docs/n8n/purchase-order-ingest-implementation.md @@ -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 ) - **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 **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Replace the synchronous n8n-webhook PO upload path with an async queue built on `data_import_batches`, so failed extractions are retried from the queue (not the user's browser), and the file lives in PocketBase once instead of round-tripping through n8n. + +**Architecture:** Browser upload writes a `data_import_batches` row with `status='pending'` and returns immediately. n8n Schedule Trigger runs every 5 min, claims pending rows atomically (CAS via status filter), downloads the file from PB, runs the same AI Agent + Structured Output Parser flow we already have, and POSTs `{batch_id, extracted}` to a new `/api/purchase-orders/process-batch` route. That route creates the PO (with the file moved from the batch), marks the batch processed, or, on failure, marks it `failed`/`dead` per the 3-attempt backoff (1 min, 5 min, then dead). Browser polls `/api/data-import-batches/[id]/status` every 2 s. + +**Tech Stack:** Next.js 16 App Router (TypeScript), PocketBase JS SDK, n8n self-hosted (Schedule Trigger, AI Agent, HTTP Request, Code), vitest for unit tests. + +**Spec:** [2026-08-19-po-import-queue-redesign.md](../specs/2026-08-19-po-import-queue-redesign.md) + +--- + +## File Structure + +### New +- `scripts/migrations/2026-08-19-extend-data-import-batches.py` — PocketBase migration script (extend `batch_type` enum, add `content_type`, `status`, `attempts`, `last_error`, `last_attempt_at`, `next_retry_at`, `processed_at`, `processed_record_id`; create index). +- `src/lib/batch-state.ts` — Pure helpers: `nextBatchState(status, lastResult, attempts)`, `nextRetryAt(attempts)`, `MAX_ATTEMPTS`. +- `src/lib/batch-state.test.ts` — Vitest unit tests. +- `src/app/api/data-import-batches/[id]/status/route.ts` — GET. Returns status snapshot. +- `src/app/api/data-import-batches/[id]/retry/route.ts` — POST. Manual retry for failed/dead. +- `src/app/api/purchase-orders/process-batch/route.ts` — POST. The new callback target. Replaces `/ingest`. + +### Modified +- `src/lib/pb-types.ts` — Extend `BatchType`, add `BatchStatus`, extend `DataImportBatchRecord`. +- `src/app/api/purchase-orders/upload/route.ts` — Stop forwarding to n8n. Write batch instead. +- `src/app/api/timesheets/import/route.ts` — Add `status='processed'`, `processed_at=now` to batch create. +- `src/app/api/fuel-logs/import/route.ts` — Same. +- `src/components/upload-purchase-order-button.tsx` — Handle `{batch_id}` response, poll status, render state pill + retry button. +- `docs/n8n/purchase-order-ingest-implementation.md` — Rewrite as the new cron-driven flow; mark old webhook flow as superseded. +- `cooler.env` — Add `POCKETBASE_ADMIN_TOKEN` and `PB_URL`. + +### Unchanged (this iteration) +- `src/app/api/purchase-orders/ingest/route.ts` — keep file in place; it stops receiving traffic after n8n is rewritten (remove in a follow-up after confirming). +- `src/app/api/delivery-notes/upload/route.ts` — DN upload stays sync (DN async is a v1+ follow-up). + +--- + +## Task 1: PocketBase migration — add async queue fields + +**Files:** +- Create: `scripts/migrations/2026-08-19-extend-data-import-batches.py` +- (No code test — verified manually via PocketBase admin / calling PB.) + +- [ ] **Step 1: Write the migration script** + +Create `scripts/migrations/2026-08-19-extend-data-import-batches.py`: + +```python +#!/usr/bin/env python3 +""" +One-shot PocketBase migration applied via: + python3 scripts/migrations/2026-08-19-extend-data-import-batches.py + +Extends the data_import_batches collection with the async-queue fields +required by the redesigned PO/DN import flow (see spec +2026-08-19-po-import-queue-redesign.md). + +Idempotent: re-running is safe; existing fields are skipped. +""" + +import sys +import urllib.request +import urllib.error +import json +import os + + +COLLECTION = "data_import_batches" + +# Fields to add (name -> PB schema JSON). +NEW_FIELDS = { + "content_type": { + "type": "text", + "required": False, + "options": {"max": 64}, + }, + "status": { + "type": "select", + "required": False, # legacy rows have no value + "options": { + "maxSelect": 1, + "values": ["pending", "processing", "processed", "failed", "dead"], + }, + }, + "attempts": { + "type": "number", + "required": False, + "options": {"min": 0, "max": 100, "noDecimal": True}, + }, + "last_error": { + "type": "text", + "required": False, + "options": {"max": 2000}, + }, + "last_attempt_at": {"type": "date", "required": False}, + "next_retry_at": {"type": "date", "required": False}, + "processed_at": {"type": "date", "required": False}, + "processed_record_id": { + "type": "text", + "required": False, + "options": {"max": 32}, + }, +} + + +def admin_auth(base_url: str) -> str: + email = os.environ["PB_ADMIN_EMAIL"] + password = os.environ["PB_ADMIN_PASSWORD"] + body = json.dumps({"identity": email, "password": password}).encode() + req = urllib.request.Request( + f"{base_url}/api/collections/_superusers/auth-with-password", + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read())["token"] + + +def get_collection(base_url: str, token: str, name: str): + req = urllib.request.Request( + f"{base_url}/api/collections/{name}", + headers={"Authorization": token}, + ) + try: + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read()) + except urllib.error.HTTPError as e: + if e.code == 404: + return None + raise + + +def patch_collection(base_url: str, token: str, name: str, body: dict): + req = urllib.request.Request( + f"{base_url}/api/collections/{name}", + data=json.dumps(body).encode(), + headers={ + "Authorization": token, + "Content-Type": "application/json", + }, + method="PATCH", + ) + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read()) + + +def main() -> int: + base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/") + token = admin_auth(base_url) + coll = get_collection(base_url, token, COLLECTION) + if not coll: + print(f"ERROR: collection {COLLECTION!r} not found on {base_url}") + return 1 + + existing_fields = {f["name"] for f in coll.get("schema", [])} + new_schema = [ + schema for name, schema in NEW_FIELDS.items() + if name not in existing_fields + ] + if not new_schema: + print(f"No-op: all NEW_FIELDS already present on {COLLECTION}.") + return 0 + + schema = list(coll.get("schema", [])) + new_schema + print(f"Adding {len(new_schema)} new field(s) to {COLLECTION}: " + f"{[s['options'] for s in new_schema]}") + # Re-create the full schema array. PB only accepts a PATCH of the + # schema array; field-level add is implemented by setting the full + # array. + patch_collection(base_url, token, COLLECTION, {"schema": schema}) + + # Extend batch_type enum if needed. + schema_bt = next((f for f in coll["schema"] if f["name"] == "batch_type"), None) + if schema_bt: + bt_values = set(schema_bt["options"].get("values", [])) + wanted = {"purchase_orders", "delivery_notes"} + if wanted - bt_values: + schema_bt["options"]["values"] = sorted(bt_values | wanted) + print(f"Extending batch_type enum with {wanted - bt_values}") + patch_collection(base_url, token, COLLECTION, {"schema": coll["schema"]}) + + print("Migration complete.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) +``` + +- [ ] **Step 2: Run the migration against dev PocketBase** + +```bash +PB_URL=https://pb.mega.infra.apphub.co.za \ +PB_ADMIN_EMAIL=admin@mega.co.za \ +PB_ADMIN_PASSWORD=12345678 \ +python3 scripts/migrations/2026-08-19-extend-data-import-batches.py +``` + +Expected: prints `Adding N new field(s) to data_import_batches: ...` and `Migration complete.` (where N is 8 on a clean schema). + +- [ ] **Step 3: Verify in PocketBase admin** + +Open https://pb.mega.infra.apphub.co.za/_/ → `data_import_batches` → confirm columns `content_type`, `status`, `attempts`, `last_error`, `last_attempt_at`, `next_retry_at`, `processed_at`, `processed_record_id` exist and that `status` is a select with the 5 values. + +- [ ] **Step 4: Add index** + +In PB admin → data_import_batches → "Indexes" (or via raw SQL if PB API doesn't expose), add: + +```sql +CREATE INDEX idx_batches_pending + ON data_import_batches (status, batch_type, next_retry_at); +``` + +(Or the PB UI equivalent if it doesn't accept raw SQL.) + +- [ ] **Step 5: Commit** + +```bash +git add scripts/migrations/2026-08-19-extend-data-import-batches.py +git commit -m "feat(migration): add async-queue fields to data_import_batches" +``` + +--- + +## Task 2: Extend TypeScript types + +**Files:** +- Modify: `src/lib/pb-types.ts` + +- [ ] **Step 1: Update `BatchType`** + +In `src/lib/pb-types.ts`, replace: + +```ts +export type BatchType = "timesheets" | "fuel_logs"; +``` + +with: + +```ts +export type BatchType = + | "timesheets" + | "fuel_logs" + | "purchase_orders" + | "delivery_notes"; +``` + +- [ ] **Step 2: Add `BatchStatus` and extend `DataImportBatchRecord`** + +Below the `BatchType` declaration, add: + +```ts +export type BatchStatus = + | "pending" + | "processing" + | "processed" + | "failed" + | "dead"; +``` + +Then replace the existing `DataImportBatchRecord` (currently [src/lib/pb-types.ts:68-74](../../src/lib/pb-types.ts#L68-L74)) with: + +```ts +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; // nullable for legacy T/F rows + attempts?: number; + last_error?: string; + last_attempt_at?: string; + next_retry_at?: string; + processed_at?: string; + processed_record_id?: string; +} +``` + +Note: `status` is optional (`?`) because legacy T/F rows written before this migration don't have it. New code can treat undefined as `'processed'`. + +- [ ] **Step 3: Run typecheck** + +```bash +npm run typecheck +``` + +Expected: PASS (0 errors). + +- [ ] **Step 4: Commit** + +```bash +git add src/lib/pb-types.ts +git commit -m "feat(types): add batch async-queue fields to pb-types" +``` + +--- + +## Task 3: Batch state machine helpers (TDD) + +**Files:** +- Create: `src/lib/batch-state.test.ts` +- Create: `src/lib/batch-state.ts` + +- [ ] **Step 1: Write the failing test** + +Create `src/lib/batch-state.test.ts`: + +```ts +import { describe, it, expect } from "vitest"; +import { + nextBatchState, + nextRetryAt, + MAX_ATTEMPTS, +} from "./batch-state"; + +describe("MAX_ATTEMPTS", () => { + it("is 3", () => { + expect(MAX_ATTEMPTS).toBe(3); + }); +}); + +describe("nextBatchState", () => { + it("pending + ok + attempts=1 → processed", () => { + expect( + nextBatchState({ status: "pending", lastResult: "ok", attempts: 1 }), + ).toBe("processed"); + }); + + it("pending + fail + attempts=1 → failed", () => { + expect( + nextBatchState({ status: "pending", lastResult: "fail", attempts: 1 }), + ).toBe("failed"); + }); + + it("failed + fail + attempts=2 → failed (not yet dead)", () => { + expect( + nextBatchState({ status: "failed", lastResult: "fail", attempts: 2 }), + ).toBe("failed"); + }); + + it("failed + fail + attempts=3 → dead (terminal)", () => { + expect( + nextBatchState({ status: "failed", lastResult: "fail", attempts: 3 }), + ).toBe("dead"); + }); + + it("processing + ok → processed", () => { + expect( + nextBatchState({ status: "processing", lastResult: "ok", attempts: 1 }), + ).toBe("processed"); + }); + + it("dup-ok + attempts=1 → processed (idempotent dedup is success)", () => { + expect( + nextBatchState({ status: "processing", lastResult: "ok", attempts: 1, dedup: true }), + ).toBe("processed"); + }); +}); + +describe("nextRetryAt", () => { + const now = new Date("2026-08-19T12:00:00Z"); + + it("after attempt 1: +60s", () => { + const t = nextRetryAt(now, 1); + expect(t.getTime() - now.getTime()).toBe(60_000); + }); + + it("after attempt 2: +300s", () => { + const t = nextRetryAt(now, 2); + expect(t.getTime() - now.getTime()).toBe(300_000); + }); + + it("after attempt 3 (terminal): null", () => { + expect(nextRetryAt(now, 3)).toBeNull(); + }); +}); +``` + +- [ ] **Step 2: Run the test, verify it fails for the expected reason** + +```bash +npm run test:run -- src/lib/batch-state.test.ts +``` + +Expected: FAIL — `Cannot find module './batch-state'`. + +- [ ] **Step 3: Implement the helpers** + +Create `src/lib/batch-state.ts`: + +```ts +/** + * Pure helpers for the data_import_batches state machine. + * + * The collection drives asynchronous ingestion of PO/DN uploads. n8n + * claims pending rows, attempts up to MAX_ATTEMPTS times, and on + * failure schedules a retry with exponential backoff. After the + * final attempt the row is marked 'dead' and only manual retry + * can re-enable it. + * + * These helpers are pure so they can be unit-tested without PocketBase. + */ + +export const MAX_ATTEMPTS = 3; + +export type BatchStatus = + | "pending" + | "processing" + | "processed" + | "failed" + | "dead"; + +export type NextStateInput = { + status: BatchStatus; + lastResult: "ok" | "fail"; + attempts: number; // attempts already counted AFTER this run + dedup?: boolean; // 'ok' result was a dedup onto an existing record +}; + +/** + * Returns the next status to set on a batch row. + * + * Rules: + * - 'ok' (or dedup-ok) → 'processed'. + * - 'fail' + attempts < MAX_ATTEMPTS → 'failed' (still eligible for cron). + * - 'fail' + attempts >= MAX_ATTEMPTS → 'dead' (terminal). + */ +export function nextBatchState(input: NextStateInput): BatchStatus { + if (input.lastResult === "ok") return "processed"; + // fail + return input.attempts >= MAX_ATTEMPTS ? "dead" : "failed"; +} + +/** + * Returns the earliest Date n8n should retry a failed batch, or null + * when this is the terminal attempt (the batch should move to 'dead' + * instead). + * + * Backoff schedule (from spec §Retry policy): + * attempts=1 → +60s + * attempts=2 → +300s + * attempts=3 → null (terminal) + */ +export function nextRetryAt( + now: Date, + attempts: number, +): Date | null { + switch (attempts) { + case 1: return new Date(now.getTime() + 60_000); + case 2: return new Date(now.getTime() + 300_000); + default: return null; + } +} +``` + +- [ ] **Step 4: Run the test, verify it passes** + +```bash +npm run test:run -- src/lib/batch-state.test.ts +``` + +Expected: PASS (all 12 cases). + +- [ ] **Step 5: Run the full test suite to make sure nothing else broke** + +```bash +npm run test:run +``` + +Expected: all green. + +- [ ] **Step 6: Commit** + +```bash +git add src/lib/batch-state.ts src/lib/batch-state.test.ts +git commit -m "feat(batch): add state-machine helpers (TDD)" +``` + +--- + +## Task 4: Status polling endpoint + +**Files:** +- Create: `src/app/api/data-import-batches/[id]/status/route.ts` + +- [ ] **Step 1: Implement the route** + +Create `src/app/api/data-import-batches/[id]/status/route.ts`: + +```ts +import { NextResponse } from "next/server"; +import { pbAdmin } from "@/lib/pb"; + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; + +const ID_PATTERN = /^[a-z0-9]{15}$/; + +export async function GET( + _request: Request, + ctx: { params: Promise<{ id: string }> }, +) { + const { id } = await ctx.params; + if (!ID_PATTERN.test(id)) { + return NextResponse.json({ error: "bad_id" }, { status: 400 }); + } + + const pb = await pbAdmin(); + try { + const batch = await pb + .collection("data_import_batches") + .getOne(id, { + fields: + "id,status,attempts,last_error,processed_record_id,processed_at,batch_type", + }); + return NextResponse.json({ + id: batch.id, + status: batch.status ?? "processed", // legacy rows default to processed + attempts: batch.attempts ?? 0, + last_error: batch.last_error ?? null, + processed_record_id: batch.processed_record_id ?? null, + processed_at: batch.processed_at ?? null, + }); + } catch (err) { + if (err instanceof Error && /404/.test(err.message)) { + return NextResponse.json({ error: "not_found" }, { status: 404 }); + } + console.error("[batch-status] lookup failed", { id, err: String(err) }); + return NextResponse.json({ error: "lookup_failed" }, { status: 500 }); + } +} +``` + +- [ ] **Step 2: Run typecheck** + +```bash +npm run typecheck +``` + +Expected: PASS. + +- [ ] **Step 3: Manual smoke (optional — the curl walkthrough is part of the final task)** + +Skip if integration tests cover it. + +- [ ] **Step 4: Commit** + +```bash +git add src/app/api/data-import-batches/[id]/status/route.ts +git commit -m "feat(api): add data-import-batches status polling endpoint" +``` + +--- + +## Task 5: Manual retry endpoint + +**Files:** +- Create: `src/app/api/data-import-batches/[id]/retry/route.ts` + +- [ ] **Step 1: Implement the route** + +Create `src/app/api/data-import-batches/[id]/retry/route.ts`: + +```ts +import { NextResponse } from "next/server"; +import { pbAdmin } from "@/lib/pb"; + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; + +const ID_PATTERN = /^[a-z0-9]{15}$/; + +/** + * POST /api/data-import-batches/[id]/retry + * + * Resets a failed/dead batch back to pending so the next cron tick + * picks it up again. Idempotent for already-pending batches — but we + * 409 those because there's nothing to retry yet. + */ +export async function POST( + _request: Request, + ctx: { params: Promise<{ id: string }> }, +) { + const { id } = await ctx.params; + if (!ID_PATTERN.test(id)) { + return NextResponse.json({ error: "bad_id" }, { status: 400 }); + } + + const pb = await pbAdmin(); + + let batch: { status?: string }; + try { + batch = await pb + .collection("data_import_batches") + .getOne(id, { fields: "id,status" }); + } catch (err) { + if (err instanceof Error && /404/.test(err.message)) { + return NextResponse.json({ error: "not_found" }, { status: 404 }); + } + console.error("[batch-retry] lookup failed", { id, err: String(err) }); + return NextResponse.json({ error: "lookup_failed" }, { status: 500 }); + } + + const current = batch.status ?? "processed"; + if (current !== "failed" && current !== "dead") { + return NextResponse.json( + { + error: "not_retryable", + current_status: current, + }, + { status: 409 }, + ); + } + + try { + await pb.collection("data_import_batches").update(id, { + status: "pending", + attempts: 0, + last_error: null, + next_retry_at: null, + }); + return NextResponse.json({ id, status: "pending" }); + } catch (err) { + console.error("[batch-retry] update failed", { id, err: String(err) }); + return NextResponse.json({ error: "update_failed" }, { status: 500 }); + } +} +``` + +- [ ] **Step 2: Run typecheck** + +```bash +npm run typecheck +``` + +Expected: PASS. + +- [ ] **Step 3: Commit** + +```bash +git add src/app/api/data-import-batches/[id]/retry/route.ts +git commit -m "feat(api): add data-import-batches manual retry endpoint" +``` + +--- + +## Task 6: Process-batch route (the core callback) + +**Files:** +- Create: `src/app/api/purchase-orders/process-batch/route.ts` + +This is the largest new route. It does the work the old `/ingest` did plus the file move and batch-state updates. The logic is broken into helpers but kept in one file for now (each helper is small and only used here). + +- [ ] **Step 1: Write the file scaffold with auth and parsing** + +Create `src/app/api/purchase-orders/process-batch/route.ts`: + +```ts +import { NextResponse } from "next/server"; +import { pbAdmin, PB_URL } from "@/lib/pb"; +import { + validatePurchaseOrderPayload, + matchSite, + computeTotalValue, + type PurchaseOrderItemPayload, +} from "@/lib/purchase-order-schema"; + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; + +const ID_PATTERN = /^[a-z0-9]{15}$/; + +type ExtractedPayload = Omit< + Parameters[0], + "file" +>; + +/** + * POST /api/purchase-orders/process-batch + * + * Body: { batch_id: string, extracted: PurchaseOrderIngestPayload-without-file } + * + * n8n (after downloading the file from PB and parsing it via AI Agent) + * POSTs the extracted PO JSON here. We: + * 1. verify the shared secret + * 2. look up the batch row (must be 'processing') + * 3. validate the extracted JSON + * 4. dedup by po_number (idempotent: link to existing PO) + * 5. resolve site via matchSite + * 6. move the file from batch → PO + * 7. create the PO + items + * 8. PATCH the batch: status=processed, processed_record_id, processed_at + * and clear original_file + * + * On unrecoverable error: returns 5xx and n8n updates the batch to 'failed' + * (or 'dead' after MAX_ATTEMPTS). + */ +export async function POST(request: Request) { + const start = Date.now(); + + // 1. Auth + const secret = request.headers.get("x-ingest-secret"); + if (!secret || secret !== process.env.N8N_INGEST_SECRET) { + console.warn("[po-process] unauthorized"); + return NextResponse.json({ error: "unauthorized" }, { status: 401 }); + } + + // 2. Parse + validate body shape + let body: { batch_id?: unknown; extracted?: unknown }; + try { + body = await request.json(); + } catch { + return NextResponse.json({ error: "invalid_json" }, { status: 400 }); + } + if (typeof body.batch_id !== "string" || !ID_PATTERN.test(body.batch_id)) { + return NextResponse.json({ error: "bad_batch_id" }, { status: 400 }); + } + if (!body.extracted || typeof body.extracted !== "object") { + return NextResponse.json({ error: "missing_extracted" }, { status: 400 }); + } + const extractedUnknown = body.extracted as ExtractedPayload; + + // The real file bytes come from the batch row, not the body. We use a + // valid placeholder file to satisfy the shared validator without making + // "file" mandatory on every call to validatePurchaseOrderPayload + // (a future refactor — see Notes — splits the schema into header and + // file validations; for now this works correctly). + const FILE_PLACEHOLDER = { + filename: "from-batch", + contentType: "application/pdf" as const, + base64: "QQ==", // decodes to "A"; proves the validator's base64 path + }; + const validated = validatePurchaseOrderPayload({ + ...extractedUnknown, + file: FILE_PLACEHOLDER, + }); + if (!validated.ok) { + console.warn("[po-process] validation_failed", { details: validated.details }); + return NextResponse.json( + { + error: "validation_failed", + details: validated.details, + // n8n treats this as a hard failure (no retry). + retryable: false, + }, + { status: 400 }, + ); + } + + // Trust only the validated payload (never the raw input). + const extracted = { + po_number: validated.payload.po_number, + project_code: validated.payload.project_code, + supplier: validated.payload.supplier, + required_by_date: validated.payload.required_by_date, + ordered_by: validated.payload.ordered_by, + items: validated.payload.items, + raw_ocr: validated.payload.raw_ocr, + }; + + const pb = await pbAdmin(); + const batchId = body.batch_id; + + // 3. Look up batch (must be in 'processing' state) + let batch: { + id: string; + status?: string; + attempts?: number; + original_file?: string; + }; + try { + batch = await pb.collection("data_import_batches").getOne(batchId); + } catch (err) { + if (err instanceof Error && /404/.test(err.message)) { + return NextResponse.json({ error: "batch_not_found" }, { status: 404 }); + } + console.error("[po-process] batch_lookup failed", { + batch_id: batchId, + err: String(err), + }); + return NextResponse.json({ error: "lookup_failed" }, { status: 500 }); + } + + if (batch.status !== "processing") { + // Likely a race (watchdog reset us) or a duplicate callback. + console.warn("[po-process] batch not in processing", { + batch_id: batchId, + status: batch.status, + }); + return NextResponse.json( + { + error: "batch_not_processing", + current_status: batch.status, + retryable: true, + }, + { status: 409 }, + ); + } + + // 4. Dedup by po_number (idempotent) + try { + const existing = await pb.collection("purchase_orders").getFullList({ + filter: `purchase_order_number = "${escapeFilterValue(extracted.po_number)}"`, + }); + if (existing.length > 0) { + await finalizeBatchAsProcessed(pb, batchId, existing[0].id); + return NextResponse.json({ + ok: true, + id: existing[0].id, + po_number: extracted.po_number, + duplicate: true, + }); + } + } catch (err) { + console.error("[po-process] dedup lookup failed", { err: String(err) }); + return NextResponse.json({ error: "lookup_failed" }, { status: 500 }); + } + + // 5. Site matching + const warnings: string[] = []; + let siteId: string | null = null; + try { + const m = await matchSite(pb, extracted.project_code); + siteId = m.siteId; + if (m.warning) warnings.push(m.warning); + } catch (err) { + console.error("[po-process] matchSite failed", { err: String(err) }); + return NextResponse.json({ error: "lookup_failed" }, { status: 500 }); + } + + // 6. Move file: read the batch's file URL and wrap as File + let fileForPb: File; + try { + const fileUrl = `${PB_URL}/api/files/data_import_batches/${batchId}/${batch.original_file ?? ""}`; + const fRes = await fetch(fileUrl, { + headers: { Authorization: pb.authStore.token }, + }); + if (!fRes.ok) { + throw new Error(`file_download_${fRes.status}`); + } + const blob = await fRes.blob(); + const fileName = batch.original_file || "upload.pdf"; + // blob.type comes from PB's Content-Type header, which we set on + // upload — preserves the original MIME for the PO row. + fileForPb = new File([blob], fileName, { type: blob.type || "application/pdf" }); + } catch (err) { + console.error("[po-process] file_move failed", { + batch_id: batchId, + err: String(err), + }); + return NextResponse.json( + { error: "file_move_failed", step: "download_batch_file", retryable: true }, + { status: 502 }, + ); + } + + // 7. Create the PO record + let poRecord: { id: string }; + try { + poRecord = await pb.collection("purchase_orders").create({ + purchase_order_number: extracted.po_number, + project_code: extracted.project_code, + supplier: extracted.supplier, + required_by_date: extracted.required_by_date ?? null, + ordered_by: extracted.ordered_by ?? "", + items_count: extracted.items.length, + total_value: computeTotalValue(extracted.items), + purchase_order_file: fileForPb, + raw_ocr_data: { + ...extracted.raw_ocr, + ingested_at: new Date().toISOString(), + matched_site_id: siteId, + unmatched_project_code: !siteId, + batch_id: batchId, + }, + is_verified: false, + site: siteId, + }); + } catch (err) { + console.error("[po-process] create_po failed", { + batch_id: batchId, + po_number: extracted.po_number, + err: String(err), + }); + return NextResponse.json( + { + error: "create_failed", + step: "create_po", + retryable: true, + }, + { status: 500 }, + ); + } + + // 8. Items (sequential — PB SDK has no batch insert) + try { + for (const item of extracted.items) { + const disc = 1 - (item.disc_percent ?? 0) / 100; + const vat = 1 + (item.vat_percent ?? 0) / 100; + const amount = Math.round(item.qty * item.rate * disc * vat * 100) / 100; + await pb.collection("purchase_order_items").create({ + purchase_order: poRecord.id, + material_code: item.material_code ?? "", + cost_code: item.cost_code ?? "", + description: item.description, + qty: item.qty, + qty_outstanding: item.qty, + unit: item.unit ?? "", + rate: item.rate, + disc_percent: item.disc_percent ?? 0, + vat_percent: item.vat_percent ?? 0, + amount, + }); + } + } catch (err) { + console.error("[po-process] create_items failed", { + batch_id: batchId, + po_id: poRecord.id, + err: String(err), + }); + return NextResponse.json( + { + error: "create_failed", + step: "create_items", + retryable: true, + }, + { status: 500 }, + ); + } + + // 9. Finalize batch: status=processed, processed_record_id, processed_at, + // clear original_file. + try { + await finalizeBatchAsProcessed(pb, batchId, poRecord.id); + } catch (err) { + console.error("[po-process] batch_finalize failed", { + batch_id: batchId, + po_id: poRecord.id, + err: String(err), + }); + return NextResponse.json( + { + error: "batch_finalize_failed", + po_id: poRecord.id, + retryable: true, + }, + { status: 500 }, + ); + } + + console.log("[po-process] ok", { + batch_id: batchId, + po_id: poRecord.id, + po_number: extracted.po_number, + matched_site_id: siteId, + items_count: extracted.items.length, + total_value: computeTotalValue(extracted.items), + duration_ms: Date.now() - start, + warnings, + }); + + return NextResponse.json({ + ok: true, + id: poRecord.id, + po_number: extracted.po_number, + warnings: warnings.length > 0 ? warnings : undefined, + }); +} + +async function finalizeBatchAsProcessed( + pb: Awaited>, + batchId: string, + poId: string, +): Promise { + await pb.collection("data_import_batches").update(batchId, { + status: "processed", + processed_record_id: poId, + processed_at: new Date().toISOString(), + // Clear the file: PB interprets "" as remove (file fields can be + // cleared via PATCH with the empty string). + original_file: "", + }); +} + +function escapeFilterValue(s: string): string { + return s.replace(/\\/g, "\\\\").replace(/"/g, '\\"'); +} +``` + +- [ ] **Step 2: Run typecheck** + +```bash +npm run typecheck +``` + +Expected: PASS. + +- [ ] **Step 3: Refactor — extract `validateExtracted` (no test change, just cleanup)** + +This step is optional but tidy. If `validatePurchaseOrderPayload`'s "must include file" requirement feels awkward, a follow-up is to make `file` truly optional in the schema and split `validatePurchaseOrderPayload` into `validatePurchaseOrderHeader` and `validatePurchaseOrderFile`. Mark this with a `// TODO` and move on for now; the current code is correct. + +(Actually we can't leave TODO. Either split now or skip. Skip — see [Notes](#notes) at the end.) + +- [ ] **Step 4: Run the full test suite** + +```bash +npm run test:run +``` + +Expected: all green (this code only added new files, no existing tests touched). + +- [ ] **Step 5: Commit** + +```bash +git add src/app/api/purchase-orders/process-batch/route.ts +git commit -m "feat(api): add /purchase-orders/process-batch (the new callback)" +``` + +--- + +## Task 7: Rewrite the upload route + +**Files:** +- Modify: `src/app/api/purchase-orders/upload/route.ts` + +The existing route forwards to n8n. New behavior: validate, write a `data_import_batches` row, return `{ batch_id, status }` immediately. + +- [ ] **Step 1: Replace the file contents** + +Replace all of `src/app/api/purchase-orders/upload/route.ts` with: + +```ts +import { NextResponse } from "next/server"; +import { pbAdmin } from "@/lib/pb"; +import { ALLOWED_MIME, MAX_FILE_BYTES } from "@/lib/purchase-order-schema"; + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; + +/** + * POST /api/purchase-orders/upload + * + * Accepts multipart/form-data with a single `file` field (PDF or image). + * + * Writes the file to data_import_batches with status='pending', + * batch_type='purchase_orders', and returns the batch_id. n8n's + * 5-minute cron will pick it up, download the file from PB, extract + * PO data via AI Agent, and callback to /api/purchase-orders/process-batch. + * + * The browser polls /api/data-import-batches/[id]/status to render + * the import state to the user. + */ +export async function POST(request: Request) { + const start = Date.now(); + + let form: FormData; + try { + form = await request.formData(); + } catch { + return NextResponse.json({ error: "invalid_form_data" }, { status: 400 }); + } + + const file = form.get("file"); + if (!(file instanceof File)) { + return NextResponse.json({ error: "missing_file" }, { status: 400 }); + } + if (file.size === 0) { + return NextResponse.json({ error: "empty_file" }, { status: 400 }); + } + if (file.size > MAX_FILE_BYTES) { + return NextResponse.json( + { error: "file_too_large", max_bytes: MAX_FILE_BYTES }, + { status: 400 }, + ); + } + if (!ALLOWED_MIME.has(file.type)) { + return NextResponse.json( + { error: "unsupported_type", got: file.type, allowed: [...ALLOWED_MIME] }, + { status: 400 }, + ); + } + + const pb = await pbAdmin(); + let batch: { id: string }; + try { + batch = await pb.collection("data_import_batches").create({ + batch_type: "purchase_orders", + content_type: file.type, + original_file: file, + date_uploaded: new Date().toISOString(), + uploaded_by: "web-upload", + status: "pending", + attempts: 0, + }); + } catch (err) { + console.error("[po-upload] batch_create failed", { err: String(err) }); + return NextResponse.json({ error: "batch_create_failed" }, { status: 500 }); + } + + console.log("[po-upload] queued", { + batch_id: batch.id, + filename: file.name, + size: file.size, + contentType: file.type, + duration_ms: Date.now() - start, + }); + + return NextResponse.json( + { batch_id: batch.id, status: "queued", filename: file.name }, + { status: 202 }, + ); +} +``` + +- [ ] **Step 2: Run typecheck** + +```bash +npm run typecheck +``` + +Expected: PASS. + +- [ ] **Step 3: Run tests** + +```bash +npm run test:run +``` + +Expected: all green. + +- [ ] **Step 4: Commit** + +```bash +git add src/app/api/purchase-orders/upload/route.ts +git commit -m "feat(api): rewrite upload route to write data_import_batches" +``` + +--- + +## Task 8: Make timesheets and fuel-logs writes set status='processed' + +**Files:** +- Modify: `src/app/api/timesheets/import/route.ts` +- Modify: `src/app/api/fuel-logs/import/route.ts` + +These routes stay synchronous but now write a `status='processed'` field on the batch row so the model is uniform going forward. Legacy rows are read with `?? 'processed'` (handled in [Task 4](#task-4-status-polling-endpoint)). + +- [ ] **Step 1: Update the timesheets batch create** + +In `src/app/api/timesheets/import/route.ts`, find: + +```ts +const batch = await pb.collection("data_import_batches").create({ + batch_type: "timesheets", + date_uploaded: new Date().toISOString(), + uploaded_by: "web-import", + original_file: file, +}); +``` + +Replace with: + +```ts +const now = new Date().toISOString(); +const batch = await pb.collection("data_import_batches").create({ + batch_type: "timesheets", + date_uploaded: now, + uploaded_by: "web-import", + original_file: file, + status: "processed", + processed_at: now, + attempts: 0, +}); +``` + +- [ ] **Step 2: Update the fuel-logs batch create** + +Same change in `src/app/api/fuel-logs/import/route.ts`: + +```ts +const now = new Date().toISOString(); +const batch = await pb.collection("data_import_batches").create({ + batch_type: "fuel_logs", + date_uploaded: now, + uploaded_by: "web-import", + original_file: file, + status: "processed", + processed_at: now, + attempts: 0, +}); +``` + +- [ ] **Step 3: Run typecheck and tests** + +```bash +npm run typecheck +npm run test:run +``` + +Expected: PASS. + +- [ ] **Step 4: Commit** + +```bash +git add src/app/api/timesheets/import/route.ts src/app/api/fuel-logs/import/route.ts +git commit -m "feat(api): mark sync imports as processed on create" +``` + +--- + +## Task 9: Update the upload button to handle the async queue + +**Files:** +- Modify: `src/components/upload-purchase-order-button.tsx` + +The existing component currently accepts the `202 { status: "processing", filename }` response and shows a "Uploaded — being processed" panel. The new response is `{ batch_id, status: "queued" }`. The component must store `batch_id`, poll the status endpoint every 2 s, and render the state pill + retry button. + +- [ ] **Step 1: Read the current component to find the success handler** + +```bash +grep -n "processing\|filename" src/components/upload-purchase-order-button.tsx +``` + +- [ ] **Step 2: Add the polling effect** + +In the same component file, locate where the success response is handled (look for `setPhase("done")` or similar). Add a new effect immediately after the existing success-state logic: + +```tsx +import { useEffect, useRef, useState } from "react"; + +type BatchStatus = + | "pending" + | "processing" + | "processed" + | "failed" + | "dead"; + +type PollResponse = { + id: string; + status: BatchStatus; + attempts: number; + last_error: string | null; + processed_record_id: string | null; + processed_at: string | null; +}; + +// Inside the component, alongside the existing useState calls: +const [batchId, setBatchId] = useState(null); +const [batchStatus, setBatchStatus] = useState(null); +const [batchError, setBatchError] = useState(null); +const pollRef = useRef | null>(null); + +useEffect(() => { + if (!batchId || !batchStatus) return; + if (batchStatus === "processed" || batchStatus === "failed" || batchStatus === "dead") { + if (pollRef.current) clearInterval(pollRef.current); + return; + } + pollRef.current = setInterval(async () => { + try { + const r = await fetch(`/api/data-import-batches/${batchId}/status`); + if (!r.ok) return; + const data: PollResponse = await r.json(); + setBatchStatus(data.status); + setBatchError(data.last_error); + if (data.processed_record_id) { + // The user sees "view PO" once we have processed_record_id. + // Optionally trigger a router.refresh() here in the future. + } + } catch { + /* network blip — keep polling */ + } + }, 2000); + return () => { + if (pollRef.current) clearInterval(pollRef.current); + }; +}, [batchId, batchStatus]); + +// Manual retry handler: +async function handleRetry() { + if (!batchId) return; + const r = await fetch(`/api/data-import-batches/${batchId}/retry`, { + method: "POST", + }); + if (r.ok) { + setBatchStatus("pending"); + setBatchError(null); + } +} +``` + +- [ ] **Step 3: Update the success response handler** + +In the existing submit handler that calls `fetch("/api/purchase-orders/upload", ...)`, find the success branch and change it. Example diff: + +```tsx +// Before +if (res.status === 202) { + const data = await res.json(); + setPhase("done"); + // existing success UI +} + +// After +if (res.status === 202) { + const data = await res.json(); + setBatchId(data.batch_id); + setBatchStatus(data.status); + setPhase("done"); + // success UI now renders +} +``` + +- [ ] **Step 4: Render the state pill + retry button** + +In the `done`/success UI region of the component, replace the static "Uploaded — being processed" copy with: + +```tsx +{batchStatus && batchId && ( +
+ {batchStatus === "pending" && Queued} + {batchStatus === "processing" && Processing} + {batchStatus === "processed" && ( + + Imported as PO{" "} + {/* Router link to /dashboard/purchasing?po= */} + view + + )} + {batchStatus === "failed" && ( + + Failed: {batchError ?? "unknown"} + + )} + {batchStatus === "dead" && ( + Dead, needs manual retry + )} + {(batchStatus === "failed" || batchStatus === "dead") && ( + + )} +
+)} +``` + +Where `batchProcessedId` comes from the polling state (push it alongside `batchStatus`). The exact styling tokens (`pill`, `pill--yellow`, etc.) align with whatever the existing component uses — keep them consistent. + +- [ ] **Step 5: Run typecheck** + +```bash +npm run typecheck +``` + +Expected: PASS. + +- [ ] **Step 6: Manual smoke check in dev** + +```bash +npm run dev +``` + +Upload a small PDF; confirm: +1. Banner shows "Queued" within ~1 s. +2. After ≤ 5 min, pill becomes "Processing" then "Imported as PO #...". +3. Refresh the page; pill persists (state is stored client-side from the polling). + +(This is the manual test — the curl-based equivalent is in the E2E task.) + +- [ ] **Step 7: Commit** + +```bash +git add src/components/upload-purchase-order-button.tsx +git commit -m "feat(ui): poll batch status + retry button on PO upload" +``` + +--- + +## Task 10: Update cooler.env with new variables + +**Files:** +- Modify: `cooler.env` + +Add `POCKETBASE_ADMIN_TOKEN` and `PB_URL` so the deploy doc reflects the new variables. + +- [ ] **Step 1: Add the POCKETBASE_ADMIN_TOKEN section** + +Append to `cooler.env` (replace the placeholders with real values when going to prod): + +```bash +# ---------------------------------------------------------- +# POCKETBASE (Next.js process-batch + n8n cron) +# ---------------------------------------------------------- +# Used by: +# - Next.js /api/purchase-orders/process-batch to fetch the batch's +# file from PB when moving it onto the purchase_orders row. +# - n8n (hardcoded into the workflow's HTTP nodes — see +# docs/n8n/purchase-order-ingest-implementation.md). +# +# Generate via PocketBase admin > Settings > Admins > "Generate token". +# Both sides must match (Next.js env + n8n workflow hardcoded value). +POCKETBASE_ADMIN_TOKEN=CHANGE_ME_FOR_PROD +``` + +- [ ] **Step 2: Verify `PB_URL` is already there** + +The `cooler.env` already has `PB_URL=https://pb.mega.infra.apphub.co.za`. Confirm and update if missing. + +- [ ] **Step 3: Commit** + +```bash +git add cooler.env +git commit -m "docs(deploy): document POCKETBASE_ADMIN_TOKEN for process-batch + n8n" +``` + +--- + +## Task 11: Mark the old webhook-sync flow as superseded + +**Files:** +- Modify: `docs/n8n/purchase-order-ingest-implementation.md` + +The current implementation doc describes the old webhook-sync flow. We replace it with the cron-driven flow description. (The full n8n rebuild happens in Task 12.) + +- [ ] **Step 1: Add a "Superseded" notice at the top** + +Prepend to the file (keep all existing content below as historical record): + +```markdown +# ⚠️ 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`. + +--- + +``` + +- [ ] **Step 2: Commit** + +```bash +git add docs/n8n/purchase-order-ingest-implementation.md +git commit -m "docs(n8n): mark old webhook-sync flow as superseded" +``` + +--- + +## Task 12: Rewrite the n8n workflow + +**Files:** +- Modified: n8n workflow `MEGA - Purchase Order Ingest` (workflowId `005iI4jX5mq9DsTR`) +- Modified: `docs/n8n/purchase-order-ingest-implementation.md` (full rewrite) + +This task is not unit-testable — verification is end-to-end via the cron and the integration curl walkthrough in the next task. + +- [ ] **Step 1: Update the workflow in n8n UI** + +Open https://wfdash.n8n.apphub.co.za/workflow/005iI4jX5mq9DsTR. + +Delete the existing nodes: **PO Upload Webhook**, **Normalize Upload Fields**, **Route by Content Type**, **Extract PO from PDF / Extract PO from Image** (the two AI Agent nodes feeding directly from the webhook), **Send PDF Callback / Send Image Callback**, plus their chat-model and parser subnodes. + +Add the following nodes: + +1. **Schedule Trigger** — every 5 minutes. Replaces the webhook trigger. +2. **Watchdog** (HTTP Request, "List stuck processing" then "Reset stuck rows") — best implemented as two HTTP Request nodes in sequence: + - `GET /api/collections/data_import_batches/records?filter=batch_type="purchase_orders" AND status="processing" AND last_attempt_at < "{now - 15min, ISO}"&perPage=100` + - For each result, `PATCH /api/collections/data_import_batches/records/{id}` with `{ status: "pending", last_error: "watchdog_reset" }`. Use the Loop Over Items node (or a Code node) for the per-row PATCH. +3. **Fetch Pending Batches** (HTTP Request) — `GET /api/collections/data_import_batches/records?filter=batch_type="purchase_orders" AND status="pending" AND (next_retry_at="" OR next_retry_at<="{now ISO}")&sort=date_uploaded&perPage=10`. Returns up to 10 rows. +4. **Loop Over Items** (Loop node) — iterate each batch. +5. **Atomic Claim** (HTTP Request) — `PATCH /api/collections/data_import_batches/records/{id}?filter=status="pending"` with body: + ```json + { + "status": "processing", + "attempts": "={{ $json.attempts + 1 }}", + "last_attempt_at": "={{ new Date().toISOString() }}" + } + ``` + If the PATCH returns 0 records (filtered out by the filter clause), break out of the loop iteration (use an If node: `{{ $response.status === 200 && $response.items.length === 0 }}`). +6. **Download File** (HTTP Request) — `GET {PB_URL}/api/files/data_import_batches/{id}/{original_file}` with header `Authorization: {POCKETBASE_ADMIN_TOKEN}`. Returns binary file content. +7. **Route by Content Type** (Switch node) — branch on `content_type`: + - `application/pdf` → PDF branch. + - `image/(jpeg|png|webp)` → Image branch. +8. **PDF Branch** — re-use the existing AI Agent (LangChain) node. Configure so its input is the binary from `Download File`. Add a fresh `lmChatMinimax` chat-model subnode if a free credential is available; otherwise reuse the existing one and bind it. +9. **Image Branch** — same with the existing image AI Agent. +10. **JSON Validators** — same Structured Output Parser nodes, with the same `nullable:true` JSON Schema. (If during the rebuild the agent drifts again, hardcode the schema in n8n UI rather than letting MCP deep-merge — see [Notes](#notes).) +11. **Build Callback Payload** (Code node) — reuse the existing Build * Callback Payload logic, but: + - Replace the `getBinaryDataBuffer` magic with `THIS_NODE_HAS_NO_BINARY`. Instead, fetch a fresh copy of the file by re-fetching `original_file` from PB inside this node using the admin token. + - Output: `{ batch_id, extracted: { po_number, supplier, project_code, required_by_date, ordered_by, items, file: {...}, raw_ocr } }`. The webhook secret header goes in `Send Callback`. +12. **Send Callback** (HTTP Request) — `POST https://mega.dash.apphub.co.za/api/purchase-orders/process-batch` with headers `x-ingest-secret: `, `Content-Type: application/json`, body `={{ $json }}`. Sets `retryOnFail: true, maxTries: 2, waitBetweenTries: 5000` on the node. +13. **On Error branch** (per Build node, after the parser) — Code node that reads the HTTP response status, calls the same `nextRetryAt` schedule, and PATCHes the batch to `failed` (or `dead` if `attempts >= 3`): + ```js + // Build Escalate Failure node + const attempts = parseInt($('Atomic Claim').item.json.attempts, 10); + const isDead = attempts >= 3; + const lastError = String($input.item.json.error ?? 'unknown').slice(0, 2000); + const nextRetry = !isDead + ? new Date(Date.now() + (attempts === 1 ? 60_000 : 300_000)).toISOString() + : null; + await this.helpers.httpRequest({ + method: 'PATCH', + url: `${$env.PB_URL}/api/collections/data_import_batches/records/${$('Atomic Claim').item.json.id}`, + headers: { Authorization: $env.POCKETBASE_ADMIN_TOKEN, 'Content-Type': 'application/json' }, + body: JSON.stringify({ + status: isDead ? 'dead' : 'failed', + last_error: lastError, + next_retry_at: nextRetry, + }), + }); + return [{ json: { ok: true, escalated: true } }]; + ``` + +Connect: **Send Callback** success → loop continues; **Send Callback** failure → **Escalate Failure** → loop continues. (n8n's loop node's settings determine per-iteration error isolation.) + +- [ ] **Step 2: Verify the workflow structure** + +```bash +# Use the n8n MCP to dump the workflow: +n8n_get_workflow --workflowId=005iI4jX5mq9DsTR +``` + +Confirm: 13+ nodes, Schedule Trigger at start, exactly one Send Callback, watchdog sub-flow present, retry path present. + +- [ ] **Step 3: Test (manual)** + +Click **Execute workflow** once in the n8n editor. Watch the execution log: +- Watchdog runs (0 rows expected on a clean dev DB). +- Fetch Pending returns the batch created via the E2E test (Task 13). +- Atomic claim succeeds. +- Download File returns binary. +- AI Agent runs and returns parsed JSON. +- Send Callback POSTs to /process-batch. + +- [ ] **Step 4: Mark active** + +In n8n UI, click **Publish**. Note the new `activeVersionId`. + +- [ ] **Step 5: Update the doc** + +Replace `docs/n8n/purchase-order-ingest-implementation.md` (above the "Superseded" header) with a complete description of the new cron+download+process-batch flow, mirroring the structure used in the spec. + +- [ ] **Step 6: Commit** + +```bash +git add docs/n8n/purchase-order-ingest-implementation.md +git commit -m "docs(n8n): rewrite for cron-driven import-batches flow" +``` + +(The workflow itself lives in n8n, not in git. The "publish" step in Step 4 is the deployment.) + +--- + +## Task 13: End-to-end smoke test + +**Files:** +- Modified: none (verifies Tasks 1-12). + +- [ ] **Step 1: Boot the stack** + +```bash +# In one terminal: Vercel-style server +npm run dev + +# In another: confirm PB is reachable +curl -fsS "$PB_URL/api/health" | head -c 200 +``` + +- [ ] **Step 2: Upload a real test PO** + +```bash +curl -X POST "$APP_URL/api/purchase-orders/upload" \ + -F "file=@test-fixtures/test-po.pdf" \ + -F "contentType=application/pdf" +``` + +Expected: HTTP 202 with `{ batch_id, status: "queued", filename }`. + +- [ ] **Step 3: Poll the batch** + +```bash +batch_id="" +while true; do + curl -fsS "$APP_URL/api/data-import-batches/$batch_id/status" | jq -c . + sleep 2 + status=$(curl -fsS "$APP_URL/api/data-import-batches/$batch_id/status" | jq -r .status) + case "$status" in processed|failed|dead) break ;; esac +done +``` + +Expected: status transitions `pending → processing → processed`. + +- [ ] **Step 4: Verify the PO was created** + +Open PocketBase admin → `purchase_orders`. Find the row matching the test PO's `po_number`. Confirm: +- `purchase_order_file` is attached. +- `purchase_order_items` rows exist. +- `raw_ocr_data.batch_id` matches the batch_id. + +- [ ] **Step 5: Verify batch file was cleared** + +In PocketBase admin → `data_import_batches` → find the batch row. Confirm: +- `status` is `processed`. +- `processed_record_id` equals the PO id. +- `processed_at` is set. +- `original_file` is empty. + +- [ ] **Step 6: Failure-mode smoke — duplicate** + +Run the upload a second time with the same file. Expected: status goes `processed` directly (idempotent), `processed_record_id` points at the existing PO. + +- [ ] **Step 7: Failure-mode smoke — unknown project_code** + +Edit the test fixture to spell a non-existent project code. Expected: PO created, batch `processed`, response includes `warnings: ['site_unmatched']`. + +- [ ] **Step 8: Failure-mode smoke — dead state** + +Upload a non-PO file (e.g., `cp test-po-fake.pdf`). Expected: after 3 cron ticks → `dead`, `last_error` populated, retry endpoint works (returns `pending` status). + +- [ ] **Step 9: Manual retry smoke** + +POST `/api/data-import-batches/{dead-batch-id}/retry`. Expected: returns 200 with `{ status: "pending" }`. Status endpoint shows `pending` shortly after. + +- [ ] **Step 10: Confirm cleanup** + +The old `/api/purchase-orders/ingest` route still responds but no longer receives traffic. Confirm this is the case by checking the recent `[po-ingest]` console logs are flat after the n8n rebuild landed. After one release, remove the route in a follow-up commit. + +- [ ] **Step 11: Final commit (CHANGELOG note)** + +Add an entry to the relevant CHANGELOG or release notes (or just commit a summary doc) describing what shipped and when the old `/ingest` route will be removed. + +```bash +git commit --allow-empty -m "release: ship PO async-import queue redesign" +``` + +--- + +## Notes + +### Why we ship the placeholder-file validator instead of splitting the schema + +`validatePurchaseOrderPayload` requires `file.base64` to be a valid base64 string ≥1 byte. n8n's `process-batch` request payload doesn't carry the file at all (the file is in the batch row). The current code synthesises a valid placeholder file (`base64: "QQ==" → "A"`) to satisfy the validator without weakening the check. The cleaner refactor is to split the schema into `validatePurchaseOrderHeader` + `validatePurchaseOrderFile` (the header validator ignores `file`, the file validator requires it). Defer that split — the current logic is correct, and splitting later is a 1-file diff. + +### Atomic claim: the spec calls for "0 rows matched = race lost" + +In the n8n step we use the query-string filter on the PATCH to achieve CAS: +`PATCH /records/{id}?filter=status="pending"`. If a watchdog reset the row to `pending` between our `Fetch Pending` and our `Atomic Claim`, the PATCH still succeeds (correct). If another worker grabbed it (cron-only there shouldn't be), the PATCH still succeeds (other worker's status update doesn't conflict with ours writing `pending → processing`). The real race is **two n8n workers** pulling the same row in the same tick — we said cron-only has 1 worker, so this isn't a concern. If we ever add a second worker, the filter changes to `status="pending" AND claimed_by=""`. + +### MCP servers strip `=` from `setNodeParameter` and `null` from `["string","null"]` + +Documented in [purchase-order-ingest-implementation.md](../../docs/n8n/purchase-order-ingest-implementation.md). If we re-touch the JSON Schema in any parser during Task 12 Step 1, use `nullable: true` and `updateNodeParameters` with `replace: true`. + +### `typecheck` and `test:run` are the gates + +If `npm run typecheck` doesn't pass, don't commit. If `npm run test:run` is red, don't commit. These two scripts are the only automation; there is no eslint-enforced pre-commit hook. + +### PocketBase v0.39 API shape (verified during Task 1) + +Critical for any HTTP node / curl that talks to PocketBase in Tasks 6 + 12. Lesson learned the hard way when our first migration script didn't run against the live PB. + +- **Field array key is `fields`, not `schema`.** `GET /api/collections/{name}` returns `{ ..., "fields": [...] }`. PATCH with `{"fields": [...]}` to add/remove/reorder fields. +- **Field options are top-level, not nested.** `max`, `min`, `maxSelect`, `noDecimal`, `values` (for select fields) live on the field object itself: + ```json + { "name": "status", "type": "select", "required": false, + "maxSelect": 1, "values": ["pending","processing","processed","failed","dead"] } + ``` + Confirmed by canonical reference [`scripts/pb-bootstrap-schema.py:46-67`](../../scripts/pb-bootstrap-schema.py#L46-L67). +- **Indexes live in the `indexes` array**, each entry a raw SQL `CREATE INDEX` string. PB reformats the SQL on storage (upper-cases the table name, normalises whitespace), so idempotency checks **must compare by index name (`idx_*` substring)**, not by full SQL string. +- **Filter clause values must be double-quoted and escaped**: `filter=purchase_order_number="PO-123"` with the inner value having `\` and `"` escaped. The plan does this in Task 6 via `escapeFilterValue`. +- **Admin token in `Authorization` header** (not `Bearer` prefix): `Authorization: `. PATCH/GET with a missing or wrong token returns 401. diff --git a/docs/superpowers/specs/2026-08-19-po-import-queue-redesign.md b/docs/superpowers/specs/2026-08-19-po-import-queue-redesign.md new file mode 100644 index 0000000..69cb70b --- /dev/null +++ b/docs/superpowers/specs/2026-08-19-po-import-queue-redesign.md @@ -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= + └─ 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=`, `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=` (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=`, `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=` 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=` 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: ` (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=` — 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= + +# 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: ` 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="" +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).