"""
The indexing dock: what it shows, and who feeds it.

Adding a 900 GB directory left the operator nothing to look at once they left
the Settings panel — and nothing at all when the directory was added from
another machine, where that panel's bar never started. The dock sits above the
music bar on every page. What it shows is worked out in `index-dock-model.js`,
which has no imports so that it runs here under node as shipped; the wiring
around it is checked in the source, as the other SPA tests do.
"""

import json
import re
import shutil
import subprocess
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
MODEL = STATIC / "index-dock-model.js"
DOCK = STATIC / "index-dock.js"
APP = STATIC / "app.js"
GROUP_PAGE = STATIC / "group-page.js"
TRANSPORT = STATIC / "transport.js"

pytestmark = pytest.mark.skipif(not MODEL.exists(), reason="SPA sources unavailable")

GB = 1024 ** 3

needs_node = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")


def _run(tmp_path, body: str):
    (tmp_path / "package.json").write_text('{"type":"module"}')
    (tmp_path / "model.js").write_text(MODEL.read_text(encoding="utf-8"))
    script = tmp_path / "case.js"
    script.write_text("import * as m from './model.js';\n" + body)
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True,
                          cwd=str(tmp_path))
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def _job(**over) -> dict:
    job = {"groupId": "g1", "groupName": "outputs", "scanning": True, "kind": "scan",
           "root": "results", "scannedBytes": 0, "totalBytes": 0, "filesDone": 0,
           "filesTotal": 0, "queued": []}
    job.update(over)
    return job


def _frames(tmp_path, frames: list[tuple[int, list[dict]]], hide_at: int | None = None):
    """Feed dockRows a sequence of (time ms, jobs) and return each frame's rows."""
    return _run(tmp_path, f"""
        const frames = {json.dumps(frames)};
        const hideAt = {json.dumps(hide_at)};
        let memory = {{}};
        const out = [];
        for (const [now, jobs] of frames) {{
          if (hideAt === now) memory = m.hideRow(memory, 'g1');
          const r = m.dockRows(jobs, memory, now);
          memory = r.memory;
          out.push(r.rows.map((row) => [row.groupId, row.state]));
        }}
        console.log(JSON.stringify(out));
    """)


# ── What is shown ───────────────────────────────────────────────────────────

@needs_node
def test_a_root_scan_shows_at_once(tmp_path):
    assert _frames(tmp_path, [(0, [_job()])]) == [[["g1", "running"]]]


@needs_node
def test_a_small_burst_waits_five_seconds_and_a_large_one_does_not(tmp_path):
    small = _job(kind="watch", root="", totalBytes=10_000)
    assert _frames(tmp_path, [(0, [small]), (4000, [small]), (5000, [small])]) == [
        [], [], [["g1", "running"]]]
    large = _job(kind="reconcile", totalBytes=2 * GB)
    assert _frames(tmp_path, [(0, [large])]) == [[["g1", "running"]]]


@needs_node
def test_a_short_burst_never_flashes_a_finished_row(tmp_path):
    small = _job(kind="watch", root="", totalBytes=10_000)
    idle = _job(scanning=False, kind="")
    assert _frames(tmp_path, [(0, [small]), (1000, [idle])]) == [[], []]


@needs_node
def test_a_group_between_two_roots_stays_on_screen(tmp_path):
    """The gap between one root's walk ending and the next taking the lock."""
    waiting = _job(scanning=False, kind="", root="", queued=["archive"])
    assert _frames(tmp_path, [(0, [_job()]), (2000, [waiting])]) == [
        [["g1", "running"]], [["g1", "running"]]]


@needs_node
def test_finished_says_so_for_four_seconds_then_goes(tmp_path):
    idle = _job(scanning=False, kind="", root="")
    assert _frames(tmp_path, [(0, [_job()]), (1000, [idle]), (4999, [idle]),
                              (5000, [idle])]) == [
        [["g1", "running"]], [["g1", "done"]], [["g1", "done"]], []]


@needs_node
def test_a_hidden_row_stays_hidden_until_the_group_is_idle(tmp_path):
    second = _job(root="archive")
    idle = _job(scanning=False, kind="", root="")
    frames = [(0, [_job()]), (1000, [_job()]), (2000, [second]), (3000, [idle]),
              (9000, [idle]), (10000, [_job(root="photos")])]
    assert _frames(tmp_path, frames, hide_at=1000) == [
        [["g1", "running"]], [], [], [], [], [["g1", "running"]]]


# ── Where it comes from ─────────────────────────────────────────────────────

@needs_node
def test_the_loopback_answer_wins_and_the_hub_names_the_group(tmp_path):
    out = _run(tmp_path, """
        const local = { g1: m.fromLoopback({ group_id: 'g1', group_name: 'node name',
          scanning: true, kind: 'scan', root: 'results', queued: ['archive'] }) };
        const pushed = {
          g1: m.fromPush('g1', { scanning: true, kind: 'rescan', root_pos: 0 }, [{ name: 'x' }]),
          g2: m.fromPush('g2', { scanning: true }, []),
        };
        console.log(JSON.stringify(
          m.mergeActivity(local, pushed, [{ id: 'g1', name: 'outputs' }])));
    """)
    by_id = {j["groupId"]: j for j in out}
    assert (by_id["g1"]["kind"], by_id["g1"]["root"], by_id["g1"]["groupName"]) == (
        "scan", "results", "outputs")
    assert by_id["g1"]["queued"] == ["archive"]
    assert by_id["g2"]["groupName"] == "g2"


@needs_node
def test_a_push_is_named_from_the_roots_table_it_does_not_carry(tmp_path):
    out = _run(tmp_path, """
        const roots = [{ name: 'outputs' }, { name: 'results' }];
        console.log(JSON.stringify([
          m.fromPush('g', { scanning: true, kind: 'scan', root_pos: 1, queued: 2 }, roots),
          m.fromPush('g', { scanning: true, root_pos: 7 }, roots),
          m.fromPush('g', { scanning: true, scanned_bytes: 5, total_bytes: 10 }, null),
        ]));
    """)
    assert (out[0]["root"], out[0]["queued"]) == ("results", 2)
    assert out[1]["root"] == "", "a position past the table must name nothing"
    assert (out[2]["kind"], out[2]["root"], out[2]["queued"]) == ("", "", 0), (
        "a node older than the dock must read as a plain scan")


# ── Wiring ──────────────────────────────────────────────────────────────────

def test_the_dock_is_on_every_page_not_in_a_route():
    app = APP.read_text(encoding="utf-8")
    mount = app.index("<${IndexingDock}")
    assert app.index("</main>") < mount < app.index("<${MusicPlayerBar}"), (
        "the dock must sit outside the routed page, right before the music bar")


def test_only_an_operator_connection_feeds_the_dock():
    page = GROUP_PAGE.read_text(encoding="utf-8")
    calls = [m.start() for m in re.finditer(r"reportIndexPush\(groupId, (?!null)", page)]
    assert len(calls) == 2, "the push handler and the handshake ack"
    for at in calls:
        guard = page[max(0, at - 200):at]
        assert "transport.memberRole === 'operator'" in guard, (
            "a member's connection must keep the sidebar dot only")


def test_leaving_a_group_clears_its_row():
    page = GROUP_PAGE.read_text(encoding="utf-8")
    cleanup = page[page.index("    return () => {\n      cancelled = true;"):]
    cleanup = cleanup[:cleanup.index("};")]
    assert "reportIndexPush(groupId, null)" in cleanup


def test_the_transport_passes_the_new_counters_through():
    source = TRANSPORT.read_text(encoding="utf-8")
    block = source[source.index("msg.type === 'index_progress'"):]
    block = block[:block.index("return;")]
    for field in ("files_done", "files_total", "kind", "root_pos", "queued"):
        assert f"{field}:" in block, f"{field} is dropped before it reaches the page"


def test_the_dock_polls_the_node_wide_route_not_one_group():
    source = DOCK.read_text(encoding="utf-8")
    assert "'/api/index-status'" in source
    assert "/index-status`" not in source