"""
A group whose node is down must not be something the reader waits for.

Some group is always down — a node is a machine in somebody's house. Search
dials every group the reader belongs to, so "one of them is off tonight" is the
normal case and not the exception, and what it costs is the whole of how the
page feels.

It used to cost everything. `fetchAllIndexes` went in batches of three and
waited for the slowest of each before starting the next, and it drew results
only once a batch was complete. Measured on twelve groups on a virtual clock,
with a live group answering in 200 ms and a dead one taking the full 10 s
deadline:

    all reachable          first result  200 ms    done    800 ms
    one down, first batch  first result   10 s     done   10.6 s
    four down, spread      first result   10 s     done     40 s
    three down, together   first result   10 s     with nothing to show

So a single node being off put a blank page and a progress bar in front of the
reader for ten seconds, while two groups of the same batch had answered in two
hundred milliseconds and nine others had not been dialled at all.

Three things fix it, and each is asserted below: a result is drawn when it
arrives rather than when its neighbours finish; the ceiling is a
ceiling on *concurrency* rather than a batch, so a dead group holds one place
instead of stalling a batch and everything queued behind it; and a group that
did not answer last time is dialled last.

The third is not a refinement of the second, it is what makes the steady state
correct. A ceiling alone still lets dead groups occupy every place at once — as
many dead groups as there are places, listed first, and nothing is drawn until
a deadline expires. Ordering by what was silent last time puts them behind
every live group. The cost is that a browser seeing these groups for the first
time has nothing to order by and pays the first sweep, once.

The ceiling is six (search-page.js, `MAX_IN_FLIGHT`). It was three, which was
also exactly what the hub admitted in flight per account, so the page ran into
the hub's refusals as soon as anything else connected beside the sweep.

These run the shipped `fetchAllIndexes`, `inFlight`, `lastKnownDown` and
`rememberDown`, lifted out of search-page.js as text, against a fake clock and a
fake browser store — see harness/search_fanout_harness.mjs. The numbers below are
the ones in the table, as assertions.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
SEARCH_PAGE = STATIC / "search-page.js"
HARNESS = Path(__file__).parent / "harness" / "search_fanout_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")

LIVE_MS = 200      # a node that answers
DEAD_MS = 10000    # a node that does not, to the connection deadline


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


def test_a_reachable_group_is_drawn_as_soon_as_it_answers():
    """Not when the two groups it happens to share a place in the queue with do."""
    out = _sweep(groups=12, dead=[])
    assert out["firstPaintAt"] == LIVE_MS
    assert out["firstPaintGroups"] == 1, (
        "a result waited for others before being drawn")


def test_one_node_being_down_does_not_delay_the_first_result():
    """The reported case: a node down in what used to be the first batch."""
    out = _sweep(groups=12, dead=[1])
    assert out["firstPaintAt"] == LIVE_MS


def test_several_nodes_down_do_not_delay_the_first_result():
    """
    Four of twelve, spread out — the shape a reader with a few groups really
    has. Batches made this ten seconds to the first result and forty to the end,
    because each dead group stalled its own batch and postponed the rest.
    """
    out = _sweep(groups=12, dead=[1, 4, 7, 10])
    assert out["firstPaintAt"] == LIVE_MS
    assert out["finishedAt"] < 4 * DEAD_MS, (
        "dead groups are being waited for one after another")


def test_on_a_first_visit_four_dead_groups_hold_nothing_back():
    """
    Four dead groups of twelve on a browser that has never swept before, so
    nothing orders them. With three places they held every place at once and
    the last three live groups waited out a deadline; with six, two places stay
    free and every live group is on screen before any deadline expires.
    """
    out = _sweep(groups=12, dead=[1, 4, 7, 10])
    assert out["firstPaintAt"] == LIVE_MS
    early = [r for r in out["renders"] if r["at"] < DEAD_MS]
    assert early[-1]["groups"] == 8, (
        "the shape of the first visit has changed — re-measure it rather than "
        "adjusting this number")


def test_once_the_silent_groups_are_known_no_live_one_waits_for_them():
    """
    The property that matters, in the state a reader is normally in: every group
    that answers is on screen before any deadline has expired.
    """
    out = _sweep(groups=12, dead=[1, 4, 7, 10], knownDown=[1, 4, 7, 10])
    live = 12 - 4
    early = [r for r in out["renders"] if r["at"] < DEAD_MS]
    assert early and early[-1]["groups"] == live, (
        "a reachable group was not on screen until a deadline expired")
    assert early[-1]["at"] <= 1000, (
        f"the live groups took {early[-1]['at']} ms to all appear")


def test_a_group_that_did_not_answer_is_dialled_last_next_time():
    """
    The limit of a ceiling, and the reason the browser remembers.

    The six groups the hub lists first are all down, so on a first visit every
    place is held at once and there is nothing to draw until one frees up. Told
    which groups were silent, the sweep puts them behind everything else and
    the first result arrives on time.
    """
    down = list(range(6))
    first_visit = _sweep(groups=12, dead=down)
    assert first_visit["firstPaintAt"] == DEAD_MS + LIVE_MS
    assert first_visit["remembered"] == [f"g-{n}" for n in down], (
        "the sweep must record who was silent, or the next visit repeats this")

    again = _sweep(groups=12, dead=down, knownDown=down)
    assert again["firstPaintAt"] == LIVE_MS


def test_the_order_is_advisory_and_survives_a_browser_that_refuses_storage():
    """
    A private window throws on both halves of `localStorage`. The sweep must
    then behave exactly as a first visit does, not fail.
    """
    out = _sweep(groups=12, dead=list(range(6)), knownDown="refuses")
    assert out["firstPaintAt"] == DEAD_MS + LIVE_MS
    assert out["remembered"] is None
    assert out["finishedAt"] is not None, "the sweep did not finish"


def test_a_node_that_comes_back_is_not_punished_for_ever():
    """
    The remembered list is rewritten from what this sweep saw, so a group that
    answers again is no longer at the back on the visit after it.
    """
    out = _sweep(groups=12, dead=[], knownDown=[1, 4])
    assert out["remembered"] == [], "a recovered node stayed on the silent list"


def test_no_more_groups_are_dialled_at_once_than_the_ceiling():
    """
    The ceiling is still a ceiling: twelve groups with six places and every
    node down cost two deadlines, not one and not twelve.
    """
    out = _sweep(groups=12, dead=list(range(12)))
    assert out["inFlight"] == 6
    assert out["finishedAt"] == 2 * DEAD_MS