"""Tests for the in-memory decision-event broker (events.py).

Offline, no network. Drive the thread-safe queue broker directly: publish,
subscribe-with-replay, unsubscribe, and the interaction where a decision is
published with no subscriber (must not error and must stay in the ring).
"""

from __future__ import annotations

import queue
import threading

import pytest

import dispatcher
import events


@pytest.fixture(autouse=True)
def _clean_broker():
    events.clear()
    yield
    events.clear()


def _decision(decision_id: int, model: str = "m") -> dict:
    return {"id": decision_id, "selected_model": model}


def test_publish_adds_to_ring_and_stays_for_replay():
    events.publish_decision(_decision(1))
    events.publish_decision(_decision(2))
    assert [d["id"] for d in events.recent_decisions()] == [1, 2]


def test_subscribe_replays_ring_into_new_queue():
    events.publish_decision(_decision(1))
    events.publish_decision(_decision(2))
    sub = events.subscribe(replay=True)
    assert [d["id"] for d in _drain(sub)] == [1, 2]


def test_subscribe_without_replay_starts_empty():
    events.publish_decision(_decision(1))
    sub = events.subscribe(replay=False)
    assert _drain(sub) == []


def test_publish_fans_out_to_live_subscribers():
    sub = events.subscribe()
    events.publish_decision(_decision(1))
    assert [d["id"] for d in _drain(sub)] == [1]


def test_unsubscribe_stops_delivery():
    sub = events.subscribe()
    events.unsubscribe(sub)
    events.publish_decision(_decision(1))
    assert _drain(sub) == []


def test_publish_with_no_subscriber_keeps_ring_and_does_not_raise():
    events.publish_decision(_decision(1))
    assert [d["id"] for d in events.recent_decisions()] == [1]


def test_full_subscriber_is_dropped_not_blocked():
    events.clear()
    subscriber = events.subscribe(replay=False, max_queue=1)
    events.publish_decision(_decision(1))
    events.publish_decision(_decision(2))  # must not raise or block; triggers eviction
    # Eviction removed the subscriber from fan-out and delivered a sentinel so
    # the SSE generator can terminate; the ring still contains both decisions.
    assert subscriber not in events._subscribers
    assert len(events.recent_decisions()) == 2


def _drain(q: queue.Queue) -> list:
    out = []
    while True:
        try:
            item = q.get_nowait()
        except queue.Empty:
            return out
        if item is events._EVICTED:
            return out
        out.append(item)


def test_concurrent_publish_subscribe_unsubscribe_raises_no_error():
    errors: list[Exception] = []

    def _publisher():
        for i in range(200):
            try:
                events.publish_decision({"id": i, "round": "pub"})
            except (RuntimeError, queue.Full) as exc:
                errors.append(exc)

    def _subscriptor():
        for _ in range(10):
            try:
                sub = events.subscribe(replay=False)
                events.unsubscribe(sub)
            except RuntimeError as exc:
                errors.append(exc)

    threads = [threading.Thread(target=_publisher) for _ in range(4)] + [
        threading.Thread(target=_subscriptor) for _ in range(4)
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert not errors, f"unexpected errors during concurrent access: {errors}"


def test_full_subscriber_eviction_terminates_decision_event_stream(monkeypatch):
    """When a subscriber is evicted (full queue), the _decision_event_stream
    generator breaks out of its loop instead of blocking forever."""
    events.clear()
    # Use a 1-slot queue we control. Monkeypatch subscribe to return it so
    # _decision_event_stream's live loop reads from our controlled queue.
    controlled = events.subscribe(replay=False, max_queue=1)
    monkeypatch.setattr(events, "subscribe", lambda *args, **kwargs: controlled)
    monkeypatch.setattr(events, "unsubscribe", lambda q: None)

    # Fill the controlled queue; publish_decision would evict on next insert.
    controlled.put_nowait(_decision(1))

    feed = dispatcher._decision_event_stream()
    assert next(feed).startswith("retry:")
    # The replay drain is empty (replay=False); the live loop drains decision(1)
    # and now blocks on controlled.get(timeout=SSE_HEARTBEAT_SECONDS).
    assert next(feed) == dispatcher._sse_data(_decision(1))

    # Now the queue is empty. Push another decision to fill it, then publish a
    # second one: the controlled subscriber is full, so publish_decision evicts
    # it and places the sentinel.
    controlled.put_nowait(_decision(2))
    events.publish_decision(_decision(3))
    # The generator should wake, see the sentinel, and terminate.
    with pytest.raises(StopIteration):
        next(feed)