"""E28 review — every request that reads a playlist's item positions and writes new ones locks the
playlist row first.

Without it two adds (two tabs; the rail and the page) both read the same `max(position)` and land
on one position, and a reorder racing an add collides with it — the list's order between the two
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.
"""

import re
from datetime import datetime, timezone

import pytest
from sqlalchemy import event

from app.models import (
    Channel,
    Playlist,
    PlaylistItem,
    PlexItem,
    PlexLibrary,
    PlexPlaylist,
    PlexPlaylistItem,
    Video,
)


@pytest.fixture
def locks(db):
    """The tables each request took a row lock on (`SELECT … FROM <t> … FOR UPDATE`)."""
    seen: list[str] = []

    def _record(conn, cursor, statement, *_):
        if "FOR UPDATE" in statement:
            seen.extend(re.findall(r"FROM\s+(\w+)", statement)[:1])

    engine = db.get_bind()
    event.listen(engine, "before_cursor_execute", _record)
    yield seen
    event.remove(engine, "before_cursor_execute", _record)


def _yt(db, user):
    db.add(Channel(id="UCtest", title="Test"))
    now = datetime.now(timezone.utc)
    for v in ("v1", "v2", "v3"):
        db.add(Video(id=v, channel_id="UCtest", published_at=now))
    pl = Playlist(user_id=user.id, name="L")
    other = Playlist(user_id=user.id, name="M")
    db.add_all([pl, other])
    db.flush()
    db.add(PlaylistItem(playlist_id=pl.id, video_id="v1", position=0))
    db.commit()
    return pl.id, other.id


def _plex(db, user):
    lib = PlexLibrary(plex_key="p1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    items = [PlexItem(rating_key=rk, library_id=lib.id, kind="movie", title=rk) for rk in "abcd"]
    db.add_all(items)
    pl = PlexPlaylist(user_id=user.id, title="L")
    db.add(pl)
    db.flush()
    for i, it in enumerate(items[:3]):
        db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=i))
    db.commit()
    return pl.id


YT = Playlist.__table__.name
PLEX = PlexPlaylist.__table__.name


@pytest.mark.parametrize(
    "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/v1/move", json={"target_playlist_id": other}
        ),
        lambda c, pid, other: c.put(f"/api/playlists/{pid}/order", json={"video_ids": ["v1"]}),
    ],
    ids=["add", "add-at", "watch-later", "move", "reorder"],
)
def test_a_youtube_playlist_write_locks_the_playlist(db, make_user, as_user, locks, call):
    user = make_user()
    pid, other = _yt(db, user)
    r = call(as_user(user), pid, other)
    assert r.status_code == 200, r.text
    assert YT in locks


@pytest.mark.parametrize(
    "call",
    [
        lambda c, pid: c.post(f"/api/plex/playlists/{pid}/items", json={"item_rating_key": "d"}),
        lambda c, pid: c.post(
            f"/api/plex/playlists/{pid}/items/bulk", json={"item_rating_keys": ["d"]}
        ),
        lambda c, pid: c.post(
            f"/api/plex/playlists/{pid}/items/bulk",
            json={"item_rating_keys": ["d"], "positions": [0]},
        ),
        lambda c, pid: c.put(
            f"/api/plex/playlists/{pid}/order", json={"item_rating_keys": ["c", "b", "a"]}
        ),
    ],
    ids=["add", "bulk", "bulk-at", "reorder"],
)
def test_a_plex_playlist_write_locks_the_playlist(db, make_user, as_user, locks, call):
    user = make_user()
    pid = _plex(db, user)
    r = call(as_user(user), pid)
    assert r.status_code == 200, r.text
    assert PLEX in locks


def test_a_plex_reorder_that_omits_a_row_does_not_collide_with_it(db, make_user, as_user):
    """The row "a" was added in another tab after this page loaded its order ["c", "b"]: it keeps
    its place after the given rows instead of sharing position 0 with "c"."""
    user = make_user()
    pid = _plex(db, user)
    r = as_user(user).put(f"/api/plex/playlists/{pid}/order", json={"item_rating_keys": ["c", "b"]})
    assert r.status_code == 200
    db.expire_all()
    rows = db.query(PlexPlaylistItem).filter_by(playlist_id=pid).all()
    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")]