"""Low level HTTP server."""

import asyncio
import warnings
from collections.abc import Awaitable, Callable
from typing import Any, Generic, TypeVar, overload

from .abc import AbstractStreamWriter
from .http_parser import RawRequestMessage
from .streams import StreamReader
from .web_exceptions import HTTPBadRequest
from .web_protocol import RequestHandler
from .web_request import BaseRequest
from .web_response import StreamResponse

__all__ = ("Server",)

_Request = TypeVar("_Request", bound=BaseRequest)
_RequestFactory = Callable[
    [
        RawRequestMessage,
        StreamReader,
        "RequestHandler[_Request]",
        AbstractStreamWriter,
        "asyncio.Task[None]",
        HTTPBadRequest | None,
    ],
    _Request,
]


class Server(Generic[_Request]):
    request_factory: _RequestFactory[_Request]

    @overload
    def __init__(
        self: "Server[BaseRequest]",
        handler: Callable[[_Request], Awaitable[StreamResponse]],
        *,
        debug: bool | None = None,
        handler_cancellation: bool = False,
        **kwargs: Any,  # TODO(PY311): Use Unpack to define kwargs from RequestHandler
    ) -> None: ...
    @overload
    def __init__(
        self,
        handler: Callable[[_Request], Awaitable[StreamResponse]],
        *,
        request_factory: _RequestFactory[_Request] | None,
        debug: bool | None = None,
        handler_cancellation: bool = False,
        **kwargs: Any,
    ) -> None: ...
    def __init__(
        self,
        handler: Callable[[_Request], Awaitable[StreamResponse]],
        *,
        request_factory: _RequestFactory[_Request] | None = None,
        debug: bool | None = None,
        handler_cancellation: bool = False,
        **kwargs: Any,
    ) -> None:
        if debug is not None:
            warnings.warn(
                "debug argument is no-op since 4.0 and scheduled for removal in 5.0",
                DeprecationWarning,
                stacklevel=2,
            )
        self._loop = asyncio.get_running_loop()
        self._connections: dict[RequestHandler[_Request], asyncio.Transport] = {}
        self._kwargs = kwargs
        self.requests_count = 0
        self.request_handler = handler
        self.request_factory = request_factory or self._make_request  # type: ignore[assignment]
        self.handler_cancellation = handler_cancellation

    @property
    def connections(self) -> list[RequestHandler[_Request]]:
        return list(self._connections.keys())

    def connection_made(
        self, handler: RequestHandler[_Request], transport: asyncio.Transport
    ) -> None:
        self._connections[handler] = transport

    def connection_lost(
        self, handler: RequestHandler[_Request], exc: BaseException | None = None
    ) -> None:
        if handler in self._connections:
            if handler._task_handler:
                handler._task_handler.add_done_callback(
                    lambda f: self._connections.pop(handler, None)
                )
            else:
                del self._connections[handler]

    def _make_request(
        self,
        message: RawRequestMessage,
        payload: StreamReader,
        protocol: RequestHandler[BaseRequest],
        writer: AbstractStreamWriter,
        task: "asyncio.Task[None]",
        pre_handler_error: HTTPBadRequest | None,
    ) -> BaseRequest:
        return BaseRequest(
            message,
            payload,
            protocol,
            writer,
            task,
            self._loop,
            pre_handler_error=pre_handler_error,
        )

    def pre_shutdown(self) -> None:
        for conn in self._connections:
            conn.close()

    async def shutdown(self, timeout: float | None = None) -> None:
        coros = (conn.shutdown(timeout) for conn in self._connections)
        await asyncio.gather(*coros)
        self._connections.clear()

    def __call__(self) -> RequestHandler[_Request]:
        try:
            return RequestHandler(self, loop=self._loop, **self._kwargs)
        except TypeError:
            # Failsafe creation: remove all custom handler_args
            kwargs = {
                k: v
                for k, v in self._kwargs.items()
                if k in ["debug", "access_log_class"]
            }
            handler = RequestHandler(self, loop=self._loop, **kwargs)
            handler.logger.warning(
                "Failed to create request handler with custom kwargs %r, "
                "falling back to filtered kwargs. This may indicate a "
                "misconfiguration.",
                self._kwargs,
            )
            return handler
