"""In-process pub/sub event bus for realtime.

Because the whole app is one Python process (react.md), realtime needs no
broker, no LISTEN/NOTIFY — just a module-level broker fanning events to the
connected SSE subscribers. Sync service code calls `broker.publish(...)`;
async SSE handlers drain a per-subscriber queue.

Perspective filtering (rep(), §4.3) is applied PER SUBSCRIBER at the SSE
boundary, not here — the bus stays dumb and broadcasts raw events. That keeps
one rep() at the API edge instead of visibility logic sprinkled around.

If the app ever scales past one process, swap this module's internals for
LISTEN/NOTIFY behind the same publish()/subscribe() interface.
"""
from __future__ import annotations

import asyncio
from typing import Any

# The main event loop, captured at startup so sync publishers can hand work to
# async subscriber queues thread-safely (sync routes run in a threadpool).
_loop: asyncio.AbstractEventLoop | None = None
_subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
_MAX_QUEUE = 1000


def set_loop(loop: asyncio.AbstractEventLoop) -> None:
    global _loop
    _loop = loop


def subscribe() -> asyncio.Queue[dict[str, Any]]:
    q: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=_MAX_QUEUE)
    _subscribers.add(q)
    return q


def unsubscribe(q: asyncio.Queue[dict[str, Any]]) -> None:
    _subscribers.discard(q)


def publish(event: dict[str, Any]) -> None:
    """Fan an event to every subscriber. Safe to call from sync code."""
    if _loop is None:
        return  # no subscribers/loop yet (e.g. during migrations)

    def _enqueue() -> None:
        for q in list(_subscribers):
            try:
                q.put_nowait(event)
            except asyncio.QueueFull:
                # A stuck client should never block the writer; drop for them.
                pass

    _loop.call_soon_threadsafe(_enqueue)


# ---------------------------------------------------------------- presence
# Plan §6.3: connect = on court, disconnect + grace = off. Connection counts
# per player (multiple tabs = one presence); the drop is announced only
# after a grace window so a reload doesn't flicker everyone's strip.
#
# The presence event carries bare player ids. The strip renders presence as
# a glow on avatars that exist in the viewer's PROJECTED board payload, so
# invisible players can't be rendered even though their ids transit the
# stream (ids-only, no names). Tightening this to per-subscriber filtering
# is a known M7 item; noted in docs/OVERNIGHT-PROGRESS.md.

_PRESENCE_GRACE_SECONDS = 10.0
_presence: dict[Any, int] = {}


def online_ids() -> list[str]:
    return sorted(str(p) for p, n in _presence.items() if n > 0)


def _publish_presence() -> None:
    publish({"type": "presence.updated", "online": online_ids()})


def presence_connect(player_id: Any) -> None:
    if player_id is None:
        return
    _presence[player_id] = _presence.get(player_id, 0) + 1
    if _presence[player_id] == 1:
        _publish_presence()


def presence_disconnect(player_id: Any) -> None:
    if player_id is None or player_id not in _presence:
        return
    _presence[player_id] -= 1
    if _presence[player_id] > 0:
        return

    def _confirm_gone() -> None:
        # Still gone after the grace window? Then it's a real departure.
        if _presence.get(player_id, 0) <= 0:
            _presence.pop(player_id, None)
            _publish_presence()

    if _loop is not None:
        _loop.call_later(_PRESENCE_GRACE_SECONDS, _confirm_gone)
    else:
        _confirm_gone()