MEGAINFRA-WEBAPP/docs/data-flows.md

# Data flows

> End-to-end ingestion pipelines with sequence diagrams and the exact site-attribution strategy each one uses.

Reflects the shape produced by the 2026-09-11 ingestion restructure
(`docs/superpowers/plans/2026-09-11-ingestion-restructure.md`). The shared
`data_import_batches` pool is gone. Two pipeline shapes replaced it:

- **The three sheet imports** (timesheets, fuel logs, plant hours) are
  still synchronous, but each now creates its own typed "uploaded document"
  row (`timesheets`, `fuel_sheets`, `plant_hours_sheets`) instead of a
  generic batch row, and computes its real outcome via
  `resolveDocumentOutcome()` (`src/lib/document-status.ts`) once parsing is
  done — never hardcoding `status: processed` before a single row exists.
- **Purchase orders** moved from synchronous-upload-then-async-batch to
  fully async against `purchase_orders` itself: the row is born at upload
  time with a placeholder number and `processing_status: pending`, and an
  n8n cron claims, extracts, and fills it in. There is no batch row at any
  point in this pipeline any more.
- **Delivery notes** keep their existing shape (site + file stamped at
  upload, OCR result posted back by webhook) minus the batch row.

---

## 1. Timesheet import (synchronous, document-first)

**Route:** `POST /api/timesheets/import` (`src/app/api/timesheets/import/route.ts`)
**Input:** `multipart/form-data` with `file` (xlsx / xls / csv), **required** `siteId`, and optional `createMissingEmployees`
**Output:** `{ documentId, totalRows, imported, skipped, errors, detectedFormat, createdEmployees }`
**Triggers:** `ImportTimesheetsButton` on the Workforce page

Unlike the fuel/plant-hours sheets, a wage sheet always belongs to exactly
one site — `siteId` is required and the upload is rejected before anything
is parsed if it's missing.

**Format auto-detection** (the route walks every sheet, first match wins):

1. **Hills Casual Matrix** — `looksLikeCasualMatrix(rows)` (`src/lib/timesheet-casual-parse.ts`) → one long row per (employee, day). Hours are `HH:MM`, "Absent" cells are skipped, names with leading commas are cleaned.
2. **Permanent wide** — `parseHillsTimesheet` (`src/lib/timesheet-grid-parse.ts`) → one row per employee with daily hours across columns.
3. **Long format** — header row `employee_no, date, hours_worked` for both CSV and XLSX fallback.

**Pipeline:**

```mermaid
sequenceDiagram
    participant U as User (browser)
    participant R as /api/timesheets/import
    participant P as PocketBase
    U->>R: multipart { file, siteId (required), createMissingEmployees? }
    R->>R: parseFile() → uniform long rows
    R->>P: create(timesheets, site=siteId, source_file, file_hash, processing_status=pending)
    alt createMissingEmployees
        R->>P: getFullList(employees, fields="id,employee_no")
        R->>P: create(missing employees, casual, site=siteId)
    end
    R->>P: getFullList(employees, fields="id,employee_no,site")
    loop for each parsed row
        R->>P: create(timesheet_entries, employee=lookup[row.employee_no], site=employeeSite.get(employeeId) || siteId)
    end
    R->>R: resolveDocumentOutcome({ rowsTotal, rowsCreated, rowErrors }, now)
    R->>P: update(timesheets doc, processing_status, rows_total, rows_created, rows_failed, row_errors, last_error)
    R-->>U: { documentId, totalRows, imported, skipped, errors, ... }
```

**Site attribution.** `timesheet_entries` (the collection that used to be
called `timesheets` before the rename) now carries a **required** `site`
relation, stamped from the employee's site at the moment of import — not
joined at read time. This is a deliberate change: the old at-read-time join
through `employees.site` meant a worker moving sites silently rewrote the
site attribution of every historical hour they'd ever logged. Stamping at
write time freezes each entry's site to what it actually was that day.

**Failure behaviour.**

- Unknown `employee_no` (and `createMissingEmployees` not set) → row errored, import continues.
- Per-row `pb.collection("timesheet_entries").create` failure → row errored, import continues.
- File parse failure → 400 with a top-level `error` and `details`.
- `resolveDocumentOutcome()` marks the document `failed` (not `processed`)
  if the file had zero data rows, or if parsing produced rows but zero of
  them landed — see `src/lib/document-status.ts`. A partial import (some
  rows landed, some failed) is `processed` with `rows_failed > 0` and
  `last_error` summarising the count; it is not silently reported clean.

---

## 2. Fuel log import (synchronous, document-first)

**Route:** `POST /api/fuel-logs/import` (`src/app/api/fuel-logs/import/route.ts`)
**Input:** `multipart/form-data` with `file` (xlsx or csv), optional `siteId`
**Output:** `{ documentId, totalRows, imported, skipped, errors, qrMatched, descMatched }`
**Triggers:** `ImportFuelLogsButton` on the Fleet page

**Expected columns** (case-insensitive, alias-resolved by `resolveColumn` in `src/lib/fuel-import-parse.ts`):

| Primary | Legacy aliases |
| --- | --- |
| `Date Logged` | `date`, `date_logged` |
| `Vehicle QR` | `vehicle_qr`, `qr`, `qr_code_id` |
| `Vehicle` | `vehicle_description`, `description` |
| `Km's/ Hours` | `odo_reading`, `odometer`, `hour_meter` |
| `Litres Pumped` | `litres`, `liters` |
| `Current Site` | `site`, `current_site` |
| `Latest Consumption` | `latest_consumption`, `consumption` (optional) |

**Pipeline:**

```mermaid
sequenceDiagram
    participant U as User (browser)
    participant R as /api/fuel-logs/import
    participant P as PocketBase
    U->>R: multipart { file, siteId? }
    R->>R: SheetJS parse → rowsRaw
    R->>P: create(fuel_sheets, site=siteId?, source_file, file_hash, processing_status=pending)
    R->>P: getFullList(vehicles, fields="id,description,site,qr_code_id")
    R->>P: getFullList(sites, fields="id,name")
    R-->>R: vehicleByQr, vehicleByDesc, siteByName maps
    loop for each row
        R->>R: parseDate, toNumber
        R->>R: vehicle = vehicleByQr[qr] ?? vehicleByDesc[desc]
        R->>R: rowSiteId = siteByName[row.site.toLowerCase()] ?? vehicle.site
        R->>P: create(fuel_logs, site=rowSiteId, vehicle, fuel_sheet=doc.id, ...)
    end
    R->>R: resolveDocumentOutcome({ rowsTotal, rowsCreated, rowErrors }, now)
    R->>P: update(fuel_sheets doc, processing_status, rows_total, rows_created, rows_failed, row_errors, last_error)
    R-->>U: { documentId, totalRows, imported, skipped, errors, qrMatched, descMatched }
```

**Site attribution strategy.** Two-tier match, same shape as before the
restructure, but the destination is a real required `site` relation on
`fuel_logs` instead of a denormalised `project_code` fallback:

1. The row's `Current Site` cell is matched case-insensitive against
   `sites.name`. If found, that site is stamped on the row.
2. If the row has no site cell (legacy CSVs) or no match, the row falls
   back to the matched vehicle's current `site`.

This is still the seam where **site attribution goes wrong** as the fleet
grows — see [multi-site-assessment §4](./multi-site-assessment.md#4-site-resolution-by-name-string).
Two sites with near-identical names, a renamed site, or a vehicle moved
between sites all still produce silently wrong attribution; what changed is
that there is no separate `project_code` fallback path to reason about any
more — one relation, one resolution order.

**Failure behaviour.**

- Unknown vehicle (no QR match, no description match) → row errored, import continues.
- Invalid date / odo / litres → row errored, import continues.
- PB insert failure → row errored, import continues; the importer extracts
  the field-level validation message via `formatPbError`.
- `resolveDocumentOutcome()` governs the document's final status exactly as
  in the timesheet pipeline (§1) — zero rows parsed or zero rows landed is
  `failed`, never `processed`.

---

## 3. Plant hours import (synchronous, document-first)

**Route:** `POST /api/plant-hours/import` (`src/app/api/plant-hours/import/route.ts`)
**Input:** `multipart/form-data` with `file` (xlsx or csv, max 10 MB), optional `siteId`
**Output:** `{ documentId, totalRows, imported, skipped, mismatched, warnings, errors }`
**Orchestrator:** `importPlantHours` in `src/lib/plant-hours-import.ts`
**Triggers:** `ImportPlantHoursButton` on the Plant page

**Pipeline:**

```mermaid
sequenceDiagram
    participant U as User
    participant R as /api/plant-hours/import
    participant O as importPlantHours (pure)
    participant P as PocketBase
    U->>R: multipart { file, siteId? }
    R->>R: SheetJS parse
    R->>P: create(plant_hours_sheets, site=siteId?, source_file, file_hash, processing_status=pending)
    R->>P: getFullList(vehicles, fields="id,description,qr_code_id,unit_type,site,cost_per_hour", perPage=5000)
    R->>P: getFullList(sites, fields="id,name", perPage=5000)
    R->>O: importPlantHours(rowsRaw, { vehicles, sites })
    O->>O: resolve each row by qr_code_id (or desc fallback)
    O->>O: compute hours, value, over_budget, hours_mismatch
    O->>O: snapshot vehicle_unit_type_at_import
    loop for each prepped row
        R->>P: create(plant_hours_log, site=vehicle.site, plant_hours_sheet=doc.id, ...)
        alt duplicate (vehicle_qr, start_date)
            P-->>R: 400 validation_not_unique
            R->>R: record as "duplicate: vehicle_qr=X start_date=Y"
        end
    end
    R->>R: resolveDocumentOutcome({ rowsTotal, rowsCreated, rowErrors }, now)
    R->>P: update(plant_hours_sheets doc, processing_status, rows_total, rows_created, rows_failed, row_errors, last_error)
    R-->>U: { documentId, totalRows, imported, skipped, mismatched, warnings, errors }
```

**Site attribution.** The orchestrator pulls the site from the matched
vehicle (`vehicle.site`) and stamps it on the row. This remains the
**only collection where `site` is required (`minSelect=1`)** on the log
itself — every `plant_hours_log` row has a non-null site.

**Composite unique index.** `idx_plant_hours_log_qr_start` on
`(vehicle_qr, start_date)` blocks accidental double-imports of the same
week for the same vehicle. The route surfaces the PB-level rejection as
a friendly `duplicate: ...` error per row, not a document-level failure.
See [multi-site-assessment §5](./multi-site-assessment.md#5-vehicles-plant-and-employees-moving-between-sites) for the cross-site implication
of this index.

**Failure behaviour.** Per-row. The orchestrator collects row-level
errors and warnings; the route appends PB insert errors. The document
itself is never aborted by one bad row — `resolveDocumentOutcome()` decides
the final status the same way as §1/§2.

**Exception report.** `GET /api/plant-hours/exception-report/[sheetId]`
reconstructs a per-row exception CSV by re-parsing the source file
from the document's `source_file` and matching against the imported rows
in `plant_hours_log.filter('plant_hours_sheet = "<sheetId>"')`. Used to
re-export a corrected sheet when the operator needs to fix and re-import.

---

## 4. Purchase order upload → ingest (async, no batch row)

Two routes cooperate; the browser never talks to n8n directly. There is no
longer a poll-for-status route, a separate batch collection, or a "move the
file onto the new row" step — the `purchase_orders` row *is* the queue item
from the moment it's created.

**Routes:**
- `POST /api/purchase-orders/upload` — `src/app/api/purchase-orders/upload/route.ts`
- `POST /api/purchase-orders/process-batch` — called by n8n — `src/app/api/purchase-orders/process-batch/route.ts`

Full implementation detail, including the n8n node map, hardcoded values,
and a manual smoke-test script, lives in
`docs/n8n/purchase-order-ingest-implementation.md` (rewritten for this
shape in Task 15 of the restructure) — this section is the short version.

**Auth model.** Both routes are server-side; `process-batch` additionally
requires the shared `N8N_INGEST_SECRET` header.

**Pipeline:**

```mermaid
sequenceDiagram
    participant U as User
    participant W as /api/purchase-orders/upload
    participant P as PocketBase
    participant N as n8n (PO ingest cron, every ~3 min)
    participant B as /api/purchase-orders/process-batch
    U->>W: multipart { file (PDF/image), siteId }
    W->>W: SHA-256 the bytes
    W->>P: getFullList(purchase_orders, file_hash=X && date_uploaded >= 30d_ago)
    alt duplicate
        W-->>U: 200 { duplicate: true, existing_id, ... }
    else new
        W->>P: create(purchase_orders, site=siteId, purchase_order_number="PENDING-...", processing_status=pending, file_hash, purchase_order_file)
        W-->>U: 202 { purchase_order_id, status: queued }
    end
    N->>P: PATCH purchase_orders/{id}?filter=processing_status="pending" (atomic claim → processing, attempts+1)
    N->>P: GET /files/purchase_orders/{id}/{purchase_order_file}
    N->>N: AI Agent extracts PO header + line items
    N->>B: POST { purchase_order_id, extracted } + x-ingest-secret
    B->>P: getOne(purchase_orders, id) — must still be 'processing'
    B->>P: getFullList(purchase_orders, filter=purchase_order_number=extracted.number) — dedup
    alt fill in (no existing PO with that number)
        B->>P: update(this row, purchase_order_number=extracted.number, processing_status=processed, raw_ocr_data, is_verified=false)
        loop for each line item
            B->>P: create(purchase_order_items, purchase_order=this row.id)
        end
    else merge (a real PO with that number already exists)
        B->>P: update(existing PO with any extracted fields it was missing)
        B->>P: delete(this placeholder row)
    else invalid extraction
        B->>P: update(this row, processing_status=failed, last_error)  // never deleted — stays visible and recoverable
    end
    B-->>N: { ok: true } or 4xx/5xx
```

**Three-outcome `process-batch` contract.** This is the core of the async
design — one route, three possible endings for the *same* placeholder row,
never a fourth row created:

1. **Fill in.** No existing PO shares the extracted number — the
   placeholder row is updated in place with the real number, OCR payload,
   and line items. This is the common case.
2. **Merge / dedup / delete placeholder.** A real PO with that number
   already exists (a genuine duplicate upload, or two near-simultaneous
   uploads of the same document). The existing PO is patched with anything
   useful the new extraction found that it was missing, and *this*
   placeholder row is deleted. The caller-visible id becomes the winner's.
3. **Mark failed.** The extraction was invalid (no PO number, garbage
   output) — the row moves to `processing_status=failed` with `last_error`
   set. It is **never deleted** on this path, so it stays visible and
   recoverable in the dashboard, and the n8n-side escalation logic can
   retry or eventually mark it `dead` after repeated failures.

**Dedup layers, in order:**

1. **Hash dedup** at upload time — SHA-256 of the file bytes, matched
   against `purchase_orders` rows created in the last 30 days. Returns
   `200 { duplicate: true }` before any new placeholder row is written.
2. **Extracted-number dedup** at process time — outcome 2 above.
3. **PB unique index** `idx_po_number` on
   `purchase_orders(purchase_order_number)` — the final backstop. If the
   dedup logic misses, this rejects the insert.

**Site attribution strategy.** `matchSite(pb, siteId or project code hint)`
in `src/lib/purchase-order-schema.ts` resolves the site once, at upload
time, before OCR ever runs — `purchase_orders.site` is a required field, so
there is no unmatched/null-site PO state to reason about downstream the way
there used to be.

**Failure behaviour.** Per the `process-batch` route's own contract, it
distinguishes `retryable: true` (5xx with a transient signal — n8n retries)
from `retryable: false` (400 — n8n marks the row `dead` after enough
attempts and surfaces it in the dashboard). Network blips and PB write
conflicts are retryable; validation failures are not.

---

## 5. Delivery note upload → OCR result (async via n8n)

**Routes:**
- `POST /api/delivery-notes/upload` — `src/app/api/delivery-notes/upload/route.ts`
- `GET /api/delivery-notes/[id]/slip` — authenticated file proxy for n8n
- `POST /api/delivery-notes/[id]/ocr-result` — webhook from n8n — `src/app/api/delivery-notes/[id]/ocr-result/route.ts`
- `POST /api/delivery-notes/[id]/assign-po` — manual fallback
- `POST /api/records/[collection]/[id]/verify` — manual verify toggle

**Pipeline:**

```mermaid
sequenceDiagram
    participant U as User
    participant U2 as /api/delivery-notes/upload
    participant P as PocketBase
    participant N as n8n (DN OCR workflow)
    participant S as /api/delivery-notes/:id/slip
    participant R as /api/delivery-notes/:id/ocr-result
    participant A as /api/delivery-notes/:id/assign-po
    U->>U2: multipart { file, site, delivery_note_number? }
    U2->>P: create(delivery_notes, slip_image=file, site, processing_status=pending, file_hash)
    U2-->>U: 201 { id, siteId, processingStatus: pending }
    N->>S: GET (x-ingest-secret)
    S->>P: getOne(delivery_notes, fields="slip_image")
    S->>P: GET /api/files/delivery_notes/:id/<filename>
    S-->>N: file bytes
    N->>N: AI Agent OCR: po_number, dn_number, items, confidence
    N->>R: POST { po_number, delivery_note_number, date_received, received_by_name, confidence, items, raw_ocr } + x-ingest-secret
    R->>P: getOne(delivery_notes, fields="id,purchase_order,...")
    R->>R: PO match (exact → normalised → Levenshtein)
    R->>P: getFullList(purchase_order_items, filter=purchase_order='<poId>')
    R->>R: resolveOcrTier(confidence, poResolved, itemsVerifiable)
    alt tier = auto
        R->>P: update(delivery_notes, processing_status=auto-verified, is_verified=true, ocr_confidence, ocr_completed_at, purchase_order, raw_ocr_data, ...)
        loop for each item
            R->>P: create(delivery_note_items, delivery_note, purchase_order_item=validatedId, qty_received, condition_notes)
        end
    else tier = needs-review
        R->>P: update(delivery_notes, processing_status=needs-review, is_verified=false, ...)
        R->>P: delete + recreate delivery_note_items (idempotent)
    else tier = manual
        R->>P: update(delivery_notes, processing_status=needs-review, ...) — no delivery_note_items
    end
    opt operator fixes
        U->>A: POST { po: <poId> } (manual fallback)
        A->>P: getOne(purchase_orders, id)
        A->>P: update(delivery_notes, purchase_order=poId, processing_status=manual-verified, is_verified=true)
    end
```

**Three-tier OCR result** (rules in `src/lib/ocr-result-tier.ts`):

| Confidence | PO resolved | Items verifiable | Tier | Effect |
| --- | --- | --- | --- | --- |
| ≥ 0.8 | ✅ | ✅ | `auto` | `is_verified=true`, `processing_status=auto-verified`, delivery_note_items created with PO links |
| ≥ 0.8 | ❌ | n/a | `needs-review` (downgraded) | `is_verified=false`, no auto-link — orphan DNs are never auto-verified |
| ≥ 0.8 | ✅ | ❌ (local fetch empty but n8n supplied ids) | `needs-review` (downgraded) | We don't claim auto when we can't cross-check the links |
| 0.5..0.79 | any | any | `needs-review` | `is_verified=false`, delivery_note_items created without PO links |
| < 0.5 | any | any | `manual` | `is_verified=false`, **no delivery_note_items** — the operator must capture manually |

**PO match priority:** exact `purchase_order_number` → normalised
(strip non-alnum, lowercase) → Levenshtein (single-char edit, threshold
0.75). The chosen path + score is recorded in
`raw_ocr_data.po_match_method` / `po_match_score` / `po_resolved_number`
so the exception inbox can show "matched by Levenshtein, 0.9" rather
than a silent "no match".

**Site attribution strategy.** Site is stamped at upload time from the
form field (`src/app/api/delivery-notes/upload/route.ts`), a required
field on the row; the OCR callback does not change it. For DNs uploaded
without a site, the route 400s.

**Idempotency.** If the OCR webhook is replayed, the route
**deletes** the existing `delivery_note_items` for this DN and recreates
them within the same call. This avoids stale matches if the OCR rerun finds
different items. The DN update is patch-style so missing fields
(delivery_note_number, date_received) are never overwritten by missing
data.

**Failure behaviour.** If the DN update itself fails, the route marks the
DN's `processing_status` as `failed` so the operator dashboard reflects the
broken state instead of leaving it pending forever. If a single
`delivery_note_items` create fails with a 400 (FK race — the PO item was
deleted between fetch and create), the route retries with the link dropped
and annotates the row as `link dropped: orphan id`.

---

## 6. Cross-cutting behaviours

- **Multipart limits.** All upload routes check `file.size` before parsing.
  PO upload enforces mime allowlist via `ALLOWED_MIME`; DN upload accepts
  any image/pdf. Plant hours import caps at 10 MB.
- **Auth headers.** `N8N_INGEST_SECRET` is checked at the top of every
  n8n-callbackable route (`process-batch`, `ocr-result`, `slip` proxy).
  The browser-facing routes do not check any auth.
- **PB ID validation.** All `[id]` path params use the PB ID pattern
  (`/^[a-z0-9]{15}$/i`) to close the filter-injection sink before the
  route does `filter: "<field>='<id>'"`.
- **Audit trail.** Every document/queue collection (`timesheets`,
  `fuel_sheets`, `plant_hours_sheets`, `purchase_orders`) keeps its own
  original file on its own row via `source_file` (or `purchase_order_file`
  for POs) — there is no separate batch collection to clear a file from any
  more. `file_hash` on each of those, plus `delivery_notes`, drives
  upload-time dedup.
- **No `created`/`updated` autodate fields.** This PocketBase instance
  never enabled them on any collection. Several routes have comments
  noting that sorting or filtering by `created`/`updated` 400s here — use
  the collection's own `date_uploaded` / `date_logged` / `date_received`
  field instead.