"""In-process pub/sub for live run streaming over WebSocket.

A run publishes events (``step`` and ``run.status``) as it executes; WebSocket
subscribers receive them live. Persisted ``Step`` rows remain the source of
truth, so a client that connects late simply replays from the database first.
"""

from __future__ import annotations

import asyncio
import uuid


class RunEventBus:
    def __init__(self) -> None:
        self._subscribers: dict[uuid.UUID, set[asyncio.Queue]] = {}

    def subscribe(self, run_id: uuid.UUID) -> asyncio.Queue:
        queue: asyncio.Queue = asyncio.Queue()
        self._subscribers.setdefault(run_id, set()).add(queue)
        return queue

    def unsubscribe(self, run_id: uuid.UUID, queue: asyncio.Queue) -> None:
        subs = self._subscribers.get(run_id)
        if not subs:
            return
        subs.discard(queue)
        if not subs:
            self._subscribers.pop(run_id, None)

    async def publish(self, run_id: uuid.UUID, event: dict) -> None:
        for queue in list(self._subscribers.get(run_id, set())):
            queue.put_nowait(event)


#: Process-wide event bus.
run_event_bus = RunEventBus()
