fix(db): re-running 0095/0096 rebuilds an invalid leftover index

A concurrent build that failed part-way leaves an INVALID index under its
name; IF NOT EXISTS kept it on the re-run, and 0095 then dropped the old
channel_id index, leaving none usable. The leftover is now dropped and
built again, and the old index goes only once the new one is valid.

backend/tests/test_migration_rerun_invalid_index.py

@@ -0,0 +1,53 @@
+"""A concurrent index build that failed part-way, then re-run (E31 S5 review), DB-lane.
+
+CREATE INDEX CONCURRENTLY that fails leaves an INVALID index under its name, and alembic_version is not
+stamped, so the next start runs the migration again. `IF NOT EXISTS` matched the name and kept the
+leftover, which the planner never uses; 0095 then dropped the old channel_id index, leaving `videos`
+with no usable one. A re-run now rebuilds the leftover and drops the old index only once the new one is
+valid.
+"""
+
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import text
+
+from app.db import engine
+
+
+def _index(name: str) -> tuple[bool, str] | None:
+    with engine.connect() as conn:
+        row = conn.execute(
+            text(
+                "SELECT i.indisvalid, pg_get_indexdef(i.indexrelid) FROM pg_index i "
+                "JOIN pg_class c ON c.oid = i.indexrelid WHERE c.relname = :name"
+            ),
+            {"name": name},
+        ).first()
+    return None if row is None else (bool(row[0]), row[1])
+
+
+def _leave_invalid(name: str) -> None:
+    """What a build that failed part-way leaves behind: the name taken, the index INVALID."""
+    with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
+        conn.execute(text(f"CREATE INDEX {name} ON videos (id)"))
+        conn.execute(
+            text(f"UPDATE pg_index SET indisvalid = false WHERE indexrelid = '{name}'::regclass")
+        )
+
+
+def test_a_rerun_rebuilds_an_invalid_leftover_before_dropping_the_old_index(db):
+    db.rollback()  # CONCURRENTLY waits for every open transaction, this session's too
+    cfg = Config("alembic.ini")
+    command.downgrade(cfg, "0094_plex_genres_hu_everywhere")
+    try:
+        _leave_invalid("ix_videos_channel_pub_cover")
+        _leave_invalid("ix_videos_views_desc")
+    finally:
+        command.upgrade(cfg, "head")
+
+    cover = _index("ix_videos_channel_pub_cover")
+    assert cover is not None and cover[0], cover
+    assert "INCLUDE (duration_seconds" in cover[1]
+    assert _index("ix_videos_channel_pub_desc") is None
+    views = _index("ix_videos_views_desc")
+    assert views is not None and views[0] and "view_count DESC NULLS LAST" in views[1], views