"""
Transfer leases over the session, rather than over `TransferSlots` alone.

test_transfer_slots.py proves the decisions; this proves the seam. Both exist
because the seam is where this repo's defects have actually lived — a reply
routed by arrival order, a session popped from a dict without its work being
stopped, a slot released by a `finally` nobody reached.

Three things can only be checked here:

  - the handlers answer under the right shape, and refuse another connection's
    transfer id;
  - **losing the connection gives everything back.** That is the primary
    reclaim, and it is a hook (`shutdown_tasks`) rather than a timer, so a test
    of the pool alone would never touch it;
  - a slot freed by one peer is *announced* to the peer waiting on it. A grant
    nobody hears about is precisely the "stuck at waiting" report the design
    exists to prevent, and it would look correct in the pool.
"""

import pytest

# Every test drives a message handler, and in the node a message handler always
# runs inside the event loop: `_do_transfer_open` starts the sweeper task there.
# Calling these synchronously tested a situation that cannot happen and failed
# on "no current event loop" the moment the sweeper stopped being faked.
pytestmark = pytest.mark.asyncio

from meshbay_common.protocol import MNP
from meshbay_node.transfers import DOWNLOAD, UPLOAD
from meshbay_node.transport.webrtc_server import WebRTCPeerSession


class _Session(WebRTCPeerSession):
    """A session with the DataChannel replaced by a list, and nothing else."""

    def __init__(self, ctx, *, key, user, group="g1"):
        self._ctx = ctx
        self._registry_key = key
        self._user_id = user
        self._group_id = group
        self.sent: list[dict] = []

    def _send(self, msg):
        self.sent.append(msg)

    def _spawn(self, coro):        # pragma: no cover - not used by these tests
        coro.close()
        return None

    def last(self, mtype=MNP.TRANSFER_STATE):
        return next(m for m in reversed(self.sent) if m.get("type") == mtype)


@pytest.fixture
def ctx():
    """A transport context, with the sweeper stopped on the way out.

    A task left running past the end of its test is a warning in the next one
    and a hang in the worst case; the sweeper is started on demand by design, so
    tearing it down is the test's job.
    """
    c: dict = {"_peers": {}}
    yield c
    task = c.get("_transfer_sweeper")
    if task is not None:
        task.cancel()


def _join(ctx, key, user, group="g1") -> _Session:
    s = _Session(ctx, key=key, user=user, group=group)
    ctx["_peers"][key] = s
    return s


async def test_a_granted_transfer_is_answered_as_granted(ctx):
    peer = _join(ctx, "s1", "alice")
    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 10})
    reply = peer.last()
    assert reply["state"] == "granted"
    assert reply["tr"] == "t1"
    assert reply["kind"] == DOWNLOAD
    assert reply["used"] == 1 and reply["cap"] >= 1


async def test_a_queued_transfer_is_told_how_many_are_ahead(ctx):
    peer = _join(ctx, "s1", "alice")
    peer._slots().per_member[DOWNLOAD] = 1
    peer._do_transfer_open({"tr": "t1"})
    peer._do_transfer_open({"tr": "t2"})
    peer._do_transfer_open({"tr": "t3"})
    assert [m["state"] for m in peer.sent] == ["granted", "queued", "queued"]
    assert peer.sent[-1]["ahead"] == 1


async def test_the_reply_carries_no_name_and_no_path(ctx):
    """A lease holds neither, and `transfer_state` stays in clear — so this is
    the message where a filename would quietly become metadata on the wire."""
    peer = _join(ctx, "s1", "alice")
    peer._do_transfer_open({"tr": "t1", "name": "Some Saga.mkv",
                            "path": "/srv/films"})
    assert set(peer.last()) <= {
        "type", "v", "tr", "state", "kind", "used", "cap", "node_used",
        "node_cap", "ahead", "reason"}


async def test_closing_frees_the_slot(ctx):
    peer = _join(ctx, "s1", "alice")
    peer._do_transfer_open({"tr": "t1"})
    peer._do_transfer_close({"tr": "t1", "reason": "done"})
    assert peer.last()["state"] == "closed"
    assert peer._slots().in_use(DOWNLOAD) == 0


async def test_one_peer_cannot_close_anothers_transfer(ctx):
    """A denial of service one random id away, otherwise."""
    alice = _join(ctx, "s1", "alice")
    bob = _join(ctx, "s2", "bob")
    alice._do_transfer_open({"tr": "t1"})
    bob._do_transfer_close({"tr": "t1"})
    assert bob.last("error")["code"] == "not_your_transfer"
    assert "t1" in alice._slots().leases


async def test_one_peer_cannot_open_on_anothers_id(ctx):
    alice = _join(ctx, "s1", "alice")
    bob = _join(ctx, "s2", "bob")
    alice._do_transfer_open({"tr": "t1"})
    bob._do_transfer_open({"tr": "t1"})
    assert bob.last("error")["code"] == "not_your_transfer"


async def test_a_transfer_with_no_id_is_refused(ctx):
    peer = _join(ctx, "s1", "alice")
    peer._do_transfer_open({"kind": DOWNLOAD})
    assert peer.last("error")["code"] == "bad_transfer_id"


# ── the reclaim that matters ────────────────────────────────────────────────

async def test_losing_the_connection_gives_everything_back(ctx):
    peer = _join(ctx, "s1", "alice")
    peer._do_transfer_open({"tr": "t1"})
    peer._do_transfer_open({"tr": "t2", "kind": UPLOAD})
    peer._release_transfers()
    slots = peer._slots()
    assert slots.leases == {}
    assert slots.in_use(DOWNLOAD) == 0 and slots.in_use(UPLOAD) == 0


async def test_the_freed_slot_reaches_the_peer_that_was_waiting(ctx):
    """
    The seam this file exists for. In the pool, granting is correct; if the
    grant is not pushed, the waiting client sits on "waiting" for ever with a
    node that believes it is streaming — and every unit test still passes.
    """
    alice = _join(ctx, "s1", "alice")
    bob = _join(ctx, "s2", "bob")
    alice._slots().caps[DOWNLOAD] = 1

    alice._do_transfer_open({"tr": "a1"})
    bob._do_transfer_open({"tr": "b1"})
    assert bob.last()["state"] == "queued"

    alice._release_transfers()
    assert bob.last()["state"] == "granted", (
        "bob was granted the slot and never told")
    assert bob.last()["tr"] == "b1"


async def test_a_grant_crossing_groups_still_reaches_its_peer(ctx):
    """The pools are node-wide and `_peer_registry` is per group (finding H1),
    so the peer to notify is not necessarily in the notifier's own registry."""
    groups = {"g1": {"_peers": {}}, "g2": {"_peers": {}}}
    ctx = {"groups": groups}
    alice = _Session(ctx, key="s1", user="alice", group="g1")
    groups["g1"]["_peers"]["s1"] = alice
    bob = _Session(ctx, key="s2", user="bob", group="g2")
    groups["g2"]["_peers"]["s2"] = bob

    alice._slots().caps[DOWNLOAD] = 1
    alice._do_transfer_open({"tr": "a1"})
    bob._do_transfer_open({"tr": "b1"})
    assert bob.last()["state"] == "queued"

    alice._release_transfers()
    assert bob.last()["state"] == "granted", (
        "a slot freed in one group never reached the peer waiting in another")


async def test_raising_the_cap_notifies_who_it_starts(ctx):
    """`set_capacity` arrives from the loopback API, with no session behind it —
    the grants it produces still have to be pushed."""
    from meshbay_node.transport.webrtc_server import WebRTCTransport

    transport = WebRTCTransport.__new__(WebRTCTransport)
    transport._ctx = ctx
    peer = _join(ctx, "s1", "alice")
    peer._slots().caps[DOWNLOAD] = 1
    peer._do_transfer_open({"tr": "t1"})
    peer._do_transfer_open({"tr": "t2"})
    assert peer.last()["state"] == "queued"

    transport.set_capacity(max_concurrent_downloads=4)
    assert peer.last()["state"] == "granted" and peer.last()["tr"] == "t2"


async def test_the_operator_can_see_the_queue(ctx):
    """`GET /api/transfers` is the answer to "was this peer ever queued", which
    a log line cannot give when the symptom is that nothing is happening."""
    from meshbay_node import ops

    peer = _join(ctx, "s1", "alice")
    peer._slots().caps[DOWNLOAD] = 1
    peer._do_transfer_open({"tr": "t1", "bytes": 5})
    peer._do_transfer_open({"tr": "t2", "bytes": 7})

    class _T:
        _ctx = ctx

    snapshot = await ops.list_transfers({"webrtc": _T()})
    assert snapshot["pools"][DOWNLOAD]["in_use"] == 1
    assert snapshot["pools"][DOWNLOAD]["queued"] == 1
    assert {x["state"] for x in snapshot["leases"]} == {"granted", "queued"}
    assert all("name" not in x and "path" not in x for x in snapshot["leases"])


async def test_asking_before_anything_has_transferred_is_not_an_error():
    from meshbay_node import ops

    class _T:
        _ctx: dict = {}

    snapshot = await ops.list_transfers({"webrtc": _T()})
    assert snapshot["leases"] == []
    assert snapshot["pools"][DOWNLOAD]["in_use"] == 0


# ── the sweeper's lifetime ──────────────────────────────────────────────────

async def test_the_sweeper_outlives_the_session_that_started_it(ctx):
    """
    It was started with `self._spawn`, which ties a task to one session's set —
    so it was cancelled the moment that peer left, and every other peer's
    abandoned lease stopped being reclaimed. Nothing else would have noticed:
    the node simply fills up over days.
    """
    import asyncio

    from meshbay_node.transport import webrtc_server as ws

    alice = _join(ctx, "s1", "alice")
    bob = _join(ctx, "s2", "bob")
    # The real _spawn, so the session genuinely owns what it starts.
    alice._tasks = set()
    alice._spawn = ws.WebRTCPeerSession._spawn.__get__(alice)
    bob._tasks = set()
    bob._spawn = ws.WebRTCPeerSession._spawn.__get__(bob)

    alice._do_transfer_open({"tr": "a1"})
    bob._do_transfer_open({"tr": "b1"})
    sweeper = ctx["_transfer_sweeper"]
    assert sweeper is not None and not sweeper.done()

    # Alice leaves, exactly as shutdown_tasks does it.
    alice._release_transfers()
    for task in list(alice._tasks):
        task.cancel()
    await asyncio.gather(*alice._tasks, return_exceptions=True)
    await asyncio.sleep(0)

    assert not sweeper.done(), (
        "the sweeper died with the session that happened to start it; bob's "
        "lease would never be reclaimed")
    sweeper.cancel()


async def test_the_sweeper_stops_when_the_last_lease_goes(ctx):
    """An idle node must run no timer — the reason this is started on demand
    rather than at boot."""
    import asyncio

    from meshbay_node.transport import webrtc_server as ws

    peer = _join(ctx, "s1", "alice")
    peer._tasks = set()
    peer._spawn = ws.WebRTCPeerSession._spawn.__get__(peer)

    original = ws.TRANSFER_SWEEP_SECS
    ws.TRANSFER_SWEEP_SECS = 0.01
    try:
        peer._do_transfer_open({"tr": "t1"})
        peer._do_transfer_close({"tr": "t1"})
        sweeper = ctx["_transfer_sweeper"]
        await asyncio.wait_for(sweeper, timeout=2)
        assert ctx.get("_transfer_sweeper") is None
    finally:
        ws.TRANSFER_SWEEP_SECS = original