siftlode/backend/app/routes/feed.py

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

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import BigInteger, Select, and_, cast, false, func, or_, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session, aliased
from sqlalchemy.orm.exc import StaleDataError

from app import quota
from app.auth import current_user
from app.db import get_db
from app.models import (
    Channel,
    ChannelTag,
    Playlist,
    PlaylistItem,
    SearchFind,
    Subscription,
    Tag,
    User,
    Video,
    VideoState,
)
from app.sync.videos import parse_iso8601_duration
from app.youtube.client import YouTubeClient, YouTubeError

router = APIRouter(prefix="/api", tags=["feed"])

# "saved" used to be a status; it's now membership in the built-in Watch later playlist.
VALID_STATES = {"new", "watched", "hidden"}
HIDDEN_LIVE = ("live", "upcoming")

# Resume-position thresholds (mirror the client): positions below this are "didn't
# really start" and within this of the end are "basically finished" — neither is worth
# storing, so we clear the position instead (the video shows no progress bar).
PROGRESS_MIN_SECONDS = 5
FINISH_MARGIN_SECONDS = 10


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(
            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": row.published_at.isoformat() if row.published_at else None,
        "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 _filtered_query(
    db: Session,
    user: User,
    *,
    tags: list[int],
    tag_mode: str,
    channel_id: str | None,
    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 watch-status 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")
    position_expr = func.coalesce(state.position_seconds, 0)

    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,
                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,
                Subscription.user_id == user.id,
                Subscription.hidden.is_(False),
            ),
        ).outerjoin(
            SearchFind,
            and_(SearchFind.video_id == Video.id, SearchFind.user_id == user.id),
        )

    query = query.outerjoin(
        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))

    # 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_id:
        query = query.where(Video.channel_id == channel_id)
    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.
    rank_expr = None
    if q:
        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),
                    func.unaccent(Channel.title).ilike(func.unaccent(f"%{q}%")),
                )
            )
            # 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.
            like = func.unaccent(f"%{q}%")
            query = query.where(
                or_(
                    func.unaccent(Video.title).ilike(like),
                    func.unaccent(Channel.title).ilike(like),
                )
            )

    if tags:
        # AND across tag categories (e.g. language AND topic narrows), OR within a
        # category; the any/all toggle controls multiple topic tags.
        cat_rows = db.execute(select(Tag.id, Tag.category).where(Tag.id.in_(tags))).all()
        by_category: dict[str, list[int]] = {}
        for tag_id, category in cat_rows:
            by_category.setdefault(category, []).append(tag_id)
        # 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 = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
        for category, ids in by_category.items():
            if category == "topic" and tag_mode == "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))

    # 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_(HIDDEN_LIVE))
            )
        if include_shorts:
            type_clauses.append(Video.is_short.is_(True))
        if include_live:
            type_clauses.append(Video.live_status.in_(HIDDEN_LIVE))
        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.
        query = query.where(
            and_(position_expr > 0, status_expr.notin_(("watched", "hidden")))
        )
    elif show == "watched":
        query = query.where(status_expr == "watched")
    elif show == "hidden":
        query = query.where(status_expr == "hidden")
    else:  # all
        query = query.where(status_expr != "hidden")

    return query, status_expr, rank_expr


def _to_tsquery_str(q: str) -> str | None:
    """Turn free-text into a to_tsquery() string: word runs ANDed together, the LAST word
    prefix-matched (`:*`) for as-you-type search — e.g. "ti amo magyar" → "ti & amo & magyar:*".
    Only `\\w` runs containing an alphanumeric are kept, so no tsquery operators can leak in
    (the string is built from sanitised tokens, never the raw query)."""
    tokens = [t for t in re.findall(r"\w+", q, flags=re.UNICODE) if any(c.isalnum() for c in t)]
    if not tokens:
        return None
    tokens[-1] = tokens[-1] + ":*"
    return " & ".join(tokens)


# Shared query parameters for /feed and /feed/count.
def _feed_params(
    tags: list[int] = Query(default=[]),
    tag_mode: str = "or",
    channel_id: str | None = None,
    q: str | None = None,
    min_duration: int | None = None,
    max_duration: int | None = None,
    max_age_days: int | None = None,
    published_after: date | None = None,
    published_before: date | None = None,
    show_normal: bool = True,
    include_shorts: bool = False,
    include_live: bool = False,
    show: str = "unwatched",
    scope: str = "my",
    library_source: str = "organic",
) -> dict:
    return {
        "tags": tags,
        "tag_mode": tag_mode,
        "channel_id": channel_id,
        "q": q,
        "min_duration": min_duration,
        "max_duration": max_duration,
        "max_age_days": max_age_days,
        "published_after": published_after,
        "published_before": published_before,
        "show_normal": show_normal,
        "include_shorts": include_shorts,
        "include_live": include_live,
        "show": show,
        "scope": scope,
        "library_source": library_source,
    }


# --- 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) -> list[dict]:
    pub = {"expr": Video.published_at, "asc": False, "nullable": True, "kind": "dt"}
    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": func.lower(Video.title), "asc": True, "nullable": True, "kind": "str"}],
        "title_desc": [{"expr": func.lower(Video.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"]


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."""
    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

    branches = []
    for j in range(len(keys)):
        branches.append(and_(*[eq(keys[i]) for i in range(j)], after(keys[j])))
    branches.append(and_(*[eq(k) for k in keys], Video.id > cursor_id))
    return or_(*branches)


def _encode_cursor(keys: list[dict], row) -> str:
    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)
    payload = json.dumps({"k": vals, "id": row.id})
    return base64.urlsafe_b64encode(payload.encode()).decode()


def _decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]:
    try:
        payload = json.loads(base64.urlsafe_b64decode(cursor.encode()))
        raw = payload["k"]
        if len(raw) != len(keys):
            raise ValueError("key arity mismatch")
        vals = [
            datetime.fromisoformat(v) if (v is not None and k["kind"] == "dt") else v
            for k, v in zip(keys, raw)
        ]
        return vals, payload["id"]
    except (binascii.Error, json.JSONDecodeError, KeyError, ValueError, TypeError):
        raise HTTPException(status_code=400, detail="Invalid feed cursor")


@router.get("/feed")
def get_feed(
    params: dict = Depends(_feed_params),
    sort: str = "newest",
    seed: int = 0,
    limit: int = Query(default=60, le=200),
    cursor: str | None = None,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    query, _status, rank_expr = _filtered_query(db, user, **params)
    keys = _sort_keys(sort, seed, rank_expr)

    # 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)])
    query = query.order_by(*_order_by(keys))

    if cursor:
        vals, cursor_id = _decode_cursor(keys, cursor)
        for k, v in zip(keys, vals):
            k["value"] = v
        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]) if (has_more and page) else None,
        "has_more": has_more,
        "limit": limit,
    }


@router.get("/feed/count")
def get_feed_count(
    params: dict = Depends(_feed_params),
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    query, _status, _rank = _filtered_query(db, user, **params)
    total = db.scalar(select(func.count()).select_from(query.subquery()))
    return {"count": total or 0}


@router.get("/facets")
def get_facets(
    params: dict = Depends(_feed_params),
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Per-tag channel counts for the *current* filter context, so the sidebar can show
    live counts and drop chips that no longer match anything. Each category is counted with
    every other filter applied (scope, channel, date, content type, search, watch state,
    and the other category's tags) but its own selections ignored — standard drill-down
    faceting, so selecting one topic doesn't zero out the rest of the topics.

    The count is the number of distinct channels that have at least one video in the current
    view, matching the existing channel-count chip semantics. JSON object keys are strings
    (tag ids)."""
    visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
    counts: dict[int, int] = {}
    for category in ("language", "topic"):
        # Disjunctive (OR) facets drop the category's own selections so its chips keep
        # independent counts (you can OR more of them in). Conjunctive (AND, topics only)
        # keeps them applied, so each remaining chip narrows to channels that ALSO have all
        # already-selected topics — and tags that can't co-occur drop to zero (hidden).
        conjunctive = category == "topic" and params.get("tag_mode") == "and"
        base, _status, _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": counts}


@router.post("/videos/{video_id}/state")
def set_video_state(
    video_id: str,
    payload: dict,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    status = payload.get("status")
    if status not in VALID_STATES:
        raise HTTPException(status_code=400, detail=f"status must be one of {VALID_STATES}")
    if db.get(Video, video_id) is None:
        raise HTTPException(status_code=404, detail="Unknown video")

    now = datetime.now(timezone.utc)

    if status == "new":
        # Un-marking: keep the row if it still holds a resume position (un-marking
        # "watched" should restore the in-progress state, not wipe where the user left
        # off); otherwise drop it. The in-app player fires this alongside a progress
        # checkpoint that may DELETE the same row, so tolerate it vanishing under us.
        row = db.execute(
            select(VideoState).where(
                VideoState.user_id == user.id, VideoState.video_id == video_id
            )
        ).scalar_one_or_none()
        if row is not None:
            if row.position_seconds:
                row.status = "new"
                row.watched_at = None
            else:
                db.delete(row)
            try:
                db.commit()
            except StaleDataError:
                db.rollback()
        return {"video_id": video_id, "status": "new"}

    # watched / hidden: atomic upsert. The player marks "watched" at end-of-playback at
    # the same instant a progress checkpoint may DELETE the in-progress row; a plain
    # SELECT-then-UPDATE then raises StaleDataError ("0 rows matched") on that race. An
    # INSERT … ON CONFLICT DO UPDATE keyed on uq_user_video is immune to it.
    set_: dict = {"status": status, "updated_at": now}
    if status == "watched":
        set_["watched_at"] = now  # hidden keeps any existing watched_at
    stmt = (
        pg_insert(VideoState)
        .values(
            user_id=user.id,
            video_id=video_id,
            status=status,
            watched_at=now if status == "watched" else None,
        )
        .on_conflict_do_update(constraint="uq_user_video", set_=set_)
    )
    db.execute(stmt)
    db.commit()
    return {"video_id": video_id, "status": status}


@router.delete("/videos/{video_id}/state")
def clear_video_state(
    video_id: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Reset a video to pristine for this user — drop the whole VideoState row (status,
    watch position and watched_at), as if it had never been opened. Saved/playlist membership
    lives elsewhere and is untouched."""
    row = db.execute(
        select(VideoState).where(
            VideoState.user_id == user.id, VideoState.video_id == video_id
        )
    ).scalar_one_or_none()
    if row is not None:
        db.delete(row)
        try:
            db.commit()
        except StaleDataError:
            db.rollback()
    return {"video_id": video_id, "status": "new", "position_seconds": 0}


@router.post("/videos/{video_id}/progress")
def set_video_progress(
    video_id: str,
    payload: dict,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Checkpoint the in-app player's resume position (called periodically while playing).
    Stores a per-user position on the video_states row without touching watch status, so a
    partially-watched video can render a progress bar and match the "in progress" filter.
    Trivially-early and near-finished positions clear the position rather than store it."""
    try:
        position = int(payload.get("position_seconds") or 0)
        duration = int(payload.get("duration_seconds") or 0)
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="position_seconds must be an integer")
    if position < 0:
        position = 0

    if db.get(Video, video_id) is None:
        raise HTTPException(status_code=404, detail="Unknown video")

    # Decide whether this position is worth keeping (else clear it).
    near_end = duration > 0 and position > duration - FINISH_MARGIN_SECONDS
    keep = position >= PROGRESS_MIN_SECONDS and not near_end

    row = db.execute(
        select(VideoState).where(
            VideoState.user_id == user.id, VideoState.video_id == video_id
        )
    ).scalar_one_or_none()

    if not keep:
        # Nothing meaningful to store: drop a status-less row entirely, otherwise just
        # zero the position (a saved/hidden video keeps its status).
        if row is not None:
            if row.status == "new":
                db.delete(row)
            else:
                row.position_seconds = 0
                row.progress_updated_at = datetime.now(timezone.utc)
            # A concurrent state change (e.g. the end-of-playback "watched" mark) may
            # have removed/updated this row already — tolerate the lost race.
            try:
                db.commit()
            except StaleDataError:
                db.rollback()
        return {"video_id": video_id, "position_seconds": 0}

    if row is None:
        row = VideoState(user_id=user.id, video_id=video_id)
        db.add(row)
    row.position_seconds = position
    row.progress_updated_at = datetime.now(timezone.utc)
    db.commit()
    return {"video_id": video_id, "position_seconds": position}


@router.get("/videos/{video_id}")
def get_video_detail(
    video_id: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """On-demand detail (description, like count) — kept out of the feed list payload
    so the feed stays lean; fetched lazily, e.g. for the title hover popover.

    Videos we already store are served from the DB for free. A video that isn't in
    our DB (e.g. a YouTube link inside another video's description that the in-app
    player navigated to) is resolved via the YouTube API (videos.list, 1 unit,
    attributed to the requesting user)."""
    v = db.get(Video, video_id)
    if v is not None:
        return {
            "id": v.id,
            "description": v.description,
            "like_count": v.like_count,
            "in_db": True,
            "channel_id": v.channel_id,
            "channel_title": v.channel.title if v.channel else None,
            "published_at": v.published_at.isoformat() if v.published_at else None,
            "view_count": v.view_count,
            "duration_seconds": v.duration_seconds,
        }

    try:
        with quota.attribute(user.id, quota.QuotaAction.VIDEOS_LOOKUP), YouTubeClient(db, user) as yt:
            items = yt.get_videos([video_id])
    except YouTubeError as exc:
        raise HTTPException(status_code=422, detail=f"YouTube lookup failed: {exc}")
    if not items:
        raise HTTPException(status_code=404, detail="Unknown video")

    snippet = items[0].get("snippet", {})
    stats = items[0].get("statistics", {})
    likes = stats.get("likeCount")
    views = stats.get("viewCount")
    return {
        "id": video_id,
        "description": snippet.get("description"),
        "like_count": int(likes) if likes is not None else None,
        "in_db": False,
        "channel_id": snippet.get("channelId"),
        "channel_title": snippet.get("channelTitle"),
        "published_at": snippet.get("publishedAt"),
        "view_count": int(views) if views is not None else None,
        "duration_seconds": parse_iso8601_duration(
            items[0].get("contentDetails", {}).get("duration")
        ),
    }