"""Native RRD empty reads are pauses, not peer EOF. No device/network involved."""
import time

from k1link.data_plane import ConsumerFrameContext, DecodedPointCloudView
from k1link.viewer.node_rerun import NodeRerunBridge, RrdSubscriber
from k1link.viewer.rerun_bridge import RerunSceneSettings


def points(sequence=1, age=0):
    now = time.monotonic_ns()
    return DecodedPointCloudView(
        context=ConsumerFrameContext(sequence, time.time_ns(), now - int(age * 1e9), now, 1, True),
        frame_id="world", positions_xyz=((1.0, 2.0, 3.0),), intensities=b"\x01",
    )


def drain_until(subscriber, predicate, seconds=3):
    deadline = time.monotonic() + seconds
    payloads = []
    while time.monotonic() < deadline:
        payload = subscriber.read()
        assert payload is not None, "idle native binary read retired the live subscriber"
        if payload:
            assert payload.startswith(b"RRF2")
            payloads.append(payload)
        if predicate():
            return payloads
    raise AssertionError("native subscriber did not make progress")


def test_native_idle_read_keeps_subscriber_alive_and_resumes():
    subscriber = RrdSubscriber(RerunSceneSettings)
    try:
        assert drain_until(subscriber, lambda: subscriber.output.empty())
        # This forces several SDK reads with no data; the previous len(None)
        # retired both RRD and camera before a first point could arrive.
        assert subscriber.read() == b""
        assert not subscriber.closed.is_set()
        subscriber.offer(points())
        assert drain_until(subscriber, lambda: subscriber.snapshot()["sequence"] == 1)
        assert subscriber.read() == b""
        subscriber.offer(points(2))
        assert drain_until(subscriber, lambda: subscriber.snapshot()["sequence"] == 2)
        assert subscriber.snapshot()["points"] == 1
        assert not subscriber.closed.is_set()
    finally:
        subscriber.close()
        subscriber.thread.join(timeout=3)
    assert not subscriber.thread.is_alive()


def test_subscriber_does_not_publish_queued_stale_points():
    subscriber = RrdSubscriber(RerunSceneSettings)
    try:
        drain_until(subscriber, lambda: subscriber.output.empty())
        subscriber.offer(points(age=5))
        assert subscriber.read() == b""
        assert subscriber.snapshot()["sequence"] == 0
        subscriber.offer(points(2))
        drain_until(subscriber, lambda: subscriber.snapshot()["sequence"] == 2)
    finally:
        subscriber.close()
        subscriber.thread.join(timeout=3)


def test_reopened_viewer_does_not_replay_cached_points_after_source_pause(monkeypatch):
    bridge = NodeRerunBridge()
    try:
        bridge.process(points())
        cached_at, frame = next(iter(bridge.latest.values()))
        bridge.latest[type(frame)] = (cached_at - 5, frame)
        sub = bridge.subscribe()
        drain_until(sub, lambda: sub.output.empty())
        assert sub.read() == b""
        assert sub.snapshot()["sequence"] == 0
        bridge.process(points(2))
        drain_until(sub, lambda: sub.snapshot()["sequence"] == 2)
    finally:
        bridge.close()
        sub.thread.join(timeout=3)