"""scheduler-job partial indexes + boolean-index cleanup (PERF-17/18/3)

Two scheduler jobs full-seq-scanned the 257k-row videos table every run:
  - PERF-17: enrich_pending (`WHERE enriched_at IS NULL LIMIT N`, every few minutes) — 44.6k buffers /
    run even when nothing is pending. A partial index over just the pending rows makes it a 1-buffer
    (empty) index scan.
  - PERF-18: maintenance revalidate (`WHERE unavailable_since IS NULL ORDER BY last_checked_at NULLS
    FIRST LIMIT N`, ~300 pages/run) — 44.7k buffers/page; ix_videos_last_checked_at had idx_scan=0. A
    partial index in that exact order serves it as an index scan (~36 buffers/page) and replaces the
    unused plain one.

PERF-3: the videos table also carried plain btrees on low-cardinality boolean columns the planner never
picks (a `col IS false` majority predicate can't ride a btree). Drop the never-useful ones (is_short,
via_explore) and slim the rest to partial indexes over the small/shrinking subset the job queries target
(via_search=true, NOT shorts_probed, unavailable_since IS NOT NULL) — less write amplification on the
biggest table, same lookups. live_status is left alone (a separate finding).

All CONCURRENTLY (autocommit_block) — downtime ~0; IF (NOT) EXISTS keeps the drift test's
downgrade->re-upgrade idempotent.
"""

from typing import Sequence, Union

from alembic import op

revision: str = "0069_job_index_hygiene"
down_revision: Union[str, None] = "0068_published_at_not_null"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    with op.get_context().autocommit_block():
        # PERF-17: enrich-pending partial (empty when nothing is pending -> 1-buffer scan).
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_enrich_pending "
            "ON videos (id) WHERE enriched_at IS NULL"
        )
        # PERF-18: maintenance revalidate order, as a partial in the exact WHERE + ORDER BY shape.
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_recheck_order "
            "ON videos (last_checked_at ASC NULLS FIRST) WHERE unavailable_since IS NULL"
        )
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_last_checked_at")
        # PERF-3: drop the never-picked boolean btrees; slim the rest to partials over the target subset.
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_is_short")
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_via_explore")
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_via_search")
        # Predicate MUST match the query's `.is_(True)`/`.is_(False)` form (SQL `IS TRUE`/`IS FALSE`):
        # Postgres's partial-index prover does NOT match a `col IS TRUE` clause to a bare `col` predicate,
        # so `WHERE via_search` / `WHERE NOT shorts_probed` would leave the queries seq-scanning.
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_via_search "
            "ON videos (id) WHERE via_search IS TRUE"
        )
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_shorts_probed")
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_shorts_probed "
            "ON videos (id) WHERE shorts_probed IS FALSE"
        )
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_unavailable_since")
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_unavailable_since "
            "ON videos (unavailable_since) WHERE unavailable_since IS NOT NULL"
        )


def downgrade() -> None:
    with op.get_context().autocommit_block():
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_enrich_pending")
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_recheck_order")
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_last_checked_at "
            "ON videos (last_checked_at)"
        )
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_is_short ON videos (is_short)"
        )
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_via_explore ON videos (via_explore)"
        )
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_via_search")
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_via_search ON videos (via_search)"
        )
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_shorts_probed")
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_shorts_probed ON videos (shorts_probed)"
        )
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_unavailable_since")
        op.execute(
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_unavailable_since "
            "ON videos (unavailable_since)"
        )