siftlode/backend/app/routes/feed.py

from datetime import date, datetime, timedelta, timezone

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import Select, and_, 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,
    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 _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",
    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(
        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,
        status_expr.label("status"),
        position_expr.label("position_seconds"),
        _saved_expr(user),
    ).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:
        # Only channels this user is subscribed to (and hasn't hidden).
        query = query.join(
            Subscription,
            and_(
                Subscription.channel_id == Video.channel_id,
                Subscription.user_id == user.id,
                Subscription.hidden.is_(False),
            ),
        )

    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))

    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)
    if q:
        like = f"%{q}%"
        query = query.where(or_(Video.title.ilike(like), 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


# 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",
) -> 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,
    }


SORTS = {
    "newest": Video.published_at.desc().nulls_last(),
    "oldest": Video.published_at.asc().nulls_last(),
    "views": Video.view_count.desc().nulls_last(),
    "views_asc": Video.view_count.asc().nulls_last(),
    "duration_desc": Video.duration_seconds.desc().nulls_last(),
    "duration_asc": Video.duration_seconds.asc().nulls_last(),
    "title": func.lower(Video.title).asc().nulls_last(),
    "title_desc": func.lower(Video.title).desc().nulls_last(),
    "subscribers": Channel.subscriber_count.desc().nulls_last(),
    "subscribers_asc": Channel.subscriber_count.asc().nulls_last(),
    # 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": func.coalesce(Subscription.priority, 0).desc(),
}


@router.get("/feed")
def get_feed(
    params: dict = Depends(_feed_params),
    sort: str = "newest",
    seed: int = 0,
    limit: int = Query(default=60, le=200),
    offset: int = 0,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    query, _status = _filtered_query(db, user, **params)
    if sort in ("priority", "priority_asc"):
        prio = func.coalesce(Subscription.priority, 0)
        query = query.order_by(
            prio.asc() if sort == "priority_asc" else prio.desc(),
            Video.published_at.desc().nulls_last(),
        )
    else:
        order = SORTS.get(sort)
        if order is None and sort == "shuffle":
            order = func.md5(func.concat(Video.id, str(seed)))
        query = query.order_by(order if order is not None else SORTS["newest"])

    rows = db.execute(query.offset(offset).limit(limit + 1)).all()
    has_more = len(rows) > limit
    return {
        "items": [_serialize(r) for r in rows[:limit]],
        "has_more": has_more,
        "offset": offset,
        "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 = _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 = _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, "video_lookup"), YouTubeClient(db, user) as yt:
            items = yt.get_videos([video_id])
    except YouTubeError as exc:
        raise HTTPException(status_code=502, 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")
        ),
    }