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.

Every importer writes into a `data_import_batches` row, which carries the
original file for the audit trail and the resulting `processed_record_id`.
Three of the pipelines are synchronous (the file is parsed and inserted in
the same request) and two are async (a separate worker, n8n, processes
them and POSTs results back).

---

## 1. Timesheet import (synchronous)

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

**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?, createMissingEmployees? }
    R->>R: parseFile() → uniform long rows
    alt createMissingEmployees && siteId
        R->>P: getFullList(employees, fields="id,employee_no")
        R->>P: create(missing employees, casual, site=siteId)
    end
    R->>P: create(data_import_batches, batch_type=timesheets, original_file)
    R->>P: getFullList(employees, fields="id,employee_no")
    loop for each parsed row
        R->>P: create(timesheets, employee=lookup[row.employee_no])
    end
    R-->>U: { batchId, totalRows, imported, skipped, errors, ... }
```

**Site attribution.** Timesheets do not carry a `site` field. The route
writes the row against the employee record; site attribution happens at
*read time* in the metrics helpers by joining through
`employeeRates.get(timesheet.employee).site`. This is the only
collection in the system with no `site` or `project_code` on the row itself.

**Failure behaviour.**

- Unknown `employee_no` (and `createMissingEmployees` not set) → row errored, import continues.
- Per-row `pb.collection("timesheets").create` failure → row errored, import continues.
- File parse failure → 400 with a top-level `error` and `details`.

---

## 2. Fuel log import (synchronous)

**Route:** `POST /api/fuel-logs/import` (`src/app/api/fuel-logs/import/route.ts`)
**Input:** `multipart/form-data` with `file` (xlsx or csv)
**Output:** `{ batchId, 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 }
    R->>R: SheetJS parse → rowsRaw
    R->>P: create(data_import_batches, batch_type=fuel_logs, original_file, status=processed)
    R->>P: getFullList(vehicles, fields="id,description,project_code,qr_code_id")
    R->>P: getFullList(sites, fields="id,name,project_code")
    R-->>R: vehicleByQr, vehicleByDesc, siteByName maps
    loop for each row
        R->>R: parseDate, toNumber
        R->>R: vehicle = vehicleByQr[qr] ?? vehicleByDesc[desc]
        R->>R: site = siteByName[row.site.toLowerCase()]
        R->>R: projectCode = site?.projectCode ?? vehicle.projectCode
        R->>P: create(fuel_logs, site, vehicle, project_code, ...)
    end
    R-->>U: { batchId, totalRows, imported, skipped, errors, qrMatched, descMatched }
```

**Site attribution strategy.** Two-tier match:

1. The row's `Current Site` cell is matched case-insensitive against
   `sites.name`. If found, the site relation and its `project_code` are
   stamped on the row.
2. If the row has no site cell (legacy CSVs), the row's `project_code`
   is set to the matched vehicle's `project_code` and the `site` relation
   is left **empty**. The fuel log exists but isn't tied to a site by
   relation — the only link is the text `project_code`.

This is 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 produce silently wrong attribution.

**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` (lines 65-78).

---

## 3. Plant hours import (synchronous)

**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)
**Output:** `{ batchId, 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 }
    R->>R: SheetJS parse
    R->>P: create(data_import_batches, batch_type=plant_hours, original_file, status=processed)
    R->>P: getFullList(vehicles, fields="id,description,project_code,qr_code_id,unit_type,site,cost_per_hour", perPage=5000)
    R->>P: getFullList(sites, fields="id,name,project_code", perPage=5000)
    R->>O: importPlantHours(rowsRaw, batchId, { 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, ...)
        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-->>U: { batchId, totalRows, imported, skipped, mismatched, warnings, errors }
```

**Site attribution.** The orchestrator pulls the site from the matched
vehicle (`vehicle.site`) and stamps both `site` and `project_code` on
the row. This is the **only collection where `site` is required
(`minSelect=1`)** — 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 batch-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 batch
itself is never aborted by one bad row.

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

---

## 4. Purchase order upload → ingest (async via n8n)

Three routes cooperate. The browser never talks to n8n directly.

**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`
- `GET /api/data-import-batches/[id]/status` — poll target for the browser

**Auth model.** All three are server-side; only `process-batch` requires
the shared `N8N_INGEST_SECRET` header. The other two run as the
PocketBase superuser.

**Pipeline:**

```mermaid
sequenceDiagram
    participant U as User
    participant W as /api/purchase-orders/upload
    participant P as PocketBase
    participant N as n8n (PO ingest workflow)
    participant B as /api/purchase-orders/process-batch
    participant B2 as /api/data-import-batches/:id/status
    U->>W: multipart { file (PDF/image) }
    W->>W: SHA-256 the bytes
    W->>P: getFullList(batches, file_hash=X && status=processed && date_uploaded >= 30d_ago)
    alt duplicate
        W-->>U: 200 { duplicate: true, existing_batch_id, ... }
    else new
        W->>P: create(data_import_batches, batch_type=purchase_orders, status=pending, file_hash)
        W-->>U: 202 { batch_id, status: queued }
    end
    N->>P: poll /api/files/data_import_batches/:id/<filename> (admin auth)
    N->>N: AI Agent extracts PO fields + line items
    N->>B: POST { batch_id, extracted } + x-ingest-secret
    B->>P: getOne(batch) — must be in 'processing'
    B->>P: getFullList(purchase_orders, filter=po_number=...) — dedup
    B->>B: matchSite(extracted.project_code) → siteId
    B->>P: download batch.original_file via /api/files
    B->>P: create(purchase_orders, file=..., site=siteId, raw_ocr_data, is_verified=false)
    loop for each line item
        B->>P: create(purchase_order_items, purchase_order=poRecord.id)
    end
    alt dedup branch
        B->>P: update(batch, status=processed, processed_record_id=existing.id)  // keep original_file
    else success branch
        B->>P: update(batch, status=processed, processed_record_id=poRecord.id, original_file="")
    end
    U->>B2: GET /api/data-import-batches/:id/status
    B2-->>U: { id, status, processed_record_id, processed_record_label, ... }
```

**Three dedup layers, in order:**

1. **Hash dedup** at upload time — SHA-256 of the file bytes, matched
   against batches in the last 30 days with `status=processed`. Returns
   `200 { duplicate: true }` before any new batch row is written.
2. **AI-extracted PO number dedup** at process time —
   `getFullList(purchase_orders, filter=po_number="...")`. If found,
   the existing PO id is recorded on the batch but **the duplicate file
   is kept on the batch row** for audit (`clearFile: false`).
3. **PB unique index** `idx_po_number` on
   `purchase_orders(purchase_order_number)` — the final backstop. If
   the AI dedup misses, this rejects the insert.

**Site attribution strategy.** `matchSite(pb, payload.project_code)` in
`src/lib/purchase-order-schema.ts`:

1. Exact match on `sites.project_code`.
2. Fuzzy match `project_code ~ "X"` (PB's `~` operator does case-insensitive substring).
3. If both miss, the PO is stored with `site=null` and a warning is
   surfaced: `unmatched_project_code: !siteId` is written into
   `raw_ocr_data`. The `unmatched_project_code` flag is the operator's
   signal that a manual assignment is needed.

**Failure behaviour.** Per the `process-batch` header comment, the route
distinguishes `retryable: true` (5xx with a transient signal — n8n
retries) from `retryable: false` (400 — n8n marks the batch `dead` 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: getOne(sites, fields="project_code")
    U2->>P: create(delivery_notes, slip_image=file, site, project_code, processing_status=pending)
    U2->>P: create(data_import_batches, batch_type=delivery_notes, processed_record_id=dn.id)
    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(received_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 received_items (idempotent)
    else tier = manual
        R->>P: update(delivery_notes, processing_status=needs-review, ...) — no received_items
    end
    R->>P: update(data_import_batches, status=processed, ...)
    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`, received_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`, received_items created without PO links |
| < 0.5 | any | any | `manual` | `is_verified=false`, **no received_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:26`); the OCR
callback does not change it. The PO the OCR picks is what links the
delivery to a site indirectly (PO.site). For DNs uploaded without a
site, the route 400s.

**Idempotency.** If the OCR webhook is replayed, the route
**deletes** the existing `received_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 batch `failed` so the operator dashboard reflects the broken state
instead of leaving it pending forever. If a single `received_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 batch keeps the original file via
  `data_import_batches.original_file`. The PO route clears this field
  on success (the file is moved to the PO row); it keeps the bytes on
  the dedup path. The fuel/plant/timesheet routes never clear the file.