feat: add mqtt pipeline tables and models

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/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)  # type: ignore
+    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)  # type: ignore
+    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
+    fix_status: str | None = Field(default=None, max_length=64)
+    lat: float | None = None
+    lon: float | None = None
+    alt_m: float | None = None
+    satellites_used: int | None = None
+    satellite_signal_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)  # type: ignore
+    rtcm_type_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)  # type: ignore
+    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/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"