"""Every feed sort on a videos column rides an index, and an ICU upgrade rebuilds the title ones.

Without an index in the exact ORDER BY, each feed page read the whole videos heap and kept a top-N
(prod: 350 ms a page for views/duration, 550–780 ms for title). Migration 0096 adds one per direction
and per reader language. The planner is told it may not sort or scan the table, so a plan that still
sorts it whole means no index matches the ORDER BY `feedquery.order_by` builds: a key, direction, NULLS
rule or collation changed on one side only. An Incremental Sort is fine: the index gives the key's order
and only its ties are sorted by id (`newest`/`oldest` ride the published_at indexes that way). DB lane.
"""

import re

import pytest
from sqlalchemy import select, text
from sqlalchemy.dialects import postgresql

from app import collation
from app.db import engine
from app.feedquery import order_by, sort_keys
from app.catalog.models import Video

VIDEO_SORTS = [
    "newest",
    "oldest",
    "views",
    "views_asc",
    "duration_desc",
    "duration_asc",
    "title",
    "title_desc",
]


def _plan(db, sort: str, lang: str) -> str:
    stmt = select(Video.id).order_by(*order_by(sort_keys(sort, 0, None, lang))).limit(60)
    sql = str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
    db.execute(text("SET LOCAL enable_seqscan = off"))
    db.execute(text("SET LOCAL enable_sort = off"))
    db.execute(text("SET LOCAL enable_bitmapscan = off"))
    return "\n".join(db.execute(text(f"EXPLAIN {sql}")).scalars())


@pytest.mark.parametrize("lang", sorted(collation.ICU))
@pytest.mark.parametrize("sort", VIDEO_SORTS)
def test_the_sort_walks_an_index(db, sort, lang):
    plan = _plan(db, sort, lang)
    db.rollback()
    full_sort = re.search(r"(?<!Incremental )Sort  \(", plan)
    assert not full_sort, f"{sort} ({lang}) sorts the table:\n{plan}"
    assert "Index Scan" in plan or "Index Only Scan" in plan, plan


def _drifted(db) -> list[str]:
    return list(
        db.execute(
            text(
                "SELECT collname FROM pg_collation WHERE collname = ANY(:names) "
                "AND collversion IS DISTINCT FROM pg_collation_actual_version(oid)"
            ),
            {"names": list(collation.ICU.values())},
        ).scalars()
    )


def _files(db) -> dict[str, int]:
    return dict(
        db.execute(
            text("SELECT relname, relfilenode FROM pg_class WHERE relname LIKE 'ix_videos_title_%'")
        ).all()
    )


def test_an_icu_upgrade_rebuilds_the_title_indexes(db):
    # What an image update with a newer ICU looks like from inside: the recorded version no longer
    # matches the library's (the test database runs as the bootstrap superuser, as the app's does).
    db.execute(text("UPDATE pg_collation SET collversion = '0.0' WHERE collname = 'hu-x-icu'"))
    db.commit()
    try:
        assert _drifted(db) == ["hu-x-icu"]
        before = _files(db)
        # A concurrent rebuild waits for every open transaction, this session's too.
        db.rollback()
        rebuilt = collation.heal_drift(engine)
        assert sorted(rebuilt) == ["ix_videos_title_hu", "ix_videos_title_hu_desc"]
        after = _files(db)
        # Rebuilt = written to new files; the English pair, on an unchanged ICU, is left alone.
        assert {n for n in before if before[n] != after[n]} == set(rebuilt)
        assert _drifted(db) == []
    finally:
        db.execute(
            text(
                "UPDATE pg_collation SET collversion = pg_collation_actual_version(oid) "
                "WHERE collname = 'hu-x-icu'"
            )
        )
        db.commit()


def test_no_drift_rebuilds_nothing(db):
    assert collation.heal_drift(engine) == []