MEGAINFRA-WEBAPP/docs/superpowers/specs/2026-08-19-po-import-queue-redesign.md

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 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:

  {
    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) 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. No schema change; we keep writing purchase_order_file, raw_ocr_data, is_verified, site as today.

TypeScript types

src/lib/pb-types.ts grows:

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)

# 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).