"""videos: the per-channel index covers what the channel manager and the feed sidebar read (E31 S4d)

The channel manager (`GET /api/channels`) sums each subscribed channel's videos: count, last upload, total
duration. The feed sidebar (`/feed/summary`, `/feed/years`) counts the reader's videos per channel and per
year under the default filters: available, not via explore, not a short, not live. Every one of them found
its rows through `ix_videos_channel_pub_desc` (channel_id, published_at DESC NULLS LAST, id) and then read
each video's heap page for the rest: ~270k rows at ~2.5 KB each, 681 MB on prod. That stays in cache only
while nothing else needs the memory. Measured on prod (2026-10-02): `/api/channels` 51–61 ms warm, but
2.25 s and 1.96 s on a first open in the browser and 6.4 s under EXPLAIN with 30k blocks read from disk;
`/feed/summary` and `/feed/years` ~140 ms warm, ~800 ms cold.

The same key with those columns INCLUDEd makes them Index Only Scans. On a prod-copy localdev, after a
VACUUM, the manager's sums read 0 heap pages and the summary reads the 25 MB index instead of the heap.
The visibility map is what lets them skip the heap; prod's autovacuum kept it complete (relallvisible ==
relpages). The key columns are unchanged, so every reader of the old index (the per-channel feed, the FK
cascade, the sync upsert, channel detail) rides this one. The old index is dropped.
tests/test_channel_list_index_only.py holds the queries and this column list together.

CONCURRENTLY (autocommit_block) builds it without blocking a live prod. IF (NOT) EXISTS keeps the drift
test's downgrade → re-upgrade round trip idempotent. A re-run after a build that failed part-way rebuilds
the INVALID leftover, and the old index is dropped only once the new one is valid: a re-run that kept the
leftover and dropped the old one left `videos` with no usable channel_id index (E31 S5 review).
"""

from alembic import op
from sqlalchemy import text

revision: str = "0095_videos_channel_cover_index"
down_revision: str | None = "0094_plex_genres_hu_everywhere"
branch_labels: str | None = None
depends_on: str | None = None


def _valid(name: str) -> bool | None:
    """The index's `indisvalid`, or None when there is no index of that name."""
    row = (
        op.get_bind()
        .execute(
            text(
                "SELECT i.indisvalid 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])


def _build(name: str, ddl: str) -> None:
    """CREATE INDEX CONCURRENTLY `name` (`ddl` is what follows the name), re-runnable. A build that
    failed part-way leaves an INVALID index under the name, and `IF NOT EXISTS` would keep it: the
    planner never uses it, and nothing says so (E31 S5 review). Such a leftover is dropped and built
    again."""
    if _valid(name) is False:
        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {name}")
    op.execute(f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {name} {ddl}")


def _swap(build: str, ddl: str, drop: str) -> None:
    _build(build, ddl)
    if not _valid(build):
        raise RuntimeError(f"{build} did not build valid; {drop} is kept")
    op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {drop}")


def upgrade() -> None:
    with op.get_context().autocommit_block():
        _swap(
            "ix_videos_channel_pub_cover",
            "ON videos (channel_id, published_at DESC NULLS LAST, id) "
            "INCLUDE (duration_seconds, unavailable_since, via_explore, is_short, live_status)",
            "ix_videos_channel_pub_desc",
        )


def downgrade() -> None:
    with op.get_context().autocommit_block():
        _swap(
            "ix_videos_channel_pub_desc",
            "ON videos (channel_id, published_at DESC NULLS LAST, id ASC)",
            "ix_videos_channel_pub_cover",
        )