feat(health): schema health / smell report

backend/app/api/snapshots.py

@@ -17,8 +17,14 @@
     SchemaSnapshotData,
 )
 from app.permissions import require_project_member
-from app.schemas import SnapshotCreateIn, SnapshotDetailOut, SnapshotOut
+from app.schemas import (
+    SchemaHealthOut,
+    SnapshotCreateIn,
+    SnapshotDetailOut,
+    SnapshotOut,
+)
 from app.ddl.export import snapshot_json_to_sql
+from app.spec.schema_health import analyze_schema_health
 from app.jobs.valkey_queue import enqueue_job_signal
 from app.spec.llm import (
     LlmConfigurationError,
@@ -161,6 +167,29 @@ async def export_snapshot_sql(
     return snapshot_json_to_sql(data.snapshot_json, target_dialect=dialect)
 
 
+@router.get("/{schema_snapshot_uuid}/schema-health", response_model=SchemaHealthOut)
+async def schema_health(
+    schema_snapshot_uuid: uuid.UUID,
+    user: CurrentUser = Depends(get_current_user),
+    session: AsyncSession = Depends(get_read_session),
+) -> SchemaHealthOut:
+    """Report schema smells for a snapshot (no-PK tables, unindexed FKs, orphans).
+
+    IDOR-safe: a uniform not-found status is returned for missing/unauthorized
+    snapshots so existence cannot be enumerated.
+    """
+    snap = await _get_authorized_snapshot(session, schema_snapshot_uuid, user)
+    if snap is None:
+        return SchemaHealthOut(
+            schema_snapshot_uuid=schema_snapshot_uuid, status="not_found", report=None
+        )
+    data = await session.get(SchemaSnapshotData, schema_snapshot_uuid)
+    report = analyze_schema_health(data.snapshot_json if data else None)
+    return SchemaHealthOut(
+        schema_snapshot_uuid=schema_snapshot_uuid, status="ok", report=report
+    )
+
+
 @router.get(
     "/{schema_snapshot_uuid}/reversing-spec.md",
     response_class=PlainTextResponse,

backend/app/schemas.py

@@ -91,6 +91,14 @@ class SnapshotDetailOut(BaseModel):
     snapshot_json: dict | None
 
 
+class SchemaHealthOut(BaseModel):
+    """Schema-smell findings for a single snapshot."""
+
+    schema_snapshot_uuid: uuid.UUID
+    status: str
+    report: dict | None
+
+
 class MeOut(BaseModel):
     """Current user payload returned by /me."""
 

backend/app/spec/schema_health.py

@@ -0,0 +1,134 @@
+"""Detect schema smells from a snapshot -- a "schema health" report.
+
+Reverse-engineering a database is most valuable when it also tells you what's
+*wrong*: tables you can't safely replicate or update (no primary key), foreign
+keys that will make joins slow (no supporting index), and tables that connect to
+nothing (possible dead weight or a missing relationship). This turns a snapshot
+into a prioritized findings list.
+
+Pure and dialect-agnostic; matches by name/oid within the snapshot only.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+WARNING = "warning"
+INFO = "info"
+
+_SEVERITY_RANK = {WARNING: 0, INFO: 1}
+
+
+def _index_columns(index_def: object) -> list[str]:
+    """Best-effort column list from an index definition.
+
+    ponytail: regex heuristic on the first ``(...)`` group. Expression indexes
+    (e.g. ``lower(email)``) yield the raw expression, which simply won't match a
+    plain FK column name -- acceptable for a "needs an index" hint.
+    """
+    text = str(index_def or "")
+    match = re.search(r"\(([^()]*)\)", text)
+    if not match:
+        return []
+    cols = []
+    for part in match.group(1).split(","):
+        token = part.strip().strip('"').split()  # drop ASC/DESC/opclass
+        if token:
+            cols.append(token[0].strip('"'))
+    return cols
+
+
+def _item(category: str, severity: str, target: str, detail: str) -> dict[str, Any]:
+    return {
+        "category": category,
+        "severity": severity,
+        "target": target,
+        "detail": detail,
+    }
+
+
+def analyze_schema_health(snapshot: dict[str, Any] | None) -> dict[str, Any]:
+    """Return schema-smell findings + a summary, most-severe first."""
+    snapshot = snapshot or {}
+    relations = snapshot.get("relations") or []
+    pk_columns = snapshot.get("pk_columns") or []
+    fk_edges = snapshot.get("fk_edges") or []
+    indexes = snapshot.get("indexes") or []
+
+    def _qname(rel: dict[str, Any]) -> str:
+        return f"{rel.get('schema_name')}.{rel.get('relation_name')}"
+
+    # Only ordinary tables ('r') can meaningfully lack a PK / be orphaned.
+    base_tables = {
+        r.get("relation_oid"): r
+        for r in relations
+        if (r.get("relation_kind") or "r") == "r"
+    }
+
+    pk_oids: set[Any] = set()
+    pk_cols_by_oid: dict[Any, set[str]] = {}
+    for pk in pk_columns:
+        oid = pk.get("relation_oid")
+        pk_oids.add(oid)
+        if pk.get("column_name") is not None:
+            pk_cols_by_oid.setdefault(oid, set()).add(str(pk["column_name"]).lower())
+
+    # Index first-columns per table (a FK is "covered" if it leads an index).
+    index_lead_by_oid: dict[Any, set[str]] = {}
+    for idx in indexes:
+        oid = idx.get("relation_oid") or idx.get("table_oid")
+        cols = _index_columns(idx.get("index_def"))
+        if cols:
+            index_lead_by_oid.setdefault(oid, set()).add(cols[0].lower())
+
+    connected: set[Any] = set()
+    for edge in fk_edges:
+        connected.add(edge.get("child_relation_oid"))
+        connected.add(edge.get("parent_relation_oid"))
+
+    items: list[dict[str, Any]] = []
+
+    for oid, rel in base_tables.items():
+        if oid not in pk_oids:
+            items.append(
+                _item(
+                    "no_primary_key", WARNING, _qname(rel),
+                    "Table has no primary key — rows can't be uniquely addressed; blocks safe updates/replication.",
+                )
+            )
+        if oid not in connected:
+            items.append(
+                _item(
+                    "orphan_table", INFO, _qname(rel),
+                    "Table has no foreign keys in or out — possibly disconnected or a missing relationship.",
+                )
+            )
+
+    for edge in fk_edges:
+        child_oid = edge.get("child_relation_oid")
+        rel = base_tables.get(child_oid)
+        if rel is None:
+            continue
+        col = str(edge.get("child_column_name") or "").lower()
+        if not col:
+            continue
+        covered = col in pk_cols_by_oid.get(child_oid, set()) or col in index_lead_by_oid.get(child_oid, set())
+        if not covered:
+            items.append(
+                _item(
+                    "unindexed_foreign_key", WARNING,
+                    f"{_qname(rel)}.{edge.get('child_column_name')}",
+                    "Foreign key column has no supporting index — joins and cascading deletes will scan the whole table.",
+                )
+            )
+
+    items.sort(key=lambda i: (_SEVERITY_RANK.get(i["severity"], 9), i["target"]))
+
+    summary = {
+        "warning": sum(1 for i in items if i["severity"] == WARNING),
+        "info": sum(1 for i in items if i["severity"] == INFO),
+        "total": len(items),
+        "tables_analyzed": len(base_tables),
+    }
+    return {"items": items, "summary": summary}

backend/tests/test_schema_health.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from app.spec.schema_health import analyze_schema_health
+
+
+def _cats(report):
+    return {(i["category"], i["severity"]) for i in report["items"]}
+
+
+def test_flags_table_without_primary_key():
+    snap = {
+        "relations": [{"relation_oid": 1, "relation_kind": "r", "schema_name": "public", "relation_name": "log"}],
+        "columns": [{"relation_oid": 1, "column_name": "msg", "data_type": "text", "is_not_null": False}],
+        "pk_columns": [],
+        "fk_edges": [],
+        "indexes": [],
+    }
+    report = analyze_schema_health(snap)
+    assert ("no_primary_key", "warning") in _cats(report)
+    assert report["summary"]["tables_analyzed"] == 1
+
+
+def test_flags_unindexed_foreign_key_but_not_indexed_one():
+    base = {
+        "relations": [
+            {"relation_oid": 1, "relation_kind": "r", "schema_name": "public", "relation_name": "member"},
+            {"relation_oid": 2, "relation_kind": "r", "schema_name": "public", "relation_name": "orders"},
+        ],
+        "columns": [],
+        "pk_columns": [
+            {"relation_oid": 1, "column_name": "member_id"},
+            {"relation_oid": 2, "column_name": "order_id"},
+        ],
+        "fk_edges": [
+            {"child_relation_oid": 2, "parent_relation_oid": 1, "child_column_name": "member_id", "parent_column_name": "member_id"},
+        ],
+        "indexes": [],
+    }
+    # No index on orders.member_id -> flagged
+    assert ("unindexed_foreign_key", "warning") in _cats(analyze_schema_health(base))
+
+    # Add a covering index -> not flagged
+    with_index = {**base, "indexes": [
+        {"relation_oid": 2, "index_name": "ix_orders_member", "index_def": "CREATE INDEX ix_orders_member ON public.orders USING btree (member_id)"},
+    ]}
+    assert ("unindexed_foreign_key", "warning") not in _cats(analyze_schema_health(with_index))
+
+
+def test_fk_column_that_is_pk_is_not_flagged_unindexed():
+    # child FK column is itself the child's PK (auto-indexed)
+    snap = {
+        "relations": [
+            {"relation_oid": 1, "relation_kind": "r", "schema_name": "public", "relation_name": "member"},
+            {"relation_oid": 2, "relation_kind": "r", "schema_name": "public", "relation_name": "profile"},
+        ],
+        "columns": [],
+        "pk_columns": [
+            {"relation_oid": 1, "column_name": "member_id"},
+            {"relation_oid": 2, "column_name": "member_id"},
+        ],
+        "fk_edges": [
+            {"child_relation_oid": 2, "parent_relation_oid": 1, "child_column_name": "member_id", "parent_column_name": "member_id"},
+        ],
+        "indexes": [],
+    }
+    assert ("unindexed_foreign_key", "warning") not in _cats(analyze_schema_health(snap))
+
+
+def test_flags_orphan_table_and_sorts_warnings_first():
+    snap = {
+        "relations": [
+            {"relation_oid": 1, "relation_kind": "r", "schema_name": "public", "relation_name": "island"},
+            {"relation_oid": 2, "relation_kind": "r", "schema_name": "public", "relation_name": "member"},
+            {"relation_oid": 3, "relation_kind": "r", "schema_name": "public", "relation_name": "orders"},
+        ],
+        "columns": [],
+        "pk_columns": [
+            {"relation_oid": 1, "column_name": "id"},
+            {"relation_oid": 2, "column_name": "member_id"},
+            {"relation_oid": 3, "column_name": "order_id"},
+        ],
+        "fk_edges": [
+            {"child_relation_oid": 3, "parent_relation_oid": 2, "child_column_name": "member_id", "parent_column_name": "member_id"},
+        ],
+        "indexes": [],
+    }
+    report = analyze_schema_health(snap)
+    cats = _cats(report)
+    assert ("orphan_table", "info") in cats  # 'island' connects to nothing
+    severities = [i["severity"] for i in report["items"]]
+    assert severities == sorted(severities, key={"warning": 0, "info": 1}.get)
+
+
+def test_views_are_not_flagged_and_empty_snapshot():
+    snap = {
+        "relations": [{"relation_oid": 9, "relation_kind": "v", "schema_name": "public", "relation_name": "v_report"}],
+        "columns": [], "pk_columns": [], "fk_edges": [], "indexes": [],
+    }
+    assert analyze_schema_health(snap)["items"] == []
+    assert analyze_schema_health({})["items"] == []