"""
A reconnect must not fail the download it is reconnecting for.

Reported live on 0.20: a large download from a node under load stopped with
"Handshake required" in the transfers panel. `_sendAndWait` let every message
through while the reconnect's own connect() ran, because the handshake it sends
has to go through; so a chunk whose retry came due in that window went out on a
channel that had opened and not finished its handshake. The node answered
"Handshake required" (webrtc/dispatch.py), which no retry treats as a dropped
connection. Now only the opening messages pass, and anything else waits for the
session or fails as a dropped connection, which the download waits out.

The replaced channel's close event is the second half: it reached the new
connection, marked it closed and failed what it had in flight.

These run the shipped transport and pipeline against a node that refuses what
comes before the handshake: harness/reconnect_race_harness.mjs.
"""

import json
import os
import re
import shutil
import subprocess
from pathlib import Path

import pytest
from spa_source import classic_scripts, transport_source

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
FILE_UTILS = STATIC / "file-utils.js"
HARNESS = Path(__file__).parent / "harness" / "reconnect_race_harness.mjs"
DISPATCH = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src" / "meshbay_node"
            / "transport" / "webrtc" / "dispatch.py")

pytestmark = pytest.mark.skipif(
    shutil.which("node") is None or not FILE_UTILS.exists(),
    reason="node or the SPA sources are not available")


def _run(**cfg) -> dict:
    transport = os.pathsep.join(
        str(STATIC / name) for name in classic_scripts() if name.startswith("transport"))
    proc = subprocess.run(
        ["node", str(HARNESS), str(FILE_UTILS), transport, json.dumps(cfg)],
        capture_output=True, text=True, encoding="utf-8", timeout=120)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_retries_that_come_due_during_the_handshake_wait_for_it():
    # The chunks in flight time out just before the connection is declared
    # failed, and their retries land while the new channel is handshaking.
    out = _run()
    assert out["reconnected"] == 1 and out["connections"] == 2
    assert out["refusedBeforeHandshake"] == 0, "a chunk was asked for before the handshake"
    assert out["outcome"] == "done"
    assert out["written"] == 400 and out["inOrder"]


@pytest.mark.parametrize("cfg", [
    {"failAfter": 20000},                 # failure declared before anything times out
    {"failAfter": 31000, "rttMs": 800},   # a slow handshake
    {"openMs": 2500},                     # a slow channel
])
def test_the_download_survives_other_timings(cfg):
    out = _run(**cfg)
    assert out["refusedBeforeHandshake"] == 0
    assert out["outcome"] == "done" and out["written"] == 400 and out["inOrder"]


@pytest.mark.parametrize("late", [800, 1500, 3000])
def test_the_replaced_channels_close_event_is_ignored(late):
    out = _run(lateCloseMs=late)
    assert out["outcome"] == "done" and out["written"] == 400


def _node_opening_types() -> set[str]:
    """What the node answers before the handshake, read from its dispatch."""
    from meshbay_common.protocol import MNP
    src = DISPATCH.read_text(encoding="utf-8")
    before = src[src.index("def _dispatch_message"):src.index("elif self._user_id is None")]
    names = set(re.findall(r"MNP\.([A-Z_]+)", before))
    return {getattr(MNP, n) for n in names}


def test_the_opening_messages_are_the_ones_the_node_takes_before_the_handshake():
    m = re.search(r"const SESSION_OPENING = new Set\(\[(.*?)\]\);", transport_source(), re.S)
    client = set(re.findall(r"'([a-z_]+)'", m.group(1)))
    # device_hello is the one connect() sends after the proof and before it
    # returns: the node takes it then, and the session is not up until it has.
    assert client - _node_opening_types() == {"device_hello"}
    assert _node_opening_types() <= client