from __future__ import annotations

import asyncio
import logging
from typing import Any
import time

import pytest
from fastapi import HTTPException, WebSocketDisconnect

from device_gateway.app import (
    Device,
    DisplayMessage,
    NotificationTest,
    TargetSelection,
    device_websocket,
    devices,
    health,
    list_devices,
    logger,
    credential_store,
    pi_event,
    pi_sessions,
    select_target,
    target_store,
    send_message,
    test_notification as send_test_notification,
)
from device_gateway.integrations.pi import PiSessionEvent


class FakeWebSocket:
    def __init__(self) -> None:
        self.incoming: asyncio.Queue[dict[str, Any] | BaseException] = asyncio.Queue()
        self.sent: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
        self.accepted = False
        self.closed: tuple[int, str] | None = None

    async def accept(self) -> None:
        self.accepted = True

    async def receive_json(self) -> dict[str, Any]:
        value = await self.incoming.get()
        if isinstance(value, BaseException):
            raise value
        return value

    async def send_json(self, value: dict[str, Any]) -> None:
        await self.sent.put(value)

    async def close(self, code: int, reason: str) -> None:
        self.closed = (code, reason)


def run(coro: Any) -> Any:
    return asyncio.run(coro)


def setup_function() -> None:
    devices.clear()
    pi_sessions.sessions.clear()


def authorize_test_device(tmp_path: Any, device_id: str = "esp32s3-test001") -> str:
    token = "unit-test-device-token"
    path = tmp_path / "device.json"
    path.write_text(
        f'{{"device_id":"{device_id}","device_token":"{token}","pi_token":"unit-test-pi-token"}}\n',
        encoding="utf-8",
    )
    credential_store.path = path
    target_store.path = tmp_path / "targets.json"
    return token


def test_health() -> None:
    assert run(health()) == {"status": "ok", "control_version": "0.5.0", "online_devices": 0}


def test_bidirectional_device_flow(caplog: pytest.LogCaptureFixture, tmp_path: Any) -> None:
    token = authorize_test_device(tmp_path)

    async def scenario() -> None:
        websocket = FakeWebSocket()
        await websocket.incoming.put(
            {
                "type": "hello",
                "device_id": "esp32s3-test001",
                "device_token": token,
                "board": "m5stack-stopwatch",
                "firmware": "0.1.0",
            }
        )
        task = asyncio.create_task(device_websocket(websocket))  # type: ignore[arg-type]

        ack = await asyncio.wait_for(websocket.sent.get(), 1)
        assert ack["type"] == "hello_ack"
        assert isinstance(ack["server_time_ms"], int)
        assert abs(ack["server_time_ms"] - time.time() * 1000) < 5000
        assert isinstance(ack["timezone_offset_minutes"], int)
        assert -720 <= ack["timezone_offset_minutes"] <= 840
        snapshot = await asyncio.wait_for(websocket.sent.get(), 1)
        assert snapshot["type"] == "server_snapshot"
        assert snapshot["network_rx_bps"] >= 0
        assert await asyncio.wait_for(websocket.sent.get(), 1) == {
            "type": "target_state",
            "target": "server",
        }
        listed = await list_devices()
        assert listed[0]["device_id"] == "esp32s3-test001"
        assert listed[0]["online"] is True

        response = await send_message(
            "esp32s3-test001", DisplayMessage(title="Hello", body="Message from cloud")
        )
        assert response["sent"] is True
        assert await asyncio.wait_for(websocket.sent.get(), 1) == {
            "type": "display_message",
            "title": "Hello",
            "body": "Message from cloud",
        }

        notification = await send_test_notification(
            "esp32s3-test001", NotificationTest(kind="done")
        )
        assert notification["sent"] is True
        assert await asyncio.wait_for(websocket.sent.get(), 1) == {
            "type": "notification_test",
            "kind": "done",
        }

        await websocket.incoming.put(
            {"type": "button_event", "button": "A", "action": "click", "timestamp": 1234}
        )
        await websocket.incoming.put({"type": "heartbeat", "uptime": 42})
        for _ in range(20):
            if devices["esp32s3-test001"].last_event == {"type": "heartbeat", "uptime": 42}:
                break
            await asyncio.sleep(0)
        assert devices["esp32s3-test001"].last_event == {"type": "heartbeat", "uptime": 42}

        await websocket.incoming.put(WebSocketDisconnect())
        await asyncio.wait_for(task, 1)
        assert devices["esp32s3-test001"].online is False

    with caplog.at_level(logging.DEBUG, logger=logger.name):
        run(scenario())

    records = [(record.levelno, record.getMessage()) for record in caplog.records]
    assert any(level == logging.INFO and "device esp32s3-test001 online" in message for level, message in records)
    assert any(level == logging.INFO and "sent display_message" in message for level, message in records)
    assert any(level == logging.INFO and "button clicked button=A" in message for level, message in records)
    assert any(level == logging.DEBUG and "heartbeat uptime=42" in message for level, message in records)
    assert not any(level == logging.INFO and "heartbeat" in message for level, message in records)
    assert any(level == logging.INFO and "device esp32s3-test001 offline" in message for level, message in records)


def test_message_to_unknown_device_is_404() -> None:
    with pytest.raises(HTTPException) as error:
        run(send_message("missing", DisplayMessage(title="Hi", body="there")))
    assert error.value.status_code == 404


def test_rejects_non_hello_first_message() -> None:
    async def scenario() -> None:
        websocket = FakeWebSocket()
        await websocket.incoming.put({"type": "heartbeat", "uptime": 1})
        await device_websocket(websocket)  # type: ignore[arg-type]
        assert websocket.accepted is True
        assert websocket.closed == (1008, "first message must be hello")

    run(scenario())


def test_rejects_invalid_device_token(tmp_path: Any) -> None:
    authorize_test_device(tmp_path)

    async def scenario() -> None:
        websocket = FakeWebSocket()
        await websocket.incoming.put(
            {
                "type": "hello",
                "device_id": "esp32s3-test001",
                "device_token": "wrong-token",
                "board": "m5stack-stopwatch",
                "firmware": "0.2.1",
            }
        )
        await device_websocket(websocket)  # type: ignore[arg-type]
        assert websocket.closed == (1008, "unauthorized device")
        assert "esp32s3-test001" not in devices

    run(scenario())


def test_selected_pi_activity_history_and_notification_deduplication(tmp_path: Any) -> None:
    authorize_test_device(tmp_path)

    def pi_event_value(
        event_name: str,
        session_id: str = "pi-session-1",
        **values: Any,
    ) -> PiSessionEvent:
        return PiSessionEvent(
            event=event_name,  # type: ignore[arg-type]
            session_id=session_id,
            name=session_id,
            cwd="/tmp",
            **values,
        )

    async def scenario() -> None:
        websocket = FakeWebSocket()
        devices["esp32s3-test001"] = Device(
            device_id="esp32s3-test001",
            board="m5stack-stopwatch",
            firmware="0.2.1",
            websocket=websocket,  # type: ignore[arg-type]
        )

        pi_sessions.apply(pi_event_value("session_start"))
        pi_sessions.apply(pi_event_value("agent_start"))
        pi_sessions.apply(
            pi_event_value(
                "activity", activity_type="text", activity="recent output tail"
            )
        )
        pi_sessions.apply(pi_event_value("session_start", "pi-session-2"))

        selected = await select_target(
            "esp32s3-test001", TargetSelection(target="pi:pi-session-1")
        )
        assert selected["delivered"] is True
        target = await asyncio.wait_for(websocket.sent.get(), 1)
        assert target["type"] == "agent_state"
        assert target["recent_text"] == "recent output tail"

        await pi_event(
            pi_event_value(
                "activity",
                "pi-session-2",
                activity_type="text",
                activity="not selected",
            ),
            "unit-test-pi-token",
        )
        assert websocket.sent.empty()

        await pi_event(
            pi_event_value(
                "activity",
                activity_type="tool",
                activity="Running\npytest -q",
                tool_name="bash",
            ),
            "unit-test-pi-token",
        )
        activity = await asyncio.wait_for(websocket.sent.get(), 1)
        assert activity == {
            "type": "agent_activity", "agent": "pi",
            "target": "agent:pi:pi-session-1",
            "session_id": "pi-session-1",
            "activity_type": "tool",
            "activity": "Running\npytest -q",
            "updated_at": activity["updated_at"],
        }

        await pi_event(pi_event_value("agent_end"), "unit-test-pi-token")
        done = await asyncio.wait_for(websocket.sent.get(), 1)
        assert done["notification"] == "done"
        await pi_event(pi_event_value("agent_end"), "unit-test-pi-token")
        duplicate = await asyncio.wait_for(websocket.sent.get(), 1)
        assert "notification" not in duplicate

        await pi_event(pi_event_value("agent_start"), "unit-test-pi-token")
        await asyncio.wait_for(websocket.sent.get(), 1)
        await pi_event(
            pi_event_value("agent_end", failed=True), "unit-test-pi-token"
        )
        failed = await asyncio.wait_for(websocket.sent.get(), 1)
        assert failed["notification"] == "failed"
        await pi_event(
            pi_event_value("agent_end"), "unit-test-pi-token"
        )
        duplicate_failed = await asyncio.wait_for(websocket.sent.get(), 1)
        assert "notification" not in duplicate_failed
        assert duplicate_failed["state"] == "FAILED"

    run(scenario())