siftlode/backend/app/routes/channels.py

"""Channel manager: the user's subscriptions with their personal overrides (priority,
hidden, tags) and per-channel sync state. Read + local-write only — YouTube-side writes
(unsubscribe) live behind the optional write scope and are added in a later phase."""
import logging
from datetime import datetime, timezone

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import and_, func, select
from sqlalchemy.orm import Session

from app import quota
from app.auth import current_user, has_write_scope
from app.db import get_db
from app.models import (
    Channel,
    ChannelTag,
    Playlist,
    PlaylistItem,
    Subscription,
    Tag,
    User,
    Video,
)
from app.routes.admin import admin_user
from app.sync.runner import run_recent_backfill
from app.sync.subscriptions import apply_channel_details
from app.youtube.client import YouTubeClient, YouTubeError

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

log = logging.getLogger("subfeed.api")


@router.get("")
def list_channels(
    user: User = Depends(current_user), db: Session = Depends(get_db)
) -> list[dict]:
    rows = db.execute(
        select(Subscription, Channel)
        .join(Channel, Channel.id == Subscription.channel_id)
        .where(Subscription.user_id == user.id)
        .order_by(Subscription.priority.desc(), func.lower(Channel.title))
    ).all()

    # Per-channel aggregates over the shared catalog in ONE grouped pass: stored count, last
    # upload, total duration, and a Normal/Shorts/Live breakdown. Cheap — a single grouped scan
    # (~35 ms over the full 233k-row catalog on the dev DB), so computed on read rather than
    # denormalised (measure-first: reads aren't frequent/hot enough to warrant cached columns).
    channel_ids = [c.id for _, c in rows]
    agg: dict[str, dict] = {}
    if channel_ids:
        for cid, n, last_up, total_dur, shorts, live in db.execute(
            select(
                Video.channel_id,
                func.count(Video.id),
                func.max(Video.published_at),
                func.coalesce(func.sum(Video.duration_seconds), 0),
                func.count(Video.id).filter(Video.is_short.is_(True)),
                func.count(Video.id).filter(Video.live_status.in_(("live", "upcoming"))),
            )
            .where(Video.channel_id.in_(channel_ids))
            .group_by(Video.channel_id)
        ).all():
            agg[cid] = {
                "stored": n,
                "last_video_at": last_up.isoformat() if last_up else None,
                "total_duration_seconds": int(total_dur or 0),
                "count_short": int(shorts),
                "count_live": int(live),
                "count_normal": max(0, int(n) - int(shorts) - int(live)),
            }

    # Channels already in the *global* deep-backfill queue — i.e. at least one user (any
    # user) has requested full history and it isn't done yet. Their whole back-catalogue is
    # coming for everyone (video data is shared), so the UI shouldn't offer "get full
    # history" to a user who simply hasn't personally opted in.
    deep_queued: set[str] = set()
    if channel_ids:
        for (cid,) in db.execute(
            select(Subscription.channel_id)
            .where(
                Subscription.channel_id.in_(channel_ids),
                Subscription.deep_requested.is_(True),
            )
            .distinct()
        ).all():
            deep_queued.add(cid)

    # The user's personal tag links, grouped by channel.
    tags_by_channel: dict[str, list[int]] = {}
    for cid, tag_id in db.execute(
        select(ChannelTag.channel_id, ChannelTag.tag_id).where(
            ChannelTag.user_id == user.id
        )
    ).all():
        tags_by_channel.setdefault(cid, []).append(tag_id)

    return [
        {
            "id": ch.id,
            "title": ch.title,
            "handle": ch.handle,
            "thumbnail_url": ch.thumbnail_url,
            "subscriber_count": ch.subscriber_count,
            "video_count": ch.video_count,
            "stored_videos": (agg.get(ch.id) or {}).get("stored", 0),
            "last_video_at": (agg.get(ch.id) or {}).get("last_video_at"),
            "total_duration_seconds": (agg.get(ch.id) or {}).get("total_duration_seconds", 0),
            "count_normal": (agg.get(ch.id) or {}).get("count_normal", 0),
            "count_short": (agg.get(ch.id) or {}).get("count_short", 0),
            "count_live": (agg.get(ch.id) or {}).get("count_live", 0),
            "priority": sub.priority,
            "hidden": sub.hidden,
            "deep_requested": sub.deep_requested,
            "deep_in_queue": ch.id in deep_queued and not ch.backfill_done,
            "tag_ids": tags_by_channel.get(ch.id, []),
            "details_synced": ch.details_synced_at is not None,
            "recent_synced": ch.recent_synced_at is not None,
            "backfill_done": ch.backfill_done,
        }
        for sub, ch in rows
    ]


# Cap how many stub channels we enrich per discovery call, to bound first-load latency and
# quota (channels.list = 1 unit / 50). Anything beyond is enriched on subsequent loads.
DISCOVERY_ENRICH_CAP = 200


def _discovery_rows(db: Session, user: User) -> list:
    """The user's playlist channels they don't subscribe to (and that aren't their own),
    each with how many playlist videos come from it and across how many playlists."""
    subscribed = select(Subscription.channel_id).where(Subscription.user_id == user.id)
    q = (
        select(
            Channel,
            func.count(func.distinct(PlaylistItem.video_id)),
            func.count(func.distinct(Playlist.id)),
        )
        .select_from(PlaylistItem)
        .join(Playlist, Playlist.id == PlaylistItem.playlist_id)
        .join(Video, Video.id == PlaylistItem.video_id)
        .join(Channel, Channel.id == Video.channel_id)
        .where(Playlist.user_id == user.id, Channel.id.not_in(subscribed))
    )
    if user.yt_channel_id:
        q = q.where(Channel.id != user.yt_channel_id)
    return db.execute(
        q.group_by(Channel.id).order_by(
            func.count(func.distinct(PlaylistItem.video_id)).desc(),
            func.lower(Channel.title),
        )
    ).all()


@router.get("/discovery")
def discover_channels(
    user: User = Depends(current_user), db: Session = Depends(get_db)
) -> list[dict]:
    """Channels that appear in the user's playlists but that they don't subscribe to (their
    own channel excluded). The base list is a local join over the shared catalog. We also
    enrich each channel's metadata up front (title/thumbnail/subscriber count) so the user
    can judge a channel BEFORE subscribing — but we do NOT pull its videos. Most-present
    channels come first; subscribing is a separate action."""
    # Resolve & cache the user's own channel id so it can be excluded — you don't subscribe
    # to yourself. Lazy + cached; needs OAuth, so demo / token-less users just skip it.
    if user.yt_channel_id is None and user.token is not None and not user.is_demo:
        try:
            with quota.attribute(user.id, "discovery"), YouTubeClient(db, user) as yt:
                own = yt.get_my_channel_id()
            if own:
                user.yt_channel_id = own
                db.commit()
        except YouTubeError as exc:
            log.warning("discovery: own-channel resolve failed (user %s): %s", user.id, exc)

    rows = _discovery_rows(db, user)

    # Enrich stub channels' metadata (uses the API key — no write scope needed). Re-read
    # the rows afterwards so the response carries the fresh subscriber counts/thumbnails.
    need = [ch.id for ch, _, _ in rows if ch.details_synced_at is None][:DISCOVERY_ENRICH_CAP]
    if need:
        try:
            with quota.attribute(user.id, "discovery"), YouTubeClient(db, user) as yt:
                apply_channel_details(db, yt.get_channels(need))
            db.commit()
            rows = _discovery_rows(db, user)
        except YouTubeError as exc:
            log.warning("discovery: channel enrich failed (user %s): %s", user.id, exc)
            db.rollback()

    return [
        {
            "id": ch.id,
            "title": ch.title,
            "handle": ch.handle,
            "thumbnail_url": ch.thumbnail_url,
            "subscriber_count": ch.subscriber_count,
            "video_count": ch.video_count,
            "playlist_video_count": int(vid_count),
            "playlist_count": int(pl_count),
            "details_synced": ch.details_synced_at is not None,
        }
        for ch, vid_count, pl_count in rows
    ]


def _user_subscription(db: Session, user: User, channel_id: str) -> Subscription:
    sub = db.execute(
        select(Subscription).where(
            Subscription.user_id == user.id, Subscription.channel_id == channel_id
        )
    ).scalar_one_or_none()
    if sub is None:
        raise HTTPException(status_code=404, detail="Not subscribed to this channel")
    return sub


@router.patch("/{channel_id}")
def update_channel(
    channel_id: str,
    payload: dict,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    sub = _user_subscription(db, user, channel_id)
    channel = db.get(Channel, channel_id)
    deep_turned_on = False
    if "priority" in payload:
        sub.priority = int(payload["priority"])
    if "hidden" in payload:
        sub.hidden = bool(payload["hidden"])
    if "deep_requested" in payload:
        # Opt this channel into (or out of) full-history backfill. The deep scheduler
        # picks the flag up on its next run; turning it off won't undo already-fetched
        # videos, it just stops further deep paging if nobody else still wants it.
        want = bool(payload["deep_requested"])
        deep_turned_on = want and not sub.deep_requested
        sub.deep_requested = want
    db.commit()

    # Opting a channel into full history should give an immediate feed: if we haven't even
    # fetched its recent uploads yet, do that one channel now (cheap) instead of waiting for
    # the scheduler. Deep paging still follows on the scheduler's next run (recent-then-deep).
    if deep_turned_on and channel is not None and channel.recent_synced_at is None:
        with quota.attribute(user.id, "backfill_recent"):
            run_recent_backfill(db, [channel], max_channels=1)

    return {
        "id": channel_id,
        "priority": sub.priority,
        "hidden": sub.hidden,
        "deep_requested": sub.deep_requested,
        "recent_synced": channel.recent_synced_at is not None if channel else None,
    }


@router.post("/{channel_id}/reset-backfill")
def reset_backfill(
    channel_id: str,
    user: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """Admin-only reset: re-fetch this channel from scratch regardless of its current sync
    state (a "reset" trigger). Clears the channel's backfill markers, opts it back into deep
    backfill, and re-runs its recent pull immediately; the deep scheduler re-pages the full
    back-catalog on its next run. Idempotent — videos upsert by id, so nothing duplicates."""
    channel = db.get(Channel, channel_id)
    if channel is None:
        raise HTTPException(status_code=404, detail="Unknown channel")
    sub = _user_subscription(db, user, channel_id)
    channel.backfill_done = False
    channel.backfill_cursor = None
    channel.recent_synced_at = None
    sub.deep_requested = True
    db.commit()
    with quota.attribute(user.id, "backfill_recent"):
        run_recent_backfill(db, [channel], max_channels=1)
    return {"id": channel_id, "reset": True}


@router.post("/{channel_id}/subscribe")
def subscribe(
    channel_id: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Subscribe to a channel discovered from the user's playlists. Gated behind the
    optional write scope (YouTube subscriptions.insert). Creates the local subscription
    with YouTube's returned resource id and enriches the channel's metadata if it's still
    a stub, so the new row renders properly in the manager. It does NOT pull the channel's
    videos — the scheduler picks those up on its next run, like any subscribed channel."""
    if not has_write_scope(user):
        raise HTTPException(
            status_code=403,
            detail="Enable playlist editing in Settings to subscribe on YouTube.",
        )
    channel = db.get(Channel, channel_id)
    if channel is None:
        raise HTTPException(status_code=404, detail="Unknown channel")
    existing = db.execute(
        select(Subscription).where(
            Subscription.user_id == user.id, Subscription.channel_id == channel_id
        )
    ).scalar_one_or_none()
    if existing is not None:
        raise HTTPException(status_code=409, detail="Already subscribed to this channel.")
    try:
        with quota.attribute(user.id, "subscribe"), YouTubeClient(db, user) as yt:
            yt_sub_id = yt.insert_subscription(channel_id)
            # Enrich a stub channel (title/thumbnail/subscriber count) so it shows up
            # properly right away; no video pull here.
            if channel.details_synced_at is None:
                apply_channel_details(db, yt.get_channels([channel_id]))
    except YouTubeError as exc:
        # Already following on YouTube but not recorded here (a desync — e.g. a stale import
        # dropped the local row, so it resurfaced in discovery). That's not a failure: the end
        # state the user wanted already holds, so record it locally and move on. The next
        # subscription resync fills in the resource id. Any other YouTube error is a real
        # problem — surface it with a clear message (422, not a 502 that the client would
        # mistake for a transient "connection lost").
        msg = str(exc).lower()
        if "already exists" in msg or "duplicate" in msg:
            yt_sub_id = ""
        else:
            raise HTTPException(status_code=422, detail=f"YouTube couldn't subscribe you: {exc}")
    db.add(
        Subscription(
            user_id=user.id,
            channel_id=channel_id,
            yt_subscription_id=yt_sub_id or None,
            subscribed_at=datetime.now(timezone.utc),
        )
    )
    db.commit()
    return {"subscribed": channel_id}


@router.delete("/{channel_id}/subscription")
def unsubscribe(
    channel_id: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Unsubscribe from this channel on YouTube and drop the local subscription. Gated
    behind the optional write scope; read-only users get a 403 and should Hide instead."""
    if not has_write_scope(user):
        raise HTTPException(
            status_code=403,
            detail="Enable playlist editing in Settings to unsubscribe on YouTube.",
        )
    sub = _user_subscription(db, user, channel_id)
    if not sub.yt_subscription_id:
        raise HTTPException(
            status_code=400,
            detail="No YouTube subscription id on record — sync subscriptions first.",
        )
    try:
        with quota.attribute(user.id, "unsubscribe"), YouTubeClient(db, user) as yt:
            yt.delete_subscription(sub.yt_subscription_id)
    except YouTubeError as exc:
        # 422 (a real, explainable error → speaking modal), not 502 which the client treats as
        # a transient "connection lost" gateway blip.
        raise HTTPException(status_code=422, detail=f"YouTube couldn't unsubscribe you: {exc}")
    db.delete(sub)
    db.commit()
    return {"unsubscribed": channel_id}


@router.post("/{channel_id}/tags")
def attach_tag(
    channel_id: str,
    payload: dict,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    _user_subscription(db, user, channel_id)
    tag_id = payload.get("tag_id")
    tag = db.get(Tag, tag_id) if tag_id is not None else None
    # Only the user's own tags may be attached as personal links.
    if tag is None or tag.user_id != user.id:
        raise HTTPException(status_code=404, detail="Unknown tag")
    exists = db.execute(
        select(ChannelTag).where(
            ChannelTag.channel_id == channel_id,
            ChannelTag.tag_id == tag_id,
            ChannelTag.user_id == user.id,
        )
    ).scalar_one_or_none()
    if exists is None:
        db.add(
            ChannelTag(
                channel_id=channel_id,
                tag_id=tag_id,
                user_id=user.id,
                source="user",
            )
        )
        db.commit()
    return {"channel_id": channel_id, "tag_id": tag_id}


@router.delete("/{channel_id}/tags/{tag_id}")
def detach_tag(
    channel_id: str,
    tag_id: int,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    row = db.execute(
        select(ChannelTag).where(
            ChannelTag.channel_id == channel_id,
            ChannelTag.tag_id == tag_id,
            ChannelTag.user_id == user.id,
        )
    ).scalar_one_or_none()
    if row is not None:
        db.delete(row)
        db.commit()
    return {"channel_id": channel_id, "tag_id": tag_id}