diff --git a/backend/app/api/snapshots.py b/backend/app/api/snapshots.py index 79f2515f..a7fe4f56 100644 --- a/backend/app/api/snapshots.py +++ b/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, diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 0359799c..6b3ff8fc 100644 --- a/backend/app/schemas.py +++ b/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.""" diff --git a/backend/app/spec/schema_health.py b/backend/app/spec/schema_health.py new file mode 100644 index 00000000..e314cc1d --- /dev/null +++ b/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} diff --git a/backend/tests/test_schema_health.py b/backend/tests/test_schema_health.py new file mode 100644 index 00000000..874ff724 --- /dev/null +++ b/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"] == []