perf(db): add hot-path indexes for matching, latest-bbox and frame lookups

Closes #663. Three query shapes on the detection hot path had no index behind
them, so each seq-scanned a growing table:

- a pose's recently-seen sequences (camera_id, pose_id, last_seen_at), run on
  every POST /detections during spatial matching
- the latest real bbox of a sequence (sequence_id, created_at), run once per
  candidate sequence per detection, and the same shape the player's sequence
  reads sort on
- sibling rows sharing a frame object (bucket_key), on DELETE /detections/{id}

Measured on the (sequence_id, created_at) shape with production-scale synthetic
data, index scan vs forced seq scan on identical rows: 7.6ms -> 0.45ms at 200k
detections, 34.8ms -> 0.42ms at 2M (roughly production today), 63.6ms -> 1.1ms
at 5M. The seq-scan side grows linearly with the table while the index scan
stays flat, so the gap widens as detections accumulate.

Built with CREATE INDEX CONCURRENTLY inside an autocommit block: detections is
the highest-write table and a plain build would hold ACCESS EXCLUSIVE against
camera ingest for its whole duration. The migration also self-heals, dropping an
index left INVALID by a cancelled build, which if_not_exists would otherwise
skip while the upgrade reported success and the planner ignored it.

The indexes are declared in models.py as well as the migration so create_all
built test databases match production, with tests pinning both sides against one
list since drift between them is otherwise invisible.

Author: Acruve15

src/app/models.py

@@ -7,6 +7,7 @@
 from enum import Enum
 from typing import Union
 
+from sqlalchemy import Index
 from sqlmodel import Field, SQLModel
 
 from app.core.config import settings
@@ -83,6 +84,15 @@ class OcclusionMask(SQLModel, table=True):
 
 class Detection(SQLModel, table=True):
     __tablename__ = "detections"
+    # Declared here as well as in the migration so create_all-built databases (the test suite)
+    # carry the same indexes as production, instead of silently planning every query differently.
+    __table_args__ = (
+        # Latest-real-bbox lookups during spatial matching, and the sequence reads the player
+        # pages through: sequence_id equality then a created_at scan.
+        Index("ix_detections_sequence_id_created_at", "sequence_id", "created_at"),
+        # Sibling-row check on deletion (multi-bbox and continuity rows share one frame object).
+        Index("ix_detections_bucket_key", "bucket_key"),
+    )
     id: int = Field(None, primary_key=True)
     camera_id: int = Field(..., foreign_key="cameras.id", nullable=False)
     pose_id: int = Field(..., foreign_key="poses.id", nullable=False)
@@ -112,6 +122,10 @@ class Detection(SQLModel, table=True):
 
 class Sequence(SQLModel, table=True):
     __tablename__ = "sequences"
+    __table_args__ = (
+        # Per-frame lookups of a pose's recently-seen sequences (spatial matching, continuity).
+        Index("ix_sequences_camera_pose_last_seen", "camera_id", "pose_id", "last_seen_at"),
+    )
     id: int = Field(None, primary_key=True)
     camera_id: int = Field(..., foreign_key="cameras.id", nullable=False)
     pose_id: Union[int, None] = Field(None, foreign_key="poses.id", nullable=True)

src/migrations/versions/2026_07_30_1000-e8f3a6c9d1b7_add_hot_path_indexes.py

@@ -0,0 +1,84 @@
+"""add hot-path indexes for sequence matching, latest-bbox and shared frame lookups
+
+Revision ID: e8f3a6c9d1b7
+Revises: c4e9f1a2b3d5
+Create Date: 2026-07-30 10:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+from sqlalchemy import text
+
+# revision identifiers, used by Alembic.
+revision: str = "e8f3a6c9d1b7"
+down_revision: Union[str, None] = "c4e9f1a2b3d5"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# (index name, table, columns). Kept in sync with the __table_args__ declarations in
+# app.models, which is what puts these indexes in create_all-built test databases too.
+INDEXES = (
+    ("ix_sequences_camera_pose_last_seen", "sequences", ["camera_id", "pose_id", "last_seen_at"]),
+    ("ix_detections_sequence_id_created_at", "detections", ["sequence_id", "created_at"]),
+    ("ix_detections_bucket_key", "detections", ["bucket_key"]),
+)
+
+
+def _drop_if_invalid(index_name: str, table_name: str) -> None:
+    """Drop an index left INVALID by a cancelled CONCURRENTLY build.
+
+    Without this the migration is not safely re-runnable: if_not_exists sees the leftover
+    relation and skips creating it, the revision stamps, and the upgrade reports success while
+    the planner ignores the invalid index, silently leaving these queries on sequential scans.
+
+    The lookup joins pg_class by name rather than casting the name to regclass, since that cast
+    raises when the relation does not exist yet (the common case) instead of returning no rows.
+    """
+    row = (
+        op
+        .get_bind()
+        .execute(
+            text(
+                "SELECT idx.indisvalid FROM pg_index idx "
+                "JOIN pg_class c ON c.oid = idx.indexrelid "
+                "WHERE c.relname = :name"
+            ),
+            {"name": index_name},
+        )
+        .first()
+    )
+    if row is not None and not row[0]:
+        op.drop_index(index_name, table_name=table_name, if_exists=True, postgresql_concurrently=True)
+
+
+def upgrade() -> None:
+    # Three query shapes on the detection hot path had no index backing them:
+    #   - a pose's recently-seen sequences (camera_id, pose_id, last_seen_at), run on every
+    #     POST /detections during spatial matching
+    #   - the latest real bbox of a sequence (sequence_id, created_at), run once per candidate
+    #     sequence per detection, and also the shape the player's sequence reads sort on
+    #   - sibling rows sharing a frame object (bucket_key), on DELETE /detections/{id}
+    #
+    # detections is the highest-write table, so a plain CREATE INDEX would hold ACCESS EXCLUSIVE
+    # against camera ingest for the whole build. CONCURRENTLY cannot run inside a transaction and
+    # env.py wraps the migration run in one, hence the autocommit block.
+    with op.get_context().autocommit_block():
+        for index_name, table_name, columns in INDEXES:
+            _drop_if_invalid(index_name, table_name)
+            op.create_index(
+                index_name,
+                table_name,
+                columns,
+                unique=False,
+                if_not_exists=True,
+                postgresql_concurrently=True,
+            )
+
+
+def downgrade() -> None:
+    # DROP INDEX CONCURRENTLY is likewise non-transactional.
+    with op.get_context().autocommit_block():
+        for index_name, table_name, _ in reversed(INDEXES):
+            op.drop_index(index_name, table_name=table_name, if_exists=True, postgresql_concurrently=True)

src/tests/test_models.py

@@ -0,0 +1,29 @@
+import pytest
+from sqlmodel import SQLModel, text
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+# Hot-path indexes, declared in two places that must not drift: the alembic migration (what
+# production runs) and __table_args__ in app.models (what create_all gives a test database built
+# without migrations). Drift is invisible at runtime, it just makes one environment plan queries
+# differently from the other, so both sides are pinned against this list.
+EXPECTED_INDEXES = {
+    "detections": {"ix_detections_sequence_id_created_at", "ix_detections_bucket_key"},
+    "sequences": {"ix_sequences_camera_pose_last_seen"},
+}
+
+
+@pytest.mark.parametrize(("table", "expected"), EXPECTED_INDEXES.items())
+def test_hot_path_indexes_are_declared_on_the_models(table: str, expected: set):
+    """Guards the model side. A DB assertion cannot cover this: the test database is migrated,
+    so the indexes are present whether or not __table_args__ still declares them."""
+    declared = {index.name for index in SQLModel.metadata.tables[table].indexes}
+    assert expected <= declared, f"not declared on {table}: {sorted(expected - declared)}"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("table", "expected"), EXPECTED_INDEXES.items())
+async def test_hot_path_indexes_exist_in_the_database(async_session: AsyncSession, table: str, expected: set):
+    """Guards the migration side: a fresh database must end up with all of them."""
+    stmt = text("SELECT indexname FROM pg_indexes WHERE tablename = :table").bindparams(table=table)
+    present = {row[0] for row in (await async_session.exec(stmt)).all()}
+    assert expected <= present, f"missing on {table}: {sorted(expected - present)}"