"""The renderers' back-channel: the registry, and the endpoints that fill and serve it.

Driven over a real socket against the real `LiveFeedServer`, because the thing being tested is a
route — an HTTP surface tested through a mocked handler proves the mock routes.
"""
import json
import sys
import urllib.error
import urllib.request
from pathlib import Path

import pytest

_LIVE = Path(__file__).resolve().parents[2] / "live"        # v3/broadcast/live
if str(_LIVE) not in sys.path:
    sys.path.insert(0, str(_LIVE))

import asrun_log  # noqa: E402
from live_feed_server import LiveFeedServer  # noqa: E402
from renderer_registry import RendererRegistry  # noqa: E402


def sample(**over):
    row = {"id": "ab12cd34", "role": "programme", "fps": 29.9, "longFrames": 2, "frames": 150,
           "windowS": 5.0, "upS": 620, "w": 1920, "h": 1080, "lang": "en", "world": "photoreal",
           "race": "Race 1009", "feedAgeMs": 380, "stale": False}
    row.update(over)
    return row


# --- the registry ---------------------------------------------------------------------------------

def test_a_renderer_is_present_because_it_is_talking_and_gone_when_it_stops():
    """Presence is a live claim, not a memory of one. A partner's box that was unplugged must stop
    being listed rather than staying for ever as one of the outputs that are fine."""
    reg = RendererRegistry(expiry_s=10.0)
    assert reg.record(sample(), now_s=100.0)
    assert len(reg.snapshot(now_s=105.0)) == 1
    assert reg.snapshot(now_s=115.0) == []



def test_the_days_page_loads_are_counted_per_world_because_that_is_how_ion_bills():
    """A Cesium ion session IS a page load: `tiles.js` fetches the world asset's `/endpoint` once per
    page and never again (for photoreal, Google's own plugin then carries the tile traffic). So a
    count of page loads by world is a count of sessions by world — the only number that maps onto the
    bill that overran in August 2026 (1118 of 1000, on one per cent of the data allowance)."""
    reg = RendererRegistry(day_fn=lambda: "2026-09-01")
    for i in range(3):
        reg.record(sample(id=f"photo{i}", world="photoreal"), now_s=100.0 + i)
    reg.record(sample(id="terra000", world="terrain"), now_s=110.0)
    # ...and a renderer that keeps talking is still ONE page load, however many beacons it sends.
    for i in range(5):
        reg.record(sample(id="photo0", world="photoreal"), now_s=120.0 + i)
    loads = reg.loads_today()
    assert loads["byWorld"] == {"photoreal": 3, "terrain": 1}
    assert loads["day"] == "2026-09-01"
    assert loads["approximate"] is False


def test_the_count_starts_from_nothing_on_a_new_day():
    """The operator asks "how many today", and their today ends at their midnight."""
    day = ["2026-08-31"]
    reg = RendererRegistry(day_fn=lambda: day[0])
    reg.record(sample(id="aaaa0000", world="photoreal"), now_s=100.0)
    assert reg.loads_today()["byWorld"] == {"photoreal": 1}
    day[0] = "2026-09-01"
    assert reg.loads_today() == {"day": "2026-09-01", "byWorld": {}, "approximate": False}
    reg.record(sample(id="bbbb0000", world="terrain"), now_s=200.0)
    assert reg.loads_today()["byWorld"] == {"terrain": 1}


def test_a_full_id_memory_says_so_rather_than_quietly_stopping():
    """The beacon is a public POST. The id memory is capped like everything else here — but a counter
    that silently stops counting is worse than one that admits it is no longer exact."""
    import renderer_registry as rr

    reg = RendererRegistry(expiry_s=1.0, day_fn=lambda: "2026-09-01")
    # A day's worth of pages that open and close: each id is seen once and has gone by the next one,
    # which is the shape the id memory has to survive (the LIVE roster stays single-digit throughout).
    n = rr.MAX_COUNTED_IDS + 5
    for i in range(n):
        reg.record(sample(id=f"{i:08x}", world="terrain"), now_s=100.0 + i * 10.0)
    loads = reg.loads_today()
    assert loads["approximate"] is True
    assert loads["byWorld"]["terrain"] == n                          # still counting, just not certain


def test_a_world_the_renderer_could_not_draw_is_reported_upward():
    """`?world=` used to accept anything and silently draw photoreal, so a typo was invisible AND
    billed. The renderer now falls to terrain and says which name it could not honour — nobody reads
    an OBS browser source's console."""
    reg = RendererRegistry()
    reg.record(sample(world="terrain", worldRequested="phtoreal"), now_s=100.0)
    assert reg.snapshot(now_s=100.0)[0]["worldRequested"] == "phtoreal"


def test_the_newest_sample_wins_and_the_age_is_carried_not_the_timestamp():
    """Two machines do not share a clock, so an age travels correctly where a timestamp would not."""
    reg = RendererRegistry()
    reg.record(sample(fps=10.0), now_s=100.0)
    reg.record(sample(fps=30.0), now_s=104.0)
    rows = reg.snapshot(now_s=105.0)
    assert len(rows) == 1 and rows[0]["fps"] == 30.0
    assert rows[0]["ageMs"] == 1000


def test_race_bound_renderer_health_fields_are_preserved_for_a_monitor():
    """A hand-over diagnosis needs both the immediate hitch window and the current-race totals."""
    reg = RendererRegistry()
    reg.record(sample(frameMaxMs=302, hitch250=2, raceAgeS=97, raceHitch250=11,
                      raceMaxFrameMs=1043, sceneGeometries=85, jsHeapMB=1420,
                      feedHandovers=2, feedLastHandoverAgeMs=600,
                      cursorSec=318.25, frontierSec=340.5, presentationLagS=22.25), now_s=1.0)
    row = reg.snapshot(now_s=1.0)[0]
    assert row["frameMaxMs"] == 302
    assert row["raceHitch250"] == 11
    assert row["raceMaxFrameMs"] == 1043
    assert row["sceneGeometries"] == 85
    assert row["jsHeapMB"] == 1420
    assert row["feedHandovers"] == 2
    assert row["cursorSec"] == 318.25
    assert row["frontierSec"] == 340.5
    assert row["presentationLagS"] == 22.25


def test_a_renderer_that_cannot_speak_says_so():
    """The one fault nothing outside the renderer can see.

    A browser source draws a flawless picture with its commentary scheduler shut, and OBS, the
    producer and YouTube's own health check all read green over pure silence — which is how a live
    stream went out with no sound and nothing reported it (2026-08-11). The page's own word for what
    its voice is doing therefore has to survive the registry rather than being dropped as an unknown
    field, and `blocked` (will never speak) has to stay distinguishable from `none` (nothing yet)."""
    reg = RendererRegistry()
    reg.record(sample(id="mute", audioBackend="blocked"), now_s=1.0)
    reg.record(sample(id="live", audioBackend="neural"), now_s=1.0)
    by_id = {r["id"]: r for r in reg.snapshot(now_s=1.0)}
    assert by_id["mute"]["audioBackend"] == "blocked"
    assert by_id["live"]["audioBackend"] == "neural"


def test_the_slowest_renderer_sorts_first():
    reg = RendererRegistry()
    reg.record(sample(id="fast", fps=30.0), now_s=1.0)
    reg.record(sample(id="slow", fps=8.0), now_s=1.0)
    assert [r["id"] for r in reg.snapshot(now_s=1.0)] == ["slow", "fast"]


@pytest.mark.parametrize("payload", [None, "a string", 42, [], {}, {"id": ""}, {"id": "   "}])
def test_an_unusable_payload_is_refused_rather_than_stored(payload):
    assert RendererRegistry().record(payload) is False


def test_untrusted_fields_are_clamped():
    """This is a public POST from a browser we do not control — under a partner arrangement, from a
    browser on someone else's premises."""
    reg = RendererRegistry()
    reg.record({"id": "x" * 200, "role": "r" * 500, "fps": "not a number",
                "lang": "en\x00\x07evil", "w": float("nan")}, now_s=1.0)
    row = reg.snapshot(now_s=1.0)[0]
    assert len(row["id"]) <= 16
    assert len(row["role"]) <= 40
    assert "fps" not in row, "a non-numeric fps must be dropped, not coerced"
    assert "w" not in row, "NaN must never reach a threshold"
    assert "\x00" not in row["lang"]


def test_the_roster_is_capped_so_the_endpoint_cannot_grow_the_process():
    reg = RendererRegistry(max_renderers=3)
    for i in range(10):
        reg.record(sample(id=f"r{i}"), now_s=1.0)
    assert len(reg.snapshot(now_s=1.0)) == 3


def test_a_known_renderer_still_updates_when_the_registry_is_full():
    reg = RendererRegistry(max_renderers=2)
    reg.record(sample(id="a", fps=30.0), now_s=1.0)
    reg.record(sample(id="b", fps=30.0), now_s=1.0)
    assert reg.record(sample(id="a", fps=12.0), now_s=2.0) is True
    assert reg.record(sample(id="c"), now_s=2.0) is False, "a full registry refuses a NEW one"
    assert {r["id"]: r["fps"] for r in reg.snapshot(now_s=2.0)}["a"] == 12.0


# --- the served surface ---------------------------------------------------------------------------

@pytest.fixture
def server():
    srv = LiveFeedServer(host="127.0.0.1", port=0)
    import threading
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{srv.port}", srv
    srv.shutdown()
    srv.server_close()


def _post(url, body: bytes):
    """POST `body` and return the status code, retrying once if the connection is aborted.

    Every refusal path here answers and then closes without draining the request body, so on Windows
    the close can win the race against the client still writing: the client gets
    `ConnectionAbortedError` (WinError 10053) and never reads the status it was refused with. That
    made several tests in this file fail intermittently under I/O load, a different one each run —
    and with CI off, a suite that cries wolf is how people learn to push past red.

    A retry is safe *here specifically* because the requests that hit this are the ones the server
    rejects: nothing was processed, so nothing is repeated. Do not reuse this helper for a POST whose
    success has side effects. Retrying rather than accepting the abort keeps every assertion exact —
    the tests still demand their 400 or 404 rather than settling for "something went wrong".
    """
    req = urllib.request.Request(url, data=body, method="POST",
                                 headers={"Content-Type": "application/json"})
    for attempt in (1, 2):
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                return resp.status
        except urllib.error.HTTPError as exc:
            return exc.code
        except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError):
            if attempt == 2:
                raise
    raise AssertionError("unreachable")


def _get(url):
    try:
        with urllib.request.urlopen(url, timeout=5) as resp:
            return resp.status, json.loads(resp.read() or b"null")
    except urllib.error.HTTPError as exc:
        return exc.code, None


def test_a_beacon_lands_and_appears_in_the_stats(server):
    url, _ = server
    assert _post(f"{url}/client-stats", json.dumps(sample()).encode()) == 204
    status, body = _get(f"{url}/stats.json")
    assert status == 200
    assert [r["id"] for r in body["renderers"]] == ["ab12cd34"]
    assert body["renderers"][0]["race"] == "Race 1009"


def test_the_beacon_answers_204_with_no_body_so_the_channel_stays_one_way(server):
    """Nothing is ever read back by a renderer: the channel cannot become a route for telling the
    picture what to do."""
    url, _ = server
    req = urllib.request.Request(f"{url}/client-stats", data=json.dumps(sample()).encode(),
                                 method="POST")
    with urllib.request.urlopen(req, timeout=5) as resp:
        assert resp.status == 204
        assert resp.read() == b""


@pytest.mark.parametrize("body", [b"", b"not json", b"null", b'{"no":"id"}'])
def test_a_malformed_beacon_is_refused_without_disturbing_the_server(server, body):
    url, _ = server
    assert _post(f"{url}/client-stats", body) == 400
    assert _get(f"{url}/stats.json")[1]["renderers"] == []
    assert _get(f"{url}/healthz")[0] == 200, "the server is still serving"


def test_an_oversized_beacon_is_refused(server):
    url, _ = server
    assert _post(f"{url}/client-stats", b'{"id":"a","role":"' + b"x" * 8000 + b'"}') == 400


def test_commentary_selection_and_actual_audio_start_land_in_the_asrun(server, tmp_path):
    url, srv = server
    log = tmp_path / "asrun.jsonl"
    srv._asrun_path = log
    srv._now_ms = lambda: 2_000
    variant = {
        "tSec": 12, "text": "Next mark for Boat 4 is M2.",
        "commentaryVariantCategory": "heading", "commentaryVariantIndex": 2,
        "commentaryVariantCount": 4, "commentaryVariantId": "heading-race7-12",
    }
    srv.publish_init({"instanceId": "race-7", "raceName": "Race 7"}, {}, [], {})
    srv.publish_delta(12, {}, [variant], {})
    assert _post(f"{url}/commentary-playback", json.dumps({
        **variant, "rendererId": "ab12cd34", "rendererRole": "programme",
        "audioBackend": "neural", "lang": "en", "clientUtcMs": 1_990,
        "instanceId": "race-7", "raceName": "Race 7", "sourceSec": 12,
    }).encode()) == 204

    rows = asrun_log.read_rows(log)
    assert [row["event"] for row in rows] == [
        asrun_log.COMMENTARY_SELECTED, asrun_log.COMMENTARY_STARTED]
    assert rows[1]["tUtcMs"] == 2_000
    assert rows[1]["rendererRole"] == "programme"
    assert rows[1]["commentaryVariantIndex"] == 2

    # Browser callbacks can repeat across pause/resume. The route remains successful but one
    # renderer + variant contributes only one actual-start row.
    assert _post(f"{url}/commentary-playback", json.dumps({
        **variant, "rendererId": "ab12cd34", "rendererRole": "programme",
        "audioBackend": "neural", "lang": "en", "clientUtcMs": 2_010,
        "instanceId": "wrong-current-race", "raceName": "Wrong", "sourceSec": 12,
    }).encode()) == 204
    assert len(asrun_log.read_rows(log)) == 2


def test_malformed_commentary_playback_is_refused(server):
    url, _ = server
    assert _post(f"{url}/commentary-playback", b'{"rendererId":"x"}') == 400


def test_a_browser_cannot_invent_or_relabel_a_commentary_variant(server):
    url, srv = server
    variant = {
        "tSec": 12, "text": "Selected line.",
        "commentaryVariantCategory": "heading", "commentaryVariantIndex": 2,
        "commentaryVariantCount": 4, "commentaryVariantId": "selected-variant",
    }
    srv.publish_init({"instanceId": "race-7", "raceName": "Race 7"}, {}, [], {})
    srv.publish_delta(12, {}, [variant], {})
    base = {
        **variant, "rendererId": "ab12cd34", "rendererRole": "programme",
        "audioBackend": "neural", "lang": "en", "sourceSec": 12,
    }
    assert _post(f"{url}/commentary-playback", json.dumps({
        **base, "commentaryVariantId": "never-published",
    }).encode()) == 400
    assert _post(f"{url}/commentary-playback", json.dumps({
        **base, "commentaryVariantIndex": 3,
    }).encode()) == 400


@pytest.mark.parametrize("path", ["/healthz", "/livez"])
def test_liveness_answers_on_both_paths(server, path):
    """Board item T-020's other half: what a supervisor restarts the producer against.

    BOTH paths, and `/livez` is the one that matters in production: Google's front end swallows
    `/healthz` on Cloud Run — measured on the deployed ops service, it returned Google's own 404 and
    the request never reached the container, while every other path arrived normally. `/healthz` is
    kept for a local run and for anything behind our own proxy.
    """
    url, _ = server
    status, body = _get(f"{url}{path}")
    assert status == 200 and body["ok"] is True and body["clients"] == 0


def test_liveness_reports_the_producer_build_when_deployed(server, monkeypatch):
    url, _ = server
    monkeypatch.setenv("VR_BUILD_SHA", "a" * 40)

    status, body = _get(f"{url}/livez")

    assert status == 200
    assert body["buildSha"] == "a" * 40


def test_posting_anywhere_else_is_a_404(server):
    url, _ = server
    assert _post(f"{url}/events", b"{}") == 404
    assert _post(f"{url}/", b"{}") == 404


def test_stats_is_behind_the_feed_token_when_one_is_set(server, monkeypatch):
    """Who is rendering the championship, and how well, is not a fact for whoever guesses the
    hostname — so it sits behind the same gate as the feed itself."""
    url, _ = server
    monkeypatch.setenv("VR_FEED_TOKEN", "s3cret")
    assert _get(f"{url}/stats.json")[0] == 403
    assert _get(f"{url}/stats.json?token=s3cret")[0] == 200
    assert _get(f"{url}/healthz")[0] == 200, "liveness discloses nothing and stays open"