from __future__ import annotations

import asyncio
import logging
from collections.abc import AsyncIterator

import pytest

from musicmouse.bus import EventBus
from musicmouse.events import (
    ButtonEvent,
    Event,
    InputEvent,
    NextTrackRequested,
    RfidTokenRead,
    VolumeChanged,
)
from musicmouse.hardware import Button, ButtonAction


@pytest.fixture
async def bus() -> AsyncIterator[EventBus]:
    async with EventBus() as running_bus:
        yield running_bus


def _rfid(tag: str = "04a1b2c3d4") -> RfidTokenRead:
    return RfidTokenRead(tag_id=bytes.fromhex(tag), figure="fuchs", source="device")


async def test_handler_receives_its_event(bus: EventBus) -> None:
    seen: list[Event] = []
    bus.subscribe(RfidTokenRead, seen.append)

    event = _rfid()
    await bus.emit_and_wait(event)

    assert seen == [event]


async def test_unrelated_handlers_are_not_called(bus: EventBus) -> None:
    seen: list[Event] = []
    bus.subscribe(NextTrackRequested, seen.append)

    await bus.emit_and_wait(_rfid())

    assert seen == []


async def test_subscribing_to_a_base_class_catches_subclasses(bus: EventBus) -> None:
    seen: list[Event] = []
    bus.subscribe(InputEvent, seen.append)

    await bus.emit_and_wait(_rfid())
    await bus.emit_and_wait(VolumeChanged(volume=30))

    assert [type(e) for e in seen] == [RfidTokenRead]


async def test_subscribe_all_sees_everything(bus: EventBus) -> None:
    seen: list[Event] = []
    bus.subscribe_all(seen.append)

    await bus.emit_and_wait(_rfid())
    await bus.emit_and_wait(VolumeChanged(volume=30))

    assert [type(e) for e in seen] == [RfidTokenRead, VolumeChanged]


async def test_async_handlers_are_awaited(bus: EventBus) -> None:
    seen: list[str] = []

    async def slow(_: Event) -> None:
        await asyncio.sleep(0.01)
        seen.append("slow")

    bus.subscribe(RfidTokenRead, slow)
    bus.subscribe(RfidTokenRead, lambda _: seen.append("fast"))

    await bus.emit_and_wait(_rfid())

    assert seen == ["slow", "fast"]


async def test_events_are_dispatched_in_emission_order(bus: EventBus) -> None:
    """Ordering is what makes 'last event wins' meaningful for LED arbitration."""
    seen: list[int] = []

    async def record(event: VolumeChanged) -> None:
        await asyncio.sleep(0)
        seen.append(event.volume)

    bus.subscribe(VolumeChanged, record)

    for volume in range(5):
        bus.emit(VolumeChanged(volume=volume))
    await bus.drain()

    assert seen == [0, 1, 2, 3, 4]


async def test_events_emitted_from_a_handler_are_handled_before_drain_returns(
    bus: EventBus,
) -> None:
    seen: list[str] = []

    def on_button(_: ButtonEvent) -> None:
        seen.append("button")
        bus.emit(NextTrackRequested(source="device"))

    bus.subscribe(ButtonEvent, on_button)
    bus.subscribe(NextTrackRequested, lambda _: seen.append("next"))

    await bus.emit_and_wait(
        ButtonEvent(button=Button.RIGHT, action=ButtonAction.PRESSED, source="device")
    )

    assert seen == ["button", "next"]


async def test_a_raising_handler_does_not_stop_the_others(
    bus: EventBus, caplog: pytest.LogCaptureFixture
) -> None:
    seen: list[str] = []

    def boom(_: Event) -> None:
        raise RuntimeError("handler is broken")

    bus.subscribe(RfidTokenRead, boom)
    bus.subscribe(RfidTokenRead, lambda _: seen.append("survivor"))

    with caplog.at_level(logging.ERROR):
        await bus.emit_and_wait(_rfid())
        await bus.emit_and_wait(_rfid())

    assert seen == ["survivor", "survivor"]
    assert "handler is broken" in caplog.text


async def test_unsubscribe(bus: EventBus) -> None:
    seen: list[Event] = []
    unsubscribe = bus.subscribe(RfidTokenRead, seen.append)

    await bus.emit_and_wait(_rfid())
    unsubscribe()
    await bus.emit_and_wait(_rfid())

    assert len(seen) == 1


async def test_emit_is_safe_from_another_thread(bus: EventBus) -> None:
    """libVLC fires its callbacks off-loop; this is the crossing that must be safe."""
    seen: list[VolumeChanged] = []
    done = asyncio.Event()

    def record(event: VolumeChanged) -> None:
        seen.append(event)
        done.set()

    bus.subscribe(VolumeChanged, record)

    await asyncio.to_thread(bus.emit, VolumeChanged(volume=42, source="player"))
    async with asyncio.timeout(2):
        await done.wait()

    assert [e.volume for e in seen] == [42]


async def test_emit_before_start_is_an_error() -> None:
    with pytest.raises(RuntimeError, match="before start"):
        EventBus().emit(VolumeChanged(volume=1))


async def test_stop_is_idempotent() -> None:
    stopped = EventBus()
    await stopped.start()
    await stopped.stop()
    await stopped.stop()