refactor: simplify mqtt pipeline state and remove event table

backend/README.md

@@ -176,7 +176,8 @@ Once you have the MJML extension installed, you can create a new email template
 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.
+- `mqtt-parser-worker` consumes unparsed rows (`parsed_at IS NULL` and `parse_error IS NULL`) and updates `terminal_device_state`.
+- Parse is single-pass by design in the current implementation: failed rows keep `parse_error` and are not auto-retried with backoff.
 
 ### Run locally with Docker Compose
 
@@ -199,7 +200,6 @@ $ PYTHONPATH=. ../.venv/bin/pytest tests/mqtt --confcutdir=tests/mqtt -q
 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)

backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py

@@ -0,0 +1,108 @@
+"""simplify mqtt parse tracking
+
+Revision ID: b4c5d6e7f809
+Revises: a1b2c3d4e5f6
+Create Date: 2026-03-19 17:30:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "b4c5d6e7f809"
+down_revision = "a1b2c3d4e5f6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "terminal_message_raw",
+        sa.Column("parsed_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_terminal_message_raw_parsed_at"),
+        "terminal_message_raw",
+        ["parsed_at"],
+        unique=False,
+    )
+
+    # Preserve historical parse outcomes before removing state-machine columns.
+    op.execute(
+        sa.text(
+            """
+            UPDATE terminal_message_raw
+            SET parsed_at = COALESCE(received_at, created_at)
+            WHERE parse_status <> 'pending'
+            """
+        )
+    )
+
+    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_parse_status"),
+        table_name="terminal_message_raw",
+    )
+    op.drop_column("terminal_message_raw", "next_retry_at")
+    op.drop_column("terminal_message_raw", "parse_attempts")
+    op.drop_column("terminal_message_raw", "parse_status")
+
+
+def downgrade() -> None:
+    op.add_column(
+        "terminal_message_raw",
+        sa.Column(
+            "parse_status",
+            sa.String(length=32),
+            nullable=False,
+            server_default="pending",
+        ),
+    )
+    op.add_column(
+        "terminal_message_raw",
+        sa.Column("parse_attempts", sa.Integer(), nullable=False, server_default="0"),
+    )
+    op.add_column(
+        "terminal_message_raw",
+        sa.Column("next_retry_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    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_next_retry_at"),
+        "terminal_message_raw",
+        ["next_retry_at"],
+        unique=False,
+    )
+
+    op.execute(
+        sa.text(
+            """
+            UPDATE terminal_message_raw
+            SET parse_status = CASE
+                WHEN parsed_at IS NULL THEN 'pending'
+                WHEN parse_error IS NULL THEN 'succeeded'
+                ELSE 'failed'
+            END,
+            parse_attempts = CASE
+                WHEN parsed_at IS NULL THEN 0
+                ELSE 1
+            END
+            """
+        )
+    )
+
+    op.drop_index(
+        op.f("ix_terminal_message_raw_parsed_at"),
+        table_name="terminal_message_raw",
+    )
+    op.drop_column("terminal_message_raw", "parsed_at")

backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py

@@ -0,0 +1,60 @@
+"""drop terminal observation event table
+
+Revision ID: c6d7e8f90123
+Revises: b4c5d6e7f809
+Create Date: 2026-03-19 18:20:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "c6d7e8f90123"
+down_revision = "b4c5d6e7f809"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.drop_table("terminal_observation_event")
+
+
+def downgrade() -> None:
+    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", sa.String(length=255), nullable=False),
+        sa.Column("event_time", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("event_type", sa.String(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,
+    )

backend/app/models.py

@@ -126,9 +126,7 @@ class TerminalMessageRaw(SQLModel, table=True):
     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(
+    parsed_at: datetime | None = Field(
         default=None,
         sa_type=DateTime(timezone=True),  # type: ignore
         index=True,
@@ -140,27 +138,6 @@ class TerminalMessageRaw(SQLModel, table=True):
     )
 
 
-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"
 

backend/app/mqtt/ingestor.py

@@ -1,24 +1,32 @@
 from __future__ import annotations
 
 import logging
+import re
+from typing import Any
 
-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__)
+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")
 
 
 def _on_connect(
-    client: mqtt.Client,
+    client: Any,
     _userdata: object,
     _flags: dict[str, int],
     reason_code: int,
-    _properties: mqtt.Properties | None = None,
+    _properties: object | None = None,
 ) -> None:
     if reason_code != 0:
         logger.error("MQTT connect failed: rc=%s", reason_code)
@@ -31,9 +39,9 @@ def _on_connect(
 
 
 def _on_message(
-    _client: mqtt.Client,
+    _client: Any,
     _userdata: object,
-    message: mqtt.MQTTMessage,
+    message: Any,
 ) -> None:
     try:
         device_id = extract_device_id(message.topic)
@@ -52,7 +60,14 @@ def _on_message(
         )
 
 
-def create_client() -> mqtt.Client:
+def create_client() -> Any:
+    try:
+        import paho.mqtt.client as mqtt  # type: ignore[import-not-found,import-untyped]
+    except ModuleNotFoundError as exc:  # pragma: no cover - runtime guard
+        raise RuntimeError(
+            "paho-mqtt is required to run mqtt-ingestor. Install backend dependencies first."
+        ) from exc
+
     client = mqtt.Client(
         callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
         client_id=settings.MQTT_CLIENT_ID,

backend/app/mqtt/parser_worker.py

@@ -2,45 +2,33 @@ 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.models import TerminalMessageRaw
 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,
+    update_device_state_from_events,
 )
 
 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)
+        update_device_state_from_events(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()

backend/app/mqtt/parsers/__init__.py

@@ -1,5 +1,4 @@
-from app.mqtt.parsers.mixed import parse_mixed_payload
+from app.mqtt.parsers.mixed import parse_mixed_payload, summarize_rtcm_types
 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

@@ -3,7 +3,27 @@ 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 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
 
 
 def parse_mixed_payload(payload: bytes) -> list[dict[str, Any]]:

backend/app/mqtt/parsers/rtcm.py

@@ -1,22 +0,0 @@
-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

@@ -9,7 +9,6 @@ from sqlmodel import Session, col, select
 from app.models import (
     TerminalDeviceState,
     TerminalMessageRaw,
-    TerminalObservationEvent,
     get_datetime_utc,
 )
 
@@ -34,7 +33,6 @@ def insert_raw_message(
         payload_sha256=hashlib.sha256(payload).hexdigest(),
         qos=qos,
         retain=retain,
-        parse_status="pending",
         created_at=ts,
     )
     session.add(row)
@@ -48,7 +46,8 @@ def fetch_pending_rows_for_update(
 ) -> list[TerminalMessageRaw]:
     statement = (
         select(TerminalMessageRaw)
-        .where(TerminalMessageRaw.parse_status == "pending")
+        .where(col(TerminalMessageRaw.parsed_at).is_(None))
+        .where(col(TerminalMessageRaw.parse_error).is_(None))
         .order_by(col(TerminalMessageRaw.received_at))
         .limit(batch_size)
     )
@@ -61,17 +60,9 @@ def fetch_pending_rows_for_update(
     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.parsed_at = get_datetime_utc()
     row.parse_error = None
-    row.next_retry_at = None
     session.add(row)
 
 
@@ -80,15 +71,13 @@ def mark_failed(
     session: Session,
     row: TerminalMessageRaw,
     parse_error: str,
-    next_retry_at: datetime | None,
 ) -> None:
-    row.parse_status = "failed"
+    row.parsed_at = get_datetime_utc()
     row.parse_error = parse_error[:1024]
-    row.next_retry_at = next_retry_at
     session.add(row)
 
 
-def write_events_and_update_state(
+def update_device_state_from_events(
     *,
     session: Session,
     row: TerminalMessageRaw,
@@ -110,14 +99,6 @@ def write_events_and_update_state(
     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")

backend/app/mqtt/topic.py

@@ -1,10 +0,0 @@
-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/tests/mqtt/test_models.py

@@ -18,4 +18,5 @@ def test_terminal_message_raw_model_can_persist() -> None:
         session.refresh(row)
 
     assert row.id is not None
-    assert row.parse_status == "pending"
+    assert row.parsed_at is None
+    assert row.parse_error is None

backend/tests/mqtt/test_parser_worker.py

@@ -1,6 +1,6 @@
-from sqlmodel import Session, SQLModel, create_engine, select
+from sqlmodel import Session, SQLModel, create_engine
 
-from app.models import TerminalDeviceState, TerminalMessageRaw, TerminalObservationEvent
+from app.models import TerminalDeviceState, TerminalMessageRaw
 from app.mqtt.parser_worker import process_pending_batch
 from app.mqtt.repository import insert_raw_message
 
@@ -11,7 +11,7 @@ def _session() -> Session:
     return Session(engine)
 
 
-def test_insert_raw_message_sets_pending_status() -> None:
+def test_insert_raw_message_sets_unparsed_defaults() -> None:
     with _session() as session:
         row = insert_raw_message(
             session=session,
@@ -22,12 +22,13 @@ def test_insert_raw_message_sets_pending_status() -> None:
             retain=False,
         )
 
-        assert row.parse_status == "pending"
+        assert row.parsed_at is None
+        assert row.parse_error is None
         assert row.payload_size > 0
         assert len(row.payload_sha256) == 64
 
 
-def test_worker_marks_success_and_writes_events() -> None:
+def test_worker_marks_success_and_updates_state() -> None:
     with _session() as session:
         raw = insert_raw_message(
             session=session,
@@ -44,10 +45,43 @@ def test_worker_marks_success_and_writes_events() -> None:
 
         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
+        assert refreshed.parsed_at is not None
+        assert refreshed.parse_error is None
 
         state = session.get(TerminalDeviceState, "device-001")
         assert state is not None
+        assert state.lat is not None
+        assert state.lon is not None
+
+
+def test_worker_marks_failed_without_retry_schedule(monkeypatch) -> None:
+    def _raise_parse_error(_payload: bytes) -> list[dict[str, object]]:
+        raise ValueError("boom")
+
+    monkeypatch.setattr(
+        "app.mqtt.parser_worker.parse_mixed_payload", _raise_parse_error
+    )
+
+    with _session() as session:
+        raw = insert_raw_message(
+            session=session,
+            device_id="device-err",
+            topic="terminal/device-err/raw",
+            payload=b"$GPGGA,broken*00\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.parsed_at is not None
+        assert refreshed.parse_error == "boom"
+        processed_again = process_pending_batch(session=session, batch_size=10)
+        assert processed_again == 0
+
+
+def test_schema_excludes_terminal_observation_event_table() -> None:
+    assert "terminal_observation_event" not in SQLModel.metadata.tables

backend/tests/mqtt/test_topic.py

@@ -1,6 +1,6 @@
 import pytest
 
-from app.mqtt.topic import extract_device_id
+from app.mqtt.ingestor import extract_device_id
 
 
 def test_extract_device_id_success() -> None: