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:

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:

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

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

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:

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.