diff --git a/backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py b/backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py new file mode 100644 index 0000000..f6088d9 --- /dev/null +++ b/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") diff --git a/backend/app/models.py b/backend/app/models.py index 147d591..5945c5c 100644 --- a/backend/app/models.py +++ b/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 diff --git a/backend/tests/mqtt/test_models.py b/backend/tests/mqtt/test_models.py new file mode 100644 index 0000000..5eadb7d --- /dev/null +++ b/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"