feat(playlists): YouTube playlist read-sync (backend)

Mirror each user's own YouTube playlists into local source='youtube' playlists,
one-way (YT -> local). New client methods iter_my_playlists / iter_my_playlist_video_ids
(OAuth, so private playlists work); sync/playlists.py reconciles the mirror (matching YT
order) and ingests any playlist videos not in the shared catalog yet (with stub channels).
POST /api/playlists/sync-youtube for manual sync (read-scope gated, per-user quota) plus a
scheduler job (playlist_sync_minutes, default 6h) that syncs all read-scope users. YouTube's
Watch Later / History are not API-accessible and are never synced.

backend/app/config.py

@@ -89,6 +89,7 @@ class Settings(BaseSettings):
     autotag_title_sample: int = 40
     autotag_interval_minutes: int = 30
     subscriptions_resync_minutes: int = 360
+    playlist_sync_minutes: int = 360
     # 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/playlists.py

@@ -5,10 +5,12 @@ from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import func, select, and_
 from sqlalchemy.orm import Session, aliased
 
-from app.auth import current_user
+from app import quota
+from app.auth import current_user, has_read_scope
 from app.db import get_db
 from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState
 from app.routes.feed import _saved_expr, _serialize
+from app.sync.playlists import sync_user_playlists
 
 router = APIRouter(prefix="/api/playlists", tags=["playlists"])
 
@@ -187,6 +189,22 @@ def remove_watch_later(
     return {"saved": False}
 
 
+@router.post("/sync-youtube")
+def sync_youtube(
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Pull the user's YouTube playlists into local mirrors now (read-only direction).
+    Also runs automatically on the scheduler; this is the manual 'sync now'."""
+    if not has_read_scope(user):
+        raise HTTPException(
+            status_code=403,
+            detail="Connect YouTube (read access) to sync your playlists.",
+        )
+    with quota.attribute(user.id, "playlist_sync"):
+        return sync_user_playlists(db, user)
+
+
 @router.get("/{playlist_id}")
 def get_playlist(
     playlist_id: int,

backend/app/scheduler.py

@@ -9,6 +9,7 @@ from app.config import settings
 from app.db import SessionLocal
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
+from app.sync.playlists import sync_all_playlists
 from app.sync.runner import (
     run_deep_backfill,
     run_enrich,
@@ -79,6 +80,12 @@ def _subscriptions_job() -> None:
     _job("subscriptions", work)
 
 
+def _playlist_sync_job() -> None:
+    # Mirror each read-scope user's YouTube playlists; per-user quota attribution is
+    # handled inside sync_all_playlists.
+    _job("playlist_sync", sync_all_playlists)
+
+
 def start_scheduler() -> None:
     global _scheduler
     if not settings.scheduler_enabled or _scheduler is not None:
@@ -114,6 +121,12 @@ def start_scheduler() -> None:
         minutes=settings.subscriptions_resync_minutes,
         id="subscriptions",
     )
+    scheduler.add_job(
+        _playlist_sync_job,
+        "interval",
+        minutes=settings.playlist_sync_minutes,
+        id="playlist_sync",
+    )
     scheduler.start()
     _scheduler = scheduler
     logger.info("scheduler started")

backend/app/sync/playlists.py

@@ -0,0 +1,152 @@
+"""Read-direction YouTube playlist sync: mirror each user's own YouTube playlists into
+local `source='youtube'` playlists (kept fresh by the scheduler). One-way (YT -> local);
+the write-back direction (local edits -> YouTube) is a later phase.
+
+YouTube's special Watch Later / History playlists are not exposed by the Data API and are
+never synced. Videos in a playlist that aren't in our shared catalog yet are fetched and
+ingested (with a stub channel) so the mirror is faithful."""
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy import delete, select
+from sqlalchemy.orm import Session
+
+from app import quota
+from app.auth import has_read_scope
+from app.models import Channel, OAuthToken, Playlist, PlaylistItem, User, Video
+from app.sync.videos import apply_video_details, parse_dt
+from app.youtube.client import YouTubeClient, YouTubeError
+
+log = logging.getLogger("subfeed.sync")
+
+
+def _ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None:
+    """Make sure every given video id exists in the catalog, fetching + ingesting the
+    missing ones (and a stub channel for any unknown channel). Deleted/private videos that
+    the API won't return are simply left out."""
+    if not video_ids:
+        return
+    have = set(
+        db.execute(select(Video.id).where(Video.id.in_(video_ids))).scalars().all()
+    )
+    missing = [v for v in dict.fromkeys(video_ids) if v not in have]
+    if not missing:
+        return
+    items = yt.get_videos(missing)
+    chan_ids = {
+        it.get("snippet", {}).get("channelId")
+        for it in items
+        if it.get("snippet", {}).get("channelId")
+    }
+    existing_ch = set(
+        db.execute(select(Channel.id).where(Channel.id.in_(chan_ids))).scalars().all()
+    )
+    now = datetime.now(timezone.utc)
+    for it in items:
+        sn = it.get("snippet", {})
+        cid = sn.get("channelId")
+        if cid and cid not in existing_ch:
+            db.add(Channel(id=cid, title=sn.get("channelTitle")))
+            existing_ch.add(cid)
+    db.flush()
+    seen: set[str] = set()
+    for it in items:
+        vid = it.get("id")
+        sn = it.get("snippet", {})
+        cid = sn.get("channelId")
+        if not vid or not cid or vid in seen:
+            continue
+        seen.add(vid)
+        v = Video(id=vid, channel_id=cid, published_at=parse_dt(sn.get("publishedAt")))
+        apply_video_details(v, it)
+        v.enriched_at = now
+        db.add(v)
+    db.commit()
+
+
+def sync_user_playlists(db: Session, user: User) -> dict:
+    """Mirror the user's YouTube playlists into local source='youtube' playlists. Leaves
+    the user's local playlists and the built-in Watch later untouched."""
+    if not has_read_scope(user):
+        return {"synced": 0, "reason": "no read scope"}
+    synced = 0
+    with YouTubeClient(db, user) as yt:
+        yt_playlists = list(yt.iter_my_playlists())
+        yt_ids = {p["id"] for p in yt_playlists if p.get("id")}
+        existing = {
+            pl.yt_playlist_id: pl
+            for pl in db.execute(
+                select(Playlist).where(
+                    Playlist.user_id == user.id, Playlist.source == "youtube"
+                )
+            ).scalars()
+        }
+        # Drop mirrors whose YouTube playlist no longer exists.
+        for ytid, pl in existing.items():
+            if ytid not in yt_ids:
+                db.delete(pl)
+        db.flush()
+
+        for idx, p in enumerate(yt_playlists):
+            ytid = p.get("id")
+            if not ytid:
+                continue
+            pl = existing.get(ytid)
+            if pl is None:
+                pl = Playlist(
+                    user_id=user.id,
+                    name=p.get("title") or "Untitled",
+                    kind="user",
+                    source="youtube",
+                    yt_playlist_id=ytid,
+                    position=1000 + idx,  # sort mirrored lists after local ones
+                )
+                db.add(pl)
+                db.flush()
+            else:
+                pl.name = p.get("title") or pl.name
+            ids = list(yt.iter_my_playlist_video_ids(ytid))
+            _ensure_videos(db, yt, ids)
+            present = (
+                set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars().all())
+                if ids
+                else set()
+            )
+            ordered: list[str] = []
+            seen: set[str] = set()
+            for v in ids:
+                if v in present and v not in seen:
+                    seen.add(v)
+                    ordered.append(v)
+            # Mirror is authoritative: replace the items to match YouTube's order.
+            db.execute(delete(PlaylistItem).where(PlaylistItem.playlist_id == pl.id))
+            for pos, vid in enumerate(ordered):
+                db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=pos))
+            db.commit()
+            synced += 1
+    return {"synced": synced}
+
+
+def sync_all_playlists(db: Session) -> dict:
+    """Scheduler entry point: mirror playlists for every user with a read scope + token."""
+    users = (
+        db.execute(
+            select(User).join(OAuthToken).where(OAuthToken.refresh_token_enc.is_not(None))
+        )
+        .scalars()
+        .all()
+    )
+    total = 0
+    for user in users:
+        if not has_read_scope(user):
+            continue
+        try:
+            with quota.attribute(user.id, "playlist_sync"):
+                total += sync_user_playlists(db, user).get("synced", 0)
+        except YouTubeError:
+            db.rollback()
+            log.warning("Playlist sync failed for user %s", user.id)
+        except Exception:
+            db.rollback()
+            log.exception("Playlist sync crashed for user %s", user.id)
+    return {"users": len(users), "playlists_synced": total}

backend/app/youtube/client.py

@@ -125,6 +125,49 @@ class YouTubeClient:
             if not page_token:
                 break
 
+    def iter_my_playlists(self) -> Iterator[dict]:
+        """Yield the authenticated user's own playlists (OAuth, so private ones are
+        included). Note: YouTube's special Watch Later / History playlists are NOT
+        returned by the Data API and cannot be synced."""
+        page_token = None
+        while True:
+            params = {"part": "snippet,contentDetails", "mine": "true", "maxResults": 50}
+            if page_token:
+                params["pageToken"] = page_token
+            data = self._get("playlists", params, cost=1, allow_key=False)
+            for item in data.get("items", []):
+                sn = item.get("snippet", {})
+                yield {
+                    "id": item.get("id"),
+                    "title": sn.get("title"),
+                    "item_count": item.get("contentDetails", {}).get("itemCount"),
+                    "thumbnail_url": best_thumbnail(sn.get("thumbnails")),
+                }
+            page_token = data.get("nextPageToken")
+            if not page_token:
+                break
+
+    def iter_my_playlist_video_ids(self, playlist_id: str) -> Iterator[str]:
+        """Yield the video ids of one of the user's own playlists, in playlist order
+        (OAuth, so private/unlisted playlists work)."""
+        page_token = None
+        while True:
+            params = {
+                "part": "contentDetails",
+                "playlistId": playlist_id,
+                "maxResults": 50,
+            }
+            if page_token:
+                params["pageToken"] = page_token
+            data = self._get("playlistItems", params, cost=1, allow_key=False)
+            for item in data.get("items", []):
+                vid = item.get("contentDetails", {}).get("videoId")
+                if vid:
+                    yield vid
+            page_token = data.get("nextPageToken")
+            if not page_token:
+                break
+
     def get_channels(self, channel_ids: list[str]) -> list[dict]:
         items: list[dict] = []
         for batch in _chunks(channel_ids, 50):