"""
Search reaches each group with one offer, and never more than a few at a time.

A phone on 4G reported one group of five missing from Search on half its
visits, and all but one on a fifth of them. The nodes were answering every
offer that reached them in about a second; the hub was refusing the others. The
page negotiated each group twice — the sweep opened a connection, read the
index, closed it, and the warm-up opened the same group again — and the sweep,
the warm-up and the tiles each kept a concurrency ceiling of their own, which
added up past what the hub admitted per account. Every refusal read as "node
unreachable".

Now every connection goes through one `ConnectionPool`, which holds the only
ceiling and keeps what the sweep opened. These run the shipped pool, index
fetch and sweep, lifted out of search-page.js as text, against a fake clock and
fake nodes that take two seconds to answer — see harness/search_pool_harness.mjs.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest
from spa_source import search_argv

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
SEARCH_PAGE = STATIC / "search-page.js"
HARNESS = Path(__file__).parent / "harness" / "search_pool_harness.mjs"

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


def _run(**cfg) -> dict:
    proc = subprocess.run(
        ["node", str(HARNESS), search_argv(), json.dumps(cfg)],
        capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_each_group_costs_one_offer_and_stays_open():
    out = _run(scenario="sweep", groups=5)
    assert out["found"] == 5
    assert out["offers"] == 5, "a group was negotiated more than once"
    assert out["openTransports"] == 5, "the sweep closed what the tiles will need"


def test_twenty_groups_stay_within_the_ceiling_and_the_pool():
    out = _run(scenario="sweep", groups=20)
    assert out["found"] == 20
    assert out["offers"] == 20
    assert out["peakNegotiating"] == out["maxInFlight"]
    assert out["openTransports"] == out["maxPool"], "the pool kept more than its size"


def test_an_index_is_never_read_from_a_connection_evicted_under_it():
    """With more groups than the pool keeps, and tiles on screen using the
    connections of groups already found, the connection a slow index is being
    read from is the least recently used one there — the first to be evicted,
    which would fail that group for a reason of this page's own making."""
    out = _run(scenario="scroll", groups=20, indexMs=5000)
    assert out["readsOnClosed"] == 0
    assert out["found"] == 20


def test_dead_nodes_cost_their_own_places_and_nothing_else():
    out = _run(scenario="sweep", groups=20, dead=[0, 3, 7])
    assert out["found"] == 17
    assert out["unreachable"] == 3
    # Three dead groups hold three of six places for one deadline, side by side.
    assert out["at"] < 10000 + 20 * 2000 / 6 + 1000


def test_tiles_asking_during_the_sweep_share_its_connections():
    """Tiles mount as soon as a group's index lands, while the sweep is still
    dialling the others. They must neither negotiate a group a second time nor
    add places of their own beside the sweep's."""
    out = _run(scenario="tiles", groups=20)
    assert out["offers"] == 20
    assert out["peakNegotiating"] <= out["maxInFlight"]


def test_refreshing_reuses_every_connection_that_is_alive():
    out = _run(scenario="refresh", groups=5)
    assert out["offersSecond"] == 0
    assert out["pings"] == 5, "a reused connection must prove it is alive first"
    assert out["found"] == 5


def test_a_connection_that_died_in_its_sleep_is_replaced_not_waited_on():
    """A phone that slept keeps reporting `connected` on channels that are gone.
    Those are found by a short ping and renegotiated; the rest are reused."""
    out = _run(scenario="slept", groups=5)
    assert out["offersSecond"] == 2
    assert out["found"] == 5
    assert out["unreachable"] == 0


def test_leaving_the_page_leaves_no_connection_behind():
    """Negotiations still under way when the page closes its pool close what
    they obtain, rather than adding it to a pool nobody will close again."""
    out = _run(scenario="unmount", groups=8)
    assert out["openAfter"] == 0
    assert out["offers"] == out["maxInFlight"], "groups were dialled after the page closed"
    assert out["remembered"] is None, (
        "groups the closed page never reached were remembered as down")