siftlode/backend/app/sync/maintenance.py

"""Maintenance / validation job: detect videos that can't be played anywhere and retire
them safely.

Lifecycle (no permanent soft-delete state — just a grace timer before a hard delete):

  1. Re-check flagged videos (those with unavailable_since set). If a video has recovered
     (e.g. a private video made public again), clear the flag. Otherwise, if its grace
     period has elapsed, HARD DELETE it (cascades to video_states / playlist_items). Before
     deleting, gather which users had it saved or in a playlist so we can tell them.
  2. Rolling re-validation: re-fetch the least-recently-checked N currently-available
     videos. Newly-unplayable ones get flagged (hidden from the feed immediately via
     unavailable_since); the grace timer starts now and they're deleted in a later run if
     they don't recover.

Detection is ~free: a video missing from the videos.list response is deleted-or-private.
We never delete a live broadcast that still reports viewers (a legit 24/7 stream); we only
flag an `upcoming` premiere that blew past its scheduled start and never went live.

Quota: phase 1 costs ~(flagged / 50) units, phase 2 ~(batch / 50). Both respect the same
backfill reserve as the other jobs so interactive syncs never starve.
"""
import logging
from datetime import timedelta

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import progress, quota, sysconfig
from app.config import settings
from app.models import Playlist, PlaylistItem, Video
from app.notifications import create_notification
from app.state import get_maintenance_batch
from app.sync.runner import get_service_user
from app.sync.videos import apply_video_details, parse_dt
from app.utils import now_utc as _now
from app.youtube.client import YouTubeClient

log = logging.getLogger("siftlode.maintenance")

# Reasons recorded on Video.unavailable_reason. "removed" = absent from videos.list
# (deleted or made private); "abandoned" = an upcoming premiere that never went live.
GRACE_DAYS_BY_REASON = {
    "removed": settings.maintenance_grace_days,
    "abandoned": 0,  # already past-due when detected; deleted on the confirming sweep
}
DEFAULT_GRACE_DAYS = settings.maintenance_grace_days


def _grace_days(reason: str | None) -> int:
    return GRACE_DAYS_BY_REASON.get(reason or "", DEFAULT_GRACE_DAYS)


def _is_abandoned_upcoming(item: dict) -> bool:
    """True for an `upcoming` broadcast whose scheduled start is >2 days past and which
    never actually went live (no actualStartTime)."""
    snippet = item.get("snippet", {})
    if snippet.get("liveBroadcastContent") != "upcoming":
        return False
    live = item.get("liveStreamingDetails") or {}
    if live.get("actualStartTime"):
        return False  # it did go live at some point — not abandoned
    scheduled = parse_dt(live.get("scheduledStartTime"))
    if scheduled is None:
        return False
    return _now() - scheduled > timedelta(days=2)


def _evaluate(video: Video, item: dict | None) -> str | None:
    """Given a fresh videos.list item (or None if absent), return an unavailable reason,
    or None if the video is fine. Applies fresh metadata as a side effect when present."""
    if item is None:
        return "removed"
    apply_video_details(video, item)
    if _is_abandoned_upcoming(item):
        return "abandoned"
    # A still-"live" broadcast is kept (a legit 24/7 stream); ended streams have already
    # settled to was_live via apply_video_details. Nothing else here is a delete trigger —
    # embeddable=false only blocks our in-app embed, not playback on youtube.com.
    return None


def _collect_user_impact(db: Session, videos: list[Video]) -> dict[int, list[dict]]:
    """For videos about to be deleted, map user_id -> list of {id, title, reason} for the
    videos that user had saved or in one of their playlists. Must run BEFORE the delete,
    while the video_states / playlist_items rows still exist."""
    by_id = {v.id: v for v in videos}
    ids = list(by_id)
    impact: dict[int, set[str]] = {}
    if not ids:
        return {}
    # Anyone who had it in a playlist — which now also covers "saved": Watch later became a
    # built-in playlist, so "saved" is no longer a VideoState status (the old status=="saved"
    # scan matched nothing and under-counted impacted users).
    for user_id, video_id in db.execute(
        select(Playlist.user_id, PlaylistItem.video_id)
        .join(PlaylistItem, PlaylistItem.playlist_id == Playlist.id)
        .where(PlaylistItem.video_id.in_(ids))
    ):
        impact.setdefault(user_id, set()).add(video_id)
    return {
        user_id: [
            {
                "id": vid,
                "title": by_id[vid].title,
                "reason": by_id[vid].unavailable_reason,
            }
            for vid in vids
        ]
        for user_id, vids in impact.items()
    }


def _notify_removed(db: Session, impact: dict[int, list[dict]]) -> int:
    """One batched notification per affected user (never per-video)."""
    for user_id, videos in impact.items():
        n = len(videos)
        create_notification(
            db,
            user_id=user_id,
            type="maintenance",
            title="Videos removed",
            body=(
                f"{n} saved or playlisted "
                f"{'video was' if n == 1 else 'videos were'} removed because "
                "they are no longer available on YouTube."
            ),
            data={"count": n, "videos": videos},
            commit=False,
        )
    db.commit()
    return len(impact)


def _recheck_flagged(db: Session, yt: YouTubeClient) -> dict:
    """Phase 1: recover or hard-delete videos already flagged unavailable."""
    flagged = (
        db.execute(select(Video).where(Video.unavailable_since.is_not(None)))
        .scalars()
        .all()
    )
    if not flagged:
        return {"checked": 0, "recovered": 0, "deleted": 0, "notified": 0}
    progress.report(0, len(flagged), "recheck")
    items = {it["id"]: it for it in yt.get_videos([v.id for v in flagged])}
    now = _now()
    recovered = 0
    to_delete: list[Video] = []
    for video in flagged:
        reason = _evaluate(video, items.get(video.id))
        video.last_checked_at = now
        if reason is None:
            # Recovered: unhide.
            video.unavailable_since = None
            video.unavailable_reason = None
            recovered += 1
            continue
        # Still unavailable; keep the original reason/timer unless it changed.
        if reason != video.unavailable_reason:
            video.unavailable_reason = reason
        deadline = video.unavailable_since + timedelta(days=_grace_days(reason))
        if now >= deadline:
            to_delete.append(video)
    db.commit()  # persist recoveries / reason updates / last_checked_at

    notified = 0
    deleted = 0
    if to_delete:
        impact = _collect_user_impact(db, to_delete)
        for video in to_delete:
            db.delete(video)
        db.commit()  # cascades to video_states / playlist_items
        deleted = len(to_delete)
        notified = _notify_removed(db, impact)
    return {
        "checked": len(flagged),
        "recovered": recovered,
        "deleted": deleted,
        "notified": notified,
    }


def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict:
    """Phase 2: re-check the least-recently-checked currently-available videos and flag any
    that have become unplayable (hidden immediately; deleted later if they don't recover)."""
    batch = get_maintenance_batch(db)  # admin override, else config default
    flagged_new = 0
    checked = 0
    now = _now()
    # Process in videos.list-sized pages so we commit incrementally and stop on low quota.
    while checked < batch:
        if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
            break
        page = (
            db.execute(
                select(Video)
                .where(Video.unavailable_since.is_(None))
                .order_by(
                    Video.last_checked_at.is_(None).desc(),  # never-checked first
                    Video.last_checked_at.asc(),
                )
                .limit(min(50, batch - checked))
            )
            .scalars()
            .all()
        )
        if not page:
            break
        items = {it["id"]: it for it in yt.get_videos([v.id for v in page])}
        for video in page:
            reason = _evaluate(video, items.get(video.id))
            video.last_checked_at = now
            if reason is not None:
                video.unavailable_since = now
                video.unavailable_reason = reason
                flagged_new += 1
        db.commit()
        checked += len(page)
        progress.report(checked, batch, "revalidate")
    return {"checked": checked, "flagged": flagged_new}


def run_maintenance(db: Session) -> dict:
    """Entry point invoked by the scheduler. Quota-attributed to the system (no actor)."""
    user = get_service_user(db)
    if user is None:
        return {"reason": "no service user"}
    with YouTubeClient(db, user) as yt:
        phase1 = _recheck_flagged(db, yt)
        phase2 = (
            _revalidate_rolling(db, yt)
            if quota.remaining_today(db) > sysconfig.get_int(db, "backfill_quota_reserve")
            else {"checked": 0, "flagged": 0, "skipped": "low quota"}
        )
    return {
        "rechecked": phase1["checked"],
        "recovered": phase1["recovered"],
        "deleted": phase1["deleted"],
        "notified": phase1["notified"],
        "revalidated": phase2["checked"],
        "flagged": phase2["flagged"],
    }