"""YouTube playlist sync, both directions. Read: mirror each user's own YouTube playlists
into local `source='youtube'` playlists (kept fresh by the scheduler). Write-back: two-way sync
local edits with YouTube (`plan_merge` / `merge_playlist`), behind the optional write scope; the
pure 3-way merge they run is `sync/playlist_merge.py`.

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 hashlib
import json
from collections.abc import Callable, Iterator
from contextlib import AbstractContextManager, contextmanager

from typing import TypeVar

from sqlalchemy import Engine, create_engine, delete, func, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import NullPool

from app import logs, progress, quota
from app.auth import has_read_scope
from app.models import User
from app.catalog.models import Video
from app.reader.models import Playlist, PlaylistItem
from app.sync.playlist_merge import merge_plan
from app.sync.runner import token_users
from app.sync.subscriptions import blank_listing, refuse_blank_listing, vanished
from app.sync.videos import ensure_videos
from app.youtube.client import YouTubeClient, YouTubeError, is_playlist_gone
from app.userscope import owned_by

log = logs.get_logger(__name__)
# Cost of a single YouTube write op (playlists/playlistItems insert/update/delete).
WRITE_UNIT_COST = 50

# The advisory-lock namespaces ("PL" per playlist, "PU" per user), so a key cannot meet another
# feature's lock on the same number (Postgres keeps the two-int keys apart from the one-bigint ones).
_SYNC_LOCK_NS = 0x504C
_USER_SYNC_LOCK_NS = 0x5055

# The claims' own connections, outside the app pool: a claim is held for a whole push (seconds of
# YouTube calls) beside the request's pooled session and quota's, so a burst of pushes drawing a
# third pooled connection each ran the pool dry and stalled every other request for pool_timeout.
# NullPool opens one per claim and closes it after, so a lock can never ride back into a pool either.
_claim_engines: dict[str, Engine] = {}


def _claim_engine(db: Session) -> Engine:
    url = db.get_bind().engine.url
    key = url.render_as_string(hide_password=False)
    eng = _claim_engines.get(key)
    if eng is None:
        eng = _claim_engines[key] = create_engine(url, poolclass=NullPool, hide_parameters=True)
    return eng


@contextmanager
def _advisory_claim(db: Session, ns: int, key: int) -> Iterator[bool]:
    """Try a session-level advisory lock for the block on a connection of its own; yields whether it
    was free. It survives the block's commits, and a crashed worker's lock dies with its connection
    — no claim column that a crash leaves stuck."""
    conn = _claim_engine(db).connect()
    held = False
    try:
        held = bool(conn.scalar(select(func.pg_try_advisory_lock(ns, key))))
        conn.commit()
        yield held
    finally:
        try:
            if held:
                conn.scalar(select(func.pg_advisory_unlock(ns, key)))
                conn.commit()
        finally:
            conn.close()


def sync_claim(db: Session, playlist_id: int) -> AbstractContextManager[bool]:
    """Claim one playlist's YouTube sync (push, re-pull, the scheduled mirror refresh) for the block;
    yields whether it was free. A push reads YouTube, then writes both sides for seconds and commits
    step by step, so a row lock ends at its first commit and two tabs (or a push and the scheduler)
    interleave into a doubled YouTube item or a baseline recorded over the other's writes."""
    return _advisory_claim(db, _SYNC_LOCK_NS, playlist_id)


class PlaylistGone(Exception):
    """The list a sync was working on was deleted meanwhile (another tab, the mirror drop). Its
    items went with it, so a plan made now reads "every item removed here" — and would delete them
    all from the YouTube playlist the reader chose to keep. The route answers 404."""


def lock_playlists(db: Session, *playlist_ids: int) -> set[int]:
    """Row-lock playlists before touching their items, several in id order. Every item write takes
    the PLAYLIST first — the route edits, a push, a re-pull, the scheduled mirror refresh and a
    delete alike: a writer that locked an item row first and the playlist second (a push's item
    replace beside a remove) deadlocks against one that went the other way, and the loser is a
    500. The rows are re-read, so `dirty`, the name and the sync baseline are the ones a finished
    writer committed. Held to the caller's next commit — so no commit may come between this and
    the write it guards. Returns the ids it found: a list deleted meanwhile is not among them."""
    rows = db.scalars(
        select(Playlist)
        .where(Playlist.id.in_(sorted(set(playlist_ids))))
        .order_by(Playlist.id)
        .with_for_update()
        .execution_options(populate_existing=True)
    ).all()  # consumed: rows fetched are what refreshes the loaded objects
    return {pl.id for pl in rows}


def _lock_or_gone(db: Session, pl: Playlist) -> None:
    """`lock_playlists` for a sync that goes on to write YouTube or the items: a list deleted since
    it was read raises `PlaylistGone` (rolled back) instead of being planned as an empty one."""
    pl_id = pl.id  # read first: the rollback expires `pl`, and its row is gone
    if pl_id not in lock_playlists(db, pl_id):
        db.rollback()
        raise PlaylistGone(pl_id)


def fingerprint(name: str, ids: list[str]) -> str:
    """A stable checksum of a playlist's name + ordered video ids. Stored as the synced
    baseline; the current fingerprint is compared to it to derive `dirty`."""
    h = hashlib.sha256()
    h.update((name or "").encode())
    h.update(b"\n")
    h.update(",".join(ids).encode())
    return h.hexdigest()


def current_fingerprint(db: Session, pl: Playlist) -> str:
    return fingerprint(pl.name, _local_video_ids(db, pl))


def _dump_ids(ids: list[str]) -> str:
    """Serialize the ordered id list stored as the 3-way-merge baseline (`synced_video_ids`)."""
    return json.dumps(ids)


def _load_baseline(pl: Playlist) -> tuple[list[str] | None, set[str]]:
    """Parse `synced_video_ids` into (ordered id set, un-ingestable id set). Two on-disk forms:
    a plain JSON list (legacy / no un-ingestable ids) or `{"ids": [...], "unavailable": [...]}`.
    `unavailable` = YouTube-only videos that exist on YouTube but the Data API won't return
    (private/deleted/region-locked), so they never became local rows — tracked so the merge can
    skip them instead of forever re-classifying them as a YouTube-add (they'd never converge)."""
    if not pl.synced_video_ids:
        return None, set()
    try:
        val = json.loads(pl.synced_video_ids)
    except (ValueError, TypeError) as exc:
        # No baseline means the three-way merge has no "what we last agreed on" to subtract, so
        # it degrades to a union: nothing is lost, but a delete made on either side comes back.
        log.warning(
            "the sync baseline for a playlist could not be read; merging as a union",
            extra=logs.event(
                "sync.playlist.baseline_unreadable",
                playlist_id=pl.id,
                reason=type(exc).__name__,
            ),
        )
        return None, set()
    if isinstance(val, list):
        return [str(v) for v in val], set()
    if isinstance(val, dict):
        raw_ids = val.get("ids")
        ids = [str(v) for v in raw_ids] if isinstance(raw_ids, list) else None
        raw_un = val.get("unavailable")
        un = {str(v) for v in raw_un} if isinstance(raw_un, list) else set()
        return ids, un
    return None, set()


def parse_baseline(pl: Playlist) -> list[str] | None:
    """The last-synced ordered id set, or None when no baseline has been recorded yet (a fresh
    export, or an older link). None makes the first merge union losslessly."""
    return _load_baseline(pl)[0]


def parse_unavailable(pl: Playlist) -> set[str]:
    """The last-synced set of un-ingestable-but-live YouTube ids (see `_load_baseline`)."""
    return _load_baseline(pl)[1]


def _set_baseline(
    pl: Playlist,
    ids: list[str],
    unavailable: list[str] | None = None,
    *,
    name: str | None = None,
) -> None:
    """Record `ids` as both the hash (drives `dirty`, over the local ids only) and the raw ordered
    set (drives merge). `unavailable` = un-ingestable-but-live YouTube ids to carry forward so the
    next merge skips them; stored only when non-empty (keeps the common case a plain list). `name`:
    the title YouTube now has, when that may differ from the list's own (see the push)."""
    pl.synced_fingerprint = fingerprint(pl.name if name is None else name, ids)
    if unavailable:
        pl.synced_video_ids = json.dumps({"ids": ids, "unavailable": list(unavailable)})
    else:
        pl.synced_video_ids = _dump_ids(ids)
    pl.dirty = False


def _set_baseline_under_lock(
    db: Session,
    pl: Playlist,
    ids: list[str],
    unavailable: list[str] | None = None,
    *,
    name: str,
) -> None:
    """A push's closing baseline: `ids` and `name` are what YouTube now holds. The push spent
    seconds on YouTube writes since it replaced the local items, so an edit may have landed
    meanwhile — lock the list and derive `dirty` from what is there, or the edit would read as
    synced and never reach YouTube. The name too: the lock re-reads a rename made meanwhile, and a
    baseline taken over THAT name matched it, so the list read clean with YouTube's title still the
    old one."""
    _lock_or_gone(db, pl)
    _set_baseline(pl, ids, unavailable, name=name)
    recompute_dirty(db, pl)


def recompute_dirty(db: Session, pl: Playlist) -> None:
    """Set `dirty` by comparing the current state to the synced baseline. Unlinked lists
    (and Watch later) are never dirty; a missing baseline counts as dirty (unknown YT state)."""
    if not pl.yt_playlist_id:
        pl.dirty = False
        return
    # The session has autoflush off, so the caller's pending item/order/name edits aren't
    # visible to current_fingerprint's query yet — flush first or dirty would lag one edit.
    db.flush()
    pl.dirty = pl.synced_fingerprint != current_fingerprint(db, pl)


def _replace_items_from_youtube(db: Session, pl: Playlist, ids: list[str]) -> list[str]:
    """Mirror a YouTube playlist's ordered video ids into `pl`'s local items (authoritative
    replace): keep only those present in the catalog in their YT order (de-duplicated), swap the
    items out, and reset the synced fingerprint + dirty flag. The caller has ingested `ids`
    (`sync.videos.ensure_videos`, which commits) and THEN taken `lock_playlists`, set `pl.name`, and commits
    after: an ingest in here re-fetched every id YouTube will not return (a private or deleted
    video) and its commit let go of the lock before the replace. Returns the final ordered 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)
    _set_local_items(db, pl, ordered)
    # After a pull/reset local == YouTube, so this ordered set IS the new merge baseline.
    _set_baseline(pl, ordered)
    return ordered


def _set_local_items(db: Session, pl: Playlist, ordered: list[str]) -> None:
    """Authoritatively replace `pl`'s local items with `ordered` (positions 0..n). The caller
    is responsible for having ensured every id exists in the catalog."""
    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))


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"}
    with user_sync_claim(db, user.id) as free:
        if not free:
            return {"synced": 0, "reason": "busy"}
        return pull_user_playlists(db, user)


def user_sync_claim(db: Session, user_id: int) -> AbstractContextManager[bool]:
    """Claim a user's bulk mirror pull for the block; yields whether it was free. Two runs at once
    (the scheduler beside a manual "sync now", two tabs) both saw a YouTube list unmirrored and
    each created its mirror — a duplicate row."""
    return _advisory_claim(db, _USER_SYNC_LOCK_NS, user_id)


def _mirror_listing(yt: YouTubeClient, ytid: str, count: object, pl_id: int) -> list[str] | None:
    """A mirror's video ids from YouTube, or None when there is nothing to believe: the list was
    deleted between the pull's listing and this read (the next pull drops the mirror as vanished —
    not a reason to fail every other list), or YouTube listed no items for a
    list it counts as non-empty (`blank_listing`)."""
    try:
        ids = list(yt.iter_my_playlist_video_ids(ytid))
    except YouTubeError as exc:
        if not is_playlist_gone(exc):
            raise
        log.info("YouTube playlist of %s vanished mid-pull; skipped", pl_id)
        return None
    if blank_listing(ids, count):
        log.warning("YouTube listed no items for non-empty playlist %s; kept", pl_id)
        return None
    return ids


def pull_user_playlists(db: Session, user: User) -> dict:
    """The mirror pull itself; the caller holds `user_sync_claim`."""
    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(
                    owned_by(Playlist.user_id, user.id), Playlist.source == "youtube"
                )
            ).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(
                    owned_by(Playlist.user_id, user.id),
                    Playlist.source == "local",
                    Playlist.yt_playlist_id.is_not(None),
                )
            ).scalars()
        )
        # Drop vanished mirrors (none on an empty answer) — locked first, as every item write.
        gone = vanished(existing, yt_ids)
        if gone:
            lock_playlists(db, *(pl.id for pl in gone))
        for pl in gone:
            db.delete(pl)
        # Committed now: a run where no list below commits (all dirty, or all claimed) would
        # otherwise drop the deletions with the session.
        db.commit()

        for idx, p in enumerate(yt_playlists):
            ytid = p.get("id")
            if not ytid or ytid in owned_yt_ids:
                continue
            pl = existing.get(ytid)
            if pl is not None and pl.dirty:
                # Local edits not yet pushed: don't clobber them with YouTube's state.
                # A successful 'Sync to YouTube' clears dirty and lets the mirror refresh.
                continue
            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()
            with sync_claim(db, pl.id) as free:
                if not free:
                    # The user is pushing or re-pulling this very list: it writes the same rows.
                    # The next scheduled run refreshes it.
                    continue
                ids = _mirror_listing(yt, ytid, p.get("item_count"), pl.id)
                if ids is None:
                    db.rollback()  # a mirror this run just created is not kept empty either
                    continue
                # Ingest first — it commits — so the row lock below is held from the dirty check
                # to the commit that replaces the items.
                ensure_videos(db, yt, ids)
                # `dirty` read above is a snapshot from before the YouTube calls (sessions keep
                # loaded rows past a commit): a remove or reorder made meanwhile must not be
                # replaced by YouTube's order. Item writes lock the playlist row, so this waits for
                # one in flight and sees what it committed.
                lock_playlists(db, pl.id)
                if pl.dirty:
                    db.rollback()
                    continue
                pl.name = p.get("title") or pl.name
                # Mirror is authoritative: replace the items to match YouTube's order.
                _replace_items_from_youtube(db, pl, ids)
                db.commit()
            synced += 1
    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 plan_merge(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
    """Compute (without writing) what a two-way sync of this playlist would do: create vs
    update, how many items each side gains/loses, reorders, and the quota estimate (YouTube
    writes only — local edits are free). Reads live YouTube state (1 unit/page) for a linked
    playlist; nothing for a fresh export. `live` = a pre-fetched (current_items, snippet) tuple
    so the caller can share one read with merge_playlist."""
    local = _local_video_ids(db, pl)
    if not pl.yt_playlist_id:
        ops = 1 + len(local)  # create the playlist + one insert per video
        return {
            "action": "create",
            "add_yt": len(local),
            "add_local": 0,
            "remove_yt": 0,
            "remove_local": 0,
            "to_reorder": 0,
            "rename": False,
            "units_estimate": ops * WRITE_UNIT_COST,
        }
    if live is not None:
        current, snippet = live
    else:
        with YouTubeClient(db, user) as yt:
            current = list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id))
            snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
    rename = bool(snippet) and (snippet.get("title") or "") != pl.name
    yt_ids = [it["video_id"] for it in current]
    refuse_blank_listing(yt_ids, (snippet or {}).get("item_count"))
    mp = merge_plan(parse_baseline(pl), local, yt_ids, parse_unavailable(pl))
    ops = len(mp.add_to_yt) + len(mp.remove_from_yt) + mp.reorders + (1 if rename else 0)
    return {
        "action": "update",
        "add_yt": len(mp.add_to_yt),
        "add_local": len(mp.add_to_local),
        "remove_yt": len(mp.remove_from_yt),
        "remove_local": len(mp.remove_from_local),
        "to_reorder": mp.reorders,
        "rename": rename,
        "units_estimate": ops * WRITE_UNIT_COST,
    }


# After this many consecutive write failures we stop calling YouTube for the rest of the push.
# The failure modes that matter here are all account-wide, not per-video: quota exhausted, the
# write scope revoked, the token dead. Every remaining write then fails identically — but each
# one still SPENDS 50 units, so a long playlist burns through the whole (already dead) budget,
# and tomorrow's too, to produce a failure list we knew after the fifth item.
_MAX_CONSECUTIVE_FAILURES = 5


class _Failed:
    """Type of the `FAILED` sentinel, so `run()` can stay generic in the op's return type: a call
    site gets back `str | _Failed` (not a bare `object`) and keeps its item id typed."""


#: Returned by `_Breaker.run` when the write did not happen (refused or failed).
FAILED = _Failed()

T = TypeVar("T")


class _Breaker:
    """Consecutive-failure circuit breaker for a single push run. Any success re-arms it.

    Drive it with `run()` — it owns the whole open-check / call / record-outcome sequence, so a
    write site can't accidentally implement two thirds of the protocol (forgetting to record a
    failure would silently disable the breaker for that operation)."""

    def __init__(
        self, limit: int = _MAX_CONSECUTIVE_FAILURES, playlist_id: int | None = None
    ) -> None:
        self.limit = limit
        self.playlist_id = playlist_id  # for the failure line only
        self.streak = 0
        self.tripped = False

    def open(self) -> bool:
        """True while writes may still be attempted."""
        return not self.tripped

    def hit(self, ok: bool) -> None:
        if ok:
            self.streak = 0
            return
        self.streak += 1
        if self.streak >= self.limit and not self.tripped:
            self.tripped = True
            log.warning(
                "playlist push: %d consecutive YouTube write failures — stopping this run "
                "(quota, scope or token); the remaining items are reported as failures",
                self.limit,
            )

    def run(
        self, op: Callable[[], T], on_fail: Callable[[], None], action: str = "write"
    ) -> T | _Failed:
        """Attempt one YouTube write. Returns the op's result, or `FAILED` when the breaker is
        already tripped (op not called) or the op raised YouTubeError. `on_fail` records the item
        in the caller's failure list either way."""
        if self.tripped:
            on_fail()
            return FAILED
        try:
            out = op()
        except YouTubeError as exc:
            # The client logged the call, but not which playlist it was for or what kind of write
            # — and the caller's failure list only ever reaches the user as a count. Not the video:
            # what is in someone's playlist is a row of their history, the same as a download.
            log.debug(
                "a playlist write to YouTube failed",
                extra=logs.event(
                    "sync.playlist.item_push_failed",
                    playlist_id=self.playlist_id,
                    action=action,
                    reason=exc.reason,
                ),
            )
            on_fail()
            self.hit(False)
            return FAILED
        self.hit(True)
        return out


def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
    """Two-way sync a local playlist with YouTube: create it on first push, else 3-way MERGE
    both sides so an item added (or deleted) on either side survives — no side wins wholesale.
    Local edits are applied directly; YouTube edits go through the failure breaker. On full
    success the merged set becomes the new baseline and the playlist is marked clean; on any
    YouTube failure the baseline is left untouched so the next sync retries and converges.
    Caller must have checked the write scope and the quota budget. `live` = a pre-fetched
    (current_items, snippet) tuple (from plan_merge's read) to avoid a second page-through."""
    local = _local_video_ids(db, pl)
    inserted = deleted = reordered = created = pulled = removed_local = 0
    failures: list[str] = []
    breaker = _Breaker(playlist_id=pl.id)
    with YouTubeClient(db, user) as yt:
        if not pl.yt_playlist_id:
            pushed_name = pl.name
            pl.yt_playlist_id = yt.create_playlist(pushed_name)
            db.commit()
            created = 1
            # Brand-new playlist: just append every local video in order.
            for vid in local:
                added = breaker.run(
                    lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
                    lambda v=vid: failures.append(v),
                    "add",
                )
                if not isinstance(added, _Failed):
                    inserted += 1
            if not failures:
                _set_baseline_under_lock(db, pl, local, name=pushed_name)
            db.commit()
            return {
                "created": created,
                "inserted": inserted,
                "deleted": deleted,
                "reordered": reordered,
                "pulled": pulled,
                "removed_local": removed_local,
                "failures": failures,
                "aborted": breaker.tripped,
                "yt_playlist_id": pl.yt_playlist_id,
            }

        # Existing link: 3-way merge against the live YouTube state (reuse plan_merge's read).
        if live is not None:
            current, snippet = live
        else:
            snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
            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}
        yt_ids = [it["video_id"] for it in current]
        # A merge would read "YouTube lists none of them" as "YouTube deleted them all".
        refuse_blank_listing(yt_ids, (snippet or {}).get("item_count"))
        # Re-attempt ingestion of ids previously marked un-ingestable that are still on YouTube: a
        # video that has since become available (e.g. private→public) should re-enter the merge, not
        # stay skipped forever. Those that still won't ingest remain filtered (`effective_unavailable`).
        prev_unavailable = parse_unavailable(pl)
        still_listed = [v for v in yt_ids if v in prev_unavailable]
        if still_listed:
            ensure_videos(db, yt, still_listed)
            healed = set(
                db.execute(select(Video.id).where(Video.id.in_(still_listed))).scalars().all()
            )
            effective_unavailable = prev_unavailable - healed
        else:
            effective_unavailable = prev_unavailable
        mp = merge_plan(parse_baseline(pl), local, yt_ids, effective_unavailable)

        # --- Siftlode side (free, direct): pull YouTube-adds in, drop YouTube-deletes, land the
        # merged order authoritatively. Do this first so local reflects the merge even if a later
        # YouTube write trips the breaker.
        ensure_videos(db, yt, mp.add_to_local)
        # `local` was read before the YouTube calls above: an add or remove committed meanwhile
        # (another tab) would be undone by the replace below. Lock the list — the ingest's commit is
        # behind us — and plan again from what is there now. Nothing new needs ingesting: a video
        # the re-read adds is already a local row. A list deleted meanwhile stops here, before a plan
        # over its now-empty items removes them all from YouTube.
        _lock_or_gone(db, pl)
        local = _local_video_ids(db, pl)
        mp = merge_plan(parse_baseline(pl), local, yt_ids, effective_unavailable)
        # Only the ids that actually exist in the catalog can become local rows (a private/deleted
        # YouTube video won't ingest); keep the merged order, dropping any that didn't materialize.
        present = (
            set(db.execute(select(Video.id).where(Video.id.in_(mp.final))).scalars().all())
            if mp.final
            else set()
        )
        final_local = [v for v in mp.final if v in present]
        pulled = len([v for v in mp.add_to_local if v in present])
        removed_local = len(mp.remove_from_local)
        _set_local_items(db, pl, final_local)
        db.commit()

        # --- YouTube side (metered, breaker-guarded). The name as of the lock above: the closing
        # baseline records THIS one, whatever a rename made during the writes below says.
        pushed_name = pl.name
        if snippet is not None and (snippet.get("title") or "") != pushed_name:
            try:
                yt.update_playlist_title(pl.yt_playlist_id, pushed_name)
            except YouTubeError as exc:
                # All the caller passes on is the word "title", which says the rename did not
                # happen but not that YouTube refused it, or why.
                log.debug(
                    "a playlist rename did not reach YouTube",
                    extra=logs.event(
                        "sync.playlist.title_push_failed",
                        playlist_id=pl.id,
                        reason=exc.reason,
                    ),
                )
                failures.append("title")
        # Propagate genuine local deletes.
        for vid in mp.remove_from_yt:
            item_id = item_id_by_vid.get(vid)
            if item_id is None:
                continue
            dropped = breaker.run(
                lambda i=item_id: yt.delete_playlist_item(i),
                lambda v=vid: failures.append(v),
                "delete",
            )
            if not isinstance(dropped, _Failed):
                deleted += 1
                item_id_by_vid.pop(vid, None)
        # Insert local-adds (append; the reorder pass below fixes positions).
        for vid in mp.add_to_yt:
            item_id = breaker.run(
                lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
                lambda v=vid: failures.append(v),
                "add",
            )
            if not isinstance(item_id, _Failed):
                item_id_by_vid[vid] = item_id
                inserted += 1
        # Reorder YouTube to match `final` via insertion-sort moves over a local model.
        final_set = set(mp.final)
        model = [v for v in yt_ids if v in final_set and v in item_id_by_vid] + [
            v for v in mp.add_to_yt if v in item_id_by_vid
        ]
        for i, want in enumerate(mp.final):
            if want not in item_id_by_vid:
                continue  # deleted-from-YT or failed insert; skip
            if i < len(model) and model[i] == want:
                continue
            moved = breaker.run(
                lambda w=want, pos=i: yt.move_playlist_item(
                    item_id_by_vid[w], pl.yt_playlist_id, w, pos
                ),
                lambda w=want: failures.append(w),
                "move",
            )
            if isinstance(moved, _Failed):
                continue
            reordered += 1
            j = model.index(want)
            model.pop(j)
            model.insert(i, want)
        if not failures:
            # Carry forward un-ingestable-but-live YouTube ids so the next merge skips them
            # (see merge_plan): this run's failed pulls, plus prior ones still on YouTube, minus
            # any that have since left YouTube. Recomputed each success → self-healing.
            yt_id_set = set(yt_ids)
            unavailable = sorted(
                (effective_unavailable | {v for v in mp.add_to_local if v not in present})
                & yt_id_set
            )
            _set_baseline_under_lock(db, pl, final_local, unavailable, name=pushed_name)
        db.commit()
    return {
        "created": created,
        "inserted": inserted,
        "deleted": deleted,
        "reordered": reordered,
        "pulled": pulled,
        "removed_local": removed_local,
        "failures": failures,
        "aborted": breaker.tripped,
        "yt_playlist_id": pl.yt_playlist_id,
    }


def repull_playlist(db: Session, user: User, pl: Playlist) -> dict:
    """Force-refresh ONE YouTube-linked playlist from YouTube, discarding local edits and
    clearing dirty. This is the per-playlist 'reset to YouTube' — unlike the bulk read-sync
    it deliberately does NOT skip a dirty playlist (overwriting local changes is the point).
    Works for both mirrors (source='youtube') and exported local playlists (yt_playlist_id)."""
    if not pl.yt_playlist_id:
        raise YouTubeError("Playlist is not linked to YouTube")
    with YouTubeClient(db, user) as yt:
        snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
        if snippet is None:
            raise YouTubeError(
                "Playlist no longer exists on YouTube", reason="playlistNotFound", path="playlists"
            )
        ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
        refuse_blank_listing(ids, snippet.get("item_count"))
        ensure_videos(db, yt, ids)
        _lock_or_gone(db, pl)
        pl.name = snippet.get("title") or pl.name
        ordered = _replace_items_from_youtube(db, pl, ids)
        db.commit()
    return {"items": len(ordered)}


def sync_all_playlists(db: Session) -> dict:
    """Scheduler entry point: mirror playlists for every user with a read scope + token."""
    users = token_users(db)
    total = 0
    for i, user in enumerate(users, 1):
        if not has_read_scope(user):
            progress.report(i, len(users), "playlist_sync")
            continue
        try:
            with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL):
                total += sync_user_playlists(db, user).get("synced", 0)
        except YouTubeError as exc:
            db.rollback()
            log.warning(
                "Playlist sync failed for user %s",
                user.id,
                extra=logs.event("sync.playlists.failed", user_id=user.id, reason=exc.reason),
            )
        except Exception:
            db.rollback()
            log.exception(
                "Playlist sync crashed for user %s",
                user.id,
                extra=logs.event("sync.playlists.crashed", user_id=user.id),
            )
        progress.report(i, len(users), "playlist_sync")
    return {"users": len(users), "playlists_synced": total}