"""Entry point for the GraalPy native standalone executable.

Serves the pyfconfig ASGI app with Hypercorn. ``graalpy -m standalone native``
(see ``Dockerfile.native``) bundles this file, the virtual environment and the
GraalPy runtime into a single self-contained executable.

granian and rloop are not used here: they are native extensions that cannot be
bundled into a standalone binary, and granian spawns its workers through
``multiprocessing``. Hypercorn is pure Python and runs in-process.
"""
from __future__ import annotations

import asyncio
import os
import signal
import socket
from types import FrameType
from typing import Any, Callable, Optional

from hypercorn.asyncio import serve
from hypercorn.config import Config

from pyfconfig.app import app


def _patch_socket_listen() -> None:
    """Make ``socket.listen`` retry with a blocking socket.

    In a GraalPy standalone binary the POSIX layer runs on GraalPy's Java
    backend, where asyncio's ``sock.listen()`` on the non-blocking server
    socket raises ``OSError(95, "Operation not supported on transport
    endpoint")``. Listening on a temporarily blocking socket succeeds.
    """
    original: Callable[..., None] = socket.socket.listen

    def listen(self: socket.socket, backlog: int = socket.SOMAXCONN) -> None:
        try:
            original(self, backlog)
        except OSError as exc:
            if exc.errno != 95:
                raise
            blocking = self.getblocking()
            self.setblocking(True)
            try:
                original(self, backlog)
            finally:
                self.setblocking(blocking)

    socket.socket.listen = listen  # type: ignore[method-assign]


async def _asgi(scope: Any, receive: Any, send: Any) -> None:
    """Serve the app, coercing ``str`` response bodies to ``bytes``.

    kaya's ASGI adapter sends ``''`` (a ``str``) for empty and streamed bodies
    (see ``AsgiContext.send_empty`` / ``stream_body``). The ASGI spec requires
    ``bytes``, and Hypercorn rejects a ``str`` body, so normalize it here.
    """

    async def _send(message: Any) -> None:
        body = message.get("body")
        if message.get("type") == "http.response.body" and isinstance(body, str):
            message = dict(message, body=body.encode())
        await send(message)

    await app(scope, receive, _send)


async def _serve() -> None:
    config = Config()
    config.bind = [f"{os.environ.get('HOST', '0.0.0.0')}:{os.environ.get('PORT', '8080')}"]

    stop = asyncio.Event()
    loop = asyncio.get_running_loop()

    def _shutdown(signum: int, frame: Optional[FrameType]) -> None:
        loop.call_soon_threadsafe(stop.set)

    for name in ("SIGINT", "SIGTERM"):
        sig: Any = getattr(signal, name, None)
        if sig is not None:
            signal.signal(sig, _shutdown)

    # A custom shutdown_trigger keeps Hypercorn from installing its own signal
    # handlers (loop.add_signal_handler is unreliable in the standalone runtime).
    await serve(_asgi, config, shutdown_trigger=stop.wait)


def main() -> None:
    _patch_socket_listen()
    try:
        asyncio.run(_serve())
    except asyncio.CancelledError:
        # Hypercorn cancels the serve task while shutting down after SIGTERM;
        # treat that as a clean exit.
        pass


if __name__ == "__main__":
    main()