"""Feed search + watch-state views (E2 S3), DB-lane.

Regression nets for two query rewrites that must stay behaviour-equivalent:
- PERF-12: the search channel-name branch moved from an ORed `unaccent(channels.title) ILIKE` (which
  defeated the search_vector GIN) to a pre-resolved `channel_id IN (...)`. A search must STILL return
  both videos whose text matches AND videos whose CHANNEL name matches.
- PERF-16: the watched/hidden/in_progress views moved from an outer join + `coalesce(state.status,'new')`
  to an inner join on video_states + the raw column. Each view must return exactly the same set.
"""

from datetime import datetime, timezone

import pytest

from app.models import Channel, ChannelTag, Subscription, Tag, Video, VideoState


def _dt(day: int) -> datetime:
    return datetime(2024, 1, day, tzinfo=timezone.utc)


def _video(vid: str, ch: str, title: str) -> Video:
    return Video(
        id=vid,
        channel_id=ch,
        published_at=_dt(1),
        title=title,
        is_short=False,
        live_status="none",
        via_search=False,
        via_explore=False,
    )


def _feed_ids(client, query: str) -> set[str]:
    resp = client.get(f"/api/feed?{query}&limit=50")
    assert resp.status_code == 200, resp.text
    return {it["id"] for it in resp.json()["items"]}


def test_search_matches_both_video_text_and_channel_name(db, as_user, make_user):
    # Two subscribed channels: one whose NAME matches the query (its video's text does not), and one
    # whose video TEXT matches (its name does not). The pre-resolved channel_id branch must still OR in
    # the name match, so both videos come back — and an unrelated video on a matching-name-less channel
    # stays out.
    user = make_user(email="search@example.com")
    db.add_all(
        [
            Channel(id="UCkube", title="Kubernetes Weekly", subscriber_count=10),
            Channel(id="UCother", title="Some Other Channel", subscriber_count=10),
        ]
    )
    db.add_all(
        [
            Subscription(user_id=user.id, channel_id="UCkube", priority=0, hidden=False),
            Subscription(user_id=user.id, channel_id="UCother", priority=0, hidden=False),
        ]
    )
    db.add_all(
        [
            _video("vidChan", "UCkube", "Totally unrelated content"),  # channel NAME matches
            _video("vidText", "UCother", "kubernetes deep dive"),  # video TEXT matches
            _video("vidNone", "UCother", "cooking pasta at home"),  # neither matches
        ]
    )
    db.commit()

    ids = _feed_ids(as_user(user), "q=kubernetes&show=all")
    assert ids == {"vidChan", "vidText"}


def test_watch_state_views_return_the_right_sets(db, as_user, make_user):
    user = make_user(email="views@example.com")
    db.add(Channel(id="UCv", title="Channel V", subscriber_count=10))
    db.add(Subscription(user_id=user.id, channel_id="UCv", priority=0, hidden=False))
    db.add_all(
        [
            _video("vWatch", "UCv", "watched one"),
            _video("vHide", "UCv", "hidden one"),
            _video("vProg", "UCv", "in progress one"),
            _video("vNew", "UCv", "brand new one"),
        ]
    )
    db.flush()
    db.add_all(
        [
            VideoState(user_id=user.id, video_id="vWatch", status="watched"),
            VideoState(user_id=user.id, video_id="vHide", status="hidden"),
            # in progress = a stored resume position while still "new" (not watched/hidden)
            VideoState(user_id=user.id, video_id="vProg", status="new", position_seconds=30),
            # vNew has NO state row -> "new"/unwatched
        ]
    )
    db.commit()

    client = as_user(user)
    assert _feed_ids(client, "show=watched") == {"vWatch"}
    assert _feed_ids(client, "show=hidden") == {"vHide"}
    assert _feed_ids(client, "show=in_progress") == {"vProg"}
    # unwatched excludes watched + hidden, keeps new AND in-progress
    assert _feed_ids(client, "show=unwatched") == {"vProg", "vNew"}
    # "all" shows everything except hidden
    assert _feed_ids(client, "show=all") == {"vWatch", "vProg", "vNew"}


def _seed_tagged(db, make_user):
    """A user subscribed to 4 channels with a spread of language/topic/other tags and differing video
    counts, so the count sum and the per-category facet drill-down are all non-trivial."""
    user = make_user(email="summary@example.com")
    tags = {
        "L1": Tag(user_id=None, name="Hungarian", category="language"),
        "L2": Tag(user_id=None, name="English", category="language"),
        "T1": Tag(user_id=None, name="Tech", category="topic"),
        "T2": Tag(user_id=None, name="Gaming", category="topic"),
        "O1": Tag(user_id=None, name="Favourite", category="other"),
        "O2": Tag(user_id=None, name="Docs", category="other"),
        # A USER-owned tag (+ a user-owned ChannelTag link below) so the visibility branch
        # `ChannelTag.user_id == user.id` is exercised, not only the system `user_id IS NULL` one.
        "U1": Tag(user_id=user.id, name="MyPin", category="other"),
    }
    db.add_all(tags.values())
    chans = {c: Channel(id=c, title=c, subscriber_count=10) for c in ("chA", "chB", "chC", "chD")}
    db.add_all(chans.values())
    db.flush()
    # channel -> (system) tag keys. chA carries two languages (L1+L2); chC two "other" tags (O1+O2)
    # AND chA carries O1 too — so the per-category any/all match narrows NON-TRIVIALLY (AND != OR) for
    # language and other, not only topic: e.g. other {O1,O2} → OR {chA,chC} vs AND {chC}; language
    # {L1,L2} → OR {chA,chB,chC} vs AND {chA}. (For topic, chA and chB each carry a DIFFERENT topic tag
    # — T1 vs T2 — so topic {T1,T2} → OR {chA,chB,chC} vs AND {} empty.)
    links = {
        "chA": ["L1", "L2", "T1", "O1"],
        "chB": ["L1", "T2"],
        "chC": ["L2", "T1", "O1", "O2"],
        "chD": [],
    }
    for cid, tkeys in links.items():
        db.add(Subscription(user_id=user.id, channel_id=cid, priority=0, hidden=False))
        for tk in tkeys:
            db.add(ChannelTag(channel_id=cid, tag_id=tags[tk].id, user_id=None))
    # The user's own private link (visible only to them): chB carries U1 for this user.
    db.add(ChannelTag(channel_id="chB", tag_id=tags["U1"].id, user_id=user.id))
    # differing video counts per channel (chA:3, chB:2, chC:1, chD:2)
    counts = {"chA": 3, "chB": 2, "chC": 1, "chD": 2}
    for cid, n in counts.items():
        db.add_all([_video(f"{cid}v{i}", cid, f"{cid} clip {i}") for i in range(n)])
    db.commit()
    return user, {k: v.id for k, v in tags.items()}


# Query-param combos (as a template with {L1}/{T1}/… placeholders) that stress every facet path:
# no tags, single-category, cross-category, disjunctive + conjunctive topic modes, and a search filter.
_SUMMARY_CASES = [
    "show=all",
    "show=all&tags={L1}",
    "show=all&tags={T1}",
    "show=all&tags={L1}&tags={T1}",
    "show=all&tags={T1}&tags={T2}&tag_mode_topic=or",
    "show=all&tags={T1}&tags={T2}&tag_mode_topic=and",
    # Per-category any/all is NOT topic-only — language and other must drill down identically too.
    "show=all&tags={L1}&tags={L2}&tag_mode_language=or",
    "show=all&tags={L1}&tags={L2}&tag_mode_language=and",
    "show=all&tags={O1}&tags={O2}&tag_mode_other=and",
    "show=all&tags={U1}",  # user-owned tag (visibility == user.id branch)
    "show=all&tags={U1}&tags={L1}",  # user tag AND a system tag, cross-category
    # "No tag" filter (item B): channels with no USER tag. It stays applied in the base pass, so the
    # facet/count lockstep must still hold with it on (alone, and combined with a positive tag filter).
    "show=all&untagged=true",
    "show=all&untagged=true&tags={L1}",
    "show=all&q=clip",
    "show=unwatched",
]


def test_untagged_filter_keeps_only_channels_the_user_has_not_tagged(db, as_user, make_user):
    """The "No tag" filter (item B): `untagged=true` keeps only channels with NO tag the USER personally
    attached (no user-owned ChannelTag). System/auto tags (user_id NULL) don't count, and another user's
    tag on the same channel doesn't either."""
    user = make_user(email="untagged@example.com")
    other = make_user(email="someone-else@example.com")
    mine = Tag(user_id=user.id, name="Fav", category="other")
    sys_topic = Tag(user_id=None, name="Tech", category="topic")
    theirs = Tag(user_id=other.id, name="TheirTag", category="other")
    db.add_all([mine, sys_topic, theirs])
    chans = {c: Channel(id=c, title=c, subscriber_count=10) for c in ("chMine", "chSys", "chNone")}
    db.add_all(chans.values())
    db.flush()
    # chMine: the user's own tag → tagged. chSys: only a system tag + another user's tag → still
    # "untagged" for THIS user. chNone: nothing at all → untagged.
    db.add(ChannelTag(channel_id="chMine", tag_id=mine.id, user_id=user.id, source="user"))
    db.add(ChannelTag(channel_id="chSys", tag_id=sys_topic.id, user_id=None, source="auto"))
    db.add(ChannelTag(channel_id="chSys", tag_id=theirs.id, user_id=other.id, source="user"))
    for cid in chans:
        db.add(Subscription(user_id=user.id, channel_id=cid, priority=0, hidden=False))
        db.add(_video(f"{cid}v", cid, f"{cid} clip"))
    db.commit()
    client = as_user(user)

    # Off: every subscribed channel shows.
    assert _feed_ids(client, "show=all") == {"chMinev", "chSysv", "chNonev"}
    # On: only channels the user has NOT personally tagged (a system tag / another user's tag don't count).
    assert _feed_ids(client, "show=all&untagged=true") == {"chSysv", "chNonev"}


def test_feed_years_counts_ignore_the_date_narrowing_but_honour_other_filters(
    db, as_user, make_user
):
    """The year facet (item 8 / E8·S2): `/api/feed/years` returns a per-year video count, newest first,
    with the publish-date narrowing DROPPED (so switching years stays a drill-down) but every other
    filter still applied (here: a channel pick)."""
    user = make_user(email="years@example.com")
    chans = {c: Channel(id=c, title=c, subscriber_count=10) for c in ("chX", "chY")}
    db.add_all(chans.values())
    db.flush()
    # chX: 2 videos in 2024, 1 in 2022; chY: 1 in 2023 (should drop once we scope to chX).
    spread = {
        "chX": [(2024, 1), (2024, 2), (2022, 5)],
        "chY": [(2023, 3)],
    }
    for cid, dates in spread.items():
        db.add(Subscription(user_id=user.id, channel_id=cid, priority=0, hidden=False))
        for i, (yr, day) in enumerate(dates):
            db.add(
                Video(
                    id=f"{cid}v{i}",
                    channel_id=cid,
                    published_at=datetime(yr, 1, day, tzinfo=timezone.utc),
                    title=f"{cid} clip {i}",
                    is_short=False,
                    live_status="none",
                    via_search=False,
                    via_explore=False,
                )
            )
    db.commit()
    client = as_user(user)

    # All channels: every year present, counted, newest first.
    resp = client.get("/api/feed/years?show=all")
    assert resp.status_code == 200, resp.text
    assert resp.json()["years"] == [
        {"year": 2024, "count": 2},
        {"year": 2023, "count": 1},
        {"year": 2022, "count": 1},
    ]

    # A channel filter narrows the year list; a date narrowing is ignored (the point of the facet — so
    # a selected year never zeroes out the others). Scoping to chX drops 2023, keeps 2024 (×2) + 2022.
    resp = client.get("/api/feed/years?show=all&channel_ids=chX&published_after=2024-01-01")
    assert resp.status_code == 200, resp.text
    assert resp.json()["years"] == [
        {"year": 2024, "count": 2},
        {"year": 2022, "count": 1},
    ]


def test_feed_years_buckets_by_utc_year_under_a_non_utc_session(db, make_user):
    """feed_years must bucket by UTC year to stay in lockstep with the year filter's UTC-pinned
    published_after/before bounds. published_at is timestamptz; a bare extract('year', …) would use the
    DB session timezone, so a video just before midnight UTC on New Year's Eve would bucket into the
    wrong (local) year on a non-UTC session and disagree with the feed the year chip drills into. Called
    directly on this session (the HTTP client uses a separate connection SET TIME ZONE wouldn't reach)."""
    from sqlalchemy import text

    from app.feedquery import feed_years

    user = make_user(email="tz-years@example.com")
    db.add(Channel(id="chTZ", title="chTZ", subscriber_count=1))
    db.add(Subscription(user_id=user.id, channel_id="chTZ", priority=0, hidden=False))
    # 2023-12-31 23:30 UTC — a positive-offset session tz local-shifts this into 2024.
    db.add(
        Video(
            id="tzv0",
            channel_id="chTZ",
            published_at=datetime(2023, 12, 31, 23, 30, tzinfo=timezone.utc),
            title="NYE clip",
            is_short=False,
            live_status="none",
            via_search=False,
            via_explore=False,
        )
    )
    db.commit()

    params: dict = {
        "tags": [],
        "tag_modes": {"language": "or", "topic": "or", "other": "or"},
        "untagged": False,
        "channel_ids": [],
        "q": None,
        "min_duration": None,
        "max_duration": None,
        "max_age_days": None,
        "published_after": None,
        "published_before": None,
        "show_normal": True,
        "include_shorts": False,
        "include_live": False,
        "show": "all",
        "scope": "my",
        "library_source": "organic",
    }
    db.execute(
        text("SET TIME ZONE 'Asia/Tokyo'")
    )  # +9: pushes the video's local wall-clock into 2024
    try:
        years = feed_years(db, user, **params)
    finally:
        db.execute(text("SET TIME ZONE 'UTC'"))  # SET (not SET LOCAL) survives rollback — reset it
    # UTC bucketing keeps the boundary video in 2023; a bare session-tz extract would report 2024.
    assert years == [{"year": 2023, "count": 1}]


@pytest.mark.parametrize("case", _SUMMARY_CASES)
def test_feed_summary_matches_count_plus_facets(db, as_user, make_user, case):
    # /feed/summary must return byte-identical numbers to the old /feed/count + /facets it replaces,
    # across the full matrix of tag filters — it reimplements the tag drill-down in Python, so this is
    # the equivalence net.
    user, tag_ids = _seed_tagged(db, make_user)
    client = as_user(user)
    qs = case.format(**tag_ids)

    summary = client.get(f"/api/feed/summary?{qs}")
    count = client.get(f"/api/feed/count?{qs}")
    facets = client.get(f"/api/facets?{qs}")
    assert summary.status_code == count.status_code == facets.status_code == 200

    assert summary.json()["count"] == count.json()["count"], qs
    assert summary.json()["facets"] == facets.json()["counts"], qs