fix(playlists): item writes lock the list first; one bulk pull per user; the claim on its own connection

- Every playlist item write (add, remove, move, reorder, rename; Plex remove and remove-bulk) row-locks
  the playlist before it touches an item, a move both lists in id order: remove beside reorder was a
  deadlock, and two adds of one video a unique-key 500.
- The scheduled mirror refresh re-reads dirty under that lock, so an edit made during its YouTube calls
  is not replaced by YouTube's order.
- A user's bulk pull holds a per-user advisory claim: two at once created duplicate mirrors. The manual
  sync answers 409 sync_all_busy instead of '0 synced', and like the re-pull spends its cooldown only
  on a pull that runs.
- Deleted mirrors are committed at once, not only when a later list commits.
- The claims open an unpooled connection of their own instead of a third pooled one per push.

backend/tests/test_playlist_position_lock.py

@@ -6,6 +6,11 @@ on one position, and a reorder racing an add collides with it — the list's ord
 rows is then whatever the planner returns. A true two-connection race is not reproducible on
 demand, so the test holds the mechanism: the parent row is taken `FOR UPDATE` by each such route.
 A Plex reorder that omits a row (added in another tab meanwhile) must not collide with it either.
+
+Round 2: the lock also comes FIRST — before the request reads or writes a single item row. A remove
+that deleted an item and then touched the playlist beside a reorder that went the other way was a
+deadlock (a 500 for one of them), and an add that checked "already there?" before the lock let two
+adds of one video both insert (a 500 on the unique key).
 """
 
 import re
@@ -41,6 +46,32 @@ def locks(db):
     event.remove(engine, "before_cursor_execute", _record)
 
 
+@pytest.fixture
+def statements(db):
+    """Every statement a request ran, in order."""
+    seen: list[str] = []
+
+    def _record(conn, cursor, statement, *_):
+        seen.append(" ".join(statement.split()))
+
+    engine = db.get_bind()
+    event.listen(engine, "before_cursor_execute", _record)
+    yield seen
+    event.remove(engine, "before_cursor_execute", _record)
+
+
+def _locked_before_items(seen: list[str], parent: str, items: str) -> str | None:
+    """The first statement touching `items` must come after a `FOR UPDATE` of `parent`; returns
+    that lock statement (None when it came too late or never)."""
+    lock = None
+    for st in seen:
+        if lock is None and "FOR UPDATE" in st and re.search(rf"FROM {parent}\b", st):
+            lock = st
+        if re.search(rf"\b{items}\b", st):
+            return lock
+    return lock
+
+
 def _yt(db, user):
     db.add(Channel(id="UCtest", title="Test"))
     now = datetime.now(timezone.utc)
@@ -134,3 +165,92 @@ def test_a_plex_reorder_that_omits_a_row_does_not_collide_with_it(db, make_user,
     by_id = {it.id: it.rating_key for it in db.query(PlexItem).all()}
     got = sorted((r.position, by_id[r.item_id]) for r in rows)
     assert got == [(0, "c"), (1, "b"), (2, "a")]
+
+
+YT_ITEMS = PlaylistItem.__table__.name
+PLEX_ITEMS = PlexPlaylistItem.__table__.name
+
+
+@pytest.mark.parametrize(
+    "call",
+    [
+        lambda c, pid, other: c.post(f"/api/playlists/{pid}/items", json={"video_id": "v1"}),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items", json={"video_id": "v2", "position": 0}
+        ),
+        lambda c, pid, other: c.delete(f"/api/playlists/{pid}/items/v1"),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items/v1/move", json={"target_playlist_id": other}
+        ),
+        lambda c, pid, other: c.put(f"/api/playlists/{pid}/order", json={"video_ids": ["v1"]}),
+    ],
+    ids=["add-present", "add-at", "remove", "move", "reorder"],
+)
+def test_a_youtube_item_write_locks_the_playlist_before_any_item(
+    db, make_user, as_user, statements, call
+):
+    user = make_user()
+    pid, other = _yt(db, user)
+    c = as_user(user)
+    statements.clear()
+    r = call(c, pid, other)
+    assert r.status_code == 200, r.text
+    assert _locked_before_items(statements, YT, YT_ITEMS), statements
+
+
+def test_a_move_locks_both_lists_in_one_ordered_statement(db, make_user, as_user, statements):
+    """Two moves A→B and B→A each locking "its" list first would lock crosswise."""
+    user = make_user()
+    pid, other = _yt(db, user)
+    statements.clear()
+    r = as_user(user).post(
+        f"/api/playlists/{pid}/items/v1/move", json={"target_playlist_id": other}
+    )
+    assert r.status_code == 200, r.text
+    lock = _locked_before_items(statements, YT, YT_ITEMS)
+    assert lock and " IN (" in lock and "ORDER BY" in lock, lock
+
+
+def test_watch_later_add_and_remove_lock_it_before_any_item(db, make_user, as_user, statements):
+    user = make_user()
+    _yt(db, user)
+    c = as_user(user)
+    assert c.post("/api/playlists/watch-later", json={"video_id": "v2"}).status_code == 200
+    statements.clear()
+    assert c.post("/api/playlists/watch-later", json={"video_id": "v2"}).status_code == 200
+    assert _locked_before_items(statements, YT, YT_ITEMS), statements
+    statements.clear()
+    assert c.delete("/api/playlists/watch-later/v2").status_code == 200
+    assert _locked_before_items(statements, YT, YT_ITEMS), statements
+
+
+def test_a_rename_locks_the_playlist(db, make_user, as_user, locks):
+    """A rename moves `dirty`, which the scheduled mirror refresh re-reads under this lock."""
+    user = make_user()
+    pid, _ = _yt(db, user)
+    r = as_user(user).patch(f"/api/playlists/{pid}", json={"name": "New"})
+    assert r.status_code == 200, r.text
+    assert YT in locks
+
+
+@pytest.mark.parametrize(
+    "call",
+    [
+        lambda c, pid: c.delete(f"/api/plex/playlists/{pid}/items/a"),
+        lambda c, pid: c.post(
+            f"/api/plex/playlists/{pid}/items/remove-bulk", json={"item_rating_keys": ["a", "b"]}
+        ),
+        lambda c, pid: c.post(f"/api/plex/playlists/{pid}/items", json={"item_rating_key": "a"}),
+    ],
+    ids=["remove", "remove-bulk", "add-present"],
+)
+def test_a_plex_item_write_locks_the_playlist_before_any_item(
+    db, make_user, as_user, statements, call
+):
+    user = make_user()
+    pid = _plex(db, user)
+    c = as_user(user)
+    statements.clear()
+    r = call(c, pid)
+    assert r.status_code == 200, r.text
+    assert _locked_before_items(statements, PLEX, PLEX_ITEMS), statements