ci(migrations): enforce graph + reversible cycle checks; fix FK downgrade naming

backend/scripts/check_migration_graph.py

@@ -0,0 +1,77 @@
+"""Migration graph integrity checks for CI.
+
+Checks:
+- alembic script graph can be loaded (detects broken/missing links)
+- single head by default (unless ALLOW_MULTIPLE_HEADS=true)
+- no orphan revisions (all revisions reachable from heads)
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+from alembic.config import Config
+from alembic.script import ScriptDirectory
+
+
+def _truthy(value: str | None) -> bool:
+    return (value or "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def main() -> int:
+    root = Path(__file__).resolve().parents[1]
+    alembic_ini = root / "alembic.ini"
+    cfg = Config(str(alembic_ini))
+    cfg.attributes["configure_logger"] = False
+
+    try:
+        script = ScriptDirectory.from_config(cfg)
+    except Exception as exc:  # pragma: no cover - CI path
+        print(f"ERROR: unable to load Alembic script directory: {exc}")
+        return 1
+
+    try:
+        heads = list(script.get_heads())
+    except Exception as exc:  # pragma: no cover - CI path
+        print(f"ERROR: unable to resolve Alembic heads: {exc}")
+        return 1
+
+    allow_multiple_heads = _truthy(os.getenv("ALLOW_MULTIPLE_HEADS"))
+    if not heads:
+        print("ERROR: no Alembic heads found")
+        return 1
+
+    if len(heads) > 1 and not allow_multiple_heads:
+        print("ERROR: multiple Alembic heads detected (set ALLOW_MULTIPLE_HEADS=true only for intentional merge windows)")
+        for h in heads:
+            print(f"  - {h}")
+        return 1
+
+    try:
+        reachable = {rev.revision for rev in script.walk_revisions(base="base", head="heads") if rev.revision}
+    except Exception as exc:  # pragma: no cover - CI path
+        print(f"ERROR: failed while walking Alembic revision graph: {exc}")
+        return 1
+
+    all_revisions = {
+        rev.revision
+        for rev in script.revision_map._revision_map.values()  # type: ignore[attr-defined]
+        if getattr(rev, "revision", None)
+    }
+    orphans = sorted(all_revisions - reachable)
+    if orphans:
+        print("ERROR: orphan Alembic revisions detected (not reachable from heads):")
+        for rev in orphans:
+            print(f"  - {rev}")
+        return 1
+
+    print("OK: migration graph integrity passed")
+    print(f"Heads: {', '.join(heads)}")
+    print(f"Reachable revisions: {len(reachable)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())