HTTP/1.1¶

async def read_http_request(
    reader: asyncio.StreamReader,
    *,
    max_head_size: int = 1_048_576,
    body_limit: int = 4_194_304,
    parser_mode: str = "auto",
) -> HTTPRequest | None:
    """
    Reads and parses a full HTTP request (head and body) from the reader.

    Args:
        reader (asyncio.StreamReader): The client stream reader.
        max_head_size (int): Max allowed bytes for the request line and headers.
        body_limit (int): Max allowed bytes for the body.
        parser_mode (str): Choice of parser ('httptools', 'h11', or 'auto').

    Returns:
        HTTPRequest | None: The parsed request, or None if the client disconnected.
    """
    try:
        head = await reader.readuntil(b"\r\n\r\n")
    except asyncio.LimitOverrunError as exc:
        raise ValueError("HTTP head exceeds configured limit") from exc
    except asyncio.IncompleteReadError:
        return None

    if len(head) > max_head_size:
        raise ValueError("HTTP head exceeds configured limit")

    method, target, version, raw_headers = _parse_request_head(head, parser_mode)
    headers = _normalize_header_items(raw_headers)

    content_length_raw = _header_lookup(headers, b"content-length")
    transfer_encoding = (_header_lookup(headers, b"transfer-encoding") or b"").lower()

    content_length = 0
    if content_length_raw is not None:
        try:
            content_length = int(content_length_raw)
        except ValueError as exc:
            raise ValueError("Invalid Content-Length header") from exc

    body_chunks: list[bytes] = [b""]
    if b"chunked" in transfer_encoding:
        body_chunks = await _read_chunked_body_chunks(reader, body_limit)
    else:
        body_chunks = await _read_content_length_body_chunks(reader, content_length, body_limit)
    body = b"".join(body_chunks)

    if _is_websocket_upgrade(headers):
        request_headers: list[tuple[str, str] | tuple[bytes, bytes]] = [
            (name.decode("latin-1"), value.decode("latin-1")) for name, value in headers
        ]
    else:
        request_headers = cast(list[tuple[str, str] | tuple[bytes, bytes]], headers)

    return HTTPRequest(
        method=method,
        target=target,
        http_version=version,
        headers=request_headers,
        body=body,
        body_chunks=body_chunks,
    )

def build_http_scope(
    request: HTTPRequest,
    *,
    client: ClientAddress,
    server: ServerAddress,
    root_path: str,
    is_tls: bool,
) -> Scope:
    """
    Converts an internal HTTPRequest into an ASGI 3.0 scope dictionary.

    Args:
        request (HTTPRequest): Parsed request data.
        client (ClientAddress): IP and port of the client.
        server (ServerAddress): IP and port of the server.
        root_path (str): The mounting point of the application.
        is_tls (bool): True if connection is encrypted.

    Returns:
        Scope: A dictionary conforming to the ASGI HTTP specification.
    """
    path, _, query = request.target.partition("?")
    decoded_path = unquote(path)
    raw_path = path.encode("latin-1")
    root_path_bytes = root_path.encode("latin-1")
    full_path = root_path + decoded_path
    full_raw_path = root_path_bytes + raw_path

    headers = request.headers
    scope_headers = (
        cast(list[tuple[bytes, bytes]], headers)
        if all(
            isinstance(name, bytes) and isinstance(value, bytes) and name == name.lower()
            for name, value in headers
        )
        else [
            (_coerce_header_bytes(name).lower(), _coerce_header_bytes(value))
            for name, value in headers
        ]
    )

    return {
        "type": "http",
        "asgi": {"version": "3.0", "spec_version": "2.3"},
        "http_version": request.http_version.removeprefix("HTTP/"),
        "method": request.method,
        "scheme": "https" if is_tls else "http",
        "path": full_path,
        "raw_path": full_raw_path,
        "query_string": query.encode("latin-1"),
        "root_path": root_path,
        "headers": scope_headers,
        "client": client,
        "server": server,
        "state": {},
    }

async def run_http_asgi(
    app: ASGIApplication,
    scope: Scope,
    request_body: bytes | list[bytes],
    *,
    expect_100_continue: bool = False,
    on_100_continue: Callable[[], Awaitable[None]] | None = None,
    on_response_start: Callable[[HTTPResponse], Awaitable[None]] | None = None,
    on_response_body: Callable[[HTTPResponse, bytes, bool], Awaitable[None]] | None = None,
) -> HTTPResponse:
    """
    Orchestrates the ASGI request/response lifecycle with high-performance formatting.

    This function manages the strict state machine required by the ASGI HTTP specification.
    It incorporates several critical performance optimizations:

    1. **Single-Chunk Optimization:** Defers finalizing headers until the first body chunk arrives.
       If the app yields a single chunk (`more_body=False`) without a `Content-Length`, it calculates
       it instantly to avoid the overhead of HTTP chunked transfer framing.
    2. **In-Flight Chunk Framing:** If `Transfer-Encoding: chunked` is actually required, it formats
       the hex framing continuously during the app's `send()` loop rather than post-processing.



    Args:
        app (ASGIApplication): The user-provided ASGI application callable.
        scope (Scope): The ASGI connection scope dictionary.
        request_body (bytes | list[bytes]): The pre-buffered input body chunks.
        expect_100_continue (bool): If the client expects a 100-Continue response.
        on_100_continue (Callable | None): Callback to trigger the 100 status code transmission.

    Returns:
        HTTPResponse: The resulting parsed response state ready for wire encoding.

    Raises:
        RuntimeError: If the ASGI application violates the protocol sequence or crashes.
    """
    response = HTTPResponse()
    body_chunks = request_body if isinstance(request_body, list) else [request_body]
    if not body_chunks:
        body_chunks = [b""]

    response_started = False
    response_complete = False
    waiting_for_100_continue = expect_100_continue
    body_index = 0
    message_complete = asyncio.Event()

    chunked_encoding: bool | None = None
    expected_content_length = 0

    async def _send_internal_server_error() -> None:
        """Sends a standard 500 Internal Server Error response.

        This is used as a safety mechanism when an ASGI application fails
        to start a response or crashes during its initial execution.
        """
        nonlocal response_started, response_complete, chunked_encoding, expected_content_length
        response_started = True
        response_complete = True
        chunked_encoding = False
        expected_content_length = 0
        response.status = 500
        response.headers = [
            (b"content-type", b"text/plain; charset=utf-8"),
            (b"content-length", b"21"),
            (b"connection", b"close"),
        ]
        response.body_chunks = [] if scope.get("method") == "HEAD" else [b"Internal Server Error"]
        response.chunked_encoding = False
        response.suppress_body = scope.get("method") == "HEAD"
        message_complete.set()

    async def receive() -> Message:
        """The ASGI receive channel for the application.

        Returns:
            Message: An 'http.request' or 'http.disconnect' message.
        """
        nonlocal waiting_for_100_continue, body_index
        if waiting_for_100_continue:
            waiting_for_100_continue = False
            if on_100_continue is not None:
                await on_100_continue()

        if body_index < len(body_chunks):
            body = body_chunks[body_index]
            body_index += 1
            return {
                "type": "http.request",
                "body": body,
                "more_body": body_index < len(body_chunks),
            }

        await message_complete.wait()
        return {"type": "http.disconnect"}

    async def send(message: Message) -> None:
        """The ASGI send channel for the application.

        Args:
            message (Message): The message sent by the ASGI application.

        Raises:
            RuntimeError: If the application sends invalid message sequences
                or violates the ASGI HTTP specification.
        """
        nonlocal response_started, response_complete, waiting_for_100_continue
        nonlocal chunked_encoding, expected_content_length

        msg_type = message["type"]

        if msg_type == "http.response.start":
            if response_started:
                raise RuntimeError("ASGI message 'http.response.start' sent more than once.")

            response_started = True
            waiting_for_100_continue = False

            response.status = int(message.get("status", 200))
            response.headers = [
                (_coerce_header_bytes(name), _coerce_header_bytes(value))
                for name, value in message.get("headers", [])
            ]
            response.suppress_body = scope.get("method") == "HEAD"

            # Parse headers for explicit length or encoding
            for name, value in response.headers:
                lowered_name = name.lower()
                if lowered_name == b"content-length":
                    expected_content_length = int(value.decode("latin-1"))
                    chunked_encoding = False
                elif lowered_name == b"transfer-encoding" and value.lower() == b"chunked":
                    chunked_encoding = True
                    expected_content_length = 0

            # Default to chunked if no explicit length/encoding (allows apps to stream without Content-Length)
            if (
                chunked_encoding is None
                and not response.suppress_body
                and response.status not in {204, 304}
            ):
                chunked_encoding = True
                response.headers.append((b"transfer-encoding", b"chunked"))

            response.chunked_encoding = bool(chunked_encoding)

            # HEAD request overrides validation
            if response.suppress_body:
                expected_content_length = 0

            if on_response_start is not None:
                await on_response_start(response)
                response.streamed = True
            return

        elif msg_type == "http.response.body":
            if not response_started:
                raise RuntimeError(
                    "ASGI message 'http.response.body' sent before 'http.response.start'."
                )
            if response_complete:
                raise RuntimeError(
                    "ASGI message 'http.response.body' sent after response already completed."
                )

            body = message.get("body", b"")
            if not isinstance(body, bytes):
                body = bytes(body)
            more_body = bool(message.get("more_body", False))

            if response.suppress_body:
                pass  # HEAD requests intentionally drop the body
            elif chunked_encoding:
                response.body_chunks.append(body)
            else:
                body_size = len(body)
                if body_size > expected_content_length:
                    raise RuntimeError("Response content longer than Content-Length")
                expected_content_length -= body_size
                response.body_chunks.append(body)

            if response.streamed and on_response_body is not None:
                await on_response_body(response, body, more_body)

            if not more_body:
                if not chunked_encoding and expected_content_length != 0:
                    raise RuntimeError("Response content shorter than Content-Length")
                response_complete = True
                message_complete.set()
        else:
            raise RuntimeError(f"Unexpected ASGI message type: '{msg_type}'.")

    try:
        result = await app(scope, receive, send)
    except BaseException as exc:
        if not response_started:
            await _send_internal_server_error()
        elif isinstance(exc, RuntimeError):
            raise
        else:
            raise RuntimeError("Exception in ASGI application") from exc
    else:
        if result is not None:
            raise RuntimeError(f"ASGI callable should return None, but returned '{result}'.")
        if not response_started:
            await _send_internal_server_error()
        elif not response_complete:
            raise RuntimeError("ASGI callable returned without completing response.")

    return response

def append_default_response_headers(
    response: HTTPResponse,
    config: PalfreyConfig,
    *,
    default_headers: list[tuple[bytes, bytes]] | None = None,
) -> None:
    """Applies configured default headers to the response.

    Injects headers like 'Server' and 'Date' if enabled and not already present.

    Args:
        response (HTTPResponse): The response object to update.
        config (PalfreyConfig): Application configuration.
        default_headers (list | None): Cached list of headers for fast-path insertion.
    """
    existing_headers = {name.lower() for name, _ in response.headers}

    if default_headers is not None:
        for name, value in default_headers:
            lowered_name = name.lower()
            if lowered_name in existing_headers:
                continue
            response.headers.append((name, value))
            existing_headers.add(lowered_name)
        return

    configured_headers = config.normalized_headers
    configured_header_names = {name.lower() for name, _ in configured_headers}

    if (
        config.server_header
        and b"server" not in existing_headers
        and "server" not in configured_header_names
    ):
        response.headers.append((b"server", _SERVER_HEADER_VALUE))

    if (
        config.date_header
        and b"date" not in existing_headers
        and "date" not in configured_header_names
    ):
        response.headers.append((b"date", _http_date_header()))

    for name, value in configured_headers:
        response.headers.append((name.encode("latin-1"), value.encode("latin-1")))

def encode_http_response_head(response: HTTPResponse, keep_alive: bool) -> Iterable[bytes]:
    """Serializes the HTTP response status line and headers.

    Handles status line and headers while preserving chunked transfer metadata
    for callers that stream body chunks separately.

    Args:
        response (HTTPResponse): The response to serialize.
        keep_alive (bool): Whether the connection should be kept alive.

    Yields:
        bytes: Serialized status and header fragments.
    """
    if response.status in _STATUS_LINES:
        yield _STATUS_LINES[response.status]
    else:
        try:
            reason = http.HTTPStatus(response.status).phrase
        except ValueError:
            reason = ""
        yield f"HTTP/1.1 {response.status} {reason}\r\n".encode("ascii")

    has_content_length = False
    has_transfer_encoding = False
    has_connection = False

    for name, value in response.headers:
        lowered_name = name.lower()
        if lowered_name == b"content-length":
            has_content_length = True
        elif lowered_name == b"transfer-encoding":
            has_transfer_encoding = True
        elif lowered_name == b"connection":
            has_connection = True

        yield name
        yield _HEADER_SEPARATOR
        yield value
        yield _CRLF

    if not has_content_length and not has_transfer_encoding:
        payload_len = 0 if response.suppress_body else sum(len(c) for c in response.body_chunks)
        yield b"content-length: "
        yield str(payload_len).encode("ascii")
        yield _CRLF

    if not has_connection:
        yield _CONNECTION_KEEP_ALIVE if keep_alive else _CONNECTION_CLOSE

    yield _CRLF

def encode_http_response_body_chunk(
    response: HTTPResponse,
    body: bytes,
    *,
    more_body: bool,
) -> Iterable[bytes]:
    """Serializes a single ASGI response body event into wire bytes."""
    if response.suppress_body:
        return
    if response.chunked_encoding:
        if body:
            yield f"{len(body):x}\r\n".encode("ascii")
            yield body
            yield _CRLF
        if not more_body:
            yield b"0\r\n\r\n"
        return
    if body:
        yield body

def encode_http_response_chunks(response: HTTPResponse, keep_alive: bool) -> Iterable[bytes]:
    """Serializes the HTTPResponse into a sequence of bytes chunks.

    Handles status line, headers, and body encoding including chunked transfer.

    Args:
        response (HTTPResponse): The response to serialize.
        keep_alive (bool): Whether the connection should be kept alive.

    Yields:
        bytes: Serialized fragments of the HTTP response.
    """
    yield from encode_http_response_head(response, keep_alive=keep_alive)
    if response.chunked_encoding:
        chunks = response.body_chunks or [b""]
        for index, chunk in enumerate(chunks):
            yield from encode_http_response_body_chunk(
                response,
                chunk,
                more_body=index < len(chunks) - 1,
            )
        return
    elif not response.suppress_body:
        yield from response.body_chunks

def encode_http_response(response: HTTPResponse, keep_alive: bool) -> bytes:
    """
    Zero-copy serialization of the HTTPResponse into raw wire bytes.

    This function is aggressively optimized to avoid intermediate string formatting
    and memory allocations. It appends pre-encoded bytes into a flat list and performs
    a single C-level `b"".join()`.

    Args:
        response (HTTPResponse): The finalized response data collected from the app.
        keep_alive (bool): Indicates if the connection will remain open.

    Returns:
        bytes: The fully serialized HTTP response ready for `transport.write()`.
    """
    return b"".join(encode_http_response_chunks(response, keep_alive))

def should_keep_alive(request: HTTPRequest, response: HTTPResponse) -> bool:
    """
    Determines if the TCP connection should persist based on headers and protocol version.

    Handles the nuances between HTTP/1.0 explicit keep-alive and HTTP/1.1 explicit close.

    Args:
        request (HTTPRequest): The parsed incoming request.
        response (HTTPResponse): The finalized outgoing response.

    Returns:
        bool: True if the connection should be kept alive, False to close.
    """
    request_connection = _normalize_connection_value(request.headers)
    response_connection = b""
    for name, value in response.headers:
        if name.lower() == b"connection":
            response_connection = value.lower()
            break

    if b"close" in request_connection or b"close" in response_connection:
        return False

    if request.http_version == "HTTP/1.0" and b"keep-alive" not in request_connection:
        return False

    return True

def is_websocket_upgrade(request: HTTPRequest) -> bool:
    """Checks if the request is initiating a WebSocket handshake."""
    return _is_websocket_upgrade(request.headers)

def requires_100_continue(request: HTTPRequest) -> bool:
    """Verifies if the client is waiting for a '100 Continue' response before sending the body."""
    expect = _header_lookup(request.headers, b"expect")
    if not expect:
        return False
    return expect.lower() == b"100-continue"