feat(playlists): push local playlists to YouTube (export + diff sync)

Local 'user' playlists (not Watch later, not YouTube mirrors) can now be pushed
to YouTube: create a new playlist on first push, else reconcile membership + order
so YouTube matches local (local wins). plan_push computes a dry-run estimate
(inserts/deletes/reorders + quota units + divergence) read from the live YouTube
state; push refuses if the estimate exceeds the remaining daily quota so a big
playlist can't strand half-pushed. Reorder uses an insertion-sort move count that
matches what executes. Edits to a linked playlist set dirty; a successful push
clears it. DELETE accepts on_youtube to also delete the playlist on YouTube. The
read-sync now skips YouTube playlists already owned by a local export, avoiding
duplicate read-only mirrors.

backend/app/routes/playlists.py

@@ -6,11 +6,12 @@ from sqlalchemy import func, select, and_
 from sqlalchemy.orm import Session, aliased
 
 from app import quota
-from app.auth import current_user, has_read_scope
+from app.auth import current_user, has_read_scope, has_write_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
+from app.sync.playlists import plan_push, push_playlist, sync_user_playlists
+from app.youtube.client import YouTubeClient, YouTubeError
 
 router = APIRouter(prefix="/api/playlists", tags=["playlists"])
 
@@ -22,6 +23,13 @@ def _own_playlist(db: Session, user: User, playlist_id: int) -> Playlist:
     return pl
 
 
+def _mark_dirty(pl: Playlist) -> None:
+    """Flag a YouTube-linked local playlist as having unpushed local edits, so the UI can
+    offer a 'Sync to YouTube'. No-op for un-linked or mirrored playlists."""
+    if pl.source == "local" and pl.yt_playlist_id:
+        pl.dirty = True
+
+
 def _summary(db: Session, pl: Playlist, count: int, has_video: bool | None = None) -> dict:
     cover = db.scalar(
         select(Video.thumbnail_url)
@@ -36,6 +44,7 @@ def _summary(db: Session, pl: Playlist, count: int, has_video: bool | None = Non
         "kind": pl.kind,
         "source": pl.source,
         "yt_playlist_id": pl.yt_playlist_id,
+        "dirty": pl.dirty,
         "item_count": count,
         "cover_thumbnail": cover,
     }
@@ -205,6 +214,72 @@ def sync_youtube(
         return sync_user_playlists(db, user)
 
 
+def _pushable(db: Session, user: User, playlist_id: int) -> Playlist:
+    """A local playlist the user may push to YouTube (create or update). The built-in
+    Watch later and read-only YouTube mirrors are never pushable."""
+    pl = _own_playlist(db, user, playlist_id)
+    if pl.kind == "watch_later":
+        raise HTTPException(
+            status_code=400, detail="The Watch later list can't be sent to YouTube."
+        )
+    if pl.source == "youtube":
+        raise HTTPException(
+            status_code=400, detail="This playlist already mirrors a YouTube playlist."
+        )
+    return pl
+
+
+@router.get("/{playlist_id}/push-plan")
+def push_plan(
+    playlist_id: int,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Dry run: what 'Sync to YouTube' would do + the quota estimate, so the UI can show a
+    confirm. For a linked playlist this reads the live YouTube state (cheap, 1 unit/page)."""
+    pl = _pushable(db, user, playlist_id)
+    if not has_write_scope(user):
+        raise HTTPException(
+            status_code=403, detail="Enable YouTube editing in Settings first."
+        )
+    try:
+        with quota.attribute(user.id, "playlist_push"):
+            plan = plan_push(db, user, pl)
+    except YouTubeError:
+        raise HTTPException(status_code=502, detail="Couldn't read the playlist on YouTube.")
+    plan["remaining_today"] = quota.remaining_today(db)
+    plan["affordable"] = plan["units_estimate"] <= plan["remaining_today"]
+    return plan
+
+
+@router.post("/{playlist_id}/push")
+def push(
+    playlist_id: int,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Push a local playlist to YouTube (create it on first push, else reconcile to match
+    local — local wins). Refuses if the estimated quota exceeds what's left today, so a
+    big playlist can't strand half-pushed."""
+    pl = _pushable(db, user, playlist_id)
+    if not has_write_scope(user):
+        raise HTTPException(
+            status_code=403, detail="Enable YouTube editing in Settings first."
+        )
+    try:
+        with quota.attribute(user.id, "playlist_push"):
+            plan = plan_push(db, user, pl)
+            if plan["units_estimate"] > quota.remaining_today(db):
+                raise HTTPException(
+                    status_code=429,
+                    detail="Not enough YouTube quota left today for this sync.",
+                )
+            return push_playlist(db, user, pl)
+    except YouTubeError:
+        db.rollback()
+        raise HTTPException(status_code=502, detail="YouTube playlist sync failed.")
+
+
 @router.get("/{playlist_id}")
 def get_playlist(
     playlist_id: int,
@@ -243,6 +318,7 @@ def get_playlist(
         "kind": pl.kind,
         "source": pl.source,
         "yt_playlist_id": pl.yt_playlist_id,
+        "dirty": pl.dirty,
         "items": [_serialize(r) for r in rows],
     }
 
@@ -270,12 +346,31 @@ def rename_playlist(
 @router.delete("/{playlist_id}")
 def delete_playlist(
     playlist_id: int,
+    on_youtube: bool = False,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
     pl = _own_playlist(db, user, playlist_id)
     if pl.kind == "watch_later":
         raise HTTPException(status_code=400, detail="The Watch later list can't be deleted.")
+    if on_youtube:
+        if pl.source == "youtube":
+            raise HTTPException(
+                status_code=400,
+                detail="This is a mirror of a YouTube playlist; manage it on YouTube.",
+            )
+        if not pl.yt_playlist_id:
+            raise HTTPException(status_code=400, detail="This playlist isn't on YouTube.")
+        if not has_write_scope(user):
+            raise HTTPException(
+                status_code=403, detail="Enable YouTube editing in Settings first."
+            )
+        try:
+            with quota.attribute(user.id, "playlist_delete"):
+                with YouTubeClient(db, user) as yt:
+                    yt.delete_playlist(pl.yt_playlist_id)
+        except YouTubeError:
+            raise HTTPException(status_code=502, detail="YouTube delete failed.")
     db.delete(pl)  # items cascade
     db.commit()
     return {"deleted": playlist_id}
@@ -309,6 +404,7 @@ def add_item(
         db.add(
             PlaylistItem(playlist_id=pl.id, video_id=video_id, position=next_pos)
         )
+        _mark_dirty(pl)
         db.commit()
     return {"playlist_id": pl.id, "video_id": video_id}
 
@@ -328,6 +424,7 @@ def remove_item(
     )
     if item is not None:
         db.delete(item)
+        _mark_dirty(pl)
         db.commit()
     return {"playlist_id": pl.id, "video_id": video_id}
 
@@ -353,5 +450,6 @@ def reorder_items(
         if it is not None:
             it.position = pos
             pos += 1
+    _mark_dirty(pl)
     db.commit()
     return {"playlist_id": pl.id, "count": pos}

backend/app/sync/playlists.py

@@ -12,13 +12,16 @@ from sqlalchemy import delete, select
 from sqlalchemy.orm import Session
 
 from app import quota
-from app.auth import has_read_scope
+from app.auth import has_read_scope, has_write_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")
 
+# Cost of a single YouTube write op (playlists/playlistItems insert/update/delete).
+WRITE_UNIT_COST = 50
+
 
 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
@@ -81,6 +84,19 @@ def sync_user_playlists(db: Session, user: User) -> dict:
                 )
             ).scalars()
         }
+        # YouTube playlists that we already own as local (exported) playlists must not be
+        # re-mirrored, or every exported list would also show up as a read-only duplicate.
+        owned_yt_ids = set(
+            db.execute(
+                select(Playlist.yt_playlist_id).where(
+                    Playlist.user_id == user.id,
+                    Playlist.source == "local",
+                    Playlist.yt_playlist_id.is_not(None),
+                )
+            )
+            .scalars()
+            .all()
+        )
         # Drop mirrors whose YouTube playlist no longer exists.
         for ytid, pl in existing.items():
             if ytid not in yt_ids:
@@ -89,7 +105,7 @@ def sync_user_playlists(db: Session, user: User) -> dict:
 
         for idx, p in enumerate(yt_playlists):
             ytid = p.get("id")
-            if not ytid:
+            if not ytid or ytid in owned_yt_ids:
                 continue
             pl = existing.get(ytid)
             if pl is None:
@@ -127,6 +143,154 @@ def sync_user_playlists(db: Session, user: User) -> dict:
     return {"synced": synced}
 
 
+def _local_video_ids(db: Session, pl: Playlist) -> list[str]:
+    return list(
+        db.execute(
+            select(PlaylistItem.video_id)
+            .where(PlaylistItem.playlist_id == pl.id)
+            .order_by(PlaylistItem.position)
+        )
+        .scalars()
+        .all()
+    )
+
+
+def _reorder_moves(current: list[str], desired: list[str]) -> int:
+    """Count the playlistItems.update moves an insertion-sort would need to turn
+    `current` order into `desired` order (both already the same set of video ids).
+    Items already in place are skipped; this matches what apply_push actually executes."""
+    model = list(current)
+    moves = 0
+    for i, want in enumerate(desired):
+        if i < len(model) and model[i] == want:
+            continue
+        j = model.index(want, i)
+        model.pop(j)
+        model.insert(i, want)
+        moves += 1
+    return moves
+
+
+def plan_push(db: Session, user: User, pl: Playlist) -> dict:
+    """Compute (without writing) what pushing this local playlist to YouTube would do:
+    create vs update, how many inserts/deletes/reorders, the quota estimate, and whether
+    YouTube has diverged (items present there that local will remove). Costs read units
+    (1/page) for a linked playlist; nothing for a fresh export."""
+    desired = _local_video_ids(db, pl)
+    if not pl.yt_playlist_id:
+        ops = 1 + len(desired)  # create the playlist + one insert per video
+        return {
+            "action": "create",
+            "to_insert": len(desired),
+            "to_delete": 0,
+            "to_reorder": 0,
+            "yt_extra": 0,
+            "units_estimate": ops * WRITE_UNIT_COST,
+        }
+    with YouTubeClient(db, user) as yt:
+        current = list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id))
+    cur_vids = [it["video_id"] for it in current]
+    cur_set = set(cur_vids)
+    desired_set = set(desired)
+    to_insert = [v for v in desired if v not in cur_set]
+    to_delete = [v for v in cur_vids if v not in desired_set]
+    # After membership reconciliation, the YT order is current-minus-deletes plus inserts
+    # appended; the reorder pass then moves items into the desired order.
+    after_membership = [v for v in cur_vids if v in desired_set] + to_insert
+    reorders = _reorder_moves(after_membership, desired)
+    ops = len(to_insert) + len(to_delete) + reorders
+    return {
+        "action": "update",
+        "to_insert": len(to_insert),
+        "to_delete": len(to_delete),
+        "to_reorder": reorders,
+        "yt_extra": len(to_delete),
+        "units_estimate": ops * WRITE_UNIT_COST,
+    }
+
+
+def push_playlist(db: Session, user: User, pl: Playlist) -> dict:
+    """Push a local playlist to YouTube: create it if needed, then reconcile membership
+    and order so YouTube matches local (local wins). Marks the playlist clean on success.
+    Caller must have checked the write scope and the quota budget."""
+    desired = _local_video_ids(db, pl)
+    inserted = deleted = reordered = created = 0
+    failures: list[str] = []
+    with YouTubeClient(db, user) as yt:
+        if not pl.yt_playlist_id:
+            pl.yt_playlist_id = yt.create_playlist(pl.name)
+            db.commit()
+            created = 1
+            # Brand-new playlist: just append every video in order.
+            for vid in desired:
+                try:
+                    yt.add_playlist_item(pl.yt_playlist_id, vid)
+                    inserted += 1
+                except YouTubeError:
+                    failures.append(vid)
+            pl.dirty = False
+            db.commit()
+            return {
+                "created": created,
+                "inserted": inserted,
+                "deleted": deleted,
+                "reordered": reordered,
+                "failures": failures,
+                "yt_playlist_id": pl.yt_playlist_id,
+            }
+
+        # Existing link: diff against the live YouTube state.
+        current = list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id))
+        item_id_by_vid = {it["video_id"]: it["item_id"] for it in current}
+        cur_vids = [it["video_id"] for it in current]
+        desired_set = set(desired)
+        # Delete extras (local is authoritative).
+        for it in current:
+            if it["video_id"] not in desired_set:
+                try:
+                    yt.delete_playlist_item(it["item_id"])
+                    deleted += 1
+                except YouTubeError:
+                    failures.append(it["video_id"])
+        # Insert missing (append; the reorder pass below fixes positions).
+        cur_set = set(cur_vids)
+        for vid in desired:
+            if vid not in cur_set:
+                try:
+                    item_id_by_vid[vid] = yt.add_playlist_item(pl.yt_playlist_id, vid)
+                    inserted += 1
+                except YouTubeError:
+                    failures.append(vid)
+        # Reorder to match `desired` via insertion-sort moves over a local model.
+        model = [v for v in cur_vids if v in desired_set] + [
+            v for v in desired if v not in cur_set and v in item_id_by_vid
+        ]
+        for i, want in enumerate(desired):
+            if want not in item_id_by_vid:
+                continue  # failed insert; skip
+            if i < len(model) and model[i] == want:
+                continue
+            try:
+                yt.move_playlist_item(item_id_by_vid[want], pl.yt_playlist_id, want, i)
+                reordered += 1
+            except YouTubeError:
+                failures.append(want)
+                continue
+            j = model.index(want)
+            model.pop(j)
+            model.insert(i, want)
+        pl.dirty = False
+        db.commit()
+    return {
+        "created": created,
+        "inserted": inserted,
+        "deleted": deleted,
+        "reordered": reordered,
+        "failures": failures,
+        "yt_playlist_id": pl.yt_playlist_id,
+    }
+
+
 def sync_all_playlists(db: Session) -> dict:
     """Scheduler entry point: mirror playlists for every user with a read scope + token."""
     users = (