"""PeerDrop Daemon — IPC server using trio.

Listens on a Unix socket (macOS/Linux) and routes requests to the
RequestHandler. Also pushes events to connected clients.
"""

from __future__ import annotations

import logging
import socket
from pathlib import Path
from typing import TYPE_CHECKING

import trio

from peerdrop.constants import DEFAULT_SOCK_PATH
from peerdrop.daemon.handler import RequestHandler
from peerdrop.daemon.protocol import make_response, unpack_request
from peerdrop.utils.framing import length_prefix_pack, length_prefix_unpack

if TYPE_CHECKING:
    from peerdrop.core.service import ServiceInterface

logger = logging.getLogger(__name__)


class DaemonServer:
    """IPC server using trio that accepts client connections and routes requests."""

    def __init__(
        self,
        service: ServiceInterface,
        sock_path: Path | str | None = None,
    ) -> None:
        self._service = service
        self._handler = RequestHandler(service)
        self._sock_path = Path(sock_path) if sock_path else DEFAULT_SOCK_PATH
        self._listeners: list[trio.SocketListener] = []
        self._clients: set[trio.SocketStream] = set()
        self._client_queues: dict[trio.SocketStream, trio.MemorySendChannel] = {}
        self._cancel_scope: trio.CancelScope | None = None
        self._running = False

    def set_nursery(self, nursery: trio.Nursery) -> None:
        """Set nursery for background tasks (e.g., non-blocking send_file)."""
        self._handler.set_nursery(nursery)

    async def serve(self, task_status: trio.TaskStatus[None] | None = None) -> None:
        """Start serving IPC connections. Blocks until cancelled.

        Usage:
            async with trio.open_nursery() as nursery:
                nursery.start_soon(server.serve)
        """
        # Ensure socket directory exists
        self._sock_path.parent.mkdir(parents=True, exist_ok=True)

        # Remove stale socket
        if self._sock_path.exists():
            self._sock_path.unlink()

        # Create Unix socket
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(str(self._sock_path))
        sock.listen(5)
        sock.setblocking(False)

        # Convert to trio socket
        trio_sock = trio.socket.from_stdlib_socket(sock)

        self._running = True
        logger.info(f"Daemon listening on {self._sock_path}")

        if task_status is not None:
            task_status.started()

        try:
            await trio.serve_listeners(
                self._handle_client,
                [trio.SocketListener(trio_sock)],
            )
        finally:
            self._running = False
            if self._sock_path.exists():
                self._sock_path.unlink()
            logger.info("Daemon stopped")

    async def start(self, nursery: trio.Nursery) -> None:
        """Start the server in the given nursery."""
        await nursery.start(self.serve)

    async def _handle_client(self, stream: trio.SocketStream) -> None:
        """Handle a single client connection."""
        try:
            peer = stream.socket.getpeername()
        except OSError:
            peer = "unknown"
        logger.info(f"Client connected: {peer}")
        self._clients.add(stream)

        # Per-client event queue: broadcast_event() enqueues without waiting,
        # a background writer drains the queue onto the socket. This way one
        # slow client can't block event delivery to everyone else.
        send_channel, receive_channel = trio.open_memory_channel[bytes](64)
        self._client_queues[stream] = send_channel

        buffer = b""
        try:
            async with trio.open_nursery() as nursery:
                nursery.start_soon(self._push_events, receive_channel, stream)

                try:
                    while True:
                        data = await stream.receive_some(4096)
                        if not data:
                            break

                        buffer += data

                        # Process all complete messages in buffer
                        while True:
                            try:
                                result = length_prefix_unpack(buffer)
                            except ValueError:
                                logger.error(f"Client {peer} sent invalid framing, disconnecting")
                                return
                            if result is None:
                                break
                            message, buffer = result

                            # Handle request
                            response = await self._handler.handle(message)
                            await stream.send_all(length_prefix_pack(response))
                finally:
                    # Signal the event writer to stop BEFORE the nursery exits:
                    # closing the send channel ends the writer's async-for loop,
                    # which lets the nursery close cleanly.
                    send_channel.close()

        except (trio.ClosedResourceError, trio.EndOfChannel, trio.BrokenResourceError):
            logger.debug(f"Client disconnected: {peer}")
        except Exception as e:
            logger.error(f"Client error: {e}")
        finally:
            self._clients.discard(stream)
            queue = self._client_queues.pop(stream, None)
            if queue is not None:
                queue.close()
            try:
                await stream.aclose()
            except Exception:
                pass
            logger.info(f"Client removed: {peer}")

    async def _push_events(
        self,
        receive_channel: trio.MemoryReceiveChannel,
        stream: trio.SocketStream,
    ) -> None:
        """Background writer that drains this client's event queue to the socket."""
        async with receive_channel:
            async for packed in receive_channel:
                try:
                    await stream.send_all(packed)
                except Exception:
                    break

    async def broadcast_event(self, event_type: str, data: dict | None = None) -> None:
        """Push an event to all connected clients.

        Events are best-effort and never block: each client has a bounded queue,
        and events for a slow client are dropped when its queue is full.
        """
        from peerdrop.daemon.protocol import make_event
        event_msg = make_event(event_type, data)
        packed = length_prefix_pack(event_msg)

        for stream, queue in list(self._client_queues.items()):
            try:
                queue.send_nowait(packed)
            except trio.WouldBlock:
                pass  # Slow client: drop event (best-effort)
            except (trio.BrokenResourceError, trio.ClosedResourceError):
                self._clients.discard(stream)
            except Exception:
                pass

    @property
    def sock_path(self) -> Path:
        """Return the socket path."""
        return self._sock_path

    @property
    def client_count(self) -> int:
        """Return number of connected clients."""
        return len(self._clients)