#!/usr/bin/env python3
"""Telemetry safety contract: append-only, fail-open, and control-path neutral.

No live services and no FFmpeg.  The important production guarantee is not
that storage can never fail, but that telemetry can fail/drop its own records
without raising into, blocking, or steering a stream.
"""
import json
import os
import pathlib
import sys
import tempfile
import time

ROOT = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))

from telemetry import Telemetry, http_status_from_exception


passed = failed = 0


def check(name, condition, detail=""):
    global passed, failed
    if condition:
        passed += 1
        print("PASS  %s" % name)
    else:
        failed += 1
        print("FAIL  %s  %s" % (name, detail))


def wait_records(store, count, timeout=3.0):
    deadline = time.time() + timeout
    rows = []
    while time.time() < deadline:
        rows = store.read_records(100)
        if len(rows) >= count:
            return rows
        time.sleep(0.02)
    return rows


with tempfile.TemporaryDirectory() as tmp:
    path = pathlib.Path(tmp) / "telemetry" / "telemetry.jsonl"
    store = Telemetry(path)
    key = store.start_session("alice", {
        "mode": "copy",
        "url": "https://signed.example/private?token=should-not-persist",
        "nested": {"stream_key": "should-not-persist", "room_id": "123"},
    })
    check("session starts", bool(key))
    check("observe accepts ffmpeg progress", store.observe(key, {
        "ffmpeg_speed": 0.80, "ffmpeg_fps": 30.0, "source_bps": 900_000,
        "output_seconds": 12.0,
    }))
    check("observe accepts recovered speed", store.observe(key, {
        "ffmpeg_speed": 1.01, "ffmpeg_fps": 30.0, "source_bps": 2_400_000,
        "output_seconds": 27.0,
    }))

    class Response:
        status_code = 403

    class HttpFailure(Exception):
        response = Response()

    check("http status extracted without message", http_status_from_exception(HttpFailure()) == 403)
    check("source-open event accepted", store.event(key, "alice", "SOURCE_OPEN_ERROR", {
        "stage": "initial_open", "error_type": "HTTPError", "http_status": 403,
        "url": "https://signed.example/do-not-store",
    }))
    check("resolution event accepted", store.event(key, "alice", "RESOLUTION_CHANGE", {
        "old_width": 720, "old_height": 1280, "new_width": 428, "new_height": 856,
    }))
    check("checkpoint accepted", store.checkpoint(key, "unit-test"))
    check("finish accepted", store.finish(key, "unit finished", {"output_seconds": 27.0}))

    rows = wait_records(store, 5)
    kinds = [row.get("type") for row in rows]
    check("writer persisted all record kinds", kinds == [
        "session_start", "event", "event", "checkpoint", "session_finish"
    ], str(kinds))
    raw = path.read_text(encoding="utf-8")
    check("signed URL is not persisted", "signed.example" not in raw and "should-not-persist" not in raw)
    finish = [row for row in rows if row.get("type") == "session_finish"][-1]
    summary = finish["summary"]
    check("summary identifies broadcaster", summary["broadcaster"] == "alice")
    check("summary counts source-open status", summary["counters"].get("source_open_http_403") == 1,
          str(summary["counters"]))
    check("summary counts slow ffmpeg samples", summary["counters"].get("ffmpeg_speed_below_0_95_samples") == 1,
          str(summary["counters"]))
    check("summary averages ffmpeg speed", abs(summary["stats"]["ffmpeg_speed"]["avg"] - 0.905) < 0.0001,
          str(summary["stats"].get("ffmpeg_speed")))
    check("summary keeps minute buckets", len(summary["minutes"]) == 1, str(summary["minutes"]))

    before = path.read_bytes()
    reopened = Telemetry(path)
    check("reopening telemetry does not truncate", path.read_bytes() == before)
    check("reopened store reads old records", len(reopened.read_records(10)) == len(rows))
    check("no delete/rotate/truncate API", not any(
        hasattr(Telemetry, name) for name in ("delete", "rotate", "truncate", "clear", "purge")
    ))

    os.environ["TELEMETRY_ENABLED"] = "0"
    try:
        disabled_path = pathlib.Path(tmp) / "disabled" / "telemetry.jsonl"
        disabled = Telemetry(disabled_path)
        check("disabled telemetry is a no-op", disabled.start_session("bob") is None and not disabled_path.exists())
    finally:
        os.environ.pop("TELEMETRY_ENABLED", None)

    blocker = pathlib.Path(tmp) / "blocker"
    blocker.write_text("not a directory", encoding="utf-8")
    broken = Telemetry(blocker / "telemetry" / "telemetry.jsonl")
    broken_key = broken.start_session("carol")
    broken.observe(broken_key, {"ffmpeg_speed": 1.0})
    broken.finish(broken_key, "storage unavailable")
    deadline = time.time() + 3.0
    while time.time() < deadline and broken.write_errors < 2:
        time.sleep(0.02)
    check("storage failure stays inside telemetry", broken.write_errors >= 2,
          "write_errors=%s" % broken.write_errors)


relay_source = (ROOT / "relay.py").read_text(encoding="utf-8")
check("telemetry result is never a streaming condition", "if telemetry_" not in relay_source and
      "while telemetry_" not in relay_source)
check("relay records HTTP status at initial open", "'stage':'initial_open'" in relay_source and
      "http_status_from_exception" in relay_source)

print("\n%d passed, %d failed" % (passed, failed))
sys.exit(1 if failed else 0)