"""The near-end checkpoint that finishes a Plex item.

- Exactly one scrobble per finish, even when two near-end checkpoints race (the 10 s tick and the
  debounced `seeked` flush arrive together): the transition is decided by a conditional write, not a
  read taken before it.
- The finish supersedes every resume push still in flight, so none can land at Plex after the
  scrobble and revive a resume point on a watched item.
"""

import threading
from datetime import datetime, timedelta, timezone

from sqlalchemy import select

from app.db import SessionLocal
from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState
from app.plex import service as plex_service
from app.routes.plex import playback


def _item(db) -> PlexItem:
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    item = PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="A Movie")
    db.add(item)
    db.commit()
    return item


def test_two_racing_finishes_flip_the_row_once(db, make_user):
    user = make_user()
    item = _item(db)
    db.add(PlexState(user_id=user.id, item_id=item.id, status="new", position_seconds=5000))
    db.commit()
    now = datetime.now(timezone.utc)

    first, second = SessionLocal(), SessionLocal()
    try:
        assert (
            plex_service.mark_finished(first, user.id, item.id, now) is True
        )  # holds the row lock
        out: list[bool] = []
        racer = threading.Thread(
            target=lambda: out.append(plex_service.mark_finished(second, user.id, item.id, now))
        )
        racer.start()
        racer.join(0.5)
        assert racer.is_alive(), "the second writer must wait on the first one's row lock"
        first.commit()
        racer.join(5)
        second.commit()
    finally:
        first.close()
        second.close()
    assert out == [False], "the second checkpoint saw the committed finish: no second scrobble"


def test_a_first_ever_finish_inserts_and_counts_once(db, make_user):
    user = make_user()
    item = _item(db)
    now = datetime.now(timezone.utc)
    assert plex_service.mark_finished(db, user.id, item.id, now) is True
    db.commit()
    assert plex_service.mark_finished(db, user.id, item.id, now + timedelta(seconds=10)) is False
    db.commit()
    st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one()
    assert (st.status, st.position_seconds) == ("watched", 0)


def test_the_finish_supersedes_a_resume_push_still_in_flight(db, as_user, make_user, monkeypatch):
    user = make_user()
    item = _item(db)
    monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None)
    earlier = datetime.now(timezone.utc) - timedelta(seconds=5)
    # A resume push was scheduled a moment ago and has not reached Plex yet.
    assert playback._should_push_resume(user.id, item.id, earlier, final=True)

    r = as_user(user).post(
        "/api/plex/item/m1/progress", json={"position_seconds": 2995, "duration_seconds": 3000}
    )
    assert r.json() == {"ok": True, "status": "watched", "position_seconds": 0}
    assert playback._resume_push_superseded(user.id, item.id, earlier) is True


def test_a_checkpoint_sent_before_the_finish_does_not_land_on_the_watched_row(
    db, as_user, make_user, monkeypatch
):
    """The 10 s tick posted at 2699 of 3000 s (not near the end) races the credits auto-skip's
    "watched". When the mark commits first, the late checkpoint used to write its position onto the
    watched row and schedule a resume push after the scrobble: Plex showed the item in progress, and
    a Plex-authoritative pull imported it back."""
    user = make_user()
    item = _item(db)
    monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None)
    client = as_user(user)
    assert client.post("/api/plex/item/m1/state", json={"status": "watched"}).status_code == 200

    r = client.post(
        "/api/plex/item/m1/progress", json={"position_seconds": 2699, "duration_seconds": 3000}
    )
    assert r.status_code == 200, r.text
    assert r.json() == {"ok": True, "status": "watched", "position_seconds": 0}
    db.expire_all()
    st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one()
    assert (st.status, st.position_seconds) == ("watched", 0)


def test_a_rewatch_records_its_position_once_the_finish_is_behind_it(
    db, as_user, make_user, monkeypatch
):
    """The other half: a watched item played again is still checkpointed — only a position arriving
    right on the heels of the mark is read as that mark's late tick."""
    user = make_user()
    item = _item(db)
    long_ago = datetime.now(timezone.utc) - timedelta(days=3)
    db.add(PlexState(user_id=user.id, item_id=item.id, status="watched", watched_at=long_ago))
    db.commit()
    monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None)
    r = as_user(user).post(
        "/api/plex/item/m1/progress", json={"position_seconds": 600, "duration_seconds": 3000}
    )
    assert r.json() == {"ok": True, "position_seconds": 600}
    db.expire_all()
    st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one()
    assert (st.status, st.position_seconds) == ("watched", 600)


def test_marking_watched_supersedes_a_resume_push_still_in_flight(
    db, as_user, make_user, monkeypatch
):
    """The same guard the near-end finish has: a resume push scheduled before the mark must not reach
    Plex after its scrobble."""
    user = make_user()
    item = _item(db)
    earlier = datetime.now(timezone.utc) - timedelta(seconds=5)
    assert playback._should_push_resume(user.id, item.id, earlier, final=True)
    r = as_user(user).post("/api/plex/item/m1/state", json={"status": "watched"})
    assert r.status_code == 200, r.text
    assert playback._resume_push_superseded(user.id, item.id, earlier) is True


def _state(db, user, item):
    db.expire_all()
    st = (
        db.execute(
            select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == item.id)
        )
        .scalars()
        .one()
    )
    return st.status, st.position_seconds


def test_a_mark_committed_after_the_checkpoints_read_still_refuses_it(db, make_user):
    """The race itself, not its sequential shadow: the route read the row while it was still in
    progress, and the finish committed from another session before the checkpoint wrote. Only a write
    conditional on the row as it is NOW refuses that position."""
    user = make_user()
    item = _item(db)
    db.add(PlexState(user_id=user.id, item_id=item.id, status="new", position_seconds=2690))
    db.commit()
    now = datetime.now(timezone.utc)
    read = plex_service.state_row(db, user, item.id)  # what the checkpoint's route read: "new"
    assert read.status == "new"

    other = SessionLocal()
    try:
        assert plex_service.mark_finished(other, user.id, item.id, now) is True
        other.commit()
    finally:
        other.close()

    assert plex_service.checkpoint(db, user, item.id, read, 2699, now, dirty=True) is False
    db.commit()
    assert _state(db, user, item) == ("watched", 0)


def test_a_tick_seconds_behind_the_mark_is_refused_a_rewatch_a_minute_later_is_not(db, make_user):
    """A position right behind the mark is that mark's late tick; one a minute later is a rewatch.
    Literal times on purpose: a window widened to an hour must fail here, not move with it."""
    user = make_user()
    item = _item(db)
    now = datetime.now(timezone.utc)
    st = PlexState(user_id=user.id, item_id=item.id, status="watched", watched_at=now)
    db.add(st)
    db.commit()

    st.watched_at = now - timedelta(seconds=20)
    db.commit()
    assert plex_service.checkpoint(db, user, item.id, st, 600, now, dirty=True) is False
    db.commit()
    st.watched_at = now - timedelta(seconds=60)
    db.commit()
    assert plex_service.checkpoint(db, user, item.id, st, 600, now, dirty=True) is True
    db.commit()
    assert _state(db, user, item) == ("watched", 600)


def test_marking_a_season_watched_supersedes_its_episodes_resume_pushes(db, as_user, make_user):
    """The bulk mark has the same guard as one item's: an episode's resume push scheduled before
    the season was marked must not reach Plex after the scrobble."""
    user = make_user()
    lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
    db.add(lib)
    db.flush()
    sh = PlexShow(rating_key="sh1", library_id=lib.id, title="Show")
    db.add(sh)
    db.flush()
    se = PlexSeason(rating_key="se1", show_id=sh.id, season_number=1)
    db.add(se)
    db.flush()
    ep = PlexItem(
        rating_key="e1",
        library_id=lib.id,
        kind="episode",
        title="E1",
        show_id=sh.id,
        season_id=se.id,
        season_number=1,
        episode_number=1,
    )
    db.add(ep)
    db.commit()
    earlier = datetime.now(timezone.utc) - timedelta(seconds=5)
    assert playback._should_push_resume(user.id, ep.id, earlier, final=True)

    r = as_user(user).post("/api/plex/season/se1/state", json={"watched": True})
    assert r.json() == {"ok": True, "changed": 1, "watched": True}
    assert playback._resume_push_superseded(user.id, ep.id, earlier) is True