"""
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 of three 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 —
with four of twelve down, five live groups appear straight away and the last
three wait out a deadline. Ordering by what was silent last time puts all eight
on screen in 600 ms. The cost is that a browser seeing these groups for the
first time has nothing to order by and pays the first sweep, once.

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_dead_groups_can_still_hold_the_last_few_back():
    """
    The limit of a ceiling, stated rather than glossed over.

    Four dead groups and three places: once all three are held by nodes that are
    not answering, nothing else is dialled until one of them gives up. The first
    results are immediate and most arrive at once, but the last few do wait — on
    a browser that has never run this sweep before and so has nothing to order
    by. The test after this one is the steady state, which is what a reader
    actually lives in.
    """
    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"] == 5, (
        "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 last bad case, and the reason the browser remembers.

    The three groups the hub lists first are all down, so on a first visit all
    three places are 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.
    """
    first_visit = _sweep(groups=12, dead=[0, 1, 2])
    assert first_visit["firstPaintAt"] == DEAD_MS + LIVE_MS
    assert first_visit["remembered"] == ["g-0", "g-1", "g-2"], (
        "the sweep must record who was silent, or the next visit repeats this")

    again = _sweep(groups=12, dead=[0, 1, 2], knownDown=[0, 1, 2])
    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=[0, 1, 2], 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 three places and every
    node down cost four deadlines, not one and not twelve.
    """
    out = _sweep(groups=12, dead=list(range(12)))
    assert out["inFlight"] == 3
    assert out["finishedAt"] == 4 * DEAD_MS