feat: implement mqtt ingestion, parsing, and archive pipeline

2026-03-17 13:27:39 +08:00
parent 27101e9bff
commit f02d324335
16 changed files with 896 additions and 5 deletions
--- a/.env.production.example
+++ b/.env.production.example
@@ -42,6 +42,18 @@ POSTGRES_PASSWORD=password_CYmsGt

 SENTRY_DSN=

+# MQTT ingestion pipeline
+MQTT_HOST=101.35.119.226
+MQTT_PORT=1883
+MQTT_USERNAME=
+MQTT_PASSWORD=
+MQTT_TOPIC_PATTERN=terminal/+/raw
+MQTT_CLIENT_ID=spatialhub-mqtt-ingestor
+RAW_HOT_RETENTION_DAYS=90
+RAW_ARCHIVE_BASE_DIR=/data/archive/gnss
+PARSER_BATCH_SIZE=200
+PARSER_POLL_INTERVAL_MS=300
+
 # Docker images (本地构建，不需要远程 registry)
 DOCKER_IMAGE_BACKEND=backend
 DOCKER_IMAGE_FRONTEND=frontend
--- a/backend/README.md
+++ b/backend/README.md
@@ -170,3 +170,40 @@ The email templates are in `./backend/app/email-templates/`. Here, there are two
 Before continuing, ensure you have the [MJML extension](https://github.com/mjmlio/vscode-mjml) installed in your VS Code.

 Once you have the MJML extension installed, you can create a new email template in the `src` directory. After creating the new email template and with the `.mjml` file open in your editor, open the command palette with `Ctrl+Shift+P` and search for `MJML: Export to HTML`. This will convert the `.mjml` file to a `.html` file and now you can save it in the build directory.
+
+## MQTT Ingestion Pipeline
+
+This project includes a two-stage MQTT pipeline:
+
+- `mqtt-ingestor` subscribes to `terminal/{device_id}/raw` and stores payload bytes losslessly in `terminal_message_raw`.
+- `mqtt-parser-worker` consumes pending rows and writes parsed events/state.
+
+### Run locally with Docker Compose
+
+```console
+$ docker compose up -d db backend mqtt-ingestor mqtt-parser-worker
+```
+
+### Unit tests for MQTT modules
+
+The existing `tests/conftest.py` expects a PostgreSQL connection. For parser/repository unit tests that run with in-memory SQLite fixtures, use:
+
+```console
+$ cd backend
+$ PYTHONPATH=. ../.venv/bin/pytest tests/mqtt --confcutdir=tests/mqtt -q
+```
+
+### Verify end-to-end ingestion quickly
+
+1. Start services (`db`, `backend`, `mqtt-ingestor`, `mqtt-parser-worker`).
+2. Publish a sample message to `terminal/device-001/raw`.
+3. Query PostgreSQL tables:
+   - `terminal_message_raw`
+   - `terminal_observation_event`
+   - `terminal_device_state`
+
+### Archive run (manual trigger)
+
+```console
+$ docker compose run --rm mqtt-parser-worker python -m app.mqtt.archive_job
+```
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -121,7 +121,7 @@ class TerminalMessageRaw(SQLModel, table=True):
        default_factory=get_datetime_utc,
        sa_type=DateTime(timezone=True),  # type: ignore
    )
-    payload_bytes: bytes = Field(sa_type=LargeBinary)  # type: ignore
+    payload_bytes: bytes = Field(sa_type=LargeBinary)
    payload_size: int = 0
    payload_sha256: str = Field(default="", max_length=64, index=True)
    qos: int = 0
@@ -154,7 +154,7 @@ class TerminalObservationEvent(SQLModel, table=True):
        index=True,
    )
    event_type: str = Field(max_length=64, index=True)
-    payload_json: dict[str, Any] = Field(default_factory=dict, sa_type=JSON)  # type: ignore
+    payload_json: dict[str, Any] = Field(default_factory=dict, sa_type=JSON)
    created_at: datetime = Field(
        default_factory=get_datetime_utc,
        sa_type=DateTime(timezone=True),  # type: ignore
@@ -167,14 +167,14 @@ class TerminalDeviceState(SQLModel, table=True):
    device_id: str = Field(max_length=255, primary_key=True)
    last_event_time: datetime | None = Field(
        default=None, sa_type=DateTime(timezone=True)
-    )  # type: ignore
+    )  # type: ignore[call-overload]
    fix_status: str | None = Field(default=None, max_length=64)
    lat: float | None = None
    lon: float | None = None
    alt_m: float | None = None
    satellites_used: int | None = None
-    satellite_signal_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)  # type: ignore
-    rtcm_type_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)  # type: ignore
+    satellite_signal_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)
+    rtcm_type_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    updated_at: datetime = Field(
        default_factory=get_datetime_utc,
        sa_type=DateTime(timezone=True),  # type: ignore
--- a/backend/app/mqtt/archive_job.py
+++ b/backend/app/mqtt/archive_job.py
@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import hashlib
+import logging
+from datetime import datetime, timedelta
+from pathlib import Path
+
+from sqlmodel import Session, col, select
+
+from app.core.config import settings
+from app.core.db import engine
+from app.models import TerminalMessageRaw, TerminalRawArchiveManifest, get_datetime_utc
+
+logger = logging.getLogger(__name__)
+
+
+def _format_timestamp(value: datetime) -> str:
+    return value.strftime("%Y-%m-%dT%H-%M-%S")
+
+
+def _build_archive_path(
+    *,
+    archive_base_dir: Path,
+    device_id: str,
+    range_start: datetime,
+    range_end: datetime,
+) -> Path:
+    day = range_start.strftime("%Y-%m-%d")
+    output_dir = archive_base_dir / device_id / day
+    output_dir.mkdir(parents=True, exist_ok=True)
+    filename = (
+        f"device_{device_id}_{_format_timestamp(range_start)}"
+        f"_{_format_timestamp(range_end)}.txt"
+    )
+    return output_dir / filename
+
+
+def run_archive_once(
+    *,
+    session: Session,
+    archive_base_dir: Path | str,
+    retention_days: int,
+    now: datetime | None = None,
+) -> list[TerminalRawArchiveManifest]:
+    archive_root = Path(archive_base_dir)
+    ts_now = now or get_datetime_utc()
+    cutoff = ts_now - timedelta(days=retention_days)
+
+    statement = (
+        select(TerminalMessageRaw)
+        .where(TerminalMessageRaw.received_at < cutoff)
+        .order_by(
+            col(TerminalMessageRaw.device_id), col(TerminalMessageRaw.received_at)
+        )
+    )
+    rows = session.exec(statement).all()
+    if not rows:
+        return []
+
+    grouped: dict[tuple[str, str], list[TerminalMessageRaw]] = {}
+    for row in rows:
+        key = (row.device_id, row.received_at.strftime("%Y-%m-%d"))
+        grouped.setdefault(key, []).append(row)
+
+    manifests: list[TerminalRawArchiveManifest] = []
+    for (_, _), group_rows in grouped.items():
+        range_start = group_rows[0].received_at
+        range_end = group_rows[-1].received_at
+        output_path = _build_archive_path(
+            archive_base_dir=archive_root,
+            device_id=group_rows[0].device_id,
+            range_start=range_start,
+            range_end=range_end,
+        )
+
+        digest = hashlib.sha256()
+        record_count = 0
+        with output_path.open("wb") as out:
+            for row in group_rows:
+                out.write(row.payload_bytes)
+                digest.update(row.payload_bytes)
+                record_count += 1
+
+        manifest = TerminalRawArchiveManifest(
+            device_id=group_rows[0].device_id,
+            range_start=range_start,
+            range_end=range_end,
+            file_path=str(output_path),
+            file_size=output_path.stat().st_size,
+            sha256=digest.hexdigest(),
+            record_count=record_count,
+        )
+        session.add(manifest)
+        manifests.append(manifest)
+
+        for row in group_rows:
+            session.delete(row)
+
+    session.commit()
+
+    for manifest in manifests:
+        session.refresh(manifest)
+
+    return manifests
+
+
+def run() -> None:
+    logging.basicConfig(level=logging.INFO)
+    with Session(engine) as session:
+        manifests = run_archive_once(
+            session=session,
+            archive_base_dir=settings.RAW_ARCHIVE_BASE_DIR,
+            retention_days=settings.RAW_HOT_RETENTION_DAYS,
+        )
+    logger.info("Archived %s file(s)", len(manifests))
+
+
+if __name__ == "__main__":
+    run()
--- a/backend/app/mqtt/ingestor.py
+++ b/backend/app/mqtt/ingestor.py
@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import logging
+
+import paho.mqtt.client as mqtt  # type: ignore[import-not-found,import-untyped]
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.core.db import engine
+from app.mqtt.repository import insert_raw_message
+from app.mqtt.topic import extract_device_id
+
+logger = logging.getLogger(__name__)
+
+
+def _on_connect(
+    client: mqtt.Client,
+    _userdata: object,
+    _flags: dict[str, int],
+    reason_code: int,
+    _properties: mqtt.Properties | None = None,
+) -> None:
+    if reason_code != 0:
+        logger.error("MQTT connect failed: rc=%s", reason_code)
+        return
+
+    logger.info(
+        "Connected to MQTT broker %s:%s", settings.MQTT_HOST, settings.MQTT_PORT
+    )
+    client.subscribe(settings.MQTT_TOPIC_PATTERN, qos=1)
+
+
+def _on_message(
+    _client: mqtt.Client,
+    _userdata: object,
+    message: mqtt.MQTTMessage,
+) -> None:
+    try:
+        device_id = extract_device_id(message.topic)
+    except ValueError:
+        logger.warning("Skipping unexpected topic: %s", message.topic)
+        return
+
+    with Session(engine) as session:
+        insert_raw_message(
+            session=session,
+            device_id=device_id,
+            topic=message.topic,
+            payload=bytes(message.payload),
+            qos=message.qos,
+            retain=message.retain,
+        )
+
+
+def create_client() -> mqtt.Client:
+    client = mqtt.Client(
+        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+        client_id=settings.MQTT_CLIENT_ID,
+        clean_session=False,
+    )
+    if settings.MQTT_USERNAME:
+        client.username_pw_set(settings.MQTT_USERNAME, settings.MQTT_PASSWORD)
+
+    client.on_connect = _on_connect
+    client.on_message = _on_message
+    client.reconnect_delay_set(min_delay=1, max_delay=30)
+    return client
+
+
+def run() -> None:
+    logging.basicConfig(level=logging.INFO)
+    client = create_client()
+    client.connect(settings.MQTT_HOST, settings.MQTT_PORT, keepalive=60)
+    client.loop_forever()
+
+
+if __name__ == "__main__":
+    run()
--- a/backend/app/mqtt/parser_worker.py
+++ b/backend/app/mqtt/parser_worker.py
@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import logging
+import time
+from datetime import timedelta
+
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.core.db import engine
+from app.models import TerminalMessageRaw, get_datetime_utc
+from app.mqtt.parsers.mixed import parse_mixed_payload
+from app.mqtt.repository import (
+    fetch_pending_rows_for_update,
+    mark_failed,
+    mark_processing,
+    mark_succeeded,
+    write_events_and_update_state,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _next_retry_delay_seconds(attempts: int) -> int:
+    # Bounded exponential backoff: 30s, 60s, 120s, ... capped to 15m.
+    return int(min(30 * (2 ** max(attempts - 1, 0)), 900))
+
+
+def _process_one_row(*, session: Session, row: TerminalMessageRaw) -> None:
+    mark_processing(session=session, row=row)
+    session.commit()
+
+    try:
+        events = parse_mixed_payload(row.payload_bytes)
+        write_events_and_update_state(session=session, row=row, events=events)
+        mark_succeeded(session=session, row=row)
+    except Exception as exc:  # pragma: no cover - safety net path
+        delay = _next_retry_delay_seconds(row.parse_attempts)
+        mark_failed(
+            session=session,
+            row=row,
+            parse_error=str(exc),
+            next_retry_at=get_datetime_utc() + timedelta(seconds=delay),
+        )
+
+    session.commit()
+
+
+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:
+        _process_one_row(session=session, row=row)
+    return len(rows)
+
+
+def run() -> None:
+    logging.basicConfig(level=logging.INFO)
+    sleep_seconds = max(settings.PARSER_POLL_INTERVAL_MS, 50) / 1000.0
+
+    while True:
+        with Session(engine) as session:
+            processed = process_pending_batch(
+                session=session, batch_size=settings.PARSER_BATCH_SIZE
+            )
+        if processed == 0:
+            time.sleep(sleep_seconds)
+        else:
+            logger.info("Processed %s pending MQTT raw messages", processed)
+
+
+if __name__ == "__main__":
+    run()
--- a/backend/app/mqtt/parsers/init.py
+++ b/backend/app/mqtt/parsers/init.py
@@ -0,0 +1,5 @@
+from app.mqtt.parsers.mixed import parse_mixed_payload
+from app.mqtt.parsers.nmea import parse_nmea_sentences
+from app.mqtt.parsers.rtcm import summarize_rtcm_types
+
+__all__ = ["parse_mixed_payload", "parse_nmea_sentences", "summarize_rtcm_types"]
--- a/backend/app/mqtt/parsers/mixed.py
+++ b/backend/app/mqtt/parsers/mixed.py
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from typing import Any
+
+from app.mqtt.parsers.nmea import parse_nmea_sentences
+from app.mqtt.parsers.rtcm import summarize_rtcm_types
+
+
+def parse_mixed_payload(payload: bytes) -> list[dict[str, Any]]:
+    events = parse_nmea_sentences(payload)
+
+    rtcm_types = summarize_rtcm_types(payload)
+    if rtcm_types:
+        events.append(
+            {
+                "event_type": "rtcm_summary",
+                "payload": {
+                    "rtcm_message_count": sum(rtcm_types.values()),
+                    "rtcm_types": rtcm_types,
+                },
+            }
+        )
+
+    return events
--- a/backend/app/mqtt/parsers/nmea.py
+++ b/backend/app/mqtt/parsers/nmea.py
@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+from typing import Any
+
+SUPPORTED_NMEA_TYPES = {"GGA", "RMC", "GSV", "GSA", "GST"}
+
+
+def _to_float(value: str) -> float | None:
+    if not value:
+        return None
+    try:
+        return float(value)
+    except ValueError:
+        return None
+
+
+def _to_int(value: str) -> int | None:
+    if not value:
+        return None
+    try:
+        return int(value)
+    except ValueError:
+        return None
+
+
+def _parse_lat_lon(value: str, hemisphere: str) -> float | None:
+    if not value or not hemisphere:
+        return None
+
+    try:
+        degree_len = 2 if hemisphere in {"N", "S"} else 3
+        degrees = float(value[:degree_len])
+        minutes = float(value[degree_len:])
+    except ValueError:
+        return None
+
+    decimal = degrees + (minutes / 60.0)
+    if hemisphere in {"S", "W"}:
+        decimal *= -1
+    return decimal
+
+
+def _extract_sentence_type(sentence: str) -> str | None:
+    if not sentence.startswith("$"):
+        return None
+    if len(sentence) < 6:
+        return None
+    nmea_type = sentence[3:6].upper()
+    if nmea_type not in SUPPORTED_NMEA_TYPES:
+        return None
+    return nmea_type
+
+
+def _parse_nmea_fields(nmea_type: str, fields: list[str]) -> dict[str, Any]:
+    payload: dict[str, Any] = {"sentence_type": nmea_type, "raw_fields": fields}
+
+    if nmea_type == "GGA":
+        payload.update(
+            {
+                "utc_time": fields[1] if len(fields) > 1 else None,
+                "lat": _parse_lat_lon(
+                    fields[2] if len(fields) > 2 else "",
+                    fields[3] if len(fields) > 3 else "",
+                ),
+                "lon": _parse_lat_lon(
+                    fields[4] if len(fields) > 4 else "",
+                    fields[5] if len(fields) > 5 else "",
+                ),
+                "fix_quality": _to_int(fields[6] if len(fields) > 6 else ""),
+                "satellites_used": _to_int(fields[7] if len(fields) > 7 else ""),
+                "hdop": _to_float(fields[8] if len(fields) > 8 else ""),
+                "altitude_m": _to_float(fields[9] if len(fields) > 9 else ""),
+            }
+        )
+    elif nmea_type == "RMC":
+        payload.update(
+            {
+                "utc_time": fields[1] if len(fields) > 1 else None,
+                "status": fields[2] if len(fields) > 2 else None,
+                "lat": _parse_lat_lon(
+                    fields[3] if len(fields) > 3 else "",
+                    fields[4] if len(fields) > 4 else "",
+                ),
+                "lon": _parse_lat_lon(
+                    fields[5] if len(fields) > 5 else "",
+                    fields[6] if len(fields) > 6 else "",
+                ),
+                "speed_knots": _to_float(fields[7] if len(fields) > 7 else ""),
+                "course_deg": _to_float(fields[8] if len(fields) > 8 else ""),
+                "date": fields[9] if len(fields) > 9 else None,
+            }
+        )
+    elif nmea_type == "GSV":
+        payload.update(
+            {
+                "total_messages": _to_int(fields[1] if len(fields) > 1 else ""),
+                "message_index": _to_int(fields[2] if len(fields) > 2 else ""),
+                "satellites_in_view": _to_int(fields[3] if len(fields) > 3 else ""),
+                "snr_values": [
+                    _to_int(fields[i])
+                    for i in range(7, len(fields), 4)
+                    if i < len(fields)
+                ],
+            }
+        )
+    elif nmea_type == "GSA":
+        payload.update(
+            {
+                "mode": fields[1] if len(fields) > 1 else None,
+                "fix_type": _to_int(fields[2] if len(fields) > 2 else ""),
+                "pdop": _to_float(fields[15] if len(fields) > 15 else ""),
+                "hdop": _to_float(fields[16] if len(fields) > 16 else ""),
+                "vdop": _to_float(fields[17] if len(fields) > 17 else ""),
+            }
+        )
+    elif nmea_type == "GST":
+        payload.update(
+            {
+                "utc_time": fields[1] if len(fields) > 1 else None,
+                "rms": _to_float(fields[2] if len(fields) > 2 else ""),
+                "std_major": _to_float(fields[3] if len(fields) > 3 else ""),
+                "std_minor": _to_float(fields[4] if len(fields) > 4 else ""),
+            }
+        )
+
+    return payload
+
+
+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
+
+        sentence = raw_line.decode("ascii", errors="ignore").strip()
+        nmea_type = _extract_sentence_type(sentence)
+        if nmea_type is None:
+            continue
+
+        main_part = sentence.split("*", 1)[0]
+        fields = main_part.split(",")
+        parsed_payload = _parse_nmea_fields(nmea_type, fields)
+        events.append(
+            {"event_type": f"nmea_{nmea_type.lower()}", "payload": parsed_payload}
+        )
+
+    return events
--- a/backend/app/mqtt/parsers/rtcm.py
+++ b/backend/app/mqtt/parsers/rtcm.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+
+def summarize_rtcm_types(payload: bytes) -> dict[int, int]:
+    counts: dict[int, int] = {}
+    i = 0
+
+    while i + 6 <= len(payload):
+        if payload[i] != 0xD3:
+            i += 1
+            continue
+
+        length = ((payload[i + 1] & 0x03) << 8) | payload[i + 2]
+        frame_end = i + 3 + length + 3
+        if frame_end > len(payload):
+            break
+
+        message_type = ((payload[i + 3] << 4) | (payload[i + 4] >> 4)) & 0x0FFF
+        counts[message_type] = counts.get(message_type, 0) + 1
+        i = frame_end
+
+    return counts
--- a/backend/app/mqtt/repository.py
+++ b/backend/app/mqtt/repository.py
@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+import hashlib
+from datetime import datetime
+from typing import Any
+
+from sqlmodel import Session, col, select
+
+from app.models import (
+    TerminalDeviceState,
+    TerminalMessageRaw,
+    TerminalObservationEvent,
+    get_datetime_utc,
+)
+
+
+def insert_raw_message(
+    *,
+    session: Session,
+    device_id: str,
+    topic: str,
+    payload: bytes,
+    qos: int,
+    retain: bool,
+    received_at: datetime | None = None,
+) -> TerminalMessageRaw:
+    ts = received_at or get_datetime_utc()
+    row = TerminalMessageRaw(
+        device_id=device_id,
+        topic=topic,
+        received_at=ts,
+        payload_bytes=payload,
+        payload_size=len(payload),
+        payload_sha256=hashlib.sha256(payload).hexdigest(),
+        qos=qos,
+        retain=retain,
+        parse_status="pending",
+        created_at=ts,
+    )
+    session.add(row)
+    session.commit()
+    session.refresh(row)
+    return row
+
+
+def fetch_pending_rows_for_update(
+    *, session: Session, batch_size: int
+) -> list[TerminalMessageRaw]:
+    statement = (
+        select(TerminalMessageRaw)
+        .where(TerminalMessageRaw.parse_status == "pending")
+        .order_by(col(TerminalMessageRaw.received_at))
+        .limit(batch_size)
+    )
+
+    bind = session.get_bind()
+    if bind is not None and bind.dialect.name != "sqlite":
+        statement = statement.with_for_update(skip_locked=True)
+
+    rows = session.exec(statement).all()
+    return list(rows)
+
+
+def mark_processing(*, session: Session, row: TerminalMessageRaw) -> None:
+    row.parse_status = "processing"
+    row.parse_attempts += 1
+    row.parse_error = None
+    session.add(row)
+
+
+def mark_succeeded(*, session: Session, row: TerminalMessageRaw) -> None:
+    row.parse_status = "succeeded"
+    row.parse_error = None
+    row.next_retry_at = None
+    session.add(row)
+
+
+def mark_failed(
+    *,
+    session: Session,
+    row: TerminalMessageRaw,
+    parse_error: str,
+    next_retry_at: datetime | None,
+) -> None:
+    row.parse_status = "failed"
+    row.parse_error = parse_error[:1024]
+    row.next_retry_at = next_retry_at
+    session.add(row)
+
+
+def write_events_and_update_state(
+    *,
+    session: Session,
+    row: TerminalMessageRaw,
+    events: list[dict[str, Any]],
+) -> None:
+    if not events:
+        state = session.get(TerminalDeviceState, row.device_id)
+        if state is None:
+            state = TerminalDeviceState(device_id=row.device_id)
+        state.last_event_time = row.received_at
+        state.updated_at = get_datetime_utc()
+        session.add(state)
+        return
+
+    state = session.get(TerminalDeviceState, row.device_id)
+    if state is None:
+        state = TerminalDeviceState(device_id=row.device_id)
+
+    for event in events:
+        payload = event.get("payload", {})
+        event_type = str(event.get("event_type", "unknown"))
+        db_event = TerminalObservationEvent(
+            raw_id=row.id,
+            device_id=row.device_id,
+            event_time=row.received_at,
+            event_type=event_type,
+            payload_json=payload if isinstance(payload, dict) else {"value": payload},
+        )
+        session.add(db_event)
+
+        if event_type == "nmea_gga":
+            state.lat = payload.get("lat")
+            state.lon = payload.get("lon")
+            state.alt_m = payload.get("altitude_m")
+            state.satellites_used = payload.get("satellites_used")
+            fix_quality = payload.get("fix_quality")
+            if fix_quality is not None:
+                state.fix_status = str(fix_quality)
+        elif event_type == "nmea_rmc":
+            if payload.get("lat") is not None:
+                state.lat = payload.get("lat")
+            if payload.get("lon") is not None:
+                state.lon = payload.get("lon")
+            status = payload.get("status")
+            if status is not None:
+                state.fix_status = str(status)
+        elif event_type == "nmea_gsv":
+            state.satellite_signal_summary = {
+                "snr_values": payload.get("snr_values", []),
+                "satellites_in_view": payload.get("satellites_in_view"),
+            }
+        elif event_type == "rtcm_summary":
+            state.rtcm_type_summary = payload.get("rtcm_types", {})
+
+    state.last_event_time = row.received_at
+    state.updated_at = get_datetime_utc()
+    session.add(state)
--- a/backend/tests/mqtt/test_archive_job.py
+++ b/backend/tests/mqtt/test_archive_job.py
@@ -0,0 +1,54 @@
+from datetime import timedelta
+from pathlib import Path
+
+from sqlmodel import Session, SQLModel, create_engine, select
+
+from app.models import TerminalMessageRaw, get_datetime_utc
+from app.mqtt.archive_job import run_archive_once
+from app.mqtt.repository import insert_raw_message
+
+
+def _session() -> Session:
+    engine = create_engine("sqlite://")
+    SQLModel.metadata.create_all(engine)
+    return Session(engine)
+
+
+def test_archive_writes_raw_bytes_and_manifest(tmp_path: Path) -> None:
+    with _session() as session:
+        current = get_datetime_utc()
+        old_ts = current - timedelta(days=91)
+
+        insert_raw_message(
+            session=session,
+            device_id="device-001",
+            topic="terminal/device-001/raw",
+            payload=b"abc",
+            qos=1,
+            retain=False,
+            received_at=old_ts,
+        )
+        insert_raw_message(
+            session=session,
+            device_id="device-001",
+            topic="terminal/device-001/raw",
+            payload=b"\x01\x02",
+            qos=1,
+            retain=False,
+            received_at=old_ts,
+        )
+
+        manifests = run_archive_once(
+            session=session,
+            archive_base_dir=tmp_path,
+            retention_days=90,
+            now=current,
+        )
+
+        assert len(manifests) == 1
+        archived_file = Path(manifests[0].file_path)
+        assert archived_file.exists()
+        assert archived_file.read_bytes() == b"abc\x01\x02"
+
+        remaining = session.exec(select(TerminalMessageRaw)).all()
+        assert remaining == []
--- a/backend/tests/mqtt/test_mixed_parser.py
+++ b/backend/tests/mqtt/test_mixed_parser.py
@@ -0,0 +1,27 @@
+from app.mqtt.parsers.mixed import parse_mixed_payload
+from app.mqtt.parsers.nmea import parse_nmea_sentences
+
+
+def test_parse_nmea_sentences_extracts_gga_and_rmc() -> None:
+    payload = (
+        b"$GPGGA,123519,3723.2475,N,12158.3416,W,1,12,0.8,10.0,M,-25.0,M,,*65\r\n"
+        b"$GPRMC,123520,A,3723.2475,N,12158.3416,W,0.13,309.62,120598,,,A*74\r\n"
+    )
+
+    events = parse_nmea_sentences(payload)
+
+    assert {e["event_type"] for e in events} >= {"nmea_gga", "nmea_rmc"}
+
+
+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\x04"
+        + b"\x43\x50\x00\x00"
+        + b"\x00\x00\x00"
+    )
+
+    events = parse_mixed_payload(payload)
+
+    event_types = {event["event_type"] for event in events}
+    assert "rtcm_summary" in event_types
--- a/backend/tests/mqtt/test_parser_worker.py
+++ b/backend/tests/mqtt/test_parser_worker.py
@@ -0,0 +1,53 @@
+from sqlmodel import Session, SQLModel, create_engine, select
+
+from app.models import TerminalDeviceState, TerminalMessageRaw, TerminalObservationEvent
+from app.mqtt.parser_worker import process_pending_batch
+from app.mqtt.repository import insert_raw_message
+
+
+def _session() -> Session:
+    engine = create_engine("sqlite://")
+    SQLModel.metadata.create_all(engine)
+    return Session(engine)
+
+
+def test_insert_raw_message_sets_pending_status() -> None:
+    with _session() as session:
+        row = insert_raw_message(
+            session=session,
+            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
+        assert len(row.payload_sha256) == 64
+
+
+def test_worker_marks_success_and_writes_events() -> None:
+    with _session() as session:
+        raw = insert_raw_message(
+            session=session,
+            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=session, batch_size=10)
+
+        assert processed == 1
+
+        refreshed = session.get(TerminalMessageRaw, raw.id)
+        assert refreshed is not None
+        assert refreshed.parse_status == "succeeded"
+
+        events = session.exec(select(TerminalObservationEvent)).all()
+        assert len(events) >= 1
+
+        state = session.get(TerminalDeviceState, "device-001")
+        assert state is not None
--- a/compose.override.yml
+++ b/compose.override.yml
@@ -87,6 +87,20 @@ services:
      SMTP_TLS: "false"
      EMAILS_FROM_EMAIL: "noreply@example.com"

+  mqtt-ingestor:
+    restart: "no"
+    command:
+      - python
+      - -m
+      - app.mqtt.ingestor
+
+  mqtt-parser-worker:
+    restart: "no"
+    command:
+      - python
+      - -m
+      - app.mqtt.parser_worker
+
  mailcatcher:
    image: schickling/mailcatcher
    ports:
--- a/compose.yml
+++ b/compose.yml
@@ -136,6 +136,83 @@ services:
      # Enable redirection for HTTP and HTTPS
      - traefik.http.routers.${STACK_NAME?Variable not set}-backend-http.middlewares=https-redirect

+  mqtt-ingestor:
+    image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}'
+    restart: always
+    networks:
+      - default
+    depends_on:
+      db:
+        condition: service_healthy
+        restart: true
+      prestart:
+        condition: service_completed_successfully
+    command:
+      - python
+      - -m
+      - app.mqtt.ingestor
+    env_file:
+      - .env
+    environment:
+      - ENVIRONMENT=${ENVIRONMENT}
+      - SECRET_KEY=${SECRET_KEY?Variable not set}
+      - FIRST_SUPERUSER=${FIRST_SUPERUSER?Variable not set}
+      - FIRST_SUPERUSER_PASSWORD=${FIRST_SUPERUSER_PASSWORD?Variable not set}
+      - POSTGRES_SERVER=db
+      - POSTGRES_PORT=${POSTGRES_PORT}
+      - POSTGRES_DB=${POSTGRES_DB}
+      - POSTGRES_USER=${POSTGRES_USER?Variable not set}
+      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set}
+      - MQTT_HOST=${MQTT_HOST-101.35.119.226}
+      - MQTT_PORT=${MQTT_PORT-1883}
+      - MQTT_USERNAME=${MQTT_USERNAME}
+      - MQTT_PASSWORD=${MQTT_PASSWORD}
+      - MQTT_TOPIC_PATTERN=${MQTT_TOPIC_PATTERN-terminal/+/raw}
+      - MQTT_CLIENT_ID=${MQTT_CLIENT_ID-spatialhub-mqtt-ingestor}
+      - RAW_HOT_RETENTION_DAYS=${RAW_HOT_RETENTION_DAYS-90}
+      - RAW_ARCHIVE_BASE_DIR=${RAW_ARCHIVE_BASE_DIR-/data/archive/gnss}
+      - PARSER_BATCH_SIZE=${PARSER_BATCH_SIZE-200}
+      - PARSER_POLL_INTERVAL_MS=${PARSER_POLL_INTERVAL_MS-300}
+    build:
+      context: .
+      dockerfile: backend/Dockerfile
+    volumes:
+      - mqtt-archive-data:/data/archive/gnss
+
+  mqtt-parser-worker:
+    image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}'
+    restart: always
+    networks:
+      - default
+    depends_on:
+      db:
+        condition: service_healthy
+        restart: true
+      prestart:
+        condition: service_completed_successfully
+    command:
+      - python
+      - -m
+      - app.mqtt.parser_worker
+    env_file:
+      - .env
+    environment:
+      - ENVIRONMENT=${ENVIRONMENT}
+      - SECRET_KEY=${SECRET_KEY?Variable not set}
+      - FIRST_SUPERUSER=${FIRST_SUPERUSER?Variable not set}
+      - FIRST_SUPERUSER_PASSWORD=${FIRST_SUPERUSER_PASSWORD?Variable not set}
+      - POSTGRES_SERVER=db
+      - POSTGRES_PORT=${POSTGRES_PORT}
+      - POSTGRES_DB=${POSTGRES_DB}
+      - POSTGRES_USER=${POSTGRES_USER?Variable not set}
+      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set}
+      - RAW_HOT_RETENTION_DAYS=${RAW_HOT_RETENTION_DAYS-90}
+      - PARSER_BATCH_SIZE=${PARSER_BATCH_SIZE-200}
+      - PARSER_POLL_INTERVAL_MS=${PARSER_POLL_INTERVAL_MS-300}
+    build:
+      context: .
+      dockerfile: backend/Dockerfile
+
  frontend:
    image: '${DOCKER_IMAGE_FRONTEND?Variable not set}:${TAG-latest}'
    restart: always
@@ -167,6 +244,7 @@ services:
      - traefik.http.routers.${STACK_NAME?Variable not set}-frontend-http.middlewares=https-redirect
 volumes:
  app-db-data:
+  mqtt-archive-data:

 networks:
  traefik-public: