Merge branch 'codex/mqtt-raw-ingestion'

.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

.gitignore

@@ -9,3 +9,4 @@ node_modules/
 .env.production
 .venv/
 .playwright-cli/
+.worktrees/

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
+```

backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py

@@ -0,0 +1,223 @@
+"""add mqtt pipeline tables
+
+Revision ID: a1b2c3d4e5f6
+Revises: 0b117fab4208
+Create Date: 2026-03-17 12:30:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+
+
+# revision identifiers, used by Alembic.
+revision = "a1b2c3d4e5f6"
+down_revision = "0b117fab4208"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "terminal_message_raw",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+        ),
+        sa.Column("topic", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("received_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("payload_bytes", sa.LargeBinary(), nullable=False),
+        sa.Column("payload_size", sa.Integer(), nullable=False),
+        sa.Column(
+            "payload_sha256", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False
+        ),
+        sa.Column("qos", sa.Integer(), nullable=False),
+        sa.Column("retain", sa.Boolean(), nullable=False),
+        sa.Column(
+            "parse_status", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False
+        ),
+        sa.Column("parse_attempts", sa.Integer(), nullable=False),
+        sa.Column("next_retry_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "parse_error", sqlmodel.sql.sqltypes.AutoString(length=1024), nullable=True
+        ),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_terminal_message_raw_device_id"),
+        "terminal_message_raw",
+        ["device_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_message_raw_next_retry_at"),
+        "terminal_message_raw",
+        ["next_retry_at"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_message_raw_parse_status"),
+        "terminal_message_raw",
+        ["parse_status"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_message_raw_payload_sha256"),
+        "terminal_message_raw",
+        ["payload_sha256"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_message_raw_received_at"),
+        "terminal_message_raw",
+        ["received_at"],
+        unique=False,
+    )
+
+    op.create_table(
+        "terminal_observation_event",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("raw_id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+        ),
+        sa.Column("event_time", sa.DateTime(timezone=True), nullable=False),
+        sa.Column(
+            "event_type", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False
+        ),
+        sa.Column("payload_json", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["raw_id"], ["terminal_message_raw.id"]),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_terminal_observation_event_device_id"),
+        "terminal_observation_event",
+        ["device_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_observation_event_event_time"),
+        "terminal_observation_event",
+        ["event_time"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_observation_event_event_type"),
+        "terminal_observation_event",
+        ["event_type"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_observation_event_raw_id"),
+        "terminal_observation_event",
+        ["raw_id"],
+        unique=False,
+    )
+
+    op.create_table(
+        "terminal_device_state",
+        sa.Column(
+            "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+        ),
+        sa.Column("last_event_time", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("fix_status", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True),
+        sa.Column("lat", sa.Float(), nullable=True),
+        sa.Column("lon", sa.Float(), nullable=True),
+        sa.Column("alt_m", sa.Float(), nullable=True),
+        sa.Column("satellites_used", sa.Integer(), nullable=True),
+        sa.Column("satellite_signal_summary", sa.JSON(), nullable=True),
+        sa.Column("rtcm_type_summary", sa.JSON(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("device_id"),
+    )
+
+    op.create_table(
+        "terminal_raw_archive_manifest",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+        ),
+        sa.Column("range_start", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("range_end", sa.DateTime(timezone=True), nullable=False),
+        sa.Column(
+            "file_path", sqlmodel.sql.sqltypes.AutoString(length=1024), nullable=False
+        ),
+        sa.Column("file_size", sa.Integer(), nullable=False),
+        sa.Column("sha256", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column("record_count", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("file_path"),
+    )
+    op.create_index(
+        op.f("ix_terminal_raw_archive_manifest_device_id"),
+        "terminal_raw_archive_manifest",
+        ["device_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_raw_archive_manifest_range_end"),
+        "terminal_raw_archive_manifest",
+        ["range_end"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_terminal_raw_archive_manifest_range_start"),
+        "terminal_raw_archive_manifest",
+        ["range_start"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f("ix_terminal_raw_archive_manifest_range_start"),
+        table_name="terminal_raw_archive_manifest",
+    )
+    op.drop_index(
+        op.f("ix_terminal_raw_archive_manifest_range_end"),
+        table_name="terminal_raw_archive_manifest",
+    )
+    op.drop_index(
+        op.f("ix_terminal_raw_archive_manifest_device_id"),
+        table_name="terminal_raw_archive_manifest",
+    )
+    op.drop_table("terminal_raw_archive_manifest")
+    op.drop_table("terminal_device_state")
+    op.drop_index(
+        op.f("ix_terminal_observation_event_raw_id"),
+        table_name="terminal_observation_event",
+    )
+    op.drop_index(
+        op.f("ix_terminal_observation_event_event_type"),
+        table_name="terminal_observation_event",
+    )
+    op.drop_index(
+        op.f("ix_terminal_observation_event_event_time"),
+        table_name="terminal_observation_event",
+    )
+    op.drop_index(
+        op.f("ix_terminal_observation_event_device_id"),
+        table_name="terminal_observation_event",
+    )
+    op.drop_table("terminal_observation_event")
+    op.drop_index(
+        op.f("ix_terminal_message_raw_received_at"), table_name="terminal_message_raw"
+    )
+    op.drop_index(
+        op.f("ix_terminal_message_raw_payload_sha256"),
+        table_name="terminal_message_raw",
+    )
+    op.drop_index(
+        op.f("ix_terminal_message_raw_parse_status"), table_name="terminal_message_raw"
+    )
+    op.drop_index(
+        op.f("ix_terminal_message_raw_next_retry_at"), table_name="terminal_message_raw"
+    )
+    op.drop_index(
+        op.f("ix_terminal_message_raw_device_id"), table_name="terminal_message_raw"
+    )
+    op.drop_table("terminal_message_raw")

backend/app/core/config.py

@@ -56,6 +56,17 @@ class Settings(BaseSettings):
     POSTGRES_PASSWORD: str = ""
     POSTGRES_DB: str = ""
 
+    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
+
     @computed_field  # type: ignore[prop-decorator]
     @property
     def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn:

backend/app/models.py

@@ -1,8 +1,9 @@
 import uuid
 from datetime import datetime, timezone
+from typing import Any
 
 from pydantic import EmailStr
-from sqlalchemy import DateTime
+from sqlalchemy import JSON, DateTime, LargeBinary
 from sqlmodel import Field, Relationship, SQLModel
 
 
@@ -53,7 +54,9 @@ class User(UserBase, table=True):
         default_factory=get_datetime_utc,
         sa_type=DateTime(timezone=True),  # type: ignore
     )
-    locations: list["Location"] = Relationship(back_populates="owner", cascade_delete=True)
+    locations: list["Location"] = Relationship(
+        back_populates="owner", cascade_delete=True
+    )
 
 
 # Properties to return via API, id is always required
@@ -108,6 +111,99 @@ class LocationsPublic(SQLModel):
     count: int
 
 
+class TerminalMessageRaw(SQLModel, table=True):
+    __tablename__ = "terminal_message_raw"
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    device_id: str = Field(max_length=255, index=True)
+    topic: str = Field(max_length=255)
+    received_at: datetime = Field(
+        default_factory=get_datetime_utc,
+        sa_type=DateTime(timezone=True),  # 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
+    retain: bool = False
+    parse_status: str = Field(default="pending", max_length=32, index=True)
+    parse_attempts: int = 0
+    next_retry_at: datetime | None = Field(
+        default=None,
+        sa_type=DateTime(timezone=True),  # type: ignore
+        index=True,
+    )
+    parse_error: str | None = Field(default=None, max_length=1024)
+    created_at: datetime = Field(
+        default_factory=get_datetime_utc,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+
+
+class TerminalObservationEvent(SQLModel, table=True):
+    __tablename__ = "terminal_observation_event"
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    raw_id: uuid.UUID = Field(
+        foreign_key="terminal_message_raw.id", nullable=False, index=True
+    )
+    device_id: str = Field(max_length=255, index=True)
+    event_time: datetime = Field(
+        default_factory=get_datetime_utc,
+        sa_type=DateTime(timezone=True),  # type: ignore
+        index=True,
+    )
+    event_type: str = Field(max_length=64, index=True)
+    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
+    )
+
+
+class TerminalDeviceState(SQLModel, table=True):
+    __tablename__ = "terminal_device_state"
+
+    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[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)
+    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
+    )
+
+
+class TerminalRawArchiveManifest(SQLModel, table=True):
+    __tablename__ = "terminal_raw_archive_manifest"
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    device_id: str = Field(max_length=255, index=True)
+    range_start: datetime = Field(
+        sa_type=DateTime(timezone=True),  # type: ignore
+        index=True,
+    )
+    range_end: datetime = Field(
+        sa_type=DateTime(timezone=True),  # type: ignore
+        index=True,
+    )
+    file_path: str = Field(max_length=1024, unique=True)
+    file_size: int = 0
+    sha256: str = Field(max_length=64)
+    record_count: int = 0
+    created_at: datetime = Field(
+        default_factory=get_datetime_utc,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+
+
 # Generic message
 class Message(SQLModel):
     message: str

backend/app/mqtt/__init__.py

@@ -0,0 +1 @@
+"""MQTT ingestion pipeline package."""

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()

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()

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()

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"]

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

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

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

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)

backend/app/mqtt/topic.py

@@ -0,0 +1,10 @@
+import re
+
+TOPIC_PATTERN = re.compile(r"^terminal/(?P<device_id>[^/]+)/raw$")
+
+
+def extract_device_id(topic: str) -> str:
+    match = TOPIC_PATTERN.match(topic)
+    if match is None:
+        raise ValueError("topic must match terminal/{device_id}/raw")
+    return match.group("device_id")

backend/pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
     "sentry-sdk[fastapi]>=2.0.0,<3.0.0",
     "pyjwt<3.0.0,>=2.8.0",
     "pwdlib[argon2,bcrypt]>=0.3.0",
+    "paho-mqtt<3.0.0,>=2.1.0",
 ]
 
 [dependency-groups]

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 == []

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

backend/tests/mqtt/test_models.py

@@ -0,0 +1,21 @@
+from sqlmodel import Session, SQLModel, create_engine
+
+from app.models import TerminalMessageRaw
+
+
+def test_terminal_message_raw_model_can_persist() -> None:
+    engine = create_engine("sqlite://")
+    SQLModel.metadata.create_all(engine)
+
+    with Session(engine) as session:
+        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",
+        )
+        session.add(row)
+        session.commit()
+        session.refresh(row)
+
+    assert row.id is not None
+    assert row.parse_status == "pending"

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

backend/tests/mqtt/test_topic.py

@@ -0,0 +1,12 @@
+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/\\{device_id\\}/raw"):
+        extract_device_id("terminal/device-001/nmea")

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:

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:

docs/superpowers/plans/2026-03-17-mqtt-raw-ingestion-implementation.md

@@ -0,0 +1,581 @@
+# 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.

docs/superpowers/specs/2026-03-17-mqtt-raw-ingestion-design.md

@@ -0,0 +1,252 @@
+# MQTT 原始数据接收与保存设计（双阶段）
+
+## 1. 背景与目标
+
+在现有 `full-stack-fastapi-template` 工程基础上，新增 MQTT 采集能力，用于接收终端上报的 GNSS 数据并可靠保存，支持后续：
+
+- 前端秒级实时展示（设备看板 + 逐条观测流）
+- 按天/时间段下载原始数据供外部软件分析
+
+本设计阶段仅实现“接收 + 保存 + 可解析落库链路”，不实现前端页面。
+
+## 2. 已确认约束
+
+- 部署环境：腾讯云，同机已有 EMQX
+- Broker 地址：`101.35.119.226:1883`
+- 鉴权方式：用户名/密码
+- Topic：仅 `terminal/{device_id}/raw`
+- `device_id`：直接从 topic 中提取
+- 上报形态：每 1~10 秒一条消息，每条消息为“完整帧/完整语句集合”
+- 负载内容：可能为 NMEA only、RTCM3.2 only 或二者混合
+- NMEA 类型：GGA / RMC / GSV / GSA / GST
+- RTCM：本期仅做“消息类型级别摘要”
+- 存储策略：90 天热数据 + 本地冷归档
+- 下载要求：下载内容必须是原始 MQTT payload 字节流，不附加元信息
+- 下载文件后缀：`.txt`（内容可包含二进制字节）
+
+## 3. 架构方案与结论
+
+### 3.1 备选方案
+
+1. FastAPI 内嵌 MQTT 客户端
+2. DB 队列双阶段（独立 ingestor + parser worker）
+3. 外部消息队列三阶段
+
+### 3.2 选型结论
+
+采用 **方案 2：DB 队列双阶段**。
+
+理由：
+
+- 避免 FastAPI 多 worker 下重复订阅/重复入库风险
+- 接收链路与解析链路解耦，稳定性更高
+- 原始数据先落库，可重试、可回放、可追溯
+- 为后续实时展示和下载提供稳定基础
+
+## 4. 组件职责
+
+### 4.1 mqtt-ingestor（采集层）
+
+- 常驻独立进程，连接 EMQX 并订阅 `terminal/+/raw`
+- 回调中仅执行轻量逻辑：
+  - 提取 `device_id`
+  - 记录 `topic/qos/retain/received_at`
+  - 原样保存 `payload_bytes`
+  - 计算 `payload_sha256` 与 `payload_size`
+  - 插入 `terminal_message_raw`（`parse_status='pending'`）
+- 不执行 NMEA/RTCM 深解析
+
+### 4.2 parser-worker（解析层）
+
+- 轮询 `terminal_message_raw` 中待解析记录
+- 解析 NMEA + RTCM3.2 摘要
+- 写入事件表 `terminal_observation_event`
+- Upsert 最新状态表 `terminal_device_state`
+- 失败可重试，错误可观测，不丢原始数据
+
+### 4.3 backend API（服务层）
+
+- 本期不新增前端页面
+- 后续基于 `event + state + archive manifest` 增加秒级展示与下载接口
+
+## 5. 数据模型设计
+
+### 5.1 terminal_message_raw（热数据原始表）
+
+用途：保存原始消息，作为解析队列与下载来源。
+
+建议字段：
+
+- `id` (uuid, pk)
+- `device_id` (text, index)
+- `topic` (text)
+- `received_at` (timestamptz, index)
+- `payload_bytes` (bytea)
+- `payload_size` (int)
+- `payload_sha256` (char(64), index)
+- `qos` (smallint)
+- `retain` (bool)
+- `parse_status` (`pending|processing|succeeded|failed`, index)
+- `parse_attempts` (int, default 0)
+- `next_retry_at` (timestamptz, nullable, index)
+- `parse_error` (text, nullable)
+- `created_at` (timestamptz)
+
+### 5.2 terminal_observation_event（解析事件流）
+
+用途：支撑逐条观测流展示与历史分析。
+
+建议字段：
+
+- `id` (uuid, pk)
+- `raw_id` (uuid, fk -> terminal_message_raw.id, index)
+- `device_id` (text, index)
+- `event_time` (timestamptz, index)
+- `event_type` (text)
+  示例：`nmea_gga` / `nmea_rmc` / `nmea_gsv` / `nmea_gsa` / `nmea_gst` / `rtcm_summary`
+- `payload_json` (jsonb)
+- `created_at` (timestamptz)
+
+### 5.3 terminal_device_state（设备最新状态）
+
+用途：支撑前端设备看板秒级刷新。
+
+建议字段：
+
+- `device_id` (text, pk)
+- `last_event_time` (timestamptz)
+- `fix_status` (text)
+- `lat` / `lon` (double precision)
+- `alt_m` (double precision)
+- `satellites_used` (int)
+- `satellite_signal_summary` (jsonb)
+- `rtcm_type_summary` (jsonb)
+- `updated_at` (timestamptz)
+
+### 5.4 terminal_raw_archive_manifest（冷归档索引）
+
+用途：记录本地归档文件与时间范围映射。
+
+建议字段：
+
+- `id` (uuid, pk)
+- `device_id` (text, index)
+- `range_start` / `range_end` (timestamptz, index)
+- `file_path` (text, unique)
+- `file_size` (bigint)
+- `sha256` (char(64))
+- `record_count` (bigint)
+- `created_at` (timestamptz)
+
+## 6. 双阶段流水线细节
+
+### 6.1 入站（ingestor）
+
+- MQTT 建议参数：`clean_session=False`、固定 `client_id`、QoS=1
+- 批量入库建议：每 100 条或 200ms flush 一次事务
+- 目标：尽快持久化原始字节，避免解析阻塞接收
+
+### 6.2 解析（parser-worker）
+
+批处理取数：
+
+- 条件：`parse_status='pending'` 且 `next_retry_at` 到期
+- 顺序：`ORDER BY received_at`
+- 锁：`FOR UPDATE SKIP LOCKED`
+- 批量：`LIMIT 100~500`
+
+处理规则：
+
+- 成功：`parse_status='succeeded'`
+- 失败：`parse_status='failed'` + `parse_attempts+1` + `parse_error`
+- 重试：退避后设置 `next_retry_at`
+
+幂等建议：
+
+- 事件表增加防重策略（如 `raw_id + event_type + event_time` 约束或哈希去重）
+- `terminal_device_state` 用 upsert，仅更新更晚 `event_time`
+
+## 7. 解析策略（本期）
+
+### 7.1 NMEA
+
+- 解析 GGA/RMC/GSV/GSA/GST
+- 每条语句产出一条 `terminal_observation_event`
+
+### 7.2 RTCM3.2
+
+- 仅提取消息类型摘要（如 1005/1077/1087）
+- 每条 raw payload 产出一条 `rtcm_summary` 事件，包含类型计数与最近出现时间
+
+### 7.3 混合负载
+
+- 单 payload 中可同时包含 NMEA 与 RTCM 片段
+- 解析器按字节扫描拆分，分别产出事件
+- 部分片段解析失败不影响其余片段；原始 payload 永远保留
+
+## 8. 热数据与本地冷归档
+
+### 8.1 热数据
+
+- `terminal_message_raw` 保留 90 天
+
+### 8.2 本地冷归档
+
+- 定时任务离峰执行
+- 归档粒度：`device_id + 日`（必要时可按小时分片）
+- 路径示例：`/data/archive/gnss/<device_id>/YYYY-MM-DD/`
+- 文件内容：仅原始 payload 字节流拼接（无额外元信息）
+- 文件名示例：`device_<id>_<start>_<end>.txt`
+
+### 8.3 校验与清理
+
+- 写入 `manifest`：时间范围、条数、sha256、文件路径
+- 校验通过后，清理超过 90 天热数据
+
+## 9. 下载行为约束（后续接口实现遵循）
+
+- 下载内容必须与原始 MQTT payload 字节一致
+- 近 90 天：从热表按时间顺序拼接输出
+- 超过 90 天：从本地归档读取输出
+- 跨热/冷边界：按时间顺序拼接两段数据
+
+## 10. 运维与可观测性
+
+必须监控：
+
+- ingestor：连接状态、入站速率、入库失败
+- parser：pending backlog、解析成功率、平均延迟、失败重试
+- archive：每日归档成功/失败、磁盘余量
+
+建议告警：
+
+- `pending` 队列持续增长
+- parser 失败率超阈值
+- 归档目录磁盘不足
+
+## 11. 配置项建议（.env）
+
+- `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`
+
+## 12. 风险与后续演进
+
+- 使用 `.txt` 扩展名存放二进制流可能被文本工具误打开；但不影响下游软件按字节读取
+- 当设备量增长时，可水平扩展 parser-worker（依赖 `SKIP LOCKED`）
+- 后续实时能力：新增 SSE/WebSocket 推送层，数据源直接复用 `event/state`
+
+## 13. 验收标准（本期）
+
+- 能稳定接收 `terminal/{device_id}/raw` 消息并原样入库
+- 混合 payload 可提取 NMEA 事件和 RTCM 类型摘要
+- 设备最新状态可持续更新
+- 支持 90 天热数据保留与本地冷归档落盘
+- 能按时间段导出原始数据（文件后缀 `.txt`，内容无附加字段）

frontend/src/client/schemas.gen.ts

@@ -71,131 +71,6 @@ export const HTTPValidationErrorSchema = {
     title: 'HTTPValidationError'
 } as const;
 
-export const ItemCreateSchema = {
-    properties: {
-        title: {
-            type: 'string',
-            maxLength: 255,
-            minLength: 1,
-            title: 'Title'
-        },
-        description: {
-            anyOf: [
-                {
-                    type: 'string',
-                    maxLength: 255
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            title: 'Description'
-        }
-    },
-    type: 'object',
-    required: ['title'],
-    title: 'ItemCreate'
-} as const;
-
-export const ItemPublicSchema = {
-    properties: {
-        title: {
-            type: 'string',
-            maxLength: 255,
-            minLength: 1,
-            title: 'Title'
-        },
-        description: {
-            anyOf: [
-                {
-                    type: 'string',
-                    maxLength: 255
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            title: 'Description'
-        },
-        id: {
-            type: 'string',
-            format: 'uuid',
-            title: 'Id'
-        },
-        owner_id: {
-            type: 'string',
-            format: 'uuid',
-            title: 'Owner Id'
-        },
-        created_at: {
-            anyOf: [
-                {
-                    type: 'string',
-                    format: 'date-time'
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            title: 'Created At'
-        }
-    },
-    type: 'object',
-    required: ['title', 'id', 'owner_id'],
-    title: 'ItemPublic'
-} as const;
-
-export const ItemUpdateSchema = {
-    properties: {
-        title: {
-            anyOf: [
-                {
-                    type: 'string',
-                    maxLength: 255,
-                    minLength: 1
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            title: 'Title'
-        },
-        description: {
-            anyOf: [
-                {
-                    type: 'string',
-                    maxLength: 255
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            title: 'Description'
-        }
-    },
-    type: 'object',
-    title: 'ItemUpdate'
-} as const;
-
-export const ItemsPublicSchema = {
-    properties: {
-        data: {
-            items: {
-                '$ref': '#/components/schemas/ItemPublic'
-            },
-            type: 'array',
-            title: 'Data'
-        },
-        count: {
-            type: 'integer',
-            title: 'Count'
-        }
-    },
-    type: 'object',
-    required: ['data', 'count'],
-    title: 'ItemsPublic'
-} as const;
-
 export const LocationCreateSchema = {
     properties: {
         title: {

frontend/src/client/sdk.gen.ts

@@ -3,118 +3,7 @@
 import type { CancelablePromise } from './core/CancelablePromise';
 import { OpenAPI } from './core/OpenAPI';
 import { request as __request } from './core/request';
-import type { ItemsReadItemsData, ItemsReadItemsResponse, ItemsCreateItemData, ItemsCreateItemResponse, ItemsReadItemData, ItemsReadItemResponse, ItemsUpdateItemData, ItemsUpdateItemResponse, ItemsDeleteItemData, ItemsDeleteItemResponse, LocationsReadLocationsData, LocationsReadLocationsResponse, LocationsCreateLocationData, LocationsCreateLocationResponse, LocationsReadLocationData, LocationsReadLocationResponse, LocationsUpdateLocationData, LocationsUpdateLocationResponse, LocationsDeleteLocationData, LocationsDeleteLocationResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen';
+import type { LocationsReadLocationsData, LocationsReadLocationsResponse, LocationsCreateLocationData, LocationsCreateLocationResponse, LocationsReadLocationData, LocationsReadLocationResponse, LocationsUpdateLocationData, LocationsUpdateLocationResponse, LocationsDeleteLocationData, LocationsDeleteLocationResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen';
-
-export class ItemsService {
-    /**
-     * Read Items
-     * Retrieve items.
-     * @param data The data for the request.
-     * @param data.skip
-     * @param data.limit
-     * @returns ItemsPublic Successful Response
-     * @throws ApiError
-     */
-    public static readItems(data: ItemsReadItemsData = {}): CancelablePromise<ItemsReadItemsResponse> {
-        return __request(OpenAPI, {
-            method: 'GET',
-            url: '/api/v1/items/',
-            query: {
-                skip: data.skip,
-                limit: data.limit
-            },
-            errors: {
-                422: 'Validation Error'
-            }
-        });
-    }
-    
-    /**
-     * Create Item
-     * Create new item.
-     * @param data The data for the request.
-     * @param data.requestBody
-     * @returns ItemPublic Successful Response
-     * @throws ApiError
-     */
-    public static createItem(data: ItemsCreateItemData): CancelablePromise<ItemsCreateItemResponse> {
-        return __request(OpenAPI, {
-            method: 'POST',
-            url: '/api/v1/items/',
-            body: data.requestBody,
-            mediaType: 'application/json',
-            errors: {
-                422: 'Validation Error'
-            }
-        });
-    }
-    
-    /**
-     * Read Item
-     * Get item by ID.
-     * @param data The data for the request.
-     * @param data.id
-     * @returns ItemPublic Successful Response
-     * @throws ApiError
-     */
-    public static readItem(data: ItemsReadItemData): CancelablePromise<ItemsReadItemResponse> {
-        return __request(OpenAPI, {
-            method: 'GET',
-            url: '/api/v1/items/{id}',
-            path: {
-                id: data.id
-            },
-            errors: {
-                422: 'Validation Error'
-            }
-        });
-    }
-    
-    /**
-     * Update Item
-     * Update an item.
-     * @param data The data for the request.
-     * @param data.id
-     * @param data.requestBody
-     * @returns ItemPublic Successful Response
-     * @throws ApiError
-     */
-    public static updateItem(data: ItemsUpdateItemData): CancelablePromise<ItemsUpdateItemResponse> {
-        return __request(OpenAPI, {
-            method: 'PUT',
-            url: '/api/v1/items/{id}',
-            path: {
-                id: data.id
-            },
-            body: data.requestBody,
-            mediaType: 'application/json',
-            errors: {
-                422: 'Validation Error'
-            }
-        });
-    }
-    
-    /**
-     * Delete Item
-     * Delete an item.
-     * @param data The data for the request.
-     * @param data.id
-     * @returns Message Successful Response
-     * @throws ApiError
-     */
-    public static deleteItem(data: ItemsDeleteItemData): CancelablePromise<ItemsDeleteItemResponse> {
-        return __request(OpenAPI, {
-            method: 'DELETE',
-            url: '/api/v1/items/{id}',
-            path: {
-                id: data.id
-            },
-            errors: {
-                422: 'Validation Error'
-            }
-        });
-    }
-}
 
 export class LocationsService {
     /**

frontend/src/client/types.gen.ts

@@ -13,29 +13,6 @@ export type HTTPValidationError = {
     detail?: Array<ValidationError>;
 };
 
-export type ItemCreate = {
-    title: string;
-    description?: (string | null);
-};
-
-export type ItemPublic = {
-    title: string;
-    description?: (string | null);
-    id: string;
-    owner_id: string;
-    created_at?: (string | null);
-};
-
-export type ItemsPublic = {
-    data: Array<ItemPublic>;
-    count: number;
-};
-
-export type ItemUpdate = {
-    title?: (string | null);
-    description?: (string | null);
-};
-
 export type LocationCreate = {
     title: string;
     description?: (string | null);
@@ -136,38 +113,6 @@ export type ValidationError = {
     };
 };
 
-export type ItemsReadItemsData = {
-    limit?: number;
-    skip?: number;
-};
-
-export type ItemsReadItemsResponse = (ItemsPublic);
-
-export type ItemsCreateItemData = {
-    requestBody: ItemCreate;
-};
-
-export type ItemsCreateItemResponse = (ItemPublic);
-
-export type ItemsReadItemData = {
-    id: string;
-};
-
-export type ItemsReadItemResponse = (ItemPublic);
-
-export type ItemsUpdateItemData = {
-    id: string;
-    requestBody: ItemUpdate;
-};
-
-export type ItemsUpdateItemResponse = (ItemPublic);
-
-export type ItemsDeleteItemData = {
-    id: string;
-};
-
-export type ItemsDeleteItemResponse = (Message);
-
 export type LocationsReadLocationsData = {
     limit?: number;
     skip?: number;

frontend/src/components/Locations/DeleteLocation.tsx

@@ -66,8 +66,8 @@ const DeleteLocation = ({ id, onSuccess }: DeleteLocationProps) => {
           <DialogHeader>
             <DialogTitle>Delete Location</DialogTitle>
             <DialogDescription>
-              This location will be permanently deleted. Are you sure? You will not
+              This location will be permanently deleted. Are you sure? You will
-              be able to undo this action.
+              not be able to undo this action.
             </DialogDescription>
           </DialogHeader>
 

frontend/src/routes/_layout/locations.tsx

@@ -36,8 +36,12 @@ function LocationsTableContent() {
         <div className="rounded-full bg-muted p-4 mb-4">
           <Search className="h-8 w-8 text-muted-foreground" />
         </div>
-        <h3 className="text-lg font-semibold">You don't have any locations yet</h3>
+        <h3 className="text-lg font-semibold">
-        <p className="text-muted-foreground">Add a new location to get started</p>
+          You don't have any locations yet
+        </h3>
+        <p className="text-muted-foreground">
+          Add a new location to get started
+        </p>
       </div>
     )
   }
@@ -59,7 +63,9 @@ function Locations() {
       <div className="flex items-center justify-between">
         <div>
           <h1 className="text-2xl font-bold tracking-tight">Locations</h1>
-          <p className="text-muted-foreground">Create and manage your locations</p>
+          <p className="text-muted-foreground">
+            Create and manage your locations
+          </p>
         </div>
         <AddLocation />
       </div>

uv.lock

@@ -69,6 +69,7 @@ dependencies = [
     { name = "fastapi", extra = ["standard"] },
     { name = "httpx" },
     { name = "jinja2" },
+    { name = "paho-mqtt" },
     { name = "psycopg", extra = ["binary"] },
     { name = "pwdlib", extra = ["argon2", "bcrypt"] },
     { name = "pydantic" },
@@ -97,6 +98,7 @@ requires-dist = [
     { name = "fastapi", extras = ["standard"], specifier = ">=0.114.2,<1.0.0" },
     { name = "httpx", specifier = ">=0.25.1,<1.0.0" },
     { name = "jinja2", specifier = ">=3.1.4,<4.0.0" },
+    { name = "paho-mqtt", specifier = ">=2.1.0,<3.0.0" },
     { name = "psycopg", extras = ["binary"], specifier = ">=3.1.13,<4.0.0" },
     { name = "pwdlib", extras = ["argon2", "bcrypt"], specifier = ">=0.3.0" },
     { name = "pydantic", specifier = ">2.0" },
@@ -811,7 +813,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/32/6a/33d1702184d94106d3cdd7bfb788e19723206fce152e303473ca3b946c7b/greenlet-3.3.0-cp310-cp310-macosx_11_0_universal2.whl", hash = "sha256:6f8496d434d5cb2dce025773ba5597f71f5410ae499d5dd9533e0653258cdb3d", size = 273658, upload-time = "2025-12-04T14:23:37.494Z" },
     { url = "https://files.pythonhosted.org/packages/d6/b7/2b5805bbf1907c26e434f4e448cd8b696a0b71725204fa21a211ff0c04a7/greenlet-3.3.0-cp310-cp310-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b96dc7eef78fd404e022e165ec55327f935b9b52ff355b067eb4a0267fc1cffb", size = 574810, upload-time = "2025-12-04T14:50:04.154Z" },
     { url = "https://files.pythonhosted.org/packages/94/38/343242ec12eddf3d8458c73f555c084359883d4ddc674240d9e61ec51fd6/greenlet-3.3.0-cp310-cp310-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:73631cd5cccbcfe63e3f9492aaa664d278fda0ce5c3d43aeda8e77317e38efbd", size = 586248, upload-time = "2025-12-04T14:57:39.35Z" },
-    { url = "https://files.pythonhosted.org/packages/f0/d0/0ae86792fb212e4384041e0ef8e7bc66f59a54912ce407d26a966ed2914d/greenlet-3.3.0-cp310-cp310-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:b299a0cb979f5d7197442dccc3aee67fce53500cd88951b7e6c35575701c980b", size = 597403, upload-time = "2025-12-04T15:07:10.831Z" },
     { url = "https://files.pythonhosted.org/packages/b6/a8/15d0aa26c0036a15d2659175af00954aaaa5d0d66ba538345bd88013b4d7/greenlet-3.3.0-cp310-cp310-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7dee147740789a4632cace364816046e43310b59ff8fb79833ab043aefa72fd5", size = 586910, upload-time = "2025-12-04T14:25:59.705Z" },
     { url = "https://files.pythonhosted.org/packages/e1/9b/68d5e3b7ccaba3907e5532cf8b9bf16f9ef5056a008f195a367db0ff32db/greenlet-3.3.0-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:39b28e339fc3c348427560494e28d8a6f3561c8d2bcf7d706e1c624ed8d822b9", size = 1547206, upload-time = "2025-12-04T15:04:21.027Z" },
     { url = "https://files.pythonhosted.org/packages/66/bd/e3086ccedc61e49f91e2cfb5ffad9d8d62e5dc85e512a6200f096875b60c/greenlet-3.3.0-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:b3c374782c2935cc63b2a27ba8708471de4ad1abaa862ffdb1ef45a643ddbb7d", size = 1613359, upload-time = "2025-12-04T14:27:26.548Z" },
@@ -819,7 +820,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/1f/cb/48e964c452ca2b92175a9b2dca037a553036cb053ba69e284650ce755f13/greenlet-3.3.0-cp311-cp311-macosx_11_0_universal2.whl", hash = "sha256:e29f3018580e8412d6aaf5641bb7745d38c85228dacf51a73bd4e26ddf2a6a8e", size = 274908, upload-time = "2025-12-04T14:23:26.435Z" },
     { url = "https://files.pythonhosted.org/packages/28/da/38d7bff4d0277b594ec557f479d65272a893f1f2a716cad91efeb8680953/greenlet-3.3.0-cp311-cp311-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a687205fb22794e838f947e2194c0566d3812966b41c78709554aa883183fb62", size = 577113, upload-time = "2025-12-04T14:50:05.493Z" },
     { url = "https://files.pythonhosted.org/packages/3c/f2/89c5eb0faddc3ff014f1c04467d67dee0d1d334ab81fadbf3744847f8a8a/greenlet-3.3.0-cp311-cp311-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:4243050a88ba61842186cb9e63c7dfa677ec146160b0efd73b855a3d9c7fcf32", size = 590338, upload-time = "2025-12-04T14:57:41.136Z" },
-    { url = "https://files.pythonhosted.org/packages/80/d7/db0a5085035d05134f8c089643da2b44cc9b80647c39e93129c5ef170d8f/greenlet-3.3.0-cp311-cp311-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:670d0f94cd302d81796e37299bcd04b95d62403883b24225c6b5271466612f45", size = 601098, upload-time = "2025-12-04T15:07:11.898Z" },
     { url = "https://files.pythonhosted.org/packages/dc/a6/e959a127b630a58e23529972dbc868c107f9d583b5a9f878fb858c46bc1a/greenlet-3.3.0-cp311-cp311-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:6cb3a8ec3db4a3b0eb8a3c25436c2d49e3505821802074969db017b87bc6a948", size = 590206, upload-time = "2025-12-04T14:26:01.254Z" },
     { url = "https://files.pythonhosted.org/packages/48/60/29035719feb91798693023608447283b266b12efc576ed013dd9442364bb/greenlet-3.3.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:2de5a0b09eab81fc6a382791b995b1ccf2b172a9fec934747a7a23d2ff291794", size = 1550668, upload-time = "2025-12-04T15:04:22.439Z" },
     { url = "https://files.pythonhosted.org/packages/0a/5f/783a23754b691bfa86bd72c3033aa107490deac9b2ef190837b860996c9f/greenlet-3.3.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:4449a736606bd30f27f8e1ff4678ee193bc47f6ca810d705981cfffd6ce0d8c5", size = 1615483, upload-time = "2025-12-04T14:27:28.083Z" },
@@ -827,7 +827,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f8/0a/a3871375c7b9727edaeeea994bfff7c63ff7804c9829c19309ba2e058807/greenlet-3.3.0-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:b01548f6e0b9e9784a2c99c5651e5dc89ffcbe870bc5fb2e5ef864e9cc6b5dcb", size = 276379, upload-time = "2025-12-04T14:23:30.498Z" },
     { url = "https://files.pythonhosted.org/packages/43/ab/7ebfe34dce8b87be0d11dae91acbf76f7b8246bf9d6b319c741f99fa59c6/greenlet-3.3.0-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:349345b770dc88f81506c6861d22a6ccd422207829d2c854ae2af8025af303e3", size = 597294, upload-time = "2025-12-04T14:50:06.847Z" },
     { url = "https://files.pythonhosted.org/packages/a4/39/f1c8da50024feecd0793dbd5e08f526809b8ab5609224a2da40aad3a7641/greenlet-3.3.0-cp312-cp312-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:e8e18ed6995e9e2c0b4ed264d2cf89260ab3ac7e13555b8032b25a74c6d18655", size = 607742, upload-time = "2025-12-04T14:57:42.349Z" },
-    { url = "https://files.pythonhosted.org/packages/77/cb/43692bcd5f7a0da6ec0ec6d58ee7cddb606d055ce94a62ac9b1aa481e969/greenlet-3.3.0-cp312-cp312-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:c024b1e5696626890038e34f76140ed1daf858e37496d33f2af57f06189e70d7", size = 622297, upload-time = "2025-12-04T15:07:13.552Z" },
     { url = "https://files.pythonhosted.org/packages/75/b0/6bde0b1011a60782108c01de5913c588cf51a839174538d266de15e4bf4d/greenlet-3.3.0-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:047ab3df20ede6a57c35c14bf5200fcf04039d50f908270d3f9a7a82064f543b", size = 609885, upload-time = "2025-12-04T14:26:02.368Z" },
     { url = "https://files.pythonhosted.org/packages/49/0e/49b46ac39f931f59f987b7cd9f34bfec8ef81d2a1e6e00682f55be5de9f4/greenlet-3.3.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:2d9ad37fc657b1102ec880e637cccf20191581f75c64087a549e66c57e1ceb53", size = 1567424, upload-time = "2025-12-04T15:04:23.757Z" },
     { url = "https://files.pythonhosted.org/packages/05/f5/49a9ac2dff7f10091935def9165c90236d8f175afb27cbed38fb1d61ab6b/greenlet-3.3.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:83cd0e36932e0e7f36a64b732a6f60c2fc2df28c351bae79fbaf4f8092fe7614", size = 1636017, upload-time = "2025-12-04T14:27:29.688Z" },
@@ -835,7 +834,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/02/2f/28592176381b9ab2cafa12829ba7b472d177f3acc35d8fbcf3673d966fff/greenlet-3.3.0-cp313-cp313-macosx_11_0_universal2.whl", hash = "sha256:a1e41a81c7e2825822f4e068c48cb2196002362619e2d70b148f20a831c00739", size = 275140, upload-time = "2025-12-04T14:23:01.282Z" },
     { url = "https://files.pythonhosted.org/packages/2c/80/fbe937bf81e9fca98c981fe499e59a3f45df2a04da0baa5c2be0dca0d329/greenlet-3.3.0-cp313-cp313-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9f515a47d02da4d30caaa85b69474cec77b7929b2e936ff7fb853d42f4bf8808", size = 599219, upload-time = "2025-12-04T14:50:08.309Z" },
     { url = "https://files.pythonhosted.org/packages/c2/ff/7c985128f0514271b8268476af89aee6866df5eec04ac17dcfbc676213df/greenlet-3.3.0-cp313-cp313-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:7d2d9fd66bfadf230b385fdc90426fcd6eb64db54b40c495b72ac0feb5766c54", size = 610211, upload-time = "2025-12-04T14:57:43.968Z" },
-    { url = "https://files.pythonhosted.org/packages/79/07/c47a82d881319ec18a4510bb30463ed6891f2ad2c1901ed5ec23d3de351f/greenlet-3.3.0-cp313-cp313-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:30a6e28487a790417d036088b3bcb3f3ac7d8babaa7d0139edbaddebf3af9492", size = 624311, upload-time = "2025-12-04T15:07:14.697Z" },
     { url = "https://files.pythonhosted.org/packages/fd/8e/424b8c6e78bd9837d14ff7df01a9829fc883ba2ab4ea787d4f848435f23f/greenlet-3.3.0-cp313-cp313-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:087ea5e004437321508a8d6f20efc4cfec5e3c30118e1417ea96ed1d93950527", size = 612833, upload-time = "2025-12-04T14:26:03.669Z" },
     { url = "https://files.pythonhosted.org/packages/b5/ba/56699ff9b7c76ca12f1cdc27a886d0f81f2189c3455ff9f65246780f713d/greenlet-3.3.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:ab97cf74045343f6c60a39913fa59710e4bd26a536ce7ab2397adf8b27e67c39", size = 1567256, upload-time = "2025-12-04T15:04:25.276Z" },
     { url = "https://files.pythonhosted.org/packages/1e/37/f31136132967982d698c71a281a8901daf1a8fbab935dce7c0cf15f942cc/greenlet-3.3.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:5375d2e23184629112ca1ea89a53389dddbffcf417dad40125713d88eb5f96e8", size = 1636483, upload-time = "2025-12-04T14:27:30.804Z" },
@@ -843,7 +841,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/d7/7c/f0a6d0ede2c7bf092d00bc83ad5bafb7e6ec9b4aab2fbdfa6f134dc73327/greenlet-3.3.0-cp314-cp314-macosx_11_0_universal2.whl", hash = "sha256:60c2ef0f578afb3c8d92ea07ad327f9a062547137afe91f38408f08aacab667f", size = 275671, upload-time = "2025-12-04T14:23:05.267Z" },
     { url = "https://files.pythonhosted.org/packages/44/06/dac639ae1a50f5969d82d2e3dd9767d30d6dbdbab0e1a54010c8fe90263c/greenlet-3.3.0-cp314-cp314-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0a5d554d0712ba1de0a6c94c640f7aeba3f85b3a6e1f2899c11c2c0428da9365", size = 646360, upload-time = "2025-12-04T14:50:10.026Z" },
     { url = "https://files.pythonhosted.org/packages/e0/94/0fb76fe6c5369fba9bf98529ada6f4c3a1adf19e406a47332245ef0eb357/greenlet-3.3.0-cp314-cp314-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:3a898b1e9c5f7307ebbde4102908e6cbfcb9ea16284a3abe15cab996bee8b9b3", size = 658160, upload-time = "2025-12-04T14:57:45.41Z" },
-    { url = "https://files.pythonhosted.org/packages/93/79/d2c70cae6e823fac36c3bbc9077962105052b7ef81db2f01ec3b9bf17e2b/greenlet-3.3.0-cp314-cp314-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:dcd2bdbd444ff340e8d6bdf54d2f206ccddbb3ccfdcd3c25bf4afaa7b8f0cf45", size = 671388, upload-time = "2025-12-04T15:07:15.789Z" },
     { url = "https://files.pythonhosted.org/packages/b8/14/bab308fc2c1b5228c3224ec2bf928ce2e4d21d8046c161e44a2012b5203e/greenlet-3.3.0-cp314-cp314-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5773edda4dc00e173820722711d043799d3adb4f01731f40619e07ea2750b955", size = 660166, upload-time = "2025-12-04T14:26:05.099Z" },
     { url = "https://files.pythonhosted.org/packages/4b/d2/91465d39164eaa0085177f61983d80ffe746c5a1860f009811d498e7259c/greenlet-3.3.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:ac0549373982b36d5fd5d30beb8a7a33ee541ff98d2b502714a09f1169f31b55", size = 1615193, upload-time = "2025-12-04T15:04:27.041Z" },
     { url = "https://files.pythonhosted.org/packages/42/1b/83d110a37044b92423084d52d5d5a3b3a73cafb51b547e6d7366ff62eff1/greenlet-3.3.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:d198d2d977460358c3b3a4dc844f875d1adb33817f0613f663a656f463764ccc", size = 1683653, upload-time = "2025-12-04T14:27:32.366Z" },
@@ -851,7 +848,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/a0/66/bd6317bc5932accf351fc19f177ffba53712a202f9df10587da8df257c7e/greenlet-3.3.0-cp314-cp314t-macosx_11_0_universal2.whl", hash = "sha256:d6ed6f85fae6cdfdb9ce04c9bf7a08d666cfcfb914e7d006f44f840b46741931", size = 282638, upload-time = "2025-12-04T14:25:20.941Z" },
     { url = "https://files.pythonhosted.org/packages/30/cf/cc81cb030b40e738d6e69502ccbd0dd1bced0588e958f9e757945de24404/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d9125050fcf24554e69c4cacb086b87b3b55dc395a8b3ebe6487b045b2614388", size = 651145, upload-time = "2025-12-04T14:50:11.039Z" },
     { url = "https://files.pythonhosted.org/packages/9c/ea/1020037b5ecfe95ca7df8d8549959baceb8186031da83d5ecceff8b08cd2/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:87e63ccfa13c0a0f6234ed0add552af24cc67dd886731f2261e46e241608bee3", size = 654236, upload-time = "2025-12-04T14:57:47.007Z" },
-    { url = "https://files.pythonhosted.org/packages/69/cc/1e4bae2e45ca2fa55299f4e85854606a78ecc37fead20d69322f96000504/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:2662433acbca297c9153a4023fe2161c8dcfdcc91f10433171cf7e7d94ba2221", size = 662506, upload-time = "2025-12-04T15:07:16.906Z" },
     { url = "https://files.pythonhosted.org/packages/57/b9/f8025d71a6085c441a7eaff0fd928bbb275a6633773667023d19179fe815/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:3c6e9b9c1527a78520357de498b0e709fb9e2f49c3a513afd5a249007261911b", size = 653783, upload-time = "2025-12-04T14:26:06.225Z" },
     { url = "https://files.pythonhosted.org/packages/f6/c7/876a8c7a7485d5d6b5c6821201d542ef28be645aa024cfe1145b35c120c1/greenlet-3.3.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:286d093f95ec98fdd92fcb955003b8a3d054b4e2cab3e2707a5039e7b50520fd", size = 1614857, upload-time = "2025-12-04T15:04:28.484Z" },
     { url = "https://files.pythonhosted.org/packages/4f/dc/041be1dff9f23dac5f48a43323cd0789cb798342011c19a248d9c9335536/greenlet-3.3.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:6c10513330af5b8ae16f023e8ddbfb486ab355d04467c4679c5cfe4659975dd9", size = 1676034, upload-time = "2025-12-04T14:27:33.531Z" },
@@ -1355,6 +1351,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/20/12/38679034af332785aac8774540895e234f4d07f7545804097de4b666afd8/packaging-25.0-py3-none-any.whl", hash = "sha256:29572ef2b1f17581046b3a2227d5c611fb25ec70ca1ba8554b24b0e69331a484", size = 66469, upload-time = "2025-04-19T11:48:57.875Z" },
 ]
 
+[[package]]
+name = "paho-mqtt"
+version = "2.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/39/15/0a6214e76d4d32e7f663b109cf71fb22561c2be0f701d67f93950cd40542/paho_mqtt-2.1.0.tar.gz", hash = "sha256:12d6e7511d4137555a3f6ea167ae846af2c7357b10bc6fa4f7c3968fc1723834", size = 148848, upload-time = "2024-04-29T19:52:55.591Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c4/cb/00451c3cf31790287768bb12c6bec834f5d292eaf3022afc88e14b8afc94/paho_mqtt-2.1.0-py3-none-any.whl", hash = "sha256:6db9ba9b34ed5bc6b6e3812718c7e06e2fd7444540df2455d2c51bd58808feee", size = 67219, upload-time = "2024-04-29T19:52:48.345Z" },
+]
+
 [[package]]
 name = "pathspec"
 version = "1.0.3"