siftlode/backend/app/sync/explore.py

"""Channel exploration: ingest an un-subscribed channel's uploads on demand (flagged
via_explore, kept per-explorer) and purge channels nobody ends up keeping.

A user opens a channel's page without subscribing → we pull a page of its recent uploads,
mark the new rows via_explore, and enrich them so they're immediately browsable. The videos
stay private to users who explored (or subscribe to) the channel — the feed query gates them
— so one user's curiosity never floods everyone's catalog. Subscribing is the "keep" signal;
otherwise the cleanup job hard-deletes the channel and its untouched ephemeral videos after a
grace period (mirrors the maintenance job's grace-then-delete safety)."""

from datetime import timedelta

from sqlalchemy import delete, or_, select
from sqlalchemy.orm import Session

from app import logs, sysconfig
from app.models import (
    Channel,
    ExploredChannel,
    PlaylistItem,
    SearchFind,
    Subscription,
    Video,
    VideoState,
)
from app.sync.videos import insert_stubs, stub_from_playlist_item, apply_video_details
from app.utils import now_utc as _now
from app.youtube.client import YouTubeClient

log = logs.get_logger(__name__)


def followed_channel():
    """WHERE clause: a channel somebody follows — a subscription, or an exploration still inside its
    grace window (`purge_unkept_explored` deletes the expired ExploredChannel rows first). ONE
    definition for both halves of the question: what the scheduled RSS poll and recent backfill cover
    (sync/runner.py) and what the explore purge may delete.

    The poll and backfill cover only these (owner decision, 2026-09-17). They used to cover EVERY
    channel row, but rows also come from search results, playlist sync and dropped subscriptions, and
    nothing deletes those: on prod 582 of 979 polled channels had no follower, three of them dead
    feeds answering 404 on every poll. The cost of narrowing is deliberate: the shared Library stops
    receiving new uploads from channels nobody follows."""
    return or_(
        select(Subscription.id).where(Subscription.channel_id == Channel.id).exists(),
        select(ExploredChannel.id).where(ExploredChannel.channel_id == Channel.id).exists(),
    )


def _enrich_new(db: Session, yt: YouTubeClient, video_ids: list[str]) -> int:
    """Enrich just the not-yet-enriched videos among `video_ids` (videos.list, 1 unit/50)."""
    if not video_ids:
        return 0
    videos = (
        db.execute(select(Video).where(Video.id.in_(video_ids), Video.enriched_at.is_(None)))
        .scalars()
        .all()
    )
    if not videos:
        return 0
    items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])}
    now = _now()
    for video in videos:
        item = items.get(video.id)
        if item is None:
            video.enriched_at = now  # unavailable; stop retrying
            continue
        apply_video_details(video, item)
        video.enriched_at = now
    db.commit()
    return len(videos)


def explore_ingest_page(
    db: Session, yt: YouTubeClient, channel: Channel, page_token: str | None = None
) -> dict:
    """Ingest one page (up to 50, newest-first) of the channel's uploads, marking the new
    videos via_explore, then enrich them. Returns the count ingested and the next page token
    (None when the back-catalogue is exhausted) so the caller can offer "load more". Stateless
    pagination: the caller round-trips the token, so the channel's own backfill cursor (used by
    the organic deep backfill) is never touched."""
    if not channel.uploads_playlist_id:
        return {"ingested": 0, "next_page_token": None}
    data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
    stubs: list[dict] = []
    for item in data.get("items", []):
        stub = stub_from_playlist_item(item, channel.id)
        if stub is not None:
            stub["via_explore"] = True
            stubs.append(stub)
    inserted = insert_stubs(db, stubs)
    self_enriched = _enrich_new(db, yt, [s["id"] for s in stubs])
    log.info(
        "explore ingest channel=%s ingested=%s enriched=%s", channel.id, inserted, self_enriched
    )
    return {"ingested": inserted, "next_page_token": data.get("nextPageToken")}


def purge_unkept_explored(db: Session) -> dict:
    """Cleanup job: hard-delete explored channels nobody kept and their untouched ephemeral
    videos, after a grace period (mirrors the maintenance grace-then-delete safety).

    Keep signals that spare a channel/video:
      - any user SUBSCRIBES to the channel (subscribe promotes it to organic — channel-level), or
      - a live ExploredChannel row (someone explored it within the grace window — channel-level), or
      - a video has per-user state: watched/in-progress (VideoState), in a playlist (PlaylistItem),
        or surfaced by the user's own search (SearchFind) — these survive even if the channel goes.
    A channel row is dropped only once it holds no videos at all. 0 quota."""
    grace_days = sysconfig.get_int(db, "explore_grace_days")
    cutoff = _now() - timedelta(days=grace_days)

    expired = db.execute(
        delete(ExploredChannel).where(ExploredChannel.created_at < cutoff)
    ).rowcount
    db.commit()

    orphan_ids = (
        db.execute(select(Channel.id).where(Channel.from_explore.is_(True), ~followed_channel()))
        .scalars()
        .all()
    )
    if not orphan_ids:
        return {"expired": expired, "videos_deleted": 0, "channels_deleted": 0}

    state_exists = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
    pl_exists = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
    find_exists = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists()
    videos_deleted = db.execute(
        delete(Video).where(
            Video.channel_id.in_(orphan_ids),
            Video.via_explore.is_(True),
            ~state_exists,
            ~pl_exists,
            ~find_exists,
        )
    ).rowcount
    db.commit()

    has_videos = select(Video.id).where(Video.channel_id == Channel.id).exists()
    channels_deleted = db.execute(
        delete(Channel).where(Channel.id.in_(orphan_ids), ~has_videos)
    ).rowcount
    db.commit()
    return {
        "expired": expired,
        "videos_deleted": videos_deleted,
        "channels_deleted": channels_deleted,
    }


def purge_unkept_search(db: Session, grace_days: int | None = None) -> dict:
    """Remove live-search result videos nobody kept, after a grace period — "as if never added".

    A `via_search` video is KEPT (and spared) when anyone has watched / in-progress / hidden it
    (a VideoState), saved/playlisted it (a PlaylistItem), or subscribes to its channel (it's then
    organically owned). A bare SearchFind does NOT count — it's the ephemeral "shown in search"
    marker itself — so a result you only glanced at is reclaimed. Deleting the video cascades its
    SearchFinds. `grace_days=0` purges immediately (the admin "purge now" path); None uses config.
    0 quota."""
    grace = sysconfig.get_int(db, "search_grace_days") if grace_days is None else grace_days
    cutoff = _now() - timedelta(days=grace)
    state_exists = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
    pl_exists = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
    sub_exists = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
    deleted = db.execute(
        delete(Video).where(
            Video.via_search.is_(True),
            Video.created_at < cutoff,
            ~state_exists,
            ~pl_exists,
            ~sub_exists,
        )
    ).rowcount
    db.commit()
    return {"search_videos_deleted": deleted}


def purge_ephemeral(db: Session) -> dict:
    """The scheduler's discovery-cleanup pass: reclaim un-kept explored channels AND un-kept
    live-search result videos in one run."""
    return {**purge_unkept_explored(db), **purge_unkept_search(db)}