feat(maintenance): scheduled job to retire unplayable videos

Add a daily maintenance/validation job that detects videos which can't be played
anywhere and retires them safely. Two phases: re-check already-flagged videos
(recover if available again, else hard-delete once the grace period elapses,
cascading to states/playlist items), and a rolling re-validation of the
least-recently-checked currently-available videos that flags newly-unplayable
ones (hidden from the feed immediately via unavailable_since).

Detection is ~free: a video missing from the videos.list response is
deleted-or-private; an `upcoming` premiere >2 days past its scheduled start that
never went live is abandoned. A still-live broadcast is kept (legit 24/7 stream).
Enrichment now also fetches part=status to populate the status columns. Grace is
7 days for removed videos, none for abandoned. Before deleting, affected users
get one batched notification (never per-video). Interval is admin-tunable via the
Scheduler dashboard; batch size and grace are config. Quota-attributed to the
system and bounded by the same backfill reserve as the other jobs.

backend/app/config.py

@@ -90,6 +90,17 @@ class Settings(BaseSettings):
     autotag_interval_minutes: int = 30
     subscriptions_resync_minutes: int = 360
     playlist_sync_minutes: int = 360
+    # --- Maintenance / validation job ---
+    # Runs once a day by default (admin-tunable via the Scheduler dashboard). Grace periods
+    # are in days because that's the granularity that matters; the rolling re-validation
+    # re-checks the least-recently-checked N videos per run (1 unit / 50 videos), so on a
+    # ~233k catalog ~15000/day ≈ a 15-day full cycle for ~300 quota units.
+    maintenance_interval_minutes: int = 1440
+    maintenance_revalidate_batch: int = 15000
+    # Grace before a flagged-unavailable video is hard-deleted (cascades to states/playlist
+    # items). Deleted/private/paywalled get a recovery window; abandoned upcoming and
+    # confirmed-dead stuck-live have no age grace (deleted on the sweep that confirms them).
+    maintenance_grace_days: int = 7
     # live_status values hidden from the feed by default. Completed-stream VODs
     # ("was_live") are real watchable content and stay visible.
     feed_default_hidden_live: str = "live,upcoming"

backend/app/routes/feed.py

@@ -154,6 +154,11 @@ def _filtered_query(
         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:

backend/app/scheduler.py

@@ -13,6 +13,7 @@ from app.db import SessionLocal
 from app.models import SchedulerSetting
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
+from app.sync.maintenance import run_maintenance
 from app.sync.playlists import sync_all_playlists
 from app.sync.runner import (
     run_deep_backfill,
@@ -44,6 +45,7 @@ JOB_INTERVALS: dict[str, int] = {
     "shorts": settings.shorts_probe_interval_minutes,
     "subscriptions": settings.subscriptions_resync_minutes,
     "playlist_sync": settings.playlist_sync_minutes,
+    "maintenance": settings.maintenance_interval_minutes,
 }
 
 # Sane bounds for an admin-set interval (minutes).
@@ -201,6 +203,14 @@ def _playlist_sync_job() -> None:
     _job("playlist_sync", sync_all_playlists)
 
 
+def _maintenance_job() -> None:
+    def work(db):
+        with quota.attribute(None, "maintenance"):
+            return run_maintenance(db)
+
+    _job("maintenance", work)
+
+
 def start_scheduler() -> None:
     global _scheduler
     if not settings.scheduler_enabled or _scheduler is not None:
@@ -219,6 +229,7 @@ def start_scheduler() -> None:
         "shorts": _shorts_job,
         "subscriptions": _subscriptions_job,
         "playlist_sync": _playlist_sync_job,
+        "maintenance": _maintenance_job,
     }
     scheduler = BackgroundScheduler(timezone="UTC")
     for job_id, fn in fns.items():

backend/app/sync/maintenance.py

@@ -0,0 +1,246 @@
+"""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 quota
+from app.config import settings
+from app.models import Playlist, PlaylistItem, Video, VideoState
+from app.notifications import create_notification
+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("subfeed.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}
+    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 = settings.maintenance_revalidate_batch
+    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) <= settings.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)
+    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) > settings.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"],
+    }

backend/app/sync/videos.py

@@ -217,6 +217,7 @@ def apply_video_details(video: Video, item: dict) -> None:
     stats = item.get("statistics", {})
     topics = item.get("topicDetails", {})
     live = item.get("liveStreamingDetails")
+    status = item.get("status", {})
 
     video.title = snippet.get("title") or video.title
     video.description = snippet.get("description")
@@ -232,6 +233,11 @@ def apply_video_details(video: Video, item: dict) -> None:
         "defaultAudioLanguage"
     )
     video.live_status = _live_status(snippet, live)
+    # status part: used by the maintenance/validation job to judge playability.
+    if status:
+        video.embeddable = status.get("embeddable")
+        video.privacy_status = status.get("privacyStatus")
+        video.upload_status = status.get("uploadStatus")
     # is_short is decided separately by the youtube.com/shorts probe (run_shorts_classification).
 
 

backend/app/youtube/client.py

@@ -327,7 +327,7 @@ class YouTubeClient:
             data = self._get(
                 "videos",
                 {
-                    "part": "snippet,contentDetails,statistics,topicDetails,liveStreamingDetails",
+                    "part": "snippet,contentDetails,statistics,topicDetails,liveStreamingDetails,status",
                     "id": ",".join(batch),
                     "maxResults": 50,
                 },