fleettickets/docs/phase-1-ingestion.md

# PRD (Phase 1) — INC hourly CSV ingestion → tracksolid_db → FleetOps Tickets map

> Status: **complete and deployed** (migrations 01–08, boto3 loader, geocoding,
> Coolify hourly `15 7-19 * * *` EAT). This document is the record of the Phase-1
> plan; see `README.md` and `docs/implementation.md` for the as-built state.

## Scope: INC only

**This workflow is strictly for INC** (incident / customer-fault tickets). It
ingests **only** `automations/inc/<EAT-timestamp>.csv`. CRQ (new-installation)
exports at `automations/crq/` are **out of scope** and are not processed here; the
field transforms below are likewise INC-only.

## Context

The client (Rahamafresh / Fireside) runs an n8n workflow that exports field-ops
tickets to our S3-compatible bucket **every hour**:

- `automations/inc/<EAT-timestamp>.csv` — **incidents / customer faults** *(in scope)*
- `automations/crq/<EAT-timestamp>.csv` — new-installation requests *(out of scope)*

(See `n8n-s3-ticket-exports.md`. Sample: `2026-06-15T17-00-00.csv`. Note: the
source later switched to an incremental `automations/inc/changes/` stream — that
doc has the current layout; this PRD records the original Phase-1 model.)

`fleettickets` owns the **downstream**: the `tickets` schema in the shared
`tracksolid_db` (raw-jsonb-first `tickets.inc`, geocoding gazetteers, and
`reporting.fn_tickets_for_map`, which `dashboard_api` serves to the FleetOps
"Tickets" tab). `tickets.crq` keeps existing but is not fed by this pipeline.

**The problem:** the loader was written for the *old* export model — JSON
`{metadata, records}` envelopes at a stable `automations/inc/latest.json`. That
model is gone; the new exports are **flat CSV, timestamped per hour, with no
`latest` pointer, no envelope, and no deltas** — every hourly file is a **full
current-state snapshot**.

**Two driving objectives this pipeline feeds:**

1. **SLA tracking** — contract requires tickets closed within **48h of
   `created_at_service`**; closed carry source `sla_status` + `mttr`, open need a
   derived state (`now − created_at_service` ≥48h breached / ≥36h at-risk).
2. **Vehicle routing (most important)** — accurately geocoded open tickets so
   FleetNow can route nearest vehicles; subsequent: team closure attribution.

## Data contract (verified against live snapshots)

- 32 columns; header + double-quoted values. INC sample = 31,434 rows.
- `ticket_id` is the **primary key**; the same ticket recurs across snapshots as it
  moves `open → closed`. Verified: 31,434 distinct ids per file, **0 in-file dups**,
  same id set every hour (0 added/dropped) → **upsert is the dedup mechanism, no
  TRUNCATE**. Consecutive files are often byte-identical → skip-if-unchanged.
- `is_alarm=true` (~10,132 rows, all `is_actionable=false`) → **dropped**.
- `latitude`/`longitude` are **empty** in the feed → geocoding required.
- A garbage **sentinel row** (`ticket_id = "EXPORT STOPPED DUE TO EXCESSIVE SIZE…"`)
  is commonly the first data line → filtered by `ticket_id` prefix.
- Timestamps (filenames + data) are **EAT (Africa/Nairobi, UTC+3)**.
- `bucket` is meaningful (`closed`/`pending`), distinct from `source_s3_bucket`.

## Approach

Keep the **raw-jsonb-first** model and everything downstream; only the loader's
input path changes: JSON-`latest` → **newest timestamped CSV**, plus move-on-success.

- **Newest file** per `automations/inc/` (parse `YYYY-MM-DDTHH-mm-ss.csv`), via
  **boto3** (path-style; no aws-CLI dependency).
- **Skip-if-unchanged**: compare newest S3 **ETag** to the last processed ETag
  (`tickets.import_meta.metadata.source_etag`); equal → skip DB write.
- **Cleaning at ingest**: drop `is_alarm=true` + sentinel; drop `week_start`,
  `week_end`, `source_s3_bucket`, `source_s3_key`, `source_snapshot_id`,
  `department`, `source_type`; normalize `region`→lowercase, `raw_status`→UPPERCASE;
  keep `service_type`* and `bucket`. (*`service_type` later dropped as constant.)
- **Upsert** on `ticket_id` (`ON CONFLICT DO UPDATE`); never delete → closure
  history accumulates. On success **move** the file(s) to
  `automations/inc/processed/`.
- Record snapshot freshness in `tickets.import_meta`.
- Geocoding unchanged: `--geocode-clusters` (coarse) + `--geocode-locations`
  (precise, actionable INC; keyed LocationIQ; 25 km wrong-city guard).

## Orchestration

Deployed on **Coolify** (own app, `Dockerfile`, keep-alive worker). Ingest runs as a
**Scheduled Task**: `python import_tickets.py --from-bucket --apply`, cron
`15 7-19 * * *` in **EAT**. Env: `DATABASE_URL`, `RUSTFS_*`, `GEOCODER_*`.

> **Superseded** (historical Phase-1 plan). As built: the loader is now the shared
> `pipeline.py` engine with thin entrypoints (`python -m inc.import_inc` / `-m
> crq.import_crq`), running as **two** Scheduled Tasks at cron `*/20 6-20 * * *`. See
> `implementation.md` and `deployment-and-operations.md`.

## Data-quality findings (carried into Phase 2)

- Source `sla_status` ≠ a plain 48h rule, and `mttr` is not wall-clock — pin the
  contract's SLA definition before trusting cross-field SLA math.
- `created_at_service` is null on ~30% of rows (incl. most open) → needs a fallback
  clock (`first_seen_at`).
- Split timestamp semantics: lifecycle = `created_at_service`→`closed_at`; export
  bookkeeping = `created_at`/`updated_at`/`first_seen_at`/`last_seen_at`.
- `assigned_team` missing ~34% (`owner` better).
- Content lag ~2 days (underlying `…wm_task.xlsx` source date).

## Outcome (as built)

Live in `tracksolid_db`: `tickets.inc` (raw + typed generated columns), geocoded to
~99.99%, alarm/sentinel removed, hourly refresh with ETag skip + archive. See
`docs/implementation.md`.