"""The as-run log: what went to air, and the partition it reads back as.

Pure over rows and a temp file — no servers, no sockets, no sleeps, matching how `race_queue` and
`recording_anchor` are tested.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

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

import asrun_log  # noqa: E402


def _on(t, inst, **kw):
    return asrun_log.on_air_row(now_ms=t, instance_id=inst, **kw)


def _off(t, inst):
    return asrun_log.off_air_row(now_ms=t, instance_id=inst)


# --- the shape of a row -------------------------------------------------------------------------

def test_an_on_air_row_records_the_race_it_displaced():
    """Only the producer knows what a race replaced — the queue hands over without telling either
    race. Recorded rather than inferred, so a hand-over is distinguishable from a fresh start."""
    row = _on(1000, "r2", race_name="Race 1010", gun_utc_ms=1200, leaving_instance_id="r1")
    assert row["event"] == "on_air"
    assert row["instanceId"] == "r2"
    assert row["raceName"] == "Race 1010"
    assert row["gunUtcMs"] == 1200
    assert row["leavingInstanceId"] == "r1"


def test_commentary_rows_separate_selection_from_an_audible_renderer_start():
    commentary = {
        "tSec": 42, "text": "A rotated line.",
        "commentaryVariantCategory": "heading", "commentaryVariantIndex": 3,
        "commentaryVariantCount": 4, "commentaryVariantId": "variant-42",
    }
    selected = asrun_log.commentary_selected_row(
        now_ms=1000, instance_id="race-7", race_name="Race 7", source_sec=42,
        commentary=commentary)
    started = asrun_log.commentary_started_row(now_ms=1250, payload={
        **commentary, "rendererId": "abcd1234", "rendererRole": "programme",
        "audioBackend": "neural", "lang": "en", "clientUtcMs": 1240,
        "instanceId": "race-7", "raceName": "Race 7", "sourceSec": 42,
    })
    assert selected["event"] == asrun_log.COMMENTARY_SELECTED
    assert started["event"] == asrun_log.COMMENTARY_STARTED
    assert started["tUtcMs"] == 1250, "the server clock, not the browser clock, owns the as-run"
    assert started["commentaryVariantCategory"] == "heading"
    assert started["commentaryVariantIndex"] == 3
    assert started["rendererRole"] == "programme"


@pytest.mark.parametrize("payload", [
    {},
    {"commentaryVariantCategory": "heading", "commentaryVariantIndex": 0,
     "commentaryVariantCount": 4, "commentaryVariantId": "v", "rendererId": "r",
     "audioBackend": "neural"},
    {"commentaryVariantCategory": "heading", "commentaryVariantIndex": 5,
     "commentaryVariantCount": 4, "commentaryVariantId": "v", "rendererId": "r",
     "audioBackend": "browser"},
])
def test_invalid_commentary_playback_receipts_are_refused(payload):
    assert asrun_log.commentary_started_row(now_ms=1000, payload=payload) is None


# --- reading it back ----------------------------------------------------------------------------

def test_a_race_that_is_still_on_air_has_no_end():
    segs = asrun_log.segments([_on(100, "r1")])
    assert len(segs) == 1
    assert segs[0]["fromUtcMs"] == 100
    assert segs[0]["toUtcMs"] is None          # "what is playing right now", with no second endpoint


def test_off_air_closes_the_segment():
    segs = asrun_log.segments([_on(100, "r1"), _off(900, "r1")])
    assert (segs[0]["fromUtcMs"], segs[0]["toUtcMs"]) == (100, 900)


def test_a_race_to_race_handover_closes_the_previous_race():
    """THE CASE THAT MATTERS. `take_next` swaps the race in place without releasing, so a direct
    hand-over emits an `on_air` and NO `off_air`. A reader expecting a matching close would leave
    race 1 open for ever and invent a gap that was never off air."""
    segs = asrun_log.segments([_on(100, "r1"), _on(900, "r2", leaving_instance_id="r1")])
    assert [s["instanceId"] for s in segs] == ["r1", "r2"]
    assert segs[0]["toUtcMs"] == 900           # closed by r2 taking the air
    assert segs[1]["fromUtcMs"] == 900
    assert segs[1]["toUtcMs"] is None


def test_commentary_audit_rows_do_not_change_the_race_partition():
    rows = [_on(100, "r1"), {"tUtcMs": 200, "event": asrun_log.COMMENTARY_SELECTED},
            {"tUtcMs": 300, "event": asrun_log.COMMENTARY_STARTED}, _off(900, "r1")]
    assert asrun_log.segments(rows) == [{
        "instanceId": "r1", "raceName": None, "gunUtcMs": None,
        "fromUtcMs": 100, "toUtcMs": 900,
    }]


def test_the_timeline_is_a_partition_with_no_gaps_or_overlaps():
    """The property the whole race-delimitation contract rests on: each segment ends exactly where
    the next begins, so 'which race was on screen at T' is a lookup rather than a search."""
    rows = [_on(0, "r1"), _on(500, "r2"), _off(900, "r2"), _on(1500, "r3")]
    segs = asrun_log.segments(rows, session_end_ms=2000)
    assert [(s["fromUtcMs"], s["toUtcMs"]) for s in segs] == [(0, 500), (500, 900), (1500, 2000)]
    # r2 -> r3 is a genuine gap: the producer was on the waiting picture, and it is visible as one
    # rather than being papered over.
    for a, b in zip(segs, segs[1:], strict=False):
        assert a["toUtcMs"] <= b["fromUtcMs"]


def test_session_end_closes_a_race_still_on_air():
    segs = asrun_log.segments([_on(100, "r1")], session_end_ms=800)
    assert segs[0]["toUtcMs"] == 800


def test_rows_are_ordered_by_time_not_by_file_order():
    """One loop appends these so they arrive ordered — but a reader that ASSUMED it would break
    silently on a concatenated or merged file."""
    segs = asrun_log.segments([_on(900, "r2"), _on(100, "r1")])
    assert [s["instanceId"] for s in segs] == ["r1", "r2"]
    assert segs[0]["toUtcMs"] == 900


def test_an_off_air_for_a_race_that_is_not_on_air_is_ignored():
    """A stale close must not truncate the race that IS on air."""
    segs = asrun_log.segments([_on(100, "r1"), _off(500, "r_other")])
    assert segs[0]["toUtcMs"] is None


# --- the file -----------------------------------------------------------------------------------

def test_append_then_read_round_trips(tmp_path):
    p = tmp_path / "sub" / "asrun.jsonl"                     # parent does not exist yet
    assert asrun_log.append(p, _on(100, "r1")) is True
    assert asrun_log.append(p, _off(900, "r1")) is True
    rows = asrun_log.read_rows(p)
    assert [r["event"] for r in rows] == ["on_air", "off_air"]
    assert asrun_log.segments(rows)[0]["toUtcMs"] == 900


def test_a_truncated_last_line_is_dropped_rather_than_raising(tmp_path):
    """A run killed mid-write leaves a partial line. The file is an append log and its tail is the
    least valuable part of it — same rule as `recording_anchor.read_samples`."""
    p = tmp_path / "asrun.jsonl"
    p.write_text(json.dumps(_on(100, "r1")) + "\n" + '{"tUtcMs": 900, "eve', encoding="utf-8")
    rows = asrun_log.read_rows(p)
    assert len(rows) == 1
    assert rows[0]["instanceId"] == "r1"


def test_read_rows_on_a_missing_file_is_empty_not_an_error(tmp_path):
    assert asrun_log.read_rows(tmp_path / "nope.jsonl") == []


def test_logging_is_off_when_no_path_is_configured():
    assert asrun_log.append(None, _on(100, "r1")) is False


def test_a_write_failure_never_reaches_the_broadcast_loop(tmp_path):
    """This is called from the live loop. A full disk or a read-only mount costs a log line and
    nothing else — the show outranks the log."""
    p = tmp_path / "asrun.jsonl"
    p.write_text("", encoding="utf-8")
    said = []

    def boom(*a, **k):
        raise OSError("disk full")

    orig = Path.open
    Path.open = boom                                          # noqa: SLF001 — deliberate fault injection
    try:
        assert asrun_log.append(p, _on(100, "r1"), on_log=said.append) is False
    finally:
        Path.open = orig
    assert said and "disk full" in said[0]


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q"]))