"""The row fetchers behind the Home surface — one place where "what is startable right now" is
expressed as SQL, for both sources.

Two rules shape this module:

**Reuse the query layer, not the route.** The YouTube half calls `feedquery.filtered_query` /
`sort_keys`, so the Home surface inherits the feed's own definitions of subscribed, blocked,
explore-visible and in-progress for free, and cannot drift from them. The Plex half calls
`plex.rollup` and the `plex.cards` serializers rather than `plex/discover.py`: that read-model
narrows EVERY rail by the Plex sidebar's `LibraryFilters`, which is right for a surface that HAS a
sidebar and wrong for a landing page that does not.

**One seed drives both halves.** The feed orders a shuffle by `md5(id || seed)` so a keyset cursor
stays valid across pages; the Plex surprise rail uses `ORDER BY random()`, which is fine for a single
un-paged draw but cannot be reproduced. A cross-source reroll needs both halves to move together —
otherwise "Mást kérek" re-draws the Plex picks while the YouTube ones stand still — so the Plex side
here takes the feed's seeded-md5 idiom instead of `random()`.
"""

from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session, aliased

from app.discover.budget import band
from app.feedquery import (
    feed_columns,
    filtered_query,
    not_blocked_cond,
    order_by,
    serialize,
    sort_keys,
)
from app.models import User
from app.plex.models import PlexItem, PlexShow, PlexState
from app.catalog.models import Channel, Video
from app.reader.models import PlaylistItem, VideoState
from app.plex import rollup
from app.plex.cards import display_title, leaf_card, movie_card, show_card
from app.plex.query import apply_show_state_filter, content_libs
from app.userscope import owned_by

# How many rows a single rail may carry — Home's rails and the Plex module's Discover rails alike
# (`plex/discover.py` imports it): the two surfaces should feel like the same app, and a rail
# is a rail. One definition, because a comment was all that held two copies in step.
RAIL_LIMIT = 20

# The feed parameters a Home rail runs with. Spelled out rather than borrowed from the route's
# `_feed_params` defaults, because those are an HTTP contract that may change for the feed's own
# reasons; the landing page's scope should not move with them silently.
_FEED_BASE: dict = {
    "tags": [],
    "tag_modes": {},
    "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": "unwatched",
    "scope": "my",
    "library_source": "organic",
}


def _seeded(col, seed: int):
    """The feed's deterministic shuffle expression, reused for Plex so one seed moves both halves."""
    return func.md5(func.concat(col, str(seed)))


# ── YouTube ───────────────────────────────────────────────────────────────────────────────────────


def yt_in_progress(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[tuple[object, dict]]:
    """Started-but-not-finished videos, most recently progressed first, as (progressed_at, row).

    The predicate is `feedquery`'s `show="in_progress"` verbatim (a stored resume position, status not
    watched/hidden) — but the query is built here rather than through `filtered_query` because this
    rail wants an ordering by `progress_updated_at`, and that column lives on the state alias the feed
    query keeps private. Driving from `video_states` is also the faster plan: it is a per-user table of
    a few hundred rows, against a catalog of many thousands.

    Building it here means the feed's OTHER predicates have to be chosen deliberately rather than
    inherited. `not_blocked_cond` is shared, because "never show me this channel" outranks "you
    started it" — blocking leaves the state row untouched, so without this the video the user just
    banned heads the resume rail. The feed's SCOPE clauses (subscribed / via_explore accessibility)
    are deliberately NOT applied, for the reason `yt_video_row` gives: something you have already
    started must stay resumable even after you unsubscribe or an exploration expires."""
    st = aliased(VideoState)
    q = (
        select(*feed_columns(user, st), st.progress_updated_at.label("progressed_at"))
        .join(Channel, Channel.id == Video.channel_id)
        .join(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
        .where(
            st.position_seconds > 0,
            st.status.notin_(("watched", "hidden")),
            Video.unavailable_since.is_(None),
            not_blocked_cond(user),
        )
        .order_by(st.progress_updated_at.desc().nullslast())
        .limit(limit)
    )
    return [(r.progressed_at, serialize(r)) for r in db.execute(q).all()]


def yt_surprise(db: Session, user: User, budget: str, seed: int, limit: int) -> list[dict]:
    """Unwatched videos from the user's own feed scope, in seeded-shuffle order, inside the budget.

    Shorts are opened up only for the shortest band: "I have ten minutes" is the one answer a Short is
    actually a good response to, and the feed's own default keeps them out otherwise."""
    lo, hi = band(budget)
    params = dict(_FEED_BASE, min_duration=lo, max_duration=hi, include_shorts=budget == "short")
    query, rank_expr = filtered_query(db, user, **params)
    keys = sort_keys("shuffle", seed, rank_expr)
    # Over-fetch, then drop anything already started. The feed's `unwatched` scope keeps watched and
    # hidden out but deliberately keeps in-progress rows in (they belong in the feed). Here they must
    # go: the resume tier has already had its chance at them, and it may have passed precisely because
    # the REMAINING time did not fit the budget — offering the same item back as a surprise, whose
    # runtime does fit, would label a half-finished thing "Meglepetés" and contradict the question.
    rows = db.execute(query.order_by(*order_by(keys)).limit(limit * 4)).all()
    out = [serialize(r) for r in rows]
    return [r for r in out if not r["position_seconds"]][:limit]


def yt_video_row(db: Session, user: User, video_id: str) -> dict | None:
    """One video as a feed row, with this user's watch state.

    Built directly rather than through `filtered_query` on purpose: a PINNED video must keep working
    even when it falls outside the feed's scope — the user unsubscribed from the channel, or blocked
    it, or the video only ever arrived through a search. The shelf is the surface you fall back to;
    it silently losing an entry because a filter moved is exactly the failure it must not have."""
    st = aliased(VideoState)
    row = db.execute(
        select(*feed_columns(user, st))
        .join(Channel, Channel.id == Video.channel_id)
        .outerjoin(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
        .where(Video.id == video_id)
    ).first()
    return serialize(row) if row is not None else None


def channels_summary(
    db: Session, user: User, channel_ids: list[str]
) -> dict[str, tuple[int, dict | None]]:
    """For each pinned channel: (unwatched count, its newest unwatched video) — the tile's only honest
    promise, since a channel has no runtime of its own, and what "put on that channel" resolves to.

    ONE query for the whole shelf, not two per channel: the count and the first row per channel are
    window functions over the same filtered set. `library_source="all"`: a pinned channel means the
    whole channel, including videos that only ever arrived through a search — the feed's "organic"
    default is a feed-scoping choice, not a statement about what belongs to a channel."""
    if not channel_ids:
        return {}
    params = dict(_FEED_BASE, channel_ids=list(channel_ids), scope="all", library_source="all")
    query, rank_expr = filtered_query(db, user, **params)
    newest = order_by(sort_keys("newest", 0, rank_expr))
    per_channel = {"partition_by": Video.channel_id}
    sub = query.add_columns(
        func.row_number().over(**per_channel, order_by=newest).label("_rn"),
        func.count().over(**per_channel).label("_total"),
    ).subquery()
    rows = db.execute(select(sub).where(sub.c._rn == 1)).all()
    return {r.channel_id: (int(r._total), serialize(r)) for r in rows}


def playlists_summary(
    db: Session, user: User, playlist_ids: list[int]
) -> dict[int, tuple[int, dict | None]]:
    """For each pinned playlist: (item count, the row to play next) — its first not-yet-watched item,
    or its first item once the whole list is watched. The same row doubles as the tile's artwork, so
    a shelf of playlists is scannable instead of a row of blank boxes. One query for the whole shelf."""
    if not playlist_ids:
        return {}
    st = aliased(VideoState)
    rows = db.execute(
        select(*feed_columns(user, st), PlaylistItem.playlist_id.label("_playlist"))
        .join(PlaylistItem, PlaylistItem.video_id == Video.id)
        .join(Channel, Channel.id == Video.channel_id)
        .outerjoin(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
        .where(PlaylistItem.playlist_id.in_(playlist_ids))
        .order_by(PlaylistItem.playlist_id, PlaylistItem.position.asc())
    ).all()
    grouped: dict[int, list] = {}
    for r in rows:
        grouped.setdefault(r._playlist, []).append(r)
    out: dict[int, tuple[int, dict | None]] = {}
    for pid, items in grouped.items():
        nxt = next((r for r in items if (r.status or "new") not in ("watched", "hidden")), items[0])
        out[pid] = (len(items), serialize(nxt))
    return out


# ── Plex ──────────────────────────────────────────────────────────────────────────────────────────


def _unseen_movie_conds(uid: int):
    """A movie the user has neither finished nor hidden. A LEFT-JOIN-IS-NULL would also drop rows that
    merely carry a resume position, which are exactly the ones Continue watching wants back."""
    seen = (
        select(1)
        .select_from(PlexState)
        .where(
            PlexState.item_id == PlexItem.id,
            owned_by(PlexState.user_id, uid),
            or_(
                PlexState.status.in_(("watched", "hidden")),
                # Started counts as "not a surprise" too — see the note in `yt_surprise`.
                PlexState.position_seconds > 0,
            ),
        )
        .exists()
    )
    return ~seen


def plex_in_progress(
    db: Session, uid: int, limit: int = RAIL_LIMIT, *, lang: str
) -> list[tuple[object, dict]]:
    """In-progress movies + partially-watched shows as (progressed_at, card).

    Mirrors `plex/discover.py::_continue_watching` — same `PlexState` predicate for movies (a
    NULL status still counts as started, so `notin_` alone would drop it), and the same shared
    `rollup.agg_subquery` + `apply_show_state_filter` for shows, so the two rails cannot disagree on
    what "in progress" means. What it does NOT inherit is that route's `LibraryFilters` narrowing:
    this surface has no Plex sidebar to read a filter from."""
    movie_libs, show_libs = content_libs(db)
    out: list[tuple[object, dict]] = []

    if movie_libs:
        mq = (
            select(PlexItem, PlexState)
            .join(
                PlexState, and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid))
            )
            .where(
                PlexItem.kind == "movie",
                PlexItem.library_id.in_(movie_libs),
                PlexState.position_seconds > 0,
                or_(
                    PlexState.status.is_(None),
                    PlexState.status.notin_(("watched", "hidden")),
                ),
            )
            .order_by(PlexState.progress_updated_at.desc().nullslast())
            .limit(limit)
        )
        for it, st in db.execute(mq).all():
            out.append((st.progress_updated_at, movie_card(it, st, lang=lang)))

    if show_libs:
        prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery()
        sq = apply_show_state_filter(
            select(PlexShow, prog.c.resumed_at, prog.c.total, prog.c.watched, prog.c.inprog).join(
                prog, prog.c.show_id == PlexShow.id
            ),
            prog.c.total,
            prog.c.watched,
            prog.c.inprog,
            "in_progress",
        ).where(PlexShow.library_id.in_(show_libs))
        for sh, resumed_at, total, watched, inprog in db.execute(
            sq.order_by(prog.c.resumed_at.desc().nullslast()).limit(limit)
        ).all():
            out.append(
                (resumed_at, show_card(sh, rollup.show_state(total, watched, inprog), lang=lang))
            )

    return out


def plex_surprise(
    db: Session, uid: int, budget: str, seed: int, limit: int, *, lang: str
) -> list[dict]:
    """Unseen playable movies inside the budget, in seeded-shuffle order.

    Movies only. A show is not a thing you can start playing, and resolving every candidate show to
    its next episode just to draw three cards would be a query per row for a rail whose whole value is
    that it answers instantly — shows reach this surface through Continue watching and the comfort
    shelf, which is where "put the next episode on" actually belongs.

    No `playable` filter, matching the Plex surprise rail: `PlexItem.playable` is a playback CLASS
    ("direct" / "remux" / "transcode"), not a yes/no, and NULL there means "not probed yet" rather than
    "will not play". Excluding un-probed items would quietly hide a fresh library from the one surface
    that is supposed to have an answer. The class travels on the card, so the client still shows it."""
    movie_libs, _ = content_libs(db)
    if not movie_libs:
        return []
    lo, hi = band(budget)
    q = (
        select(PlexItem)
        .where(
            PlexItem.kind == "movie",
            PlexItem.library_id.in_(movie_libs),
            _unseen_movie_conds(uid),
        )
        .order_by(_seeded(PlexItem.rating_key, seed))
        .limit(limit)
    )
    if lo is not None:
        q = q.where(PlexItem.duration_s >= lo)
    if hi is not None:
        q = q.where(PlexItem.duration_s <= hi)
    items = db.execute(q).scalars().all()
    states = rollup.item_state_map(db, uid, [it.id for it in items])
    return [movie_card(it, states.get(it.id), lang=lang) for it in items]


# Watch states that take an episode out of the running for "what would play next".
_DONE = ("watched", "hidden")


def next_episode(db: Session, uid: int, show_id: int, *, lang: str) -> dict | None:
    """What "put that show on" means: the episode in progress, else the first unwatched one, else the
    first — over the same `(season, episode)` broadcast ordering `detail.py::_episode_nav` uses.

    Three ordered `LIMIT 1` probes rather than one scan picked apart in Python. The Python version
    loaded EVERY episode of the show to choose one of them — 236 rows for Friends, 227 for JAG — which
    the comfort shelf could just about afford (a handful of hand-picked pins) and the resume rail
    could not, since it resolves a show per rail row on the landing page.

    Returned as a `leaf_card` so it carries the show title the episode row itself does not."""
    asc = (
        PlexItem.season_number.asc().nullsfirst(),
        PlexItem.episode_number.asc().nullsfirst(),
    )
    base = (
        select(PlexItem, PlexState)
        .outerjoin(
            PlexState, and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid))
        )
        .where(PlexItem.show_id == show_id, PlexItem.kind == "episode")
        .limit(1)
    )
    # In progress — the LAST one in broadcast order, which is `plex.rollup.rollup`'s `inprog[-1]`,
    # i.e. what the show page's own Resume button plays. Two episodes of one show are in progress at
    # once more often than it sounds (`watch_sync._reconcile_inprogress_movies` documents that a
    # scrub to the end leaves a stale `status='new', position>0` row until the daily reconcile), and
    # taking opposite ends of that set is how one show came to have two different "continue" answers:
    # the Home tile ranked it by its LATEST touch and then offered the EARLIEST started episode.
    # `position_seconds > 0` already requires the state row to exist, so no null guard.
    pick = db.execute(
        base.where(PlexState.position_seconds > 0, PlexState.status.notin_(_DONE)).order_by(
            PlexItem.season_number.desc().nullslast(),
            PlexItem.episode_number.desc().nullslast(),
        )
    ).first()
    # Unwatched — INCLUDING an episode with no state row at all, which is the common case and the one
    # a bare `status NOT IN (...)` silently drops (NULL is not false, it is unknown, so the row fails
    # the filter).
    pick = (
        pick
        or db.execute(
            base.where(or_(PlexState.id.is_(None), PlexState.status.notin_(_DONE))).order_by(*asc)
        ).first()
    )
    # Fully watched show: start it again from the top.
    pick = pick or db.execute(base.order_by(*asc)).first()
    if pick is None:
        return None

    ep, st = pick
    show = db.get(PlexShow, show_id)
    return leaf_card(
        ep,
        st,
        show_title=display_title(show, lang) if show else None,
        show_rk=show.rating_key if show else None,
        lang=lang,
    )