"""
A large download is spread over several connections to its node.

One WebRTC connection carries 5 to 6 MB/s at 22 ms on a line that carries 47:
the browser's UDP socket has a 64 KB kernel buffer, a burst overflows it, and
the node's SCTP sender halves its window at every loss. The ceiling is per
connection, so `MeshBayTransport.lanes` opens more of them and
`pipelinedDownload` sends chunk i over lane i modulo their number.

These run the shipped lanes (transport.js, executed whole) under the shipped
pipeline (lifted out of file-utils.js as text) against a fake clock and a fake
node. See harness/download_lanes_harness.mjs.
"""

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

import pytest
from spa_source import classic_scripts

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
FILE_UTILS = STATIC / "file-utils.js"
HARNESS = Path(__file__).parent / "harness" / "download_lanes_harness.mjs"

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")
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_a_large_download_runs_over_eight_connections_in_order():
    out = _run(chunks=200)
    assert out["outcome"] == "done"
    assert out["written"] == 200 and out["inOrder"]
    assert out["offers"] == 7, "seven lanes beside the connection already open"
    lanes = {k: v for k, v in out["carried"].items() if k != "main"}
    assert len(lanes) == 7 and min(lanes.values()) >= 15, out["carried"]
    # It starts on the main connection without waiting for the lanes.
    assert out["carried"]["main"] > max(lanes.values())
    assert out["peakInFlight"] == 16, "two chunks in flight per connection"


def test_a_lane_is_the_main_connection_again_and_never_a_join():
    args = _run(chunks=200)["laneArgs"]
    assert args == {"groupId": "group", "sameGek": True, "sameIdentity": True,
                    "joinCode": None, "nodePk": "pk"}


def test_a_small_download_opens_nothing():
    out = _run(chunks=10)
    assert out["outcome"] == "done" and out["offers"] == 0
    assert out["peakInFlight"] == 8


def test_a_read_without_a_lease_opens_nothing():
    """A preview or a track is not a transfer, and the node would count each
    lane's read against that lane's own leaseless ceiling."""
    out = _run(chunks=200, tr="")
    assert out["outcome"] == "done" and out["offers"] == 0
    assert out["carried"] == {"main": 200}


def test_a_lane_that_fails_is_dropped_and_its_chunks_asked_again():
    out = _run(chunks=200, failLaneAfter=3)
    assert out["outcome"] == "done"
    assert out["written"] == 200 and out["inOrder"]
    assert out["retried"] >= 1
    assert out["carried"]["lane1"] <= 3
    assert out["offers"] == 8, "the dropped lane is replaced, once"


def test_a_node_that_refuses_lanes_costs_one_round_of_offers():
    """`lanes` is asked at every chunk: without the retry delay a refusing node
    would be sent an offer per chunk."""
    out = _run(chunks=200, refuseLanes=True)
    assert out["outcome"] == "done"
    assert out["carried"] == {"main": 200}
    assert out["offers"] == 7


def test_idle_lanes_are_closed():
    out = _run(chunks=200)
    assert out["lanesAfterIdle"] == 0 and out["lanesLeftOpen"] == 0
    assert out["timersLeft"] == 0, "the idle timer outlived the lanes"


@pytest.mark.parametrize("close_at", [400, 1500])
def test_closing_the_connection_closes_its_lanes(close_at):
    """At 400 ms the lanes are still negotiating, at 1500 they are carrying."""
    out = _run(chunks=2000, closeAt=close_at)
    assert out["lanesLeftOpenByClose"] == 0
    assert out["offers"] == 7, "a closed connection opened more lanes"