"""In-memory decision-event broker for the dashboard's live view.

This module is deliberately small and dependency-free: it keeps a bounded
ring buffer of recent routing decisions and lets asyncio-based subscribers
receive them in near real time. Every decision inserted via
``dispatcher.persist_route_decision`` is also published here, so the TUI can
show a live feed without polling ``/metrics``.

The broker is intentionally **not** durable: if no dashboard is connected, the
ring is all that survives. The durable source of truth remains the
``route_decisions`` SQLite table; this is just a volatile fan-out helper.
"""

from __future__ import annotations

import queue
import threading
from collections import deque
from typing import Any, Final

DEFAULT_BUFFER_SIZE: Final[int] = 100
DEFAULT_QUEUE_SIZE: Final[int] = 100

# Sentinel value to signal a subscriber that has been evicted
# (its queue was full). The SSE stream generator detects this and
# terminates so the TUI can reconnect.
_EVICTED = object()

_buffer: deque[dict[str, Any]] = deque(maxlen=DEFAULT_BUFFER_SIZE)
_subscribers: set[queue.Queue[dict[str, Any]]] = set()
_subscribers_lock = threading.Lock()


def publish_decision(decision: dict[str, Any]) -> None:
    """Append ``decision`` to the ring buffer and fan it out to subscribers.

    If a subscriber's queue is full or dead it is silently removed, because
    the broker must never fail the caller (a routing decision should never
    be slowed or broken by a dashboard consumer).

    When a subscriber is evicted for being full, a sentinel value is written
    to its queue so the consuming SSE generator can detect the closure and
    terminate rather than blocking forever.
    """
    _buffer.append(decision)

    dead: set[queue.Queue[dict[str, Any]]] = set()
    with _subscribers_lock:
        for subscriber in _subscribers:
            try:
                subscriber.put_nowait(decision)
            except queue.Full:
                dead.add(subscriber)
        _subscribers.difference_update(dead)

    # Signal evicted subscribers so their SSE streams terminate.
    # The decision was never queued (queue was full), so place the sentinel
    # directly — it will displace the old item but is enough to wake up the
    # consumer and signal that the connection should be re-established.
    for subscriber in dead:
        try:
            subscriber.put_nowait(_EVICTED)
        except queue.Full:
            # Queue is still full — replace the old item with the sentinel.
            # Put non-blocking puts will never block if we're here, but the
            # queue is genuinely full so we just discard the old item.
            try:
                subscriber.get_nowait()  # discard the oldest item
                subscriber.put_nowait(_EVICTED)
            except queue.Empty:
                pass


def subscribe(
    replay: bool = True,
    max_queue: int = DEFAULT_QUEUE_SIZE,
) -> queue.Queue[dict[str, Any]]:
    """Create a thread-safe subscriber queue.

    By default the current ring-buffer contents are preloaded so a new
    connection immediately sees recent decisions. The caller must arrange to
    remove the queue with :func:`unsubscribe` when it disconnects.
    """
    new_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=max_queue)
    if replay:
        for decision in _buffer:
            try:
                new_queue.put_nowait(decision)
            except queue.Full:
                break
    with _subscribers_lock:
        _subscribers.add(new_queue)
    return new_queue


def unsubscribe(subscriber: queue.Queue[dict[str, Any]]) -> None:
    """Remove ``subscriber`` from the fan-out set."""
    with _subscribers_lock:
        _subscribers.discard(subscriber)


def recent_decisions(limit: int = DEFAULT_BUFFER_SIZE) -> list[dict[str, Any]]:
    """Return up to ``limit`` items from the ring buffer, newest last."""
    return list(_buffer)[-limit:]


def clear() -> None:
    """Drop every buffered decision and subscriber.

    Used by tests to ensure a clean broker state; production code should
    almost never call this.
    """
    with _subscribers_lock:
        _subscribers.clear()
    _buffer.clear()