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:

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:

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

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

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:

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.