"""Tearing a session down does not stall everybody else's playback.

Every segment request (`current_session`) and every checkpoint (`touch_session`) takes the one
registry `_lock`. Stopping a session waits up to 3 s for its ffmpeg to exit and then deletes a whole
segmented film, ~1200 files — and that used to happen under the same lock, so one viewer's seek, a
cap cull or a reap pass held up every other viewer's HLS requests for seconds, each one a parked
threadpool worker and its pooled DB connection.

What still has to be atomic is pinned too: a session directory is reused for the same item, offset
and tag, so a new session on that path must not start before the old ffmpeg — which writes by path
— has exited.
"""

import threading
import time

import pytest

from app.plex import stream


class _SlowProc:
    """An ffmpeg that takes a while to exit after TERM: `wait` blocks until `exited` is set."""

    def __init__(self, order: list[str], name: str):
        self.order = order
        self.name = name
        self.waiting = threading.Event()
        self.exited = threading.Event()

    def poll(self):
        return 0 if self.exited.is_set() else None

    def terminate(self):
        pass

    def kill(self):
        self.exited.set()

    def wait(self, timeout=None):
        self.waiting.set()
        if not self.exited.wait(timeout):
            raise stream.subprocess.TimeoutExpired("ffmpeg", timeout)
        self.order.append(f"{self.name} exited")
        return 0


class _Session:
    def __init__(self, directory, proc, key: str):
        self.id = f"sid-{key}"
        self.key = key
        self.dir = directory
        self.proc = proc
        self.last_access = time.time()
        self.transcode = False


@pytest.fixture
def sessions(monkeypatch):
    live: dict = {}
    monkeypatch.setattr(stream, "_sessions", live)
    monkeypatch.setattr(stream, "_dying", {}, raising=False)
    return live


def test_stopping_a_session_does_not_hold_up_another_viewers_requests(tmp_path, sessions):
    order: list[str] = []
    slow = _SlowProc(order, "a")
    d = tmp_path / "1_u1_0_0_+0.00"
    d.mkdir()
    (d / "seg_0.ts").write_bytes(b"x")
    sessions[(1, "a")] = _Session(d, slow, "a")
    sessions[(2, "b")] = _Session(tmp_path / "2_u2_0_0_+0.00", _SlowProc(order, "b"), "b")

    stopper = threading.Thread(target=stream.stop_session, args=(1, "a", "sid-a"))
    stopper.start()
    try:
        assert slow.waiting.wait(2.0), "the stop never waited for its ffmpeg"
        got: list = []
        reader = threading.Thread(target=lambda: got.append(stream.current_session(2, "b")))
        reader.start()
        reader.join(1.0)
        assert not reader.is_alive(), "a segment request waited on another viewer's teardown"
        assert got and got[0] is sessions[(2, "b")]
    finally:
        slow.exited.set()
        stopper.join(5.0)
    # The teardown still finishes: the session is gone and nothing of its directory is left behind.
    assert (1, "a") not in sessions
    assert sorted(p.name for p in tmp_path.iterdir()) == []


def test_a_reap_pass_does_not_hold_up_another_viewers_requests(tmp_path, sessions):
    order: list[str] = []
    slow = _SlowProc(order, "a")
    d = tmp_path / "1_u1_0_0_+0.00"
    d.mkdir()
    gone = _Session(d, slow, "a")
    gone.last_access = time.time() - stream._SESSION_IDLE_S - 5
    sessions[(1, "a")] = gone
    sessions[(2, "b")] = _Session(tmp_path / "2_u2_0_0_+0.00", _SlowProc(order, "b"), "b")

    reaper = threading.Thread(target=stream.reap_idle)
    reaper.start()
    try:
        assert slow.waiting.wait(2.0), "the reap never waited for its ffmpeg"
        toucher = threading.Thread(target=stream.touch_session, args=(2, "b"))
        toucher.start()
        toucher.join(1.0)
        assert not toucher.is_alive(), "a checkpoint waited on another viewer's teardown"
    finally:
        slow.exited.set()
        reaper.join(5.0)
    assert list(sessions) == [(2, "b")]
    assert not d.exists()


class _FakeItem:
    rating_key = "12345"
    file_path = "/media/movie.mkv"
    playable = "remux"
    codec_video = "h264"


def test_a_new_session_on_a_dying_ones_path_starts_only_after_it_exited(
    tmp_path, sessions, monkeypatch
):
    """The same item at the same second gets the same directory. The old ffmpeg writes by PATH, so
    while it is still winding down after TERM it can write a segment — or, on a clean exit, a final
    playlist with ENDLIST — into the NEW session's directory."""
    order: list[str] = []
    procs = iter([_SlowProc(order, "old"), _SlowProc(order, "new")])

    def popen(*a, **k):
        order.append("popen")
        return next(procs)

    monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path)
    src = tmp_path / "movie.mkv"
    src.write_bytes(b"x")
    monkeypatch.setattr(stream.paths, "local_media_path", lambda db, fp: src)
    monkeypatch.setattr(stream.sysconfig, "get_int", lambda db, k: 2)
    monkeypatch.setattr(stream.sysconfig, "get_bool", lambda db, k: False)
    monkeypatch.setattr(stream.subprocess, "Popen", popen)
    monkeypatch.setattr(stream, "wait_for", lambda *a, **k: True)

    first = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0)
    assert first is not None
    stopper = threading.Thread(target=stream.stop_session, args=(7, "12345", first.id))
    stopper.start()
    assert first.proc.waiting.wait(2.0)
    threading.Timer(0.3, first.proc.exited.set).start()
    second = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0)
    stopper.join(5.0)
    assert second is not None and second.dir == first.dir
    assert order.index("old exited") < order.index("popen", 1)
    assert second.dir.is_dir(), "the old teardown deleted the new session's directory"
    second.proc.exited.set()