"""What the bus costs the event loop, and that it still delivers everything."""

import asyncio
import threading

from fluksio.flow.events import QUEUE_SIZE, EventBus


def _event(i: int) -> dict[str, object]:
    return {"type": "node_executed", "node": f"n{i}", "outputs": 1}


def test_a_cascade_costs_one_loop_callback() -> None:
    """The whole point of coalescing.

    A three-node cascade publishes 13-16 events and each used to cross to the
    event loop on its own. They are one callback now — and still every event.
    """

    async def run() -> tuple[int, int]:
        bus = EventBus()
        loop = asyncio.get_running_loop()
        bus.bind(loop)
        calls = 0
        real = loop.call_soon_threadsafe

        def counting(*args: object, **kwargs: object) -> object:
            nonlocal calls
            calls += 1
            return real(*args, **kwargs)  # type: ignore[arg-type]

        loop.call_soon_threadsafe = counting  # type: ignore[method-assign]
        try:
            async with bus.subscribe() as queue:
                # From a worker thread, which is where nodes run.
                thread = threading.Thread(
                    target=lambda: [bus.publish(_event(i)) for i in range(16)]
                )
                thread.start()
                thread.join()
                await asyncio.sleep(0.05)
                return calls, queue.qsize()
        finally:
            loop.call_soon_threadsafe = real  # type: ignore[method-assign]

    calls, delivered = asyncio.run(run())
    assert calls == 1
    assert delivered == 16


def test_a_slow_subscriber_still_loses_its_oldest() -> None:
    """Coalescing must not turn the drop-oldest bound into unbounded growth."""

    async def run() -> tuple[int, object]:
        bus = EventBus()
        bus.bind(asyncio.get_running_loop())
        async with bus.subscribe() as queue:
            thread = threading.Thread(
                target=lambda: [bus.publish(_event(i)) for i in range(QUEUE_SIZE + 50)]
            )
            thread.start()
            thread.join()
            await asyncio.sleep(0.05)
            return queue.qsize(), (await queue.get())["node"]

    size, oldest = asyncio.run(run())
    assert size == QUEUE_SIZE
    # The first fifty were dropped, not the last.
    assert oldest == "n50"


def test_nothing_is_buffered_with_nobody_listening() -> None:
    """An engine with no browser open holds no events."""

    async def run() -> int:
        bus = EventBus()
        bus.bind(asyncio.get_running_loop())
        for i in range(10):
            bus.publish(_event(i))
        await asyncio.sleep(0.01)
        return len(bus._pending)

    assert asyncio.run(run()) == 0