"""
What an automatic reconnect has to re-read, and who is allowed to stop
listening for one.

Both defects here produce a plausible screen rather than an error, so they read
the source — the only evidence available for the SPA (CLAUDE.md), and the right
kind for a fault whose whole symptom is a page that looks fine and is stale.

  - `_reconnectLoop` re-did the handshake and nothing else. Nobody asked for an
    index again, and a push sent while the old channel was dying reached
    nobody, so the page stayed frozen at whatever it last saw until someone
    reloaded it. Survivable while a node that came back came back with the same
    answers; a *restarted* one does not. It rebuilds its index from
    `index_cache.db` — path/mtime/size/hash/type, no enrichment — so during its
    re-enrichment pass the index it serves carries no artist and no album on
    any track, and a client that reconnected inside that window drew an empty
    Music grid for as long as it stayed open.

  - `onReconnected` was one setter. The video player took it on open and set it
    back to `null` on close, which silently disabled every other consumer's
    handler — including, once the group page had one, the refresh above.
"""

import re
from pathlib import Path

import pytest
from spa_source import transport_source

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
TRANSPORT = STATIC / "transport.js"
GROUP_PAGE = STATIC / "group-page.js"
VIDEO_PLAYER = STATIC / "video-player.js"

pytestmark = pytest.mark.skipif(
    not TRANSPORT.exists(), reason="the SPA sources are not available")


@pytest.fixture(scope="module")
def transport():
    return transport_source()


@pytest.fixture(scope="module")
def group_page():
    return GROUP_PAGE.read_text()


@pytest.fixture(scope="module")
def video_player():
    return VIDEO_PLAYER.read_text()


def _reconnect_loop(transport: str) -> str:
    body = transport[transport.index("async _reconnectLoop()"):]
    return body[:body.index("\n  /**")]


def test_a_reconnect_is_announced_to_every_listener(transport):
    """One consumer unsubscribing must not silence the others."""
    assert "addReconnectListener(fn)" in transport
    assert "set onReconnected(" not in transport, (
        "a single slot is what let the video player unset the group page's handler")
    assert "this._reconnectListeners = new Set()" in transport


def test_addReconnectListener_hands_back_its_own_unsubscribe(transport):
    body = transport[transport.index("addReconnectListener(fn) {"):]
    body = body[:body.index("\n  }")]
    assert "this._reconnectListeners.add(fn)" in body
    assert "return () => this._reconnectListeners.delete(fn)" in body, (
        "without it a caller can only stop listening by clearing the whole set")


def test_the_reconnect_carries_the_fresh_ack(transport):
    """
    The page's whole view of the node — which folders each app reads, which
    apps are on, the roots — was answered by a process that may since have
    restarted.
    """
    loop = _reconnect_loop(transport)
    assert re.search(r"ack\s*=\s*await this\.connect\(", loop), (
        "the reconnect's own handshake answer was thrown away")
    assert re.search(r"fn\(ack\)", loop)


def test_one_listener_throwing_does_not_rob_the_next(transport):
    loop = _reconnect_loop(transport)
    notify = loop[loop.index("_reconnectListeners"):]
    assert "try {" in notify and "catch" in notify


def test_nothing_assigns_the_old_setter():
    """`grep onReconnected =` is what this is, spelled so it cannot rot."""
    offenders = [p.name for p in STATIC.glob("*.js")
                 if re.search(r"\.onReconnected\s*=", p.read_text())]
    assert offenders == [], (
        f"{offenders} still assign a slot that no longer exists")


def test_the_group_page_refetches_the_index_on_reconnect(group_page):
    assert "addReconnectListener(" in group_page, (
        "a reconnect that re-reads nothing leaves the page a node-restart old")
    listener = group_page[group_page.index("addReconnectListener("):]
    listener = listener[:listener.index("\n        });")]
    assert "applyAck(" in listener, "the ack is a restarted node's answers, not the old one's"
    assert "transport.fetchIndex()" in listener, (
        "a delta is computed against a snapshot only the node has, and this "
        "session was never told what it missed")
    assert "applyIndex(" in listener


def test_the_group_page_reimports_the_gek_on_reconnect(group_page):
    """A chat epoch or a re-key while we were away hands back a different GEK."""
    listener = group_page[group_page.index("addReconnectListener("):]
    listener = listener[:listener.index("\n        });")]
    assert "importGEK" in listener


def test_the_ack_is_applied_by_one_implementation(group_page):
    """
    Two copies drifting is how `helloworld`'s directories went missing from one
    of them; the connect path and the reconnect path read the same function.
    """
    assert group_page.count("const applyAck = useCallback(") == 1
    assert group_page.count("applyAck(ack);") == 1
    assert group_page.count("applyAck(reack);") == 1
    body = group_page[group_page.index("const applyAck = useCallback("):]
    body = body[:body.index("\n  }, [")]
    for setter in ("setEnabledApps", "setAppDirectories", "setMusicbrainzConfig",
                   "setTmdbConfig", "setChatDirectory", "setSearchListed"):
        assert setter in body, f"{setter} is not re-read on a reconnect"


def test_every_listener_is_dropped_by_whoever_registered_it(group_page, video_player):
    """
    A transport handed on to a running download (`releaseWhenIdle`) outlives
    the page that opened it, and would go on driving a component that is gone.
    """
    for name, source in (("group-page.js", group_page),
                         ("video-player.js", video_player)):
        assert re.search(r"=\s*transport\.addReconnectListener\(", source), (
            f"{name} must keep the unsubscribe it is handed")
        assert "offReconnect()" in source, (
            f"{name} registers a reconnect listener it never drops")


def test_every_harness_that_renders_the_group_page_stubs_the_subscription():
    """
    The fast half of a guard `test_group_tab_fallback` already provides slowly.

    GroupPage subscribes on connect and calls the unsubscribe on unmount, so a
    stub node without this throws inside `connect()` and the page renders its
    error state — a whole harness reporting a layout it never drew. Three of
    these stubs answer an unknown method through a Proxy, which hands back a
    promise where an unsubscribe belongs and defers the same failure to the
    teardown. Nine minutes of browser tests found it; this finds it in a
    fraction of a second.
    """
    harness = Path(__file__).parent / "harness"
    stubs = [p for p in harness.glob("*.py")
             if "window.MeshBayTransport" in p.read_text()
             and "GroupPage" in p.read_text()]
    assert stubs, "no harness stubs the transport any more — has this moved?"
    missing = [p.name for p in stubs
               if "addReconnectListener" not in p.read_text()]
    assert missing == [], (
        f"{missing} render GroupPage against a node that cannot be subscribed to")