# MQTT Raw Ingestion Pipeline Implementation Plan

> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build a production-safe MQTT ingestion pipeline that subscribes to `terminal/{device_id}/raw`, stores payload bytes losslessly, parses NMEA + RTCM 3.2 type summaries asynchronously, and supports 90-day hot data plus local cold archive.

**Architecture:** Use a DB-queue two-stage pipeline: `mqtt-ingestor` writes raw payloads to PostgreSQL immediately, and `parser-worker` asynchronously consumes pending rows to produce event/state tables. A separate archive job writes local cold files and manifest records, then prunes hot data beyond retention.

**Tech Stack:** FastAPI/SQLModel/Alembic/PostgreSQL, paho-mqtt, pytest, Docker Compose.

---

## Scope Check

This plan covers one cohesive subsystem (MQTT ingest + parse + archive pipeline). It intentionally excludes frontend UI and realtime push implementation, but produces backend data structures required for both.

## Execution Skills

- `@superpowers/test-driven-development`
- `@superpowers/systematic-debugging`
- `@superpowers/verification-before-completion`

## File Structure Map

### Create

- `backend/app/mqtt/__init__.py` — MQTT package marker.
- `backend/app/mqtt/topic.py` — topic parsing and validation (`device_id` extraction).
- `backend/app/mqtt/repository.py` — DB helpers for raw insert, pending fetch, status update, event/state write, archive manifest.
- `backend/app/mqtt/parsers/__init__.py` — parser package exports.
- `backend/app/mqtt/parsers/nmea.py` — NMEA sentence extraction + structured fields.
- `backend/app/mqtt/parsers/rtcm.py` — RTCM 3.x frame scan + type summary.
- `backend/app/mqtt/parsers/mixed.py` — mixed payload scanner orchestrating NMEA + RTCM parsers.
- `backend/app/mqtt/ingestor.py` — MQTT client callbacks + buffered DB flush.
- `backend/app/mqtt/parser_worker.py` — pending job polling loop (`FOR UPDATE SKIP LOCKED`).
- `backend/app/mqtt/archive_job.py` — local archive writer and retention cleanup.
- `backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py` — migration for raw/event/state/archive tables.
- `backend/tests/mqtt/test_topic.py` — unit tests for topic parser.
- `backend/tests/mqtt/test_mixed_parser.py` — unit tests for mixed payload parse behavior.
- `backend/tests/mqtt/test_parser_worker.py` — DB-backed tests for pending->parsed transitions.
- `backend/tests/mqtt/test_archive_job.py` — archive/manifest/cleanup tests.

### Modify

- `backend/pyproject.toml` — add MQTT dependency (`paho-mqtt`).
- `backend/app/core/config.py` — add MQTT and pipeline configuration env vars.
- `backend/app/models.py` — add `TerminalMessageRaw`, `TerminalObservationEvent`, `TerminalDeviceState`, `TerminalRawArchiveManifest` models.
- `compose.yml` — add production `mqtt-ingestor` and `mqtt-parser-worker` services.
- `compose.override.yml` — add local dev services and optional archive runner profile.
- `.env.production.example` — add MQTT and archive env examples.
- `backend/README.md` — operational docs for running ingestor/worker/archive and verification commands.

### Optional (if command wiring is needed)

- `backend/pyproject.toml` entry points for CLI commands (`mqtt-ingestor`, `mqtt-parser-worker`, `mqtt-archive`).

---

### Task 1: Configuration + Topic Parsing Foundation

**Files:**
- Create: `backend/app/mqtt/__init__.py`
- Create: `backend/app/mqtt/topic.py`
- Test: `backend/tests/mqtt/test_topic.py`
- Modify: `backend/app/core/config.py`
- Modify: `backend/pyproject.toml`

- [ ] **Step 1: Write failing tests for topic parsing**

```python
# backend/tests/mqtt/test_topic.py
import pytest

from app.mqtt.topic import extract_device_id


def test_extract_device_id_success() -> None:
    assert extract_device_id("terminal/device-001/raw") == "device-001"


def test_extract_device_id_rejects_invalid_topic() -> None:
    with pytest.raises(ValueError, match="terminal/.+/raw"):
        extract_device_id("terminal/device-001/nmea")
```

- [ ] **Step 2: Run test to verify it fails**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_topic.py -v`
Expected: FAIL with `ModuleNotFoundError: No module named 'app.mqtt.topic'`

- [ ] **Step 3: Implement minimal topic parser + config fields**

```python
# backend/app/mqtt/topic.py
import re

TOPIC_RE = re.compile(r"^terminal/(?P<device_id>[^/]+)/raw$")


def extract_device_id(topic: str) -> str:
    match = TOPIC_RE.match(topic)
    if not match:
        raise ValueError("topic must match terminal/{device_id}/raw")
    return match.group("device_id")
```

```python
# backend/app/core/config.py (additions)
MQTT_HOST: str = "101.35.119.226"
MQTT_PORT: int = 1883
MQTT_USERNAME: str = ""
MQTT_PASSWORD: str = ""
MQTT_TOPIC_PATTERN: str = "terminal/+/raw"
MQTT_CLIENT_ID: str = "spatialhub-mqtt-ingestor"
RAW_HOT_RETENTION_DAYS: int = 90
RAW_ARCHIVE_BASE_DIR: str = "/data/archive/gnss"
PARSER_BATCH_SIZE: int = 200
PARSER_POLL_INTERVAL_MS: int = 300
```

- [ ] **Step 4: Run tests to verify pass**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_topic.py -v`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add backend/app/mqtt/__init__.py backend/app/mqtt/topic.py backend/tests/mqtt/test_topic.py backend/app/core/config.py backend/pyproject.toml
git commit -m "feat: add MQTT topic parsing and pipeline config"
```

### Task 2: Add DB Schema for Raw/Event/State/Archive

**Files:**
- Modify: `backend/app/models.py`
- Create: `backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py`
- Test: `backend/tests/mqtt/test_parser_worker.py`

- [ ] **Step 1: Write failing DB test for raw row persistence**

```python
# backend/tests/mqtt/test_parser_worker.py
from sqlmodel import Session

from app.models import TerminalMessageRaw


def test_terminal_message_raw_model_can_persist(db: Session) -> None:
    row = TerminalMessageRaw(
        device_id="device-001",
        topic="terminal/device-001/raw",
        payload_bytes=b"$GPGGA,123519,3723.2475,N,12158.3416,W,1,12,0.8,10.0,M,-25.0,M,,*65\r\n",
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    assert row.id is not None
```

- [ ] **Step 2: Run test to verify it fails**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_parser_worker.py::test_terminal_message_raw_model_can_persist -v`
Expected: FAIL with `ImportError`/`AttributeError` for missing `TerminalMessageRaw`

- [ ] **Step 3: Implement models and migration**

```python
# backend/app/models.py (new table skeleton example)
class TerminalMessageRaw(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    device_id: str = Field(index=True)
    topic: str
    received_at: datetime = Field(default_factory=get_datetime_utc, sa_type=DateTime(timezone=True))
    payload_bytes: bytes
    payload_size: int
    payload_sha256: str = Field(max_length=64, index=True)
    parse_status: str = Field(default="pending", index=True)
```

```python
# alembic upgrade should create:
# terminal_message_raw, terminal_observation_event, terminal_device_state, terminal_raw_archive_manifest
```

```bash
# if creating from CLI
cd backend && ../.venv/bin/alembic revision --autogenerate -m "add mqtt pipeline tables" --rev-id a1b2c3d4e5f6
```

- [ ] **Step 4: Run migration + targeted test**

Run: `cd backend && ../.venv/bin/alembic upgrade head`
Expected: migration applies without errors

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_parser_worker.py::test_terminal_message_raw_model_can_persist -v`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add backend/app/models.py backend/app/alembic/versions/*.py backend/tests/mqtt/test_parser_worker.py
git commit -m "feat: add MQTT pipeline database schema"
```

### Task 3: Implement Raw Ingestion Repository + MQTT Ingestor

**Files:**
- Create: `backend/app/mqtt/repository.py`
- Create: `backend/app/mqtt/ingestor.py`
- Test: `backend/tests/mqtt/test_parser_worker.py`

- [ ] **Step 1: Write failing test for raw insert helper**

```python
def test_insert_raw_message_sets_pending_status(db: Session) -> None:
    row = insert_raw_message(
        session=db,
        device_id="device-001",
        topic="terminal/device-001/raw",
        payload=b"$GPRMC,123520,A,3723.2475,N,12158.3416,W,0.13,309.62,120598,,,A*74\r\n",
        qos=1,
        retain=False,
    )
    assert row.parse_status == "pending"
    assert row.payload_size > 0
```

- [ ] **Step 2: Run test to verify failure**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_parser_worker.py::test_insert_raw_message_sets_pending_status -v`
Expected: FAIL with `NameError`/missing repository function

- [ ] **Step 3: Implement repository + ingestor callback path**

```python
# backend/app/mqtt/repository.py

def insert_raw_message(*, session: Session, device_id: str, topic: str, payload: bytes, qos: int, retain: bool) -> TerminalMessageRaw:
    row = TerminalMessageRaw(
        device_id=device_id,
        topic=topic,
        payload_bytes=payload,
        payload_size=len(payload),
        payload_sha256=hashlib.sha256(payload).hexdigest(),
        qos=qos,
        retain=retain,
        parse_status="pending",
    )
    session.add(row)
    session.commit()
    session.refresh(row)
    return row
```

```python
# backend/app/mqtt/ingestor.py (callback shape)
def on_message(client: Client, userdata: object, message: MQTTMessage) -> None:
    device_id = extract_device_id(message.topic)
    with Session(engine) as session:
        insert_raw_message(
            session=session,
            device_id=device_id,
            topic=message.topic,
            payload=message.payload,
            qos=message.qos,
            retain=message.retain,
        )
```

- [ ] **Step 4: Run tests for repository path**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_parser_worker.py -k insert_raw_message -v`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add backend/app/mqtt/repository.py backend/app/mqtt/ingestor.py backend/tests/mqtt/test_parser_worker.py
git commit -m "feat: persist MQTT raw messages via ingestor repository"
```

### Task 4: Build NMEA Parser for GGA/RMC/GSV/GSA/GST

**Files:**
- Create: `backend/app/mqtt/parsers/__init__.py`
- Create: `backend/app/mqtt/parsers/nmea.py`
- Test: `backend/tests/mqtt/test_mixed_parser.py`

- [ ] **Step 1: Write failing tests for NMEA extraction**

```python
def test_parse_nmea_sentences_extracts_gga_and_rmc() -> None:
    payload = b"$GPGGA,123519,4807.038,N,01131.000,E,1,08,0.9,545.4,M,46.9,M,,*47\r\n$GPRMC,123520,A,4807.038,N,01131.000,E,0.13,309.62,120598,,,A*10\r\n"
    events = parse_nmea_sentences(payload)
    assert {e["event_type"] for e in events} >= {"nmea_gga", "nmea_rmc"}
```

- [ ] **Step 2: Run test to verify failure**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_mixed_parser.py -k nmea -v`
Expected: FAIL with missing parser function

- [ ] **Step 3: Implement minimal NMEA parser**

```python
# backend/app/mqtt/parsers/nmea.py

def parse_nmea_sentences(payload: bytes) -> list[dict[str, Any]]:
    events: list[dict[str, Any]] = []
    for raw_line in payload.splitlines():
        if not raw_line.startswith(b"$"):
            continue
        line = raw_line.decode("ascii", errors="ignore")
        talker_and_type = line[3:6].lower()
        if talker_and_type in {"gga", "rmc", "gsv", "gsa", "gst"}:
            events.append({"event_type": f"nmea_{talker_and_type}", "raw": line})
    return events
```

- [ ] **Step 4: Run NMEA tests**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_mixed_parser.py -k nmea -v`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add backend/app/mqtt/parsers/__init__.py backend/app/mqtt/parsers/nmea.py backend/tests/mqtt/test_mixed_parser.py
git commit -m "feat: parse NMEA sentences into observation events"
```

### Task 5: Build RTCM Summary + Mixed Payload Scanner

**Files:**
- Create: `backend/app/mqtt/parsers/rtcm.py`
- Create: `backend/app/mqtt/parsers/mixed.py`
- Test: `backend/tests/mqtt/test_mixed_parser.py`

- [ ] **Step 1: Write failing test for RTCM type summary in mixed payload**

```python
def test_parse_mixed_payload_returns_rtcm_summary() -> None:
    payload = (
        b"$GPGSV,1,1,04,02,17,213,40,05,29,121,37,12,66,314,44,25,14,048,39*71\r\n"
        + b"\xd3\x00\x13"
        + b"\x43\x50\x00"
        + b"\x00" * 16
    )
    result = parse_mixed_payload(payload)
    assert "rtcm_summary" in {e["event_type"] for e in result}
```

- [ ] **Step 2: Run test to verify failure**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_mixed_parser.py -k rtcm -v`
Expected: FAIL with missing mixed parser

- [ ] **Step 3: Implement RTCM scanner and orchestrator**

```python
# backend/app/mqtt/parsers/rtcm.py

def summarize_rtcm_types(payload: bytes) -> dict[int, int]:
    counts: dict[int, int] = {}
    i = 0
    while i + 5 < len(payload):
        if payload[i] != 0xD3:
            i += 1
            continue
        length = ((payload[i + 1] & 0x03) << 8) | payload[i + 2]
        end = i + 3 + length + 3
        if end > len(payload):
            break
        msg_type = ((payload[i + 3] << 4) | (payload[i + 4] >> 4)) & 0x0FFF
        counts[msg_type] = counts.get(msg_type, 0) + 1
        i = end
    return counts
```

```python
# backend/app/mqtt/parsers/mixed.py

def parse_mixed_payload(payload: bytes) -> list[dict[str, Any]]:
    events = parse_nmea_sentences(payload)
    rtcm_counts = summarize_rtcm_types(payload)
    if rtcm_counts:
        events.append({"event_type": "rtcm_summary", "rtcm_types": rtcm_counts})
    return events
```

- [ ] **Step 4: Run mixed parser tests**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_mixed_parser.py -v`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add backend/app/mqtt/parsers/rtcm.py backend/app/mqtt/parsers/mixed.py backend/tests/mqtt/test_mixed_parser.py
git commit -m "feat: add RTCM type summary and mixed payload parser"
```

### Task 6: Implement Parser Worker (Pending Queue -> Event/State)

**Files:**
- Create: `backend/app/mqtt/parser_worker.py`
- Modify: `backend/app/mqtt/repository.py`
- Test: `backend/tests/mqtt/test_parser_worker.py`

- [ ] **Step 1: Write failing worker test for status transitions and event writes**

```python
def test_worker_marks_success_and_writes_events(db: Session) -> None:
    raw = insert_raw_message(
        session=db,
        device_id="device-001",
        topic="terminal/device-001/raw",
        payload=b"$GPGGA,123519,3723.2475,N,12158.3416,W,1,12,0.8,10.0,M,-25.0,M,,*65\r\n",
        qos=1,
        retain=False,
    )
    processed = process_pending_batch(session=db, batch_size=10)
    assert processed == 1
    db.refresh(raw)
    assert raw.parse_status == "succeeded"
```

- [ ] **Step 2: Run test to verify failure**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_parser_worker.py -k worker -v`
Expected: FAIL with missing `process_pending_batch`

- [ ] **Step 3: Implement pending fetch with row locking and worker loop**

```python
# repository fetch query shape
select(TerminalMessageRaw).where(
    TerminalMessageRaw.parse_status == "pending"
).order_by(TerminalMessageRaw.received_at).with_for_update(
    skip_locked=True
).limit(batch_size)
```

```python
# parser_worker.py

def process_pending_batch(*, session: Session, batch_size: int) -> int:
    rows = fetch_pending_rows_for_update(session=session, batch_size=batch_size)
    for row in rows:
        mark_processing(session=session, row=row)
        events = parse_mixed_payload(row.payload_bytes)
        write_events_and_update_state(session=session, row=row, events=events)
        mark_succeeded(session=session, row=row)
    return len(rows)
```

- [ ] **Step 4: Run parser worker tests**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_parser_worker.py -v`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add backend/app/mqtt/parser_worker.py backend/app/mqtt/repository.py backend/tests/mqtt/test_parser_worker.py
git commit -m "feat: add async parser worker for pending raw messages"
```

### Task 7: Implement Local Archive Job + 90-Day Hot Retention Cleanup

**Files:**
- Create: `backend/app/mqtt/archive_job.py`
- Test: `backend/tests/mqtt/test_archive_job.py`
- Modify: `backend/app/mqtt/repository.py`

- [ ] **Step 1: Write failing archive test for byte-accurate output and manifest**

```python
def test_archive_writes_raw_bytes_and_manifest(tmp_path: Path, db: Session) -> None:
    seed_old_raw_rows(db, days_ago=91, payloads=[b"abc", b"\x01\x02"])
    output = run_archive_once(session=db, archive_base_dir=tmp_path)
    archived_file = Path(output[0].file_path)
    assert archived_file.read_bytes() == b"abc\x01\x02"
```

- [ ] **Step 2: Run test to verify failure**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_archive_job.py -v`
Expected: FAIL with missing archive job implementation

- [ ] **Step 3: Implement archive writer and retention cleanup**

```python
# backend/app/mqtt/archive_job.py

def archive_day(*, session: Session, device_id: str, day_start: datetime, output_path: Path) -> None:
    with output_path.open("wb") as f:
        rows = fetch_rows_for_archive(session=session, device_id=device_id, day_start=day_start)
        for row in rows:
            f.write(row.payload_bytes)
    write_manifest(session=session, device_id=device_id, day_start=day_start, output_path=output_path, rows=rows)
    delete_archived_hot_rows(session=session, archived_rows=rows)
```

- [ ] **Step 4: Run archive tests**

Run: `cd backend && ../.venv/bin/pytest tests/mqtt/test_archive_job.py -v`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add backend/app/mqtt/archive_job.py backend/app/mqtt/repository.py backend/tests/mqtt/test_archive_job.py
git commit -m "feat: add local archive job with hot retention cleanup"
```

### Task 8: Runtime Wiring, Compose Services, and Operational Docs

**Files:**
- Modify: `compose.yml`
- Modify: `compose.override.yml`
- Modify: `.env.production.example`
- Modify: `backend/README.md`
- Modify: `backend/pyproject.toml` (CLI scripts if used)

- [ ] **Step 1: Write failing smoke checklist (documented commands) for ingestor/worker startup**

```text
# backend/README.md verification block should include:
1) start db/backend/mqtt services
2) publish sample payload to terminal/device-001/raw
3) verify raw/event/state rows in postgres
```

- [ ] **Step 2: Run local startup command and capture expected failure before wiring**

Run: `docker compose up -d mqtt-ingestor mqtt-parser-worker`
Expected: FAIL with `no such service` before compose changes

- [ ] **Step 3: Add compose services + env vars + runbook commands**

```yaml
# compose service shape
mqtt-ingestor:
  command: ["python", "-m", "app.mqtt.ingestor"]
mqtt-parser-worker:
  command: ["python", "-m", "app.mqtt.parser_worker"]
```

- [ ] **Step 4: Run end-to-end verification commands**

Run:
- `cd backend && ../.venv/bin/pytest tests/mqtt -v`
- `cd backend && ../.venv/bin/pytest tests/api/routes -v`
- `cd backend && ../.venv/bin/mypy app`
- `cd backend && ../.venv/bin/ruff check app`

Expected: all PASS

- [ ] **Step 5: Commit**

```bash
git add compose.yml compose.override.yml .env.production.example backend/README.md backend/pyproject.toml
git commit -m "chore: wire mqtt ingestor and parser worker runtime configuration"
```

## Final Verification Gate

- [ ] Run: `cd backend && ../.venv/bin/pytest tests/mqtt tests/api/routes -v`
- [ ] Run: `cd backend && ../.venv/bin/mypy app`
- [ ] Run: `cd backend && ../.venv/bin/ruff check app && ../.venv/bin/ruff format app --check`
- [ ] Run a manual MQTT publish and verify DB rows (`raw`, `event`, `state`) update.
- [ ] Confirm archive output filename format: `device_<id>_<start>_<end>.txt` and content is byte-identical concatenation.

## Implementation Notes

- Keep ingestor lightweight; no heavy parsing inside MQTT callbacks.
- Any parse failure must retain raw row and set retry metadata.
- Do not add metadata into archive download file bytes; only payload concatenation is allowed.
- Use UTC timestamps everywhere for consistent retention and download range slicing.