"""Feed read-model: Reader's query-building, paging and row serialization behind the feed.

One home for the read-model rather than the router module: the feed routes, the playlist-detail
query and Home share these helpers (the projection, the filter builder, and keyset paging), and the
feed's routes only validate and answer — a page, its counts and its facets are built here. Nothing
here defines a route or reaches for request state; it's all `(db, user, params) -> query`, and a
malformed cursor raises `BadCursor` for the route handler to translate (the tsquery sanitizer lives
in `app.utils`).
"""

import base64
import binascii
import json
from datetime import date, datetime, timedelta, timezone

from sqlalchemy import BigInteger, Select, and_, case, cast, false, func, or_, select
from sqlalchemy.orm import Session, aliased

from app.collation import collated
from app.models import (
    LIVE_OR_UPCOMING,
    BlockedChannel,
    Channel,
    ExploredChannel,
    SearchFind,
    Subscription,
    User,
    Video,
)
from app.reader.models import ChannelTag, Playlist, PlaylistItem, Tag, VideoState
from app.utils import iso, like_escape, to_tsquery_str
from app.userscope import owned_by


def _channel_url(channel_id: str, handle: str | None) -> str:
    if handle and handle.startswith("@"):
        return f"https://www.youtube.com/{handle}"
    return f"https://www.youtube.com/channel/{channel_id}"


def _saved_expr(user: User):
    """A correlated EXISTS labelled `saved`: is this video in the user's Watch later list?
    (Watch later is the built-in playlist that replaced the old per-video 'saved' status.)"""
    return (
        select(1)
        .select_from(PlaylistItem)
        .join(Playlist, Playlist.id == PlaylistItem.playlist_id)
        .where(
            owned_by(Playlist.user_id, user.id),
            Playlist.kind == "watch_later",
            PlaylistItem.video_id == Video.id,
        )
        .exists()
        .label("saved")
    )


def feed_columns(user: User, state) -> list:
    """The shared feed-row projection (joined Channel fields + per-user watch state +
    `saved`), consumed by both the feed query and the playlist-detail query so a new column
    is added in exactly one place. `state` is the caller's aliased VideoState (outer-joined)."""
    return [
        Video.id,
        Video.title,
        Video.channel_id,
        Channel.title.label("channel_title"),
        Channel.thumbnail_url.label("channel_thumbnail"),
        Channel.handle.label("channel_handle"),
        Video.published_at,
        Video.thumbnail_url,
        Video.duration_seconds,
        Video.view_count,
        Video.is_short,
        Video.live_status,
        func.coalesce(state.status, "new").label("status"),
        func.coalesce(state.position_seconds, 0).label("position_seconds"),
        _saved_expr(user),
    ]


def serialize(row) -> dict:
    return {
        "id": row.id,
        "title": row.title,
        "channel_id": row.channel_id,
        "channel_title": row.channel_title,
        "channel_thumbnail": row.channel_thumbnail,
        "channel_url": _channel_url(row.channel_id, row.channel_handle),
        "published_at": iso(row.published_at),
        "thumbnail_url": row.thumbnail_url,
        "duration_seconds": row.duration_seconds,
        "view_count": row.view_count,
        "is_short": row.is_short,
        "live_status": row.live_status,
        "status": row.status or "new",
        "position_seconds": row.position_seconds or 0,
        "saved": bool(getattr(row, "saved", False)),
        "watch_url": f"https://www.youtube.com/watch?v={row.id}",
    }


def _tag_visible(user: User):
    """A ChannelTag row this user may see: a shared system tag (user_id NULL) or their own."""
    return or_(ChannelTag.user_id.is_(None), owned_by(ChannelTag.user_id, user.id))


# The pseudo-category of a selected tag id no Tag row answers (a saved view or the feed state outlives
# a deleted tag). It filters like a category whose tags no channel carries — to nothing, in the SQL and
# in feed_summary's mirror alike — instead of the id being dropped, which showed the WHOLE feed under
# the view's old name. No real category is called this (language/topic/other).
MISSING_TAG_CATEGORY = "missing"


def tags_by_category(db: Session, user: User, tag_ids: list[int]) -> dict[str, list[int]]:
    """Group tag ids by their category (language/topic/other). The single source for the category split
    shared by the feed's tag WHERE (filtered_query) and the facet/count summary (feed_summary), so the
    two can't disagree on which selected tag belongs to which category. An unknown id lands in
    MISSING_TAG_CATEGORY — and so does another user's private tag: resolving it to its real category
    told the caller which foreign ids exist."""
    if not tag_ids:
        return {}  # skip the pointless `WHERE id IN ()` round-trip on the no-tag-filter hot path
    out: dict[str, list[int]] = {}
    visible = or_(Tag.user_id.is_(None), owned_by(Tag.user_id, user.id))
    for tid, cat in db.execute(
        select(Tag.id, Tag.category).where(Tag.id.in_(tag_ids), visible)
    ).all():
        out.setdefault(cat, []).append(tid)
    known = {tid for ids in out.values() for tid in ids}
    if missing := [tid for tid in tag_ids if tid not in known]:
        out[MISSING_TAG_CATEGORY] = missing
    return out


def not_blocked_cond(user: User):
    """Per-user channel blocklist: never show this user videos from a channel they blocked
    (applies to feed, Library, explore, the channel page and the Home resume rail alike).

    A shared expression rather than a clause each surface writes for itself. Blocking does not touch
    `video_states`, so a half-watched video from a newly blocked channel survives with its resume
    position — and the one query that copied only PART of this predicate set is exactly how that
    video came back as the landing page's suggestion after the user had said "never again"."""
    return Video.channel_id.notin_(
        select(BlockedChannel.channel_id).where(owned_by(BlockedChannel.user_id, user.id))
    )


def filtered_query(
    db: Session,
    user: User,
    *,
    tags: list[int],
    tag_modes: dict[str, str],
    untagged: bool = False,
    channel_ids: list[str],
    q: str | None,
    min_duration: int | None,
    max_duration: int | None,
    max_age_days: int | None,
    published_after: date | None,
    published_before: date | None,
    show_normal: bool,
    include_shorts: bool,
    include_live: bool,
    show: str,
    scope: str = "my",
    library_source: str = "organic",
    exclude_tag_category: str | None = None,
) -> tuple[Select, object]:
    """Build the feed query (joins + all WHERE filters), shared by /feed and /feed/count.
    Returns the column-bearing select plus the priority/relevance rank expression for sorting.

    `scope="my"` (default) restricts the feed to the user's own non-hidden subscriptions.
    `scope="all"` shows every video in the shared catalog (any user's ingested channels);
    the subscription is then only LEFT-joined so per-channel priority still resolves for
    channels the user happens to be subscribed to. Per-user watch state stays private in
    either mode via the VideoState outer join."""
    state = aliased(VideoState)
    status_expr = func.coalesce(state.status, "new")

    query = select(*feed_columns(user, state)).join(Channel, Channel.id == Video.channel_id)

    if scope == "all":
        # Whole shared catalog; subscription is optional (only for priority sort).
        query = query.outerjoin(
            Subscription,
            and_(
                Subscription.channel_id == Video.channel_id,
                owned_by(Subscription.user_id, user.id),
            ),
        )
    else:
        # "Mine": the user's own videos = their non-hidden subscriptions OR videos they
        # surfaced via their own live YouTube search. Both are LEFT-joined so the Source
        # filter below can include either or both, and the subscription row stays available
        # for the priority sort.
        query = query.outerjoin(
            Subscription,
            and_(
                Subscription.channel_id == Video.channel_id,
                owned_by(Subscription.user_id, user.id),
                Subscription.hidden.is_(False),
            ),
        ).outerjoin(
            SearchFind,
            and_(SearchFind.video_id == Video.id, owned_by(SearchFind.user_id, user.id)),
        )

    # The state-gated views (watched/hidden/in_progress) only ever return videos that HAVE a
    # video_states row, so an INNER join lets the planner drive from the user's own small video_states
    # set (~100 rows) instead of scanning all videos and outer-joining (PERF-16). status is NOT NULL on
    # a present row (server_default 'new'), so the raw-column filters below match the coalesce form
    # exactly. 'unwatched'/'all' include row-less 'new' videos, so they keep the outer join.
    state_required = show in ("watched", "hidden", "in_progress")
    join_state = query.join if state_required else query.outerjoin
    query = join_state(state, and_(state.video_id == Video.id, state.user_id == user.id))

    # Hide videos the maintenance job has flagged as currently unplayable (deleted/private/
    # abandoned). They're either deleted after a grace period or unhidden if they recover;
    # either way they shouldn't show in any feed view meanwhile.
    query = query.where(Video.unavailable_since.is_(None))

    # Explored videos are PER-EXPLORER private: a via_explore video (ingested when someone
    # opened an un-subscribed channel's page) is visible only to users who explored or
    # subscribe to its channel, so one user's curiosity never leaks into everyone else's
    # catalog. Applied in BOTH scopes; the channel page reaches its ephemeral videos by
    # asking for scope=all + channel_ids (the channel is in this accessible set).
    accessible_explore = (
        select(Subscription.channel_id).where(owned_by(Subscription.user_id, user.id))
    ).union(select(ExploredChannel.channel_id).where(owned_by(ExploredChannel.user_id, user.id)))
    query = query.where(
        or_(
            Video.via_explore.is_(False),
            Video.channel_id.in_(accessible_explore),
        )
    )

    query = query.where(not_blocked_cond(user))

    # Source filter. In the shared Library ("all" scope) it uses the GLOBAL via_search flag
    # (search-discovered by anyone). In "my" scope it's the user's OWN provenance:
    # "organic" = your subscriptions, "search" = videos you found via your own search,
    # "all" = both. Default "organic" keeps the Mine feed = subscriptions only.
    if scope == "all":
        if library_source == "organic":
            query = query.where(Video.via_search.is_(False))
        elif library_source == "search":
            query = query.where(Video.via_search.is_(True))
        # "all" → both organic and search-discovered videos
    else:
        subscribed = Subscription.channel_id.isnot(None)
        searched = SearchFind.video_id.isnot(None)
        if library_source == "search":
            query = query.where(searched)
        elif library_source == "all":
            query = query.where(or_(subscribed, searched))
        else:  # "organic" (default): subscriptions only
            query = query.where(subscribed)

    if channel_ids:
        # OR across the picked channels; every other filter still applies on top.
        query = query.where(Video.channel_id.in_(channel_ids))
    if min_duration is not None:
        query = query.where(Video.duration_seconds >= min_duration)
    if max_duration is not None:
        query = query.where(Video.duration_seconds <= max_duration)
    if max_age_days is not None:
        cutoff = datetime.now(timezone.utc).timestamp() - max_age_days * 86400
        query = query.where(Video.published_at >= datetime.fromtimestamp(cutoff, tz=timezone.utc))
    if published_after is not None:
        start = datetime.combine(published_after, datetime.min.time(), tzinfo=timezone.utc)
        query = query.where(Video.published_at >= start)
    if published_before is not None:
        end = datetime.combine(
            published_before, datetime.min.time(), tzinfo=timezone.utc
        ) + timedelta(days=1)
        query = query.where(Video.published_at < end)
    # Full-text relevance search over the weighted search document (title > creator keywords +
    # the queries that surfaced the video > description; see Video.search_vector). YouTube-like
    # "fuzzy" matching: word-order-independent, multi-word AND, prefix on the word being typed,
    # accent-insensitive (unaccent_simple). The channel name still matches as a substring so
    # channel searches work. `rank_expr` (weight-aware ts_rank) drives the "relevance" sort, over the
    # newest RANK_WINDOW matches only (see there).
    rank_expr = None
    if q:
        # Resolve the channels whose NAME matches up front (there are only a few hundred channels, so
        # this is a tiny standalone scan) and match videos by channel_id. ORing an unindexed
        # `unaccent(channels.title) ILIKE '%q%'` straight into the per-video predicate defeated the
        # search_vector GIN — the planner detoasted every non-matching tsvector to test the `@@`
        # (~2.6s/search on the prod-sized set); channel_id.in_(...) keeps the GIN as a BitmapOr branch
        # (PERF-12). Empty match -> false() so the OR collapses to the text match alone. The query
        # is a literal inside the pattern: `_` in `my_channel` is not a wildcard.
        like = func.unaccent(f"%{like_escape(q)}%")
        matching_channel_ids = (
            db.execute(select(Channel.id).where(func.unaccent(Channel.title).ilike(like)))
            .scalars()
            .all()
        )
        channel_match = (
            Video.channel_id.in_(matching_channel_ids) if matching_channel_ids else false()
        )
        ts_str = to_tsquery_str(q)
        if ts_str:
            tsq = func.to_tsquery("public.unaccent_simple", ts_str)
            query = query.where(or_(Video.search_vector.op("@@")(tsq), channel_match))
            # Scale ts_rank (a float4) to an integer score so the keyset cursor compares it
            # EXACTLY — a raw float4 vs the float8 cursor value mismatches on round-trip and
            # breaks paging (the same page repeats). 1e6 granularity is ample; ties break on
            # published_at then id.
            rank_expr = cast(func.ts_rank(Video.search_vector, tsq) * 1000000, BigInteger)
        else:
            # No usable tokens (e.g. only punctuation): fall back to a plain substring match.
            query = query.where(or_(func.unaccent(Video.title).ilike(like), channel_match))

    if tags:
        # AND across tag categories (e.g. language AND topic narrows), OR within a category — unless that
        # category's mode is "and" (its per-category any/all toggle), which requires a channel to carry
        # ALL its selected tags. ⚠️ feed_summary.passes() reimplements this exact rule in Python for the
        # sidebar count/facets — keep the two in lockstep (equivalence-tested by
        # test_feed_summary_matches_count_plus_facets).
        by_category = tags_by_category(db, user, tags)
        # Facet counting drops the category being counted so its own chips don't zero each
        # other out (standard drill-down faceting: a category's count ignores its own
        # selections but still honours the other categories' filters).
        if exclude_tag_category:
            by_category.pop(exclude_tag_category, None)
        visible = _tag_visible(user)
        for category, ids in by_category.items():
            if tag_modes.get(category, "or") == "and" and len(ids) > 1:
                sub = (
                    select(ChannelTag.channel_id)
                    .where(ChannelTag.tag_id.in_(ids), visible)
                    .group_by(ChannelTag.channel_id)
                    .having(func.count(func.distinct(ChannelTag.tag_id)) == len(set(ids)))
                )
            else:
                sub = select(ChannelTag.channel_id).where(ChannelTag.tag_id.in_(ids), visible)
            query = query.where(Video.channel_id.in_(sub))

    # "No tag" filter (item B): keep only channels the USER has not personally tagged in the channel
    # manager — i.e. channels with no ChannelTag row owned by this user (user_id == me, source "user").
    # System/auto tags (user_id NULL) don't count: this is "channels I haven't tagged yet". It's a plain
    # channel-level narrowing (like channel_ids/date), NOT a tag-category selection, so feed_summary
    # leaves it applied in the base pass and needs no mirror in passes().
    if untagged:
        query = query.where(
            Video.channel_id.notin_(
                select(ChannelTag.channel_id).where(owned_by(ChannelTag.user_id, user.id))
            )
        )

    # Content type: Normal / Shorts / Live·Upcoming as a union of enabled types.
    # Explicit Watched/Saved/Hidden views show every type so nothing goes missing.
    explicit_view = show in ("watched", "hidden")
    if not explicit_view:
        type_clauses = []
        if show_normal:
            type_clauses.append(
                and_(Video.is_short.is_(False), Video.live_status.notin_(LIVE_OR_UPCOMING))
            )
        if include_shorts:
            type_clauses.append(Video.is_short.is_(True))
        if include_live:
            type_clauses.append(Video.live_status.in_(LIVE_OR_UPCOMING))
        query = query.where(or_(*type_clauses) if type_clauses else false())

    if show == "unwatched":
        query = query.where(status_expr.notin_(("watched", "hidden")))
    elif show == "in_progress":
        # Started but not finished: a stored resume position, not yet watched/hidden. INNER-joined
        # (state_required) so the raw state columns are safe and the planner can use video_states.
        query = query.where(
            and_(state.position_seconds > 0, state.status.notin_(("watched", "hidden")))
        )
    elif show == "watched":
        query = query.where(state.status == "watched")
    elif show == "hidden":
        query = query.where(state.status == "hidden")
    else:  # all
        query = query.where(status_expr != "hidden")

    return query, rank_expr


def feed_summary(db: Session, user: User, **params) -> dict:
    """Single-pass replacement for /feed/count + 3x /facets (PERF-13): the sidebar's total video count
    and every tag's distinct-channel facet count, derived in Python from ONE grouped query instead of
    four full passes over the subscribed set on every filter change.

    The one query is the base filter with the TAG predicate removed, grouped by channel: it yields every
    channel that has a matching video, with its video count. From that ~few-hundred-row map we compute:
      - count  = sum of the video counts of the channels that pass the SELECTED tag filter;
      - facets = per tag, the number of distinct channels that carry it AND pass the drill-down filter
                 (the tag's own category dropped, unless it's a conjunctive topic-AND — mirroring the
                 old /facets exclude_tag_category logic exactly).
    Tag matching is channel-based, identical to filtered_query's tag block (AND across categories, OR
    within, a category in "and" mode requires ALL its selected tags)."""
    tag_modes = params.get("tag_modes", {})

    # ONE pass: per-channel video counts for the base filter, the positive tag filter stripped so every
    # candidate channel is present to facet over. The "No tag" (untagged) filter is a plain channel-level
    # narrowing, so it STAYS applied here (not stripped) and the facets naturally drill down within it.
    base, _rank = filtered_query(db, user, **{**params, "tags": []})
    ch_col = base.with_only_columns(Video.channel_id).subquery()
    channel_counts: dict[str, int] = dict(
        db.execute(select(ch_col.c.channel_id, func.count()).group_by(ch_col.c.channel_id)).all()
    )
    if not channel_counts:
        return {"count": 0, "facets": {}}

    # channel -> its visible tag ids, plus each tag's category (for grouping the facet output).
    visible = _tag_visible(user)
    channel_tags: dict[str, set[int]] = {}
    tag_category: dict[int, str] = {}
    for cid, tid, cat in db.execute(
        select(ChannelTag.channel_id, ChannelTag.tag_id, Tag.category)
        .join(Tag, Tag.id == ChannelTag.tag_id)
        .where(visible, ChannelTag.channel_id.in_(channel_counts.keys()))
    ).all():
        channel_tags.setdefault(cid, set()).add(tid)
        tag_category[tid] = cat

    # Selected tags grouped by category (read from Tag via the shared helper, so a selected tag that no
    # base channel carries still filters everything out for its category — matching the SQL `IN (empty)`
    # behaviour).
    selected_by_cat: dict[str, set[int]] = {
        cat: set(ids) for cat, ids in tags_by_category(db, user, params.get("tags") or []).items()
    }

    def passes(ch_tags: set[int], exclude_cat: str | None) -> bool:
        # The Python mirror of filtered_query's SQL tag block (AND across categories, OR within,
        # topic-AND requires all selected) — keep in lockstep; see the ⚠️ note there. (The "No tag"
        # filter is applied in the base query above, so it needs no mirror here.)
        for cat, sel in selected_by_cat.items():
            if cat == exclude_cat:
                continue
            if tag_modes.get(cat, "or") == "and" and len(sel) > 1:
                if (
                    not sel <= ch_tags
                ):  # conjunctive: channel must carry ALL this category's selected tags
                    return False
            elif not (sel & ch_tags):  # disjunctive: channel must carry ANY selected tag
                return False
        return True

    count = sum(
        n for cid, n in channel_counts.items() if passes(channel_tags.get(cid, set()), None)
    )

    facets: dict[int, int] = {}
    for category in ("language", "topic", "other"):
        # Drop this category's own selections so its chips keep independent counts; a conjunctive
        # topic-AND keeps them (each remaining chip narrows to channels that ALSO have the selected).
        exclude = None if tag_modes.get(category, "or") == "and" else category
        for ch_tags in channel_tags.values():
            if not passes(ch_tags, exclude):
                continue
            for tid in ch_tags:
                if tag_category.get(tid) == category:
                    facets[tid] = facets.get(tid, 0) + 1
    return {"count": count, "facets": facets}


def feed_years(db: Session, user: User, **params) -> list[dict]:
    """Per-year video counts for the current filter context: powers the year control's
    quick chips + the "more…" per-year-counts dialog. Every OTHER filter (tags, channels, content type,
    search, watch state, scope) still applies, but the publish-date narrowing itself is DROPPED —
    published_after/before and max_age_days — so the year list is the full set of years available and
    switching years stays a live drill-down (the same principle as the tag facets dropping their own
    category). One grouped query, newest year first."""
    base, _rank = filtered_query(
        db,
        user,
        **{**params, "published_after": None, "published_before": None, "max_age_days": None},
    )
    pub = base.with_only_columns(Video.published_at.label("pub")).subquery()
    # Bucket by UTC year to stay in lockstep with the year filter's boundaries (published_after/before
    # are pinned to UTC in filtered_query). published_at is timestamptz, so a bare extract('year', …)
    # would use the DB session timezone and, on a non-UTC session, bucket a video near a year boundary
    # into a different year than the filter includes — the count and the drill-down feed would disagree.
    year = func.extract("year", func.timezone("UTC", pub.c.pub))
    rows = db.execute(
        select(year.label("y"), func.count()).group_by(year).order_by(year.desc())
    ).all()
    return [{"year": int(y), "count": n} for y, n in rows if y is not None]


# --- Keyset (cursor) pagination ---------------------------------------------
#
# Each sort is described as an ordered list of key columns (expression + direction
# + whether it can be NULL + a JSON round-trip kind), always with Video.id as the
# final, unique tiebreaker. This drives BOTH the ORDER BY and the keyset WHERE so
# they can never drift apart — the prerequisite for correct cursor paging. Keyset
# replaces OFFSET/LIMIT: deep scrolling no longer scans-and-discards skipped rows,
# and the page boundary is stable even as the scheduler ingests new videos mid-scroll
# (OFFSET would shift and duplicate/skip rows).
def sort_keys(sort: str, seed: int, rank_expr=None, lang: str | None = None) -> list[dict]:
    """`lang` orders the title sorts in the reader's alphabet (`collation`); the other keys ignore it."""
    pub = {"expr": Video.published_at, "asc": False, "nullable": False, "kind": "dt"}
    title = collated(Video.title, lang)
    table: dict[str, list[dict]] = {
        "newest": [pub],
        "oldest": [{**pub, "asc": True}],
        "views": [{"expr": Video.view_count, "asc": False, "nullable": True, "kind": "num"}],
        "views_asc": [{"expr": Video.view_count, "asc": True, "nullable": True, "kind": "num"}],
        "duration_desc": [
            {"expr": Video.duration_seconds, "asc": False, "nullable": True, "kind": "num"}
        ],
        "duration_asc": [
            {"expr": Video.duration_seconds, "asc": True, "nullable": True, "kind": "num"}
        ],
        "title": [{"expr": title, "asc": True, "nullable": True, "kind": "str"}],
        "title_desc": [{"expr": title, "asc": False, "nullable": True, "kind": "str"}],
        "subscribers": [
            {"expr": Channel.subscriber_count, "asc": False, "nullable": True, "kind": "num"}
        ],
        "subscribers_asc": [
            {"expr": Channel.subscriber_count, "asc": True, "nullable": True, "kind": "num"}
        ],
        # Your per-channel priority (set in the channel manager), newest first within a tier.
        # coalesce keeps it null-safe in "all" scope where unsubscribed channels have no row.
        "priority": [
            {
                "expr": func.coalesce(Subscription.priority, 0),
                "asc": False,
                "nullable": False,
                "kind": "num",
            },
            pub,
        ],
        "priority_asc": [
            {
                "expr": func.coalesce(Subscription.priority, 0),
                "asc": True,
                "nullable": False,
                "kind": "num",
            },
            pub,
        ],
        # Deterministic per-seed shuffle: the same seed yields the same total order, so
        # the cursor stays valid across pages of one shuffled run.
        "shuffle": [
            {
                "expr": func.md5(func.concat(Video.id, str(seed))),
                "asc": True,
                "nullable": False,
                "kind": "str",
            }
        ],
    }
    # Relevance (search-result ranking) only exists when there's a query to rank against;
    # without `rank_expr` it falls through to newest. Ties break by newest, then id.
    if rank_expr is not None:
        table["relevance"] = [
            {"expr": rank_expr, "asc": False, "nullable": False, "kind": "num"},
            pub,
        ]
    return table.get(sort) or table["newest"]


# Relevance ranks only the newest RANK_WINDOW matches: ts_rank reads every match's tsvector, and
# a broad word matches most of the library ("a" ranked 193k rows in 6.8 s on prod, against 1.1 s to just
# find them). The rows past the window keep rank 0, so they follow the ranked ones newest first; a word
# that broad has no meaningful relevance order anyway, and a specific one fits the window whole.
RANK_WINDOW = 2000


def rank_window_start(db: Session, query: Select) -> datetime | None:
    """The published_at the ranked window starts at: the RANK_WINDOW-th newest match, None when fewer
    match (rank them all). The newest-first walk stops after RANK_WINDOW matches (40 ms for "a")."""
    return db.scalar(
        query.with_only_columns(Video.published_at)
        .order_by(Video.published_at.desc())
        .offset(RANK_WINDOW - 1)
        .limit(1)
    )


def windowed_rank(rank_expr, since: datetime | None):
    """`rank_expr` inside the window, 0 past it. CASE evaluates ts_rank only for the rows inside."""
    if since is None:
        return rank_expr
    return case((Video.published_at >= since, rank_expr), else_=0)


def windowed_relevance_rows(
    db: Session, query: Select, keys: list[dict], window: datetime, cursor_id: str | None, n: int
) -> list:
    """Up to `n` rows of the windowed relevance order, read as its two halves: the window ranked, then
    the rest newest first (every window row sorts before every row past it). As one ORDER BY the
    planner computes the CASE for every match and sorts them all (2.7 s for "a" on prod); split, each
    half rides an index (80 ms + 5 ms). With a cursor, `keys` carry its values (keyset_predicate)."""
    pub_key = keys[1]  # relevance: [rank, published_at]
    rows: list = []
    if cursor_id is None or pub_key["value"] >= window:
        inside = query.where(Video.published_at >= window).order_by(*order_by(keys))
        if cursor_id is not None:
            inside = inside.where(keyset_predicate(keys, cursor_id))
        rows = list(db.execute(inside.limit(n)).all())
    if len(rows) < n:
        newest = [pub_key]  # past the window every rank is 0: the order is newest first
        rest = query.where(Video.published_at < window).order_by(*order_by(newest))
        if cursor_id is not None:
            rest = rest.where(keyset_predicate(newest, cursor_id))
        rows += db.execute(rest.limit(n - len(rows))).all()
    return rows


def order_by(keys: list[dict]):
    cols = []
    for k in keys:
        c = k["expr"].asc() if k["asc"] else k["expr"].desc()
        cols.append(c.nulls_last() if k["nullable"] else c)
    cols.append(Video.id.asc())  # unique final tiebreaker
    return cols


def keyset_predicate(keys: list[dict], cursor_id: str):
    """WHERE clause selecting rows strictly AFTER the cursor in the sort order.

    Lexicographic over the key columns then id, NULL-aware for `nulls_last`: a NULL
    sorts after every non-NULL value, so a cursor sitting on a non-NULL value must
    also pull in the NULL group, and equality treats NULL == NULL.

    Built recursively so that for a NON-nullable leading key we can prepend a non-strict
    range bound (`col <= v` for desc / `col >= v` for asc) as an AND conjunct — logically
    a no-op over the OR of strict-after branches, but it lets the planner bound the index
    scan instead of scanning from the top and filtering every earlier row (PERF-15: deep
    `newest` pages went 51.8k -> ~0.5k buffers). A NULLABLE key can't take that bound (the
    NULL group sorts after every value and `col <= v` would drop it), so it keeps the pure
    NULL-aware OR form."""

    def eq(k):
        return k["expr"].is_(None) if k["value"] is None else k["expr"] == k["value"]

    def after(k):
        if k["value"] is None:
            return false()  # nulls are last; nothing sorts strictly after them
        cmp = k["expr"] > k["value"] if k["asc"] else k["expr"] < k["value"]
        return or_(cmp, k["expr"].is_(None)) if k["nullable"] else cmp

    def rec(i: int):
        if i == len(keys):
            return Video.id > cursor_id  # all keys tied -> break by the unique id
        k = keys[i]
        tail = or_(after(k), and_(eq(k), rec(i + 1)))
        if not k["nullable"] and k["value"] is not None:
            bound = k["expr"] >= k["value"] if k["asc"] else k["expr"] <= k["value"]
            return and_(bound, tail)
        return tail

    return rec(0)


def encode_cursor(keys: list[dict], row, window: datetime | None = None) -> str:
    """`window` is the relevance window's start: the cursor carries it, so every page ranks against the
    window page one did. Recomputed per page, a video ingested mid-scroll would push the window's oldest
    row to rank 0, behind the cursor, and the reader would see it twice."""
    vals = []
    for i, k in enumerate(keys):
        v = getattr(row, f"_k{i}")
        vals.append(v.isoformat() if (v is not None and k["kind"] == "dt") else v)
    body: dict = {"k": vals, "id": row.id}
    if window is not None:
        body["w"] = window.isoformat()
    payload = json.dumps(body)
    return base64.urlsafe_b64encode(payload.encode()).decode()


def _cursor_dt(v):
    """A cursor's timestamp: aware or refused. The server writes only aware ones (timestamptz), and
    windowed_relevance_rows compares them in Python, where a naive one beside an aware one raises — a
    500 for a hand-made cursor instead of the 400."""
    if v is None:
        return None
    dt = datetime.fromisoformat(v)
    if dt.tzinfo is None:
        raise ValueError("naive cursor timestamp")
    return dt


def decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str, datetime | None]:
    try:
        payload = json.loads(base64.urlsafe_b64decode(cursor.encode()))
        raw = payload["k"]
        if len(raw) != len(keys):
            raise ValueError("key arity mismatch")
        vals = [_cursor_dt(v) if k["kind"] == "dt" else v for k, v in zip(keys, raw, strict=True)]
        for k, v in zip(keys, vals, strict=True):
            if v is None and not k["nullable"]:
                raise ValueError("null value for a non-nullable key")
        window = payload.get("w")
        return vals, payload["id"], _cursor_dt(window) if window else None
    except (binascii.Error, json.JSONDecodeError, KeyError, ValueError, TypeError) as exc:
        # Framework-free: the route handler translates this into an HTTP 400 so the
        # read-model stays web-layer-agnostic (see routes/feed.get_feed).
        raise ValueError("Invalid feed cursor") from exc


class BadCursor(ValueError):
    """A `cursor` this module did not write (or one from another sort): the caller's bad input."""


def feed_page(
    db: Session,
    user: User,
    params: dict,
    *,
    sort: str,
    seed: int,
    limit: int,
    cursor: str | None,
    lang: str,
) -> dict:
    """One page of the feed, and the cursor for the next. A malformed cursor raises `BadCursor`."""
    query, rank_expr = filtered_query(db, user, **params)
    keys = sort_keys(sort, seed, rank_expr, lang)

    vals, cursor_id, window = None, None, None
    if cursor:
        try:
            vals, cursor_id, window = decode_cursor(keys, cursor)
        except ValueError as e:
            raise BadCursor(str(e)) from e
    ranked = sort == "relevance" and rank_expr is not None
    if ranked and not cursor:
        window = rank_window_start(db, query)
    if ranked and window is not None:
        keys = sort_keys(sort, seed, windowed_rank(rank_expr, window), lang)

    # Expose each key column on the row so we can build the next cursor from the last item.
    query = query.add_columns(*[k["expr"].label(f"_k{i}") for i, k in enumerate(keys)])
    if vals is not None:
        for k, v in zip(keys, vals, strict=True):
            k["value"] = v
    if ranked and window is not None:
        rows = windowed_relevance_rows(db, query, keys, window, cursor_id, limit + 1)
    else:
        query = query.order_by(*order_by(keys))
        if cursor_id is not None:
            query = query.where(keyset_predicate(keys, cursor_id))
        rows = db.execute(query.limit(limit + 1)).all()
    has_more = len(rows) > limit
    page = rows[:limit]
    return {
        "items": [serialize(r) for r in page],
        "next_cursor": (
            encode_cursor(keys, page[-1], window if ranked else None)
            if (has_more and page)
            else None
        ),
        "has_more": has_more,
        "limit": limit,
    }


def feed_count(db: Session, user: User, **params) -> int:
    """How many videos the filters match — the reference oracle `feed_summary`'s count is tested
    against."""
    query, _rank = filtered_query(db, user, **params)
    return db.scalar(select(func.count()).select_from(query.subquery())) or 0


def facet_counts(db: Session, user: User, **params) -> dict[int, int]:
    """Per-tag channel counts for the current filter context, one pass per category — the reference
    oracle `feed_summary`'s facets are tested against.

    Each category is counted with every other filter applied but its own selections ignored
    (drill-down faceting: selecting one topic doesn't zero out the rest). A conjunctive (AND)
    category keeps them applied, so each remaining chip narrows to channels that ALSO have every
    selected tag, and tags that can't co-occur drop to zero."""
    visible = _tag_visible(user)
    counts: dict[int, int] = {}
    for category in ("language", "topic", "other"):
        conjunctive = params.get("tag_modes", {}).get(category) == "and"
        base, _rank = filtered_query(
            db,
            user,
            **{**params, "exclude_tag_category": None if conjunctive else category},
        )
        channels = base.with_only_columns(Video.channel_id).distinct().subquery()
        rows = db.execute(
            select(ChannelTag.tag_id, func.count(func.distinct(ChannelTag.channel_id)))
            .select_from(channels)
            .join(ChannelTag, ChannelTag.channel_id == channels.c.channel_id)
            .join(Tag, and_(Tag.id == ChannelTag.tag_id, Tag.category == category))
            .where(visible)
            .group_by(ChannelTag.tag_id)
        ).all()
        for tag_id, count in rows:
            counts[tag_id] = count
    return counts