MEGAINFRA-WEBAPP/docs/superpowers/plans/2026-08-19-po-import-queue-implementation.md

# PO Import Queue Redesign — Implementation Plan

> **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<typeof validatePurchaseOrderPayload>[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<ReturnType<typeof pbAdmin>>,
  batchId: string,
  poId: string,
): Promise<void> {
  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<string | null>(null);
const [batchStatus, setBatchStatus] = useState<BatchStatus | null>(null);
const [batchError, setBatchError] = useState<string | null>(null);
const pollRef = useRef<ReturnType<typeof setInterval> | 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 <BatchStatusPill ... />
}
```

- [ ] **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 && (
  <div data-testid="batch-status">
    {batchStatus === "pending" && <span className="pill pill--yellow">Queued</span>}
    {batchStatus === "processing" && <span className="pill pill--blue">Processing</span>}
    {batchStatus === "processed" && (
      <span className="pill pill--green">
        Imported as PO{" "}
        {/* Router link to /dashboard/purchasing?po=<processed_record_id> */}
        <a href={`/dashboard/purchasing?po=${batchProcessedId}`}>view</a>
      </span>
    )}
    {batchStatus === "failed" && (
      <span className="pill pill--red">
        Failed: {batchError ?? "unknown"}
      </span>
    )}
    {batchStatus === "dead" && (
      <span className="pill pill--red">Dead, needs manual retry</span>
    )}
    {(batchStatus === "failed" || batchStatus === "dead") && (
      <button type="button" onClick={handleRetry} className="btn-secondary">
        Retry
      </button>
    )}
  </div>
)}
```

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: <hardcoded>`, `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="<from step 2>"
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 <some-jpg> 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: <token>`. PATCH/GET with a missing or wrong token returns 401.