"""The printer camera hub (api/printer_camera.py): guest pacing and the pull API.

Regressions this pins, all measured on the live page 2026-09-26:
  - "5fps" guests got 3.3fps: a strict `gap < interval` throttle skipped the
    frame landing at 199ms of a 200ms interval, so every other gap was 300ms.
  - a slow guest fell seconds behind: pushed MJPEG queues in socket buffers the
    hub cannot see. Guests now pull one frame per request (`next_frame`), which
    cannot queue, and share ONE server-side sample however fast they ask.
  - still laggy on an iPhone (Safari and Chrome): guests were capped at 5fps
    against the owner's 10, and one pull at a time capped a phone at
    1/(RTT + transfer). Guests now get the full rate with PULL_DEPTH pulls in
    flight; the live test below runs AS A GUEST at a phone's round trip, and
    skips (not fails) when the camera itself is delivering under the bar.
"""

import asyncio
import os
import sys

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "api"))

import printer_camera  # noqa: E402
from printer_camera import GUEST_FRAME_INTERVAL, PULL_DEPTH, _Camera  # noqa: E402

UPSTREAM_FPS = 10  # what the Centauri Carbon pushes


class _Clock:
    def __init__(self):
        self.t = 1000.0

    def __call__(self):
        return self.t


@pytest.fixture
def clock(monkeypatch):
    c = _Clock()
    monkeypatch.setattr(printer_camera, "monotonic", c)
    return c


@pytest.fixture
def cam():
    c = _Camera()
    c._ensure_running = lambda: None  # never dial a printer from a test
    return c


def _jittered_gaps(n, seed=7):
    """Upstream gaps: 100ms +-5ms, the jitter measured off the real camera."""
    import random
    rng = random.Random(seed)
    return [1 / UPSTREAM_FPS + rng.uniform(-0.005, 0.005) for _ in range(n)]


def _promoted(cam, clock, n=200):
    out = []
    for i, gap in enumerate(_jittered_gaps(n)):
        clock.t += gap
        before = cam._guest_seq
        cam._publish(b"f%d" % i)
        if cam._guest_seq != before:
            out.append(clock.t)
    return out


def test_guests_get_every_upstream_frame(cam, clock):
    """The guest sample is the printer's full rate, not a fraction of it."""
    promoted = _promoted(cam, clock)
    assert len(promoted) == 200


def test_guest_ceiling_under_jitter_skips_no_extra_frame(cam, clock, monkeypatch):
    """Were the ceiling ever to bite (interval 2x the upstream's), it must
    pass every other frame and never leave a 3-frame hole."""
    monkeypatch.setattr(printer_camera, "GUEST_FRAME_INTERVAL", 0.2)
    promoted = _promoted(cam, clock)
    gaps = [b - a for a, b in zip(promoted, promoted[1:])]
    fps = len(gaps) / (promoted[-1] - promoted[0])
    assert 4.75 <= fps <= 5.25
    assert max(gaps) < 0.25


def test_mjpeg_throttle_is_a_true_5fps_under_jitter(cam, clock):
    """The pushed body's own throttle carries the same slack."""
    async def run():
        gen = cam.frames(min_interval=0.2)
        sent = []
        first = asyncio.ensure_future(gen.__anext__())
        await asyncio.sleep(0)
        for i, gap in enumerate(_jittered_gaps(120)):
            clock.t += gap
            cam._publish(b"f%d" % i)
            await asyncio.sleep(0)
            if first is not None and first.done():
                sent.append(clock.t)
                first = asyncio.ensure_future(gen.__anext__())
                await asyncio.sleep(0)
        first.cancel()
        try:
            await first
        except (asyncio.CancelledError, StopAsyncIteration):
            pass
        return sent

    sent = asyncio.run(run())
    gaps = [b - a for a, b in zip(sent, sent[1:])]
    assert max(gaps) < 0.25
    assert len(sent) >= 55  # 12s of upstream at 5fps, less the first frame


def test_first_pull_gets_the_current_frame_at_once(cam, clock):
    cam._publish(b"jpeg-1")
    seq, frame = asyncio.run(cam.next_frame(after=0, wait=0.01))
    assert frame == b"jpeg-1" and seq == 1


def test_pull_waits_for_a_newer_frame(cam, clock):
    async def run():
        cam._publish(b"old")
        waiter = asyncio.ensure_future(cam.next_frame(after=1, wait=5))
        await asyncio.sleep(0)
        assert not waiter.done()  # nothing newer than what the caller holds
        clock.t += GUEST_FRAME_INTERVAL
        cam._publish(b"new")
        return await waiter

    assert asyncio.run(run()) == (2, b"new")


def test_pull_times_out_to_none_not_a_stale_repeat(cam, clock):
    cam._publish(b"only")
    assert asyncio.run(cam.next_frame(after=1, wait=0.01)) is None


def test_polling_fast_cannot_beat_the_guest_rate(cam, clock, monkeypatch):
    """A guest asking as fast as it likes still sees only the shared sample."""
    monkeypatch.setattr(printer_camera, "GUEST_FRAME_INTERVAL", 0.2)

    async def run():
        seen, seq = [], 0
        for i in range(40):
            clock.t += 1 / UPSTREAM_FPS
            cam._publish(b"f%d" % i)
            got = await cam.next_frame(after=seq, wait=0.001)
            if got:
                seq = got[0]
                seen.append(got[1])
        return seen

    assert len(asyncio.run(run())) == 20


def test_a_pipelined_guest_counts_once_against_the_cap(cam, clock, monkeypatch):
    monkeypatch.setattr(printer_camera, "MAX_VIEWERS", 2)

    async def run():
        one_guest = [asyncio.ensure_future(cam.next_frame(after=0, wait=5))
                     for _ in range(PULL_DEPTH)]
        await asyncio.sleep(0)
        room_for_second = cam.can_admit()
        second = asyncio.ensure_future(cam.next_frame(after=0, wait=5))
        await asyncio.sleep(0)
        full = not cam.can_admit()
        for f in one_guest + [second]:
            f.cancel()
        await asyncio.sleep(0)
        return room_for_second, full

    assert asyncio.run(run()) == (True, True)
    assert cam.can_admit()  # cancelled pulls release their slots


def test_upstream_stays_wanted_between_pulls_then_lets_go(cam, clock):
    cam._publish(b"x")
    asyncio.run(cam.next_frame(after=0, wait=0.01))
    clock.t += 1.0
    assert cam._wanted()  # the gap between two pulls must not hang up
    clock.t += printer_camera._IDLE_STOP_S
    assert not cam._wanted()


# ── the route, against the live container (conftest `client`) ─────────────
def test_frame_route_refuses_anonymous(client):
    assert client.get("/printer/frame").status_code == 401


def _printer_up(client, guest_cookie):
    r = client.get("/api/printer/health", headers=guest_cookie)
    if r.status_code != 200:
        pytest.skip("printer / home tunnel offline — no camera to measure")


def test_frame_route_serves_a_guest_one_jpeg_with_its_seq(client, guest_cookie):
    """Also proves /printer/frame is not shadowed by the owner-only
    /printer/{path} catch-all, which would bounce a guest to the admin login."""
    _printer_up(client, guest_cookie)
    r = client.get("/printer/frame?after=0", headers=guest_cookie)
    assert r.status_code == 200, r.text[:200]
    assert r.headers["content-type"] == "image/jpeg"
    assert r.content[:2] == b"\xff\xd8"
    assert int(r.headers["x-frame-seq"]) > 0
    assert r.headers["cache-control"] == "no-store"
    assert "content-encoding" not in r.headers  # a JPEG is never re-gzipped


def test_guest_on_a_phone_round_trip_gets_the_full_rate(guest_cookie):
    """What web/printer.js does, AS A GUEST, over the public edge, with a
    phone's round trip added (150ms, split either side of each request):
    PULL_DEPTH pulls in flight, each asking for the frame after the last one
    asked for. Must hold near the camera's ~10fps with no visible hole, and
    every frame must arrive fresh."""
    import threading
    import time

    import httpx
    from conftest import BASE_URL
    RTT = 0.15
    with httpx.Client(base_url=BASE_URL, timeout=15.0) as c:
        if c.get("/api/printer/health", headers=guest_cookie).status_code != 200:
            pytest.skip("printer / home tunnel offline — no camera to measure")
        lock = threading.Lock()
        state = {"asked": 0, "shown": 0}
        got, t0 = [], time.monotonic()

        def worker():
            while time.monotonic() - t0 < 5.0:
                with lock:
                    after = state["asked"]
                    if after:
                        state["asked"] += 1
                time.sleep(RTT / 2)
                r = c.get(f"/printer/frame?after={after}", headers=guest_cookie)
                time.sleep(RTT / 2)
                with lock:
                    if r.status_code != 200:
                        state["asked"] = state["shown"]
                        continue
                    seq = int(r.headers["x-frame-seq"])
                    state["asked"] = max(state["asked"], seq)
                    if seq > state["shown"]:
                        state["shown"] = seq
                        got.append((time.monotonic(), seq))

        threads = [threading.Thread(target=worker) for _ in range(PULL_DEPTH)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
    got = got[3:]  # the first pulls both take the frame already there
    span = got[-1][0] - got[0][0]
    # The guest seq advances once per frame the hub PROMOTED, so its span over
    # the run is the rate the camera actually delivered. A slow home uplink or
    # camera (measured 6.6fps for minutes on 2026-10-07) is not a regression
    # here and must not block unrelated commits: skip, don't fail.
    hub_fps = (got[-1][1] - got[0][1]) / span
    if hub_fps < 1 / GUEST_FRAME_INTERVAL * 0.8:
        pytest.skip(f"camera upstream slow ({hub_fps:.2f}fps) — nothing to measure the pull against")
    times = [t for t, _ in got]
    gaps = [b - a for a, b in zip(times, times[1:])]
    fps = len(gaps) / span
    assert fps >= 1 / GUEST_FRAME_INTERVAL * 0.8, f"{fps:.2f}fps"
    assert max(gaps) < 0.3, f"max gap {max(gaps) * 1000:.0f}ms"