"""
The browser half of MNP 1.0's sealed index, measured rather than read.

`test_index_no_cleartext.py` proves the node sends no filename in the clear. This
proves the client can still read one — and, the part that matters more, that it
*stops* when it cannot instead of reporting an empty group.

Driven through `harness/index_seal_probe.mjs`, which runs the shipped
`transport.js` over the shipped `crypto.js` and is fed real frames built here.
A source-reading test could show that `openGroup` is called; only this can show
what a waiting `fetchIndex()` is told when it throws.
"""

import json
import shutil
import struct
import subprocess
import tempfile
from pathlib import Path

import msgpack
import pytest
from meshbay_common.crypto import generate_gek
from meshbay_common.groupbox import PURPOSE_INDEX, seal

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
PROBE = Path(__file__).resolve().parent / "harness" / "index_seal_probe.mjs"

pytestmark = pytest.mark.skipif(
    shutil.which("node") is None or not PROBE.exists(),
    reason="node unavailable — the client half cannot be measured",
)

GROUP = "g-probe"
GEK = generate_gek()


def _entry(name: str) -> dict:
    return {"id": "ab" * 32, "name": name, "path": "library", "size": 10,
            "type": "file", "added_at": 0, "uploader_id": ""}


def _frame(msg: dict) -> str:
    body = msgpack.packb(msg, use_bin_type=True)
    return (struct.pack(">I", len(body)) + body).hex()


def _sync_frame(gek: bytes, *names: str, version: int = 3,
                req_id: int | None = None) -> str:
    payload = {"version": version, "entries": [_entry(n) for n in names],
               "dirs": ["library"], "roots": [{"name": "library"}]}
    # `req_id` is what a current node stamps on a *reply*; the push it sends a
    # newly connected peer answers no request and carries none. Both shapes
    # arrive here, and only one of them may resolve a waiting fetchIndex.
    return _frame({"type": "index_sync", "v": "1.0", "group_id": GROUP,
                   **({"req_id": req_id} if req_id is not None else {}),
                   **seal(gek, PURPOSE_INDEX, "index_sync", GROUP, payload)})


def _delta_frame(gek: bytes, name: str, base: int, version: int) -> str:
    payload = {"base_version": base, "version": version,
               "additions": [_entry(name)], "deletions": [], "updates": []}
    return _frame({"type": "index_delta", "v": "1.0", "group_id": GROUP,
                   **seal(gek, PURPOSE_INDEX, "index_delta", GROUP, payload)})


def _run(frames: list[str], gek: bytes = GEK) -> dict:
    with tempfile.TemporaryDirectory() as tmp:
        vectors = Path(tmp) / "vectors.json"
        vectors.write_text(json.dumps(
            {"gek": gek.hex(), "group_id": GROUP, "frames": frames}))
        proc = subprocess.run(
            ["node", str(PROBE), str(STATIC), str(vectors)],
            capture_output=True, text=True, timeout=120)
    if proc.returncode != 0:
        pytest.fail(f"probe failed:\n{proc.stderr}")
    return json.loads(proc.stdout)


def test_a_sealed_index_reaches_the_consumer_intact():
    out = _run([_sync_frame(GEK, "a-film.mkv", "another.mkv")])

    assert [e["event"] for e in out["events"]] == ["index_sync"]
    sync = out["events"][0]
    assert sync["entries"] == ["a-film.mkv", "another.mkv"]
    assert sync["dirs"] == ["library"]
    # Moved inside the payload (D4) and still delivered flat, so no consumer had
    # to change: it reads the same message it always read.
    assert sync["version"] == 3
    assert not sync["hasCiphertext"], "the envelope's own fields leaked to consumers"

    # And the waiting caller gets the opened form, not the envelope.
    assert out["fetchIndex"]["state"] == "resolved"
    assert out["fetchIndex"]["entries"] == ["a-film.mkv", "another.mkv"]


def test_an_index_that_does_not_open_ends_the_session():
    """
    §3.4, and the reason it is a rule rather than a preference. An empty
    `entries` is a legitimate state — a group whose operator has shared nothing
    yet — so a client that fell back to one would show the same screen for
    "nothing here" and for "we could not decrypt anything this node sent".
    """
    out = _run([_sync_frame(generate_gek(), "a-film.mkv")])

    kinds = [e["event"] for e in out["events"]]
    assert "index_sync" not in kinds, "a failed decrypt was reported as an index"
    assert kinds == ["session_failed"]
    assert "index_sync" in out["events"][0]["message"], (
        "the failure must name the message type that could not be opened")

    # The caller is told, rather than left to time out 30 s later.
    assert out["fetchIndex"]["state"] == "rejected"
    assert "index_sync" in out["fetchIndex"]["message"]
    assert out["closed"] == 1, "the session carried on after an unopenable message"


def test_a_delta_that_does_not_open_ends_the_session_too():
    """
    The delta has no caller waiting on it — it is pushed — so a silent failure
    here would leave a browser showing a stale index with nothing wrong on
    screen, which is the worst of the three shapes.
    """
    out = _run([_sync_frame(GEK, "a-film.mkv"),
                _delta_frame(generate_gek(), "new.mkv", 3, 4)])

    assert [e["event"] for e in out["events"]] == ["index_sync", "session_failed"]
    assert "index_delta" in out["events"][1]["message"]
    assert out["closed"] == 1


def test_deltas_are_applied_in_arrival_order():
    """
    Opening is asynchronous and `_dispatch` is not. Two messages opened
    independently settle in whichever order WebCrypto finishes them, and a delta
    applied before the one it follows is a wrong view of the group that nothing
    reports. Three deltas in one burst is the cheapest way to force the race.
    """
    frames = [_sync_frame(GEK, "a-film.mkv")]
    frames += [_delta_frame(GEK, f"added-{i}.mkv", 3 + i, 4 + i) for i in range(3)]

    out = _run(frames)

    assert [e["event"] for e in out["events"]] == [
        "index_sync", "index_delta", "index_delta", "index_delta"]
    assert [e["additions"][0] for e in out["events"][1:]] == [
        "added-0.mkv", "added-1.mkv", "added-2.mkv"]
    assert [e["base_version"] for e in out["events"][1:]] == [3, 4, 5]


def test_a_sealed_reply_is_opened_before_it_reaches_its_caller():
    """A stamped index_sync must not be short-circuited by its `req_id`.

    Every other reply a node stamps is resolved straight out of the pending
    map, which is the whole point of the id. An index message cannot be: it is
    sealed, opening it is asynchronous, and `_dispatch` is not. Handing it over
    on the strength of the id alone gives `fetchIndex` the envelope — nonce and
    ciphertext, no entries — and never calls `onIndexSync` at all.

    `req_id` is 0 here because it is the transport's first request, and a
    falsy id is exactly the one a presence check gets wrong.
    """
    out = _run([_sync_frame(GEK, "a-film.mkv", req_id=0)])

    assert [e["event"] for e in out["events"]] == ["index_sync"], (
        "the consumer was never told about an index that arrived as a reply")
    assert out["events"][0]["entries"] == ["a-film.mkv"]
    assert out["events"][0]["hasCiphertext"] is False
    assert out["fetchIndex"]["state"] == "resolved"
    assert out["fetchIndex"]["entries"] == ["a-film.mkv"], (
        "the caller was handed the sealed envelope instead of the index")