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 datetime, timedelta, timezone

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

from app import progress, quota, sysconfig
from app.config import settings
from app.models import Playlist, PlaylistItem, Video, VideoState
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.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 _now() -> datetime:
    return datetime.now(timezone.utc)


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 {}
    # Saved (VideoState.status == "saved").
    for user_id, video_id in db.execute(
        select(VideoState.user_id, VideoState.video_id).where(
            VideoState.video_id.in_(ids), VideoState.status == "saved"
        )
    ):
        impact.setdefault(user_id, set()).add(video_id)
    # In any of the user's playlists (covers watch-later, which is a playlist).
    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"],
    }