fix(playlists): what the first review of bulk actions found

An Esc during a pointer drag is spent on the drag and leaves the selection.
A wheel over the rail re-reads the target each frame, so a drop lands on the
row under the pointer. After a bulk removal the focus goes to the next row
left, and to the list's menu when the list is emptied.

remove-bulk names the videos that went, and its Undo brings back only those.
An Undo of a move keeps each row (and when it was saved), leaves out a video
that moved on from the target, closes up its place, and says how many came
back. Delete with every tick hidden removes nothing and says why. A list gone
before the lock answers 404, and a body without its videos 422.

One removal path now: the single row goes through the bulk one, the same
toast and undo, which also retires the add-at-position route. railDrop reuses
the store and scrollParent, the drag image the overlay and the z-index token,
and the sensors keep their number. The count is read out from a region that
outlives the band. Tests fill the gaps the review named, each seen failing.

backend/app/reader/service.py

@@ -406,30 +406,6 @@ def remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -
     return True
 
 
-def place_item(db: Session, pl: Playlist, video_id: str, position: int) -> None:
-    """Put a video already in the list at `position` (clamped to the end), renumbering the whole
-    list: positions can have gaps after removals, so "index N" is only meaningful over the
-    ordered rows."""
-    rows = queries.items(db, pl.id)
-    new = next(r for r in rows if r.video_id == video_id)
-    rows.remove(new)
-    rows.insert(min(position, len(rows)), new)
-    for pos, row in enumerate(rows):
-        row.position = pos
-    db.flush()
-
-
-def add_to_playlist(db: Session, pl: Playlist, video_id: str, position: int | None) -> None:
-    """Add a video to one of the user's lists, at `position` when given (the list page's "Undo" of
-    a removal puts the row back where it was in ONE request), else at the end. A video already in
-    the list stays where it is."""
-    added = add_item(db, pl, video_id, mark_dirty=False)
-    if added and position is not None:
-        place_item(db, pl, video_id, position)
-    recompute_dirty(db, pl)
-    db.flush()
-
-
 def rename_playlist(db: Session, pl: Playlist, name: str) -> None:
     """A new name moves `dirty`; the caller holds the playlist's lock (the scheduled mirror refresh
     reads `dirty` under it)."""
@@ -447,9 +423,9 @@ def delete_playlist(db: Session, pl: Playlist) -> None:
     db.flush()
 
 
-def remove_items(db: Session, pl: Playlist, video_ids: list[str]) -> int:
+def remove_items(db: Session, pl: Playlist, video_ids: list[str]) -> list[str]:
     """Remove these videos from a playlist, in one transaction; a video not in it any more is simply
-    not counted. Says how many went."""
+    not among them. Says which went: an Undo must not bring back one this removal never took."""
     lock_playlists(db, pl.id)
     wanted = set(video_ids)
     rows = [it for it in queries.items(db, pl.id) if it.video_id in wanted]
@@ -459,24 +435,38 @@ def remove_items(db: Session, pl: Playlist, video_ids: list[str]) -> int:
         recompute_dirty(db, pl)
         touch(pl)
     db.flush()
-    return len(rows)
+    return [row.video_id for row in rows]
 
 
-def _insert_at(db: Session, pl: Playlist, placed: list[tuple[int, str]]) -> int:
+def _insert_at(
+    db: Session,
+    pl: Playlist,
+    placed: list[tuple[int, str]],
+    leaving: dict[str, PlaylistItem] | None = None,
+    staying_out: frozenset[str] = frozenset(),
+) -> int:
     """Add videos at their indices (ascending, so each index is where the row stood when the rows
     before it were back) and renumber the list 0..n-1, as positions can have gaps. A video already in
-    the list stays where it is, and one the catalog has lost is left out. Says how many were added;
-    the caller holds the playlist's lock."""
+    the list stays where it is, and one the catalog has lost or in `staying_out` is left out, the
+    ones after it closing up. A video with a row in `leaving` brings that row along, for the reason
+    `move_items` re-parents. Says how many were added; the caller holds the playlist's lock."""
+    leaving = leaving or {}
     rows = queries.items(db, pl.id)
     present = {r.video_id for r in rows}
     known = queries.known_video_ids(db, [vid for _, vid in placed])
     added = 0
+    left_out = 0
     for at, vid in sorted(placed, key=lambda p: p[0]):
-        if vid in present or vid not in known:
+        if vid in present or vid not in known or vid in staying_out:
+            left_out += 1
             continue
-        row = PlaylistItem(playlist_id=pl.id, video_id=vid, position=0)
-        db.add(row)
-        rows.insert(min(at, len(rows)), row)
+        row = leaving.pop(vid, None)
+        if row is None:
+            row = PlaylistItem(playlist_id=pl.id, video_id=vid, position=0)
+            db.add(row)
+        else:
+            row.playlist_id = pl.id
+        rows.insert(min(at - left_out, len(rows)), row)
         present.add(vid)
         added += 1
     for pos, row in enumerate(rows):
@@ -496,6 +486,24 @@ def restore_items(db: Session, pl: Playlist, video_ids: list[str], positions: li
     return added
 
 
+class ListGone(Exception):
+    """A list of a two-list write was deleted between the route's read and the lock."""
+
+    def __init__(self, *, source: bool):
+        super().__init__("source" if source else "target")
+        self.source = source
+
+
+def _lock_both(db: Session, src: Playlist, dst: Playlist) -> None:
+    """Both lists, in id order: a move A→B beside a move B→A would otherwise lock crosswise. A list
+    already gone is said here: a write into it would answer with its foreign key's 500."""
+    found = lock_playlists(db, src.id, dst.id)
+    if src.id not in found:
+        raise ListGone(source=True)
+    if dst.id not in found:
+        raise ListGone(source=False)
+
+
 def move_items(
     db: Session, src: Playlist, dst: Playlist, video_ids: list[str]
 ) -> tuple[list[str], list[str]]:
@@ -506,8 +514,7 @@ def move_items(
     list twice."""
     if src.id == dst.id:
         raise MoveRefused("same_playlist")
-    # Both lists, in id order: a move A→B beside a move B→A would otherwise lock crosswise.
-    lock_playlists(db, src.id, dst.id)
+    _lock_both(db, src, dst)
     wanted = set(video_ids)
     rows = [it for it in queries.items(db, src.id) if it.video_id in wanted]
     held = queries.in_playlist(db, dst.id, [r.video_id for r in rows])
@@ -547,17 +554,23 @@ def undo_move(
     already_there: list[str],
     positions: list[int],
 ) -> int:
-    """Take a move back in one transaction: the `moved` videos leave `dst` and every video returns to
-    its place in `src`, while the `already_there` ones stay in `dst`, which held them before.
-    `positions` pairs with `moved` + `already_there`. Says how many came back."""
-    lock_playlists(db, src.id, dst.id)
+    """Take a move back in one transaction: the `moved` videos leave `dst` and return to their places
+    in `src`, keeping their rows, as the move did. The `already_there` ones return too and stay in
+    `dst`, which held them before. A moved video `dst` no longer holds has gone on from there, and
+    bringing it back would leave it in lists the reader never put it in, so it stays out, as a
+    move skips a video a concurrent edit took out. `positions` pairs with `moved` + `already_there`.
+    Says how many came back."""
+    _lock_both(db, src, dst)
     wanted = set(moved)
-    rows = [it for it in queries.items(db, dst.id) if it.video_id in wanted]
-    for row in rows:
+    rows = {it.video_id: it for it in queries.items(db, dst.id) if it.video_id in wanted}
+    changed = bool(rows)
+    placed = list(zip(positions, [*moved, *already_there], strict=True))
+    back = _insert_at(db, src, placed, leaving=rows, staying_out=frozenset(wanted - rows.keys()))
+    # What `src` holds again by now only leaves `dst`.
+    for row in rows.values():
         db.delete(row)
     db.flush()
-    back = _insert_at(db, src, list(zip(positions, [*moved, *already_there], strict=True)))
-    if rows:
+    if changed:
         recompute_dirty(db, dst)
         touch(dst)
     if back:

backend/app/routes/playlists.py

@@ -6,7 +6,7 @@ out (`/{id}/push-plan`, `/{id}/revert-youtube`), behind the optional write scope
 from typing import TYPE_CHECKING, Literal
 
 from fastapi import APIRouter, Depends, HTTPException
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, model_validator
 from sqlalchemy.orm import Session
 
 from app import quota
@@ -386,24 +386,17 @@ def delete_playlist(
     return {"deleted": playlist_id}
 
 
-class AddItemIn(VideoRefIn):
-    # Where the video lands among the list's items, in their stored order (0 = first). Omitted, it
-    # is appended. The list page's "Undo" of a removal puts the row back where it was with this —
-    # ONE request, because append-then-reorder leaves it at the end whenever the second call fails.
-    position: int | None = Field(default=None, ge=0)
-
-
 @router.post("/{playlist_id}/items")
 def add_item(
     playlist_id: int,
-    body: AddItemIn,
+    body: VideoRefIn,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pl = _own_playlist(db, user, playlist_id)
+    pl = _own_playlist(db, user, playlist_id, lock=True)
     video_id = body.video_id
     _video_or_404(db, video_id)
-    service.add_to_playlist(db, pl, video_id, body.position)
+    service.add_item(db, pl, video_id, mark_dirty=True)
     db.commit()
     return {"ok": True, "playlist_id": pl.id, "video_id": video_id}
 
@@ -415,14 +408,15 @@ def remove_item(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pl = _own_playlist(db, user, playlist_id)
+    pl = _own_playlist(db, user, playlist_id, lock=True)
     service.remove_item(db, pl, video_id, mark_dirty=True)
     db.commit()
     return {"ok": True, "playlist_id": pl.id, "video_id": video_id}
 
 
 class VideoIdsIn(BaseModel):
-    video_ids: list[str] = []
+    # Required: with a default, a misnamed key answered 200 having acted on no video.
+    video_ids: list[str]
 
 
 def _paired(video_count: int, positions: list[int]) -> None:
@@ -460,6 +454,11 @@ class UndoMoveIn(BaseModel):
         return self
 
 
+def _list_gone(e: service.ListGone) -> HTTPException:
+    """Each list of a two-list write under the reason the route's own read gives it."""
+    return not_found("playlist") if e.source else not_found("move_target")
+
+
 def _other_list(db: Session, user: User, playlist_id: int) -> "Playlist":
     """The other list of a move. Its own reason: "this playlist doesn't exist any more" about a list
     that is gone was read as the one on screen, which is still there."""
@@ -476,8 +475,8 @@ def remove_items(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    """Remove many videos from a playlist at once. Returns how many went."""
-    pl = _own_playlist(db, user, playlist_id)
+    """Remove many videos from a playlist at once. Returns the ones that went."""
+    pl = _own_playlist(db, user, playlist_id, lock=True)
     removed = service.remove_items(db, pl, body.video_ids)
     db.commit()
     return {"ok": True, "playlist_id": pl.id, "removed": removed}
@@ -491,7 +490,7 @@ def restore_items(
     db: Session = Depends(get_db),
 ) -> dict:
     """The "Undo" of a removal: the videos back at their places. Returns how many came back."""
-    pl = _own_playlist(db, user, playlist_id)
+    pl = _own_playlist(db, user, playlist_id, lock=True)
     added = service.restore_items(db, pl, body.video_ids, body.positions)
     db.commit()
     return {"ok": True, "playlist_id": pl.id, "added": added}
@@ -519,6 +518,8 @@ def move_items(
         raise HTTPException(
             status_code=400, detail={"code": "playlist_move", "reason": e.reason}
         ) from None
+    except service.ListGone as e:
+        raise _list_gone(e) from None
     db.commit()
     return {
         "ok": True,
@@ -539,19 +540,18 @@ def undo_move(
     """The "Undo" of a move, back into this playlist in one transaction. Returns how many came back."""
     src = _own_playlist(db, user, playlist_id)
     dst = _other_list(db, user, body.from_playlist_id)
-    added = service.undo_move(db, src, dst, body.moved, body.already_there, body.positions)
+    try:
+        added = service.undo_move(db, src, dst, body.moved, body.already_there, body.positions)
+    except service.ListGone as e:
+        raise _list_gone(e) from None
     db.commit()
     return {"ok": True, "playlist_id": src.id, "added": added}
 
 
-class ReorderItemsIn(BaseModel):
-    video_ids: list[str] = []
-
-
 @router.put("/{playlist_id}/order")
 def reorder_items(
     playlist_id: int,
-    body: ReorderItemsIn,
+    body: VideoIdsIn,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:

backend/tests/test_playlist_add_position.py

@@ -1,4 +1,4 @@
-"""Adding a video AT a position: the list page's "Undo" of a removal.
+"""Putting a video back AT a position: the list page's "Undo" of a removal.
 
 The row has to come back where it was in ONE request. The two-call shape (append, then send the
 whole order) leaves it at the end of the list whenever the second call fails, and the reorder would
@@ -38,14 +38,17 @@ def _ids(db, pl_id):
     ]
 
 
-def _add(client, pl_id, body):
-    return client.post(f"/api/playlists/{pl_id}/items", json=body)
+def _restore(client, pl_id, video_id, position):
+    return client.post(
+        f"/api/playlists/{pl_id}/items/restore",
+        json={"video_ids": [video_id], "positions": [position]},
+    )
 
 
 def test_a_position_puts_the_video_back_in_its_place(db, make_user, as_user):
     user = make_user()
     pl = _seed(db, user, ["a", "b", "c"])
-    r = _add(as_user(user), pl.id, {"video_id": "vX", "position": 1})
+    r = _restore(as_user(user), pl.id, "vX", 1)
     assert r.status_code == 200
     assert _ids(db, pl.id) == ["a", "vX", "b", "c"]
 
@@ -54,36 +57,37 @@ def test_the_position_counts_rows_not_stored_numbers(db, make_user, as_user):
     # A removal leaves a gap (0, 5, 9): index 2 is before the THIRD row, not "position 2".
     user = make_user()
     pl = _seed(db, user, ["a", "b", "c"], positions=[0, 5, 9])
-    _add(as_user(user), pl.id, {"video_id": "vX", "position": 2})
+    _restore(as_user(user), pl.id, "vX", 2)
     assert _ids(db, pl.id) == ["a", "b", "vX", "c"]
 
 
 def test_a_position_past_the_end_appends(db, make_user, as_user):
     user = make_user()
     pl = _seed(db, user, ["a", "b"])
-    _add(as_user(user), pl.id, {"video_id": "vX", "position": 40})
+    _restore(as_user(user), pl.id, "vX", 40)
     assert _ids(db, pl.id) == ["a", "b", "vX"]
 
 
-def test_no_position_still_appends(db, make_user, as_user):
+def test_an_add_appends(db, make_user, as_user):
     user = make_user()
     pl = _seed(db, user, ["a", "b"])
-    _add(as_user(user), pl.id, {"video_id": "vX"})
+    r = as_user(user).post(f"/api/playlists/{pl.id}/items", json={"video_id": "vX"})
+    assert r.status_code == 200
     assert _ids(db, pl.id) == ["a", "b", "vX"]
 
 
 def test_a_video_already_in_the_list_does_not_move(db, make_user, as_user):
-    # The add is idempotent; a position must not turn it into a reorder.
+    # A restore is idempotent; a position must not turn it into a reorder.
     user = make_user()
     pl = _seed(db, user, ["a", "b", "c"])
-    _add(as_user(user), pl.id, {"video_id": "c", "position": 0})
+    _restore(as_user(user), pl.id, "c", 0)
     assert _ids(db, pl.id) == ["a", "b", "c"]
 
 
 def test_a_negative_position_is_refused(db, make_user, as_user):
     user = make_user()
     pl = _seed(db, user, ["a"])
-    r = _add(as_user(user), pl.id, {"video_id": "vX", "position": -1})
+    r = _restore(as_user(user), pl.id, "vX", -1)
     assert r.status_code == 422
     assert _ids(db, pl.id) == ["a"]
 
@@ -101,6 +105,6 @@ def test_putting_a_removed_video_back_leaves_a_linked_list_clean(db, make_user,
     assert client.delete(f"/api/playlists/{pl.id}/items/b").status_code == 200
     db.expire_all()
     assert db.get(Playlist, pl.id).dirty is True
-    _add(client, pl.id, {"video_id": "b", "position": 1})
+    _restore(client, pl.id, "b", 1)
     db.expire_all()
     assert db.get(Playlist, pl.id).dirty is False

backend/tests/test_playlist_bulk.py

@@ -7,8 +7,9 @@ import pytest
 from sqlalchemy import select
 
 from app.catalog.models import Channel, Video
-from app.reader import service
+from app.reader import queries, service
 from app.reader.models import Playlist, PlaylistItem
+from app.sync.playlists import current_fingerprint
 
 
 def _seed_videos(db, ids):
@@ -61,7 +62,7 @@ def test_remove_bulk_takes_out_the_named_videos_and_counts_what_was_there(db, ma
     )
 
     assert r.status_code == 200
-    assert r.json() == {"ok": True, "playlist_id": pl.id, "removed": 2}
+    assert r.json() == {"ok": True, "playlist_id": pl.id, "removed": ["b", "d"]}
     assert _ids(db, pl.id) == ["a", "c"]
 
 
@@ -77,7 +78,7 @@ def test_remove_bulk_of_nothing_leaves_the_list_untouched(db, make_user, as_user
         f"/api/playlists/{pl.id}/items/remove-bulk", json={"video_ids": ["gone"]}
     )
 
-    assert r.json()["removed"] == 0
+    assert r.json()["removed"] == []
     db.expire_all()
     assert db.get(Playlist, pl.id).updated_at == stamp
 
@@ -276,6 +277,238 @@ def test_undo_move_restores_both_lists_as_they_stood(db, make_user, as_user):
     assert _ids(db, dst.id) == ["c"]
 
 
+def test_undo_move_keeps_when_each_video_was_saved(db, make_user, as_user):
+    user = make_user()
+    _seed_videos(db, ["a"])
+    src = _playlist(db, user, "Source", [])
+    dst = _playlist(db, user, "Target", ["a"])
+    saved = datetime(2020, 1, 2, tzinfo=timezone.utc)
+    row = db.scalar(select(PlaylistItem).where(PlaylistItem.playlist_id == dst.id))
+    row.added_at = saved
+    db.commit()
+
+    r = as_user(user).post(
+        f"/api/playlists/{src.id}/items/undo-move",
+        json={"from_playlist_id": dst.id, "moved": ["a"], "positions": [0]},
+    )
+
+    assert r.json()["added"] == 1
+    db.expire_all()
+    back = db.scalar(select(PlaylistItem).where(PlaylistItem.playlist_id == src.id))
+    assert back.added_at == saved
+
+
+def test_undo_move_leaves_a_video_that_moved_on_where_it_is(db, make_user, as_user):
+    """A to B, then B to C, then the first Undo: the video is not in the list the Undo takes it from,
+    and coming back would leave it in two lists the reader never put it in."""
+    user = make_user()
+    _seed_videos(db, ["a", "b"])
+    src = _playlist(db, user, "Source", [])
+    dst = _playlist(db, user, "Target", ["b"])
+    third = _playlist(db, user, "Third", ["a"])
+
+    r = as_user(user).post(
+        f"/api/playlists/{src.id}/items/undo-move",
+        json={"from_playlist_id": dst.id, "moved": ["a", "b"], "positions": [0, 1]},
+    )
+
+    assert r.json()["added"] == 1
+    assert _ids(db, src.id) == ["b"]
+    assert _ids(db, dst.id) == []
+    assert _ids(db, third.id) == ["a"]
+
+
+def test_an_undo_closes_up_the_places_of_a_video_that_stayed_out(db, make_user, as_user):
+    """Each index counts the rows put back before it, so one that stays out moves the rest up."""
+    user = make_user()
+    _seed_videos(db, ["a", "b", "c", "d"])
+    src = _playlist(db, user, "Source", ["c", "d"])
+    dst = _playlist(db, user, "Target", ["b"])
+
+    r = as_user(user).post(
+        f"/api/playlists/{src.id}/items/undo-move",
+        json={"from_playlist_id": dst.id, "moved": ["a", "b"], "positions": [0, 1]},
+    )
+
+    assert r.json()["added"] == 1
+    assert _ids(db, src.id) == ["b", "c", "d"]
+
+
+def test_a_restore_closes_up_the_places_of_a_video_the_catalog_lost(db, make_user, as_user):
+    user = make_user()
+    _seed_videos(db, ["b", "c", "d"])
+    pl = _playlist(db, user, "L", ["c", "d"])
+
+    r = as_user(user).post(
+        f"/api/playlists/{pl.id}/items/restore",
+        json={"video_ids": ["lost", "b"], "positions": [0, 1]},
+    )
+
+    assert r.json()["added"] == 1
+    assert _ids(db, pl.id) == ["b", "c", "d"]
+
+
+def test_undo_move_takes_a_video_out_of_the_target_when_the_source_has_it_again(
+    db, make_user, as_user
+):
+    user = make_user()
+    _seed_videos(db, ["a"])
+    src = _playlist(db, user, "Source", ["a"])
+    dst = _playlist(db, user, "Target", ["a"])
+
+    r = as_user(user).post(
+        f"/api/playlists/{src.id}/items/undo-move",
+        json={"from_playlist_id": dst.id, "moved": ["a"], "positions": [0]},
+    )
+
+    assert r.json()["added"] == 0
+    assert _ids(db, src.id) == ["a"]
+    assert _ids(db, dst.id) == []
+
+
+@pytest.mark.parametrize(
+    ("path", "body", "gone", "reason"),
+    [
+        ("move-bulk", {"video_ids": ["a"], "target_playlist_id": "dst"}, "dst", "move_target"),
+        ("move-bulk", {"video_ids": ["a"], "target_playlist_id": "dst"}, "src", "playlist"),
+        (
+            "undo-move",
+            {"from_playlist_id": "dst", "moved": ["a"], "positions": [0]},
+            "src",
+            "playlist",
+        ),
+        (
+            "undo-move",
+            {"from_playlist_id": "dst", "moved": ["a"], "positions": [0]},
+            "dst",
+            "move_target",
+        ),
+    ],
+)
+def test_a_list_deleted_before_the_lock_is_not_found(
+    db, make_user, as_user, monkeypatch, path, body, gone, reason
+):
+    user = make_user()
+    _seed_videos(db, ["a"])
+    src = _playlist(db, user, "Source", ["a"])
+    dst = _playlist(db, user, "Target", ["a"])
+    ids = {"src": src.id, "dst": dst.id}
+    body = {k: (dst.id if v == "dst" else v) for k, v in body.items()}
+    # Another tab's delete committed between the route's read and this lock.
+    monkeypatch.setattr(service, "lock_playlists", lambda _db, *got: set(got) - {ids[gone]})
+
+    r = as_user(user).post(f"/api/playlists/{src.id}/items/{path}", json=body)
+
+    assert r.status_code == 404
+    assert reason in str(r.json()["detail"])
+
+
+@pytest.mark.parametrize(
+    ("method", "path", "body"),
+    [
+        ("post", "items", {"video_id": "a"}),
+        ("delete", "items/a", None),
+        ("post", "items/remove-bulk", {"video_ids": ["a"]}),
+        ("post", "items/restore", {"video_ids": ["a"], "positions": [0]}),
+    ],
+)
+def test_a_one_list_write_finds_its_list_under_the_lock(
+    db, make_user, as_user, monkeypatch, method, path, body
+):
+    """Read with its lock, so a delete that won the row answers 404 rather than a foreign key's 500."""
+    user = make_user()
+    _seed_videos(db, ["a"])
+    pl = _playlist(db, user, "L", [])
+    real = queries.own_playlist
+
+    def locked_read_finds_nothing(db_, user_, playlist_id, *, lock=False):
+        return None if lock else real(db_, user_, playlist_id)
+
+    monkeypatch.setattr(queries, "own_playlist", locked_read_finds_nothing)
+    kw = {"json": body} if body is not None else {}
+
+    r = getattr(as_user(user), method)(f"/api/playlists/{pl.id}/{path}", **kw)
+
+    assert r.status_code == 404
+
+
+@pytest.mark.parametrize(
+    ("path", "body"),
+    [
+        ("items/remove-bulk", {"ids": ["a"]}),
+        ("items/move-bulk", {"ids": ["a"], "target_playlist_id": 1}),
+    ],
+)
+def test_a_body_without_its_videos_is_refused_rather_than_acting_on_none(
+    db, make_user, as_user, path, body
+):
+    user = make_user()
+    pl = _playlist(db, user, "L", [])
+
+    assert as_user(user).post(f"/api/playlists/{pl.id}/{path}", json=body).status_code == 422
+
+
+def test_a_reorder_without_its_videos_is_refused(db, make_user, as_user):
+    user = make_user()
+    pl = _playlist(db, user, "L", [])
+
+    assert as_user(user).put(f"/api/playlists/{pl.id}/order", json={"ids": []}).status_code == 422
+
+
+def _synced(db, pl):
+    pl.synced_fingerprint = current_fingerprint(db, pl)
+    pl.dirty = False
+    db.commit()
+
+
+def test_a_bulk_remove_and_its_undo_leave_a_linked_list_as_it_was(db, make_user, as_user):
+    user = make_user()
+    _seed_videos(db, ["a", "b"])
+    pl = _playlist(db, user, "L", ["a", "b"], yt_playlist_id="PLx")
+    _synced(db, pl)
+    client = as_user(user)
+
+    r = client.post(f"/api/playlists/{pl.id}/items/remove-bulk", json={"video_ids": ["a"]})
+    assert r.status_code == 200
+    db.expire_all()
+    assert db.get(Playlist, pl.id).dirty is True
+
+    r = client.post(
+        f"/api/playlists/{pl.id}/items/restore", json={"video_ids": ["a"], "positions": [0]}
+    )
+    assert r.status_code == 200
+    db.expire_all()
+    assert db.get(Playlist, pl.id).dirty is False
+
+
+def test_a_move_and_its_undo_leave_both_linked_lists_as_they_were(db, make_user, as_user):
+    user = make_user()
+    _seed_videos(db, ["a", "b"])
+    src = _playlist(db, user, "Source", ["a", "b"], yt_playlist_id="PLs")
+    dst = _playlist(db, user, "Target", ["b"], yt_playlist_id="PLd")
+    _synced(db, src)
+    _synced(db, dst)
+    client = as_user(user)
+
+    moved = client.post(
+        f"/api/playlists/{src.id}/items/move-bulk",
+        json={"video_ids": ["a"], "target_playlist_id": dst.id},
+    ).json()
+    assert moved["moved"] == ["a"]
+    db.expire_all()
+    assert db.get(Playlist, src.id).dirty is True
+    assert db.get(Playlist, dst.id).dirty is True
+
+    r = client.post(
+        f"/api/playlists/{src.id}/items/undo-move",
+        json={"from_playlist_id": dst.id, "moved": ["a"], "positions": [0]},
+    )
+    assert r.status_code == 200
+    db.expire_all()
+    assert db.get(Playlist, src.id).dirty is False
+    assert db.get(Playlist, dst.id).dirty is False
+
+
 def test_undo_move_is_one_transaction(db, make_user, as_user, monkeypatch):
     user = make_user()
     _seed_videos(db, ["a"])

backend/tests/test_playlist_position_lock.py

@@ -102,17 +102,24 @@ PLEX = PlexPlaylist.__table__.name
     "call",
     [
         lambda c, pid, other: c.post(f"/api/playlists/{pid}/items", json={"video_id": "v2"}),
-        lambda c, pid, other: c.post(
-            f"/api/playlists/{pid}/items", json={"video_id": "v2", "position": 0}
-        ),
         lambda c, pid, other: c.post("/api/playlists/watch-later", json={"video_id": "v2"}),
         lambda c, pid, other: c.post(
             f"/api/playlists/{pid}/items/move-bulk",
             json={"video_ids": ["v1"], "target_playlist_id": other},
         ),
         lambda c, pid, other: c.put(f"/api/playlists/{pid}/order", json={"video_ids": ["v1"]}),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items/remove-bulk", json={"video_ids": ["v1"]}
+        ),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items/restore", json={"video_ids": ["v2"], "positions": [0]}
+        ),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items/undo-move",
+            json={"from_playlist_id": other, "moved": ["v2"], "positions": [0]},
+        ),
     ],
-    ids=["add", "add-at", "watch-later", "move", "reorder"],
+    ids=["add", "watch-later", "move", "reorder", "remove-bulk", "restore", "undo-move"],
 )
 def test_a_youtube_playlist_write_locks_the_playlist(db, make_user, as_user, locks, call):
     user = make_user()
@@ -169,9 +176,6 @@ PLEX_ITEMS = PlexPlaylistItem.__table__.name
     "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/move-bulk",
@@ -180,8 +184,27 @@ PLEX_ITEMS = PlexPlaylistItem.__table__.name
         lambda c, pid, other: c.put(f"/api/playlists/{pid}/order", json={"video_ids": ["v1"]}),
         # the ORM cascade deletes the items before the list row
         lambda c, pid, other: c.delete(f"/api/playlists/{pid}"),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items/remove-bulk", json={"video_ids": ["v1"]}
+        ),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items/restore", json={"video_ids": ["v2"], "positions": [0]}
+        ),
+        lambda c, pid, other: c.post(
+            f"/api/playlists/{pid}/items/undo-move",
+            json={"from_playlist_id": other, "moved": ["v2"], "positions": [0]},
+        ),
+    ],
+    ids=[
+        "add-present",
+        "remove",
+        "move",
+        "reorder",
+        "delete",
+        "remove-bulk",
+        "restore",
+        "undo-move",
     ],
-    ids=["add-present", "add-at", "remove", "move", "reorder", "delete"],
 )
 def test_a_youtube_item_write_locks_the_playlist_before_any_item(
     db, make_user, as_user, statements, call
@@ -195,15 +218,23 @@ def test_a_youtube_item_write_locks_the_playlist_before_any_item(
     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."""
+@pytest.mark.parametrize(
+    ("path", "body"),
+    [
+        ("move-bulk", lambda other: {"video_ids": ["v1"], "target_playlist_id": other}),
+        ("undo-move", lambda other: {"from_playlist_id": other, "moved": ["v2"], "positions": [0]}),
+    ],
+    ids=["move", "undo"],
+)
+def test_a_move_locks_both_lists_in_one_ordered_statement(
+    db, make_user, as_user, statements, path, body
+):
+    """Two moves A→B and B→A each locking "its" list first would lock crosswise, and an Undo runs the
+    other way round from a move by its nature."""
     user = make_user()
     pid, other = _yt(db, user)
     statements.clear()
-    r = as_user(user).post(
-        f"/api/playlists/{pid}/items/move-bulk",
-        json={"video_ids": ["v1"], "target_playlist_id": other},
-    )
+    r = as_user(user).post(f"/api/playlists/{pid}/items/{path}", json=body(other))
     assert r.status_code == 200, r.text
     lock = _locked_before_items(statements, YT, YT_ITEMS)
     assert lock and "ORDER BY" in lock, lock

backend/tests/test_reader_service.py

@@ -210,9 +210,8 @@ def test_an_add_is_idempotent_and_lands_where_asked(db, make_user):
     pl = service.create_playlist(db, user, "L")
     assert service.add_item(db, pl, "a", mark_dirty=False) is True
     assert service.add_item(db, pl, "a", mark_dirty=False) is False
-    service.add_to_playlist(db, pl, "b", None)
-    service.add_to_playlist(db, pl, "c", 0)
-    service.add_to_playlist(db, pl, "a", 2)  # already there: stays put
+    assert service.add_item(db, pl, "b", mark_dirty=True) is True
+    assert service.restore_items(db, pl, ["c", "a"], [0, 2]) == 1  # "a" is there: it stays put
     assert _uncommitted(db)
     db.commit()
     assert _ids(db, pl) == ["c", "a", "b"]