webtransport

%% Copyright (c) 2026, Benoit Chesneau.
%% Licensed under the Apache License, Version 2.0.
%%
%% @doc WebTransport Public API
%%
%% This module provides the public API for WebTransport over HTTP/2 and HTTP/3.
%%
%% == Server Usage ==
%%
%% ```
%% %% Start a WebTransport listener
%% {ok, Listener} = webtransport:start_listener(my_listener, #{
%%     transport => h3,
%%     port => 8443,
%%     certfile => "cert.pem",
%%     keyfile => "key.pem",
%%     handler => my_wt_handler
%% }).
%%
%% %% Stop the listener
%% ok = webtransport:stop_listener(my_listener).
%% '''
%%
%% == Client Usage ==
%%
%% ```
%% %% Connect to a WebTransport server
%% {ok, Session} = webtransport:connect("example.com", 443, "/wt", #{
%%     transport => h3
%% }).
%%
%% %% Open a bidirectional stream
%% {ok, Stream} = webtransport:open_stream(Session, bidi).
%%
%% %% Send data
%% ok = webtransport:send(Session, Stream, <<"Hello">>).
%%
%% %% Send with FIN flag
%% ok = webtransport:send(Session, Stream, <<"World">>, fin).
%%
%% %% Send unreliable datagram
%% ok = webtransport:send_datagram(Session, <<"ping">>).
%%
%% %% Close stream
%% ok = webtransport:close_stream(Session, Stream).
%%
%% %% Close session
%% ok = webtransport:close_session(Session).
%% '''
%%
%% == Handler Behaviour ==
%%
%% See {@link webtransport_handler} for the callback behaviour.
%%
-module(webtransport).

%% Integration API (for embedding in a generic HTTP server)
-export([h3_settings/0, h3_settings/1]).
-export([h2_settings/0, h2_settings/1]).
-export([accept/4]).

%% Server API (convenience: standalone listener)
-export([start_listener/2, stop_listener/1]).
-export([listeners/0, listener_info/1]).
-export([listener_loop/1]).

%% Client API
-export([connect/4, connect/5]).

%% Session API
-export([open_stream/2]).
-export([send/3, send/4]).
-export([send_datagram/2]).
-export([close_stream/2]).
-export([reset_stream/3, stop_sending/3]).
-export([drain_session/1]).
-export([close_session/1, close_session/2, close_session/3]).
-export([session_info/1]).

%% Types
-type session() :: pid().
-type stream() :: non_neg_integer().
-type listener_name() :: atom().
-type request() :: #{
    path := binary(),
    authority := binary(),
    headers => [{binary(), binary()}]
}.

-type listener_opts() :: #{
    transport := h2 | h3,
    port := inet:port_number(),
    certfile := file:filename(),
    keyfile := file:filename(),
    handler := module(),
    handler_opts => term(),
    max_data => non_neg_integer(),
    max_streams_bidi => non_neg_integer(),
    max_streams_uni => non_neg_integer(),
    %% `compat_mode' selects which WebTransport HTTP/3 draft the listener
    %% accepts. Default `auto' accepts draft-15 and draft-02 and picks the
    %% matching code path per CONNECT. Pin to `latest' to refuse draft-02
    %% clients or to `legacy_browser_compat' to refuse draft-15 clients.
    compat_mode => latest | legacy_browser_compat | auto
}.

-type connect_opts() :: #{
    transport => h2 | h3,
    certfile => file:filename(),
    keyfile => file:filename(),
    cacertfile => file:filename(),
    verify => verify_none | verify_peer,
    headers => [{binary(), binary()}],
    timeout => timeout(),
    %% Clients pick an explicit compat mode. Default `latest' sends
    %% draft-15 SETTINGS and `:protocol = webtransport-h3'. Use
    %% `legacy_browser_compat' only when talking to a known draft-02 peer
    %% (Chrome / Firefox / quic-go v0.9).
    compat_mode => latest | legacy_browser_compat
}.

-export_type([session/0, stream/0, request/0]).
-export_type([listener_name/0, listener_opts/0, connect_opts/0]).

-include("webtransport.hrl").

%% ============================================================================
%% Integration API
%% ============================================================================

%% @doc Return HTTP/3 settings to merge into a quic_h3 server configuration.
%%
%% Use this when embedding WebTransport into an existing HTTP/3 server
%% instead of using `start_listener/2'. The returned map contains:
%%
%% - `settings' -- H3 SETTINGS to advertise (wt_enabled, flow control, etc.)
%% - `stream_type_handler' -- claims WT extension streams (0x41 bidi, 0x54 uni)
%% - `h3_datagram_enabled' -- enables QUIC datagrams
%% - `quic_opts' -- QUIC transport params (max_datagram_frame_size, reset_stream_at)
%% - `connection_handler' -- per-connection setup (creates the WT stream router)
%%
%% Merge the result into your quic_h3 server opts:
%% ```
%% Opts = maps:merge(webtransport:h3_settings(), #{
%%     cert => CertDer, key => PrivateKey,
%%     handler => fun my_handler/5
%% }),
%% quic_h3:start_server(my_server, 443, Opts).
%% '''
-spec h3_settings() -> map().
h3_settings() ->
    h3_settings(#{}).

-spec h3_settings(map()) -> map().
h3_settings(Opts) ->
    CompatMode = maps:get(compat_mode, Opts, auto),
    Settings = case CompatMode of
        auto ->
            maps:merge(wt_h3:default_settings(latest),
                       wt_h3:default_settings(legacy_browser_compat));
        Mode ->
            wt_h3:default_settings(Mode)
    end,
    Claim = wt_stream_type_handler(),
    ConnectionHandler = fun(QuicConnPid) ->
        {ok, Router} = webtransport_h3_router:start(undefined),
        ensure_router_table(),
        %% Key the router by the QUIC connection pid. accept/4 derives the
        %% same key from its H3 connection via quic_h3:get_quic_conn/1, so
        %% it resolves this exact router no matter which process calls
        %% accept/4. quic_h3 runs request handlers in a spawned process and
        %% this callback runs in the listener process, so there is no shared
        %% process dictionary to rely on; the embedder may also dispatch
        %% accept/4 to its own per-request worker.
        ets:insert(webtransport_routers, {QuicConnPid, Router}),
        spawn(fun() -> reap_router(QuicConnPid, Router) end),
        #{
            owner => Router,
            stream_type_handler => Claim,
            h3_datagram_enabled => true
        }
    end,
    #{
        settings => Settings,
        stream_type_handler => Claim,
        h3_datagram_enabled => true,
        quic_opts => #{
            max_datagram_frame_size => 65535,
            reset_stream_at => true
        },
        connection_handler => ConnectionHandler
    }.

%% @doc Return HTTP/2 settings to merge into an h2 server configuration.
%%
%% ```
%% Opts = maps:merge(webtransport:h2_settings(), #{
%%     cert => "cert.pem", key => "key.pem",
%%     handler => fun my_handler/5,
%%     enable_connect_protocol => true
%% }),
%% h2:start_server(443, Opts).
%% '''
-spec h2_settings() -> map().
h2_settings() ->
    h2_settings(#{}).

-spec h2_settings(map()) -> map().
h2_settings(Opts) ->
    #{
        enable_connect_protocol => true,
        settings => #{
            enable_connect_protocol => 1,
            wt_initial_max_data =>
                maps:get(max_data, Opts, ?DEFAULT_MAX_DATA),
            wt_initial_max_streams_bidi =>
                maps:get(max_streams_bidi, Opts, ?DEFAULT_MAX_STREAMS_BIDI),
            wt_initial_max_streams_uni =>
                maps:get(max_streams_uni, Opts, ?DEFAULT_MAX_STREAMS_UNI)
        }
    }.

%% @doc Accept a WebTransport session on an incoming CONNECT request.
%%
%% Call this from your HTTP request handler when you want a particular
%% CONNECT request to become a WebTransport session. The function validates
%% headers, starts the session gen_statem, registers it as the stream
%% handler (like `quic_h3:set_stream_handler/3'), sends 200, and returns
%% `{ok, Session}'.
%%
%% ```
%% my_handler(Conn, StreamId, <<"CONNECT">>, <<"/wt">>, Headers) ->
%%     webtransport:accept(Conn, StreamId, Headers, #{
%%         transport => h3,
%%         handler => my_wt_handler,
%%         handler_opts => #{owner => self()}
%%     });
%% my_handler(Conn, StreamId, <<"GET">>, _Path, _Headers) ->
%%     quic_h3:send_response(Conn, StreamId, 200, []),
%%     quic_h3:send_data(Conn, StreamId, <<"hello">>, true).
%% '''
-spec accept(pid(), non_neg_integer(), [{binary(), binary()}], map()) ->
    {ok, session()} | {error, term()}.
accept(Conn, StreamId, Headers, Opts) ->
    Transport = maps:get(transport, Opts, h3),
    Handler = maps:get(handler, Opts, undefined),
    HandlerOpts = maps:get(handler_opts, Opts, #{}),
    Method = proplists:get_value(<<":method">>, Headers, <<>>),
    Path = proplists:get_value(<<":path">>, Headers, <<"/">>),
    case {Handler, Method} of
        {undefined, _} ->
            {error, missing_handler};
        {_, <<"CONNECT">>} ->
            case run_origin_check(Handler, Headers, HandlerOpts) of
                accept ->
                    do_accept(Transport, Conn, StreamId, Path,
                              Headers, Handler, HandlerOpts, Opts);
                {reject, Status, Reason} ->
                    send_reject(Transport, Conn, StreamId, Status, Reason),
                    {error, {rejected, Status}}
            end;
        {_, _} ->
            {error, not_connect_method}
    end.

do_accept(h3, H3Conn, StreamId, Path, Headers, Handler, HandlerOpts, Opts) ->
    CompatMode = maps:get(compat_mode, Opts, auto),
    case classify_h3_connect(Headers, CompatMode) of
        {ok, ClientMode} ->
            Router = get_or_create_router(H3Conn),
            Opts1 = Opts#{compat_mode => ClientMode},
            do_accept_h3(H3Conn, StreamId, Path, Headers,
                         Handler, HandlerOpts, Opts1, Router);
        {error, Reason} ->
            send_reject(h3, H3Conn, StreamId, 400,
                        iolist_to_binary(io_lib:format("~p", [Reason]))),
            {error, Reason}
    end;
do_accept(h2, Conn, StreamId, Path, Headers, Handler, HandlerOpts, Opts) ->
    case is_webtransport_request(Headers) of
        true ->
            do_accept_h2(Conn, StreamId, Path, Headers,
                         Handler, HandlerOpts, Opts);
        false ->
            send_reject(h2, Conn, StreamId, 400, <<"Bad Request">>),
            {error, not_webtransport}
    end.

do_accept_h3(H3Conn, StreamId, Path, Headers, Handler, HandlerOpts, Opts, Router) ->
    TransportState = webtransport_h3:new(H3Conn, StreamId, Router),
    Authority = proplists:get_value(<<":authority">>, Headers, <<>>),
    Request = #{path => Path, authority => Authority, headers => Headers},
    SessionOpts = maps:merge(HandlerOpts, #{
        request => Request,
        is_server => true,
        handler_opts => HandlerOpts,
        max_data => maps:get(max_data, Opts, ?DEFAULT_MAX_DATA),
        max_streams_bidi => maps:get(max_streams_bidi, Opts, ?DEFAULT_MAX_STREAMS_BIDI),
        max_streams_uni => maps:get(max_streams_uni, Opts, ?DEFAULT_MAX_STREAMS_UNI)
    }),
    case webtransport_session:start_link(h3, TransportState, Handler, SessionOpts) of
        {ok, Session} ->
            quic_h3:set_stream_handler(H3Conn, StreamId, Session),
            webtransport_h3_router:register_session(Router, StreamId, Session),
            quic_h3:send_response(H3Conn, StreamId, 200, []),
            {ok, Session};
        {error, Reason} ->
            send_reject(h3, H3Conn, StreamId, 500,
                        iolist_to_binary(io_lib:format("~p", [Reason]))),
            {error, Reason}
    end.

do_accept_h2(Conn, StreamId, Path, Headers, Handler, HandlerOpts, Opts) ->
    h2:send_response(Conn, StreamId, 200, []),
    TransportState = webtransport_h2:new(Conn, StreamId),
    Authority = proplists:get_value(<<":authority">>, Headers, <<>>),
    Request = #{path => Path, authority => Authority, headers => Headers},
    ServerDefaults = #{
        u  => maps:get(max_streams_uni, Opts, ?DEFAULT_MAX_STREAMS_UNI),
        bl => maps:get(max_streams_bidi, Opts, ?DEFAULT_MAX_STREAMS_BIDI),
        br => maps:get(max_data, Opts, ?DEFAULT_MAX_DATA)
    },
    Negotiated = case proplists:get_value(<<"webtransport-init">>, Headers) of
        undefined -> ServerDefaults;
        InitBin ->
            case wt_h2_init:parse(InitBin) of
                {ok, Init} -> wt_h2_init:apply_greater_of(Init, ServerDefaults);
                {error, _} -> ServerDefaults
            end
    end,
    SessionOpts = maps:merge(HandlerOpts, #{
        request => Request,
        is_server => true,
        handler_opts => HandlerOpts,
        max_data => maps:get(br, Negotiated, ?DEFAULT_MAX_DATA),
        max_streams_bidi => maps:get(bl, Negotiated, ?DEFAULT_MAX_STREAMS_BIDI),
        max_streams_uni => maps:get(u, Negotiated, ?DEFAULT_MAX_STREAMS_UNI)
    }),
    case webtransport_session:start_link(h2, TransportState, Handler, SessionOpts) of
        {ok, Session} ->
            LoopPid = spawn(fun() -> h2_data_loop(Conn, StreamId, Session) end),
            _ = h2:set_stream_handler(Conn, StreamId, LoopPid),
            {ok, Session};
        {error, Reason} ->
            h2:send_data(Conn, StreamId,
                         iolist_to_binary(io_lib:format("~p", [Reason])), true),
            {error, Reason}
    end.

send_reject(h3, H3Conn, StreamId, Status, Body) ->
    quic_h3:send_response(H3Conn, StreamId, Status, []),
    quic_h3:send_data(H3Conn, StreamId, Body, true);
send_reject(h2, Conn, StreamId, Status, Body) ->
    h2:send_response(Conn, StreamId, Status, []),
    h2:send_data(Conn, StreamId, Body, true).

%% Find the WT stream router for an H3 connection.
%%
%% The router demuxes extension streams (uni 0x54, bidi 0x41) to sessions.
%% It is keyed by the QUIC connection pid: the `h3_settings/1'
%% connection_handler registers it under the pid it is handed, and accept/4
%% derives the same pid from its H3 connection via `quic_h3:get_quic_conn/1'.
%% Both agree on one router per connection without a shared process
%% dictionary, so accept/4 works from any process (e.g. an embedder's
%% per-request worker).
get_or_create_router(H3Conn) ->
    ensure_router_table(),
    Key = router_key(H3Conn),
    case ets:lookup(webtransport_routers, Key) of
        [{_, Router}] ->
            case is_process_alive(Router) of
                true -> Router;
                false ->
                    ets:delete(webtransport_routers, Key),
                    create_and_register_router(Key)
            end;
        [] ->
            %% No connection_handler ran for this connection (the server was
            %% wired without h3_settings/1). Create a router so the session
            %% still starts; lacking the connection_handler's `owner' it
            %% won't receive peer-opened streams, but that is a caller-side
            %% misconfiguration.
            create_and_register_router(Key)
    end.

%% The registry key shared by the connection_handler and accept/4: the QUIC
%% connection pid behind the H3 connection. The connection_handler is handed
%% this same pid directly, so both sides resolve one router per connection.
router_key(H3Conn) ->
    quic_h3:get_quic_conn(H3Conn).

create_and_register_router(Key) ->
    {ok, Router} = webtransport_h3_router:start(undefined),
    ets:insert(webtransport_routers, {Key, Router}),
    Router.

%% Drop a per-connection router's registry row and stop the router when its
%% QUIC connection ends, so a long-lived embedder doesn't accumulate stale
%% rows or idle router processes. Runs in a throwaway process: the
%% connection_handler that spawns it runs inside the quic listener, which
%% must not block or monitor on its behalf. The {Key, Router} match means a
%% later connection that reuses the pid (and overwrote the row) is left
%% untouched.
reap_router(QuicConnPid, Router) ->
    Ref = erlang:monitor(process, QuicConnPid),
    receive
        {'DOWN', Ref, process, _, _} ->
            ets:match_delete(webtransport_routers, {QuicConnPid, Router}),
            catch gen_server:stop(Router, normal, 5000),
            ok
    end.

ensure_router_table() ->
    case ets:whereis(webtransport_routers) of
        undefined ->
            try
                ets:new(webtransport_routers, [named_table, public, set])
            catch
                error:badarg -> ok  %% race: another process created it
            end;
        _ ->
            ok
    end.

%% ============================================================================
%% Server API
%% ============================================================================

%% @doc Start a WebTransport listener.
%%
%% Options:
%% - `transport' - Required. Either `h2' (HTTP/2) or `h3' (HTTP/3)
%% - `port' - Required. Port to listen on
%% - `certfile' - Required. Path to TLS certificate
%% - `keyfile' - Required. Path to TLS private key
%% - `handler' - Required. Module implementing webtransport_handler behaviour
%% - `handler_opts' - Optional. Initial options passed to handler:init/2
%% - `max_data' - Optional. Initial max session data (default 1MB)
%% - `max_streams_bidi' - Optional. Initial max bidi streams (default 100)
%% - `max_streams_uni' - Optional. Initial max uni streams (default 100)
%%
-spec start_listener(listener_name(), listener_opts()) ->
    {ok, pid()} | {error, term()}.
start_listener(Name, #{transport := Transport} = Opts) when is_atom(Name) ->
    case validate_listener_opts(Opts) of
        ok ->
            case Transport of
                h2 -> start_h2_listener(Name, Opts);
                h3 -> start_h3_listener(Name, Opts)
            end;
        {error, _} = Err ->
            Err
    end.

%% @doc Stop a WebTransport listener.
-spec stop_listener(listener_name()) -> ok | {error, term()}.
stop_listener(Name) when is_atom(Name) ->
    case whereis(Name) of
        undefined ->
            {error, not_found};
        Pid ->
            case persistent_term:get({webtransport_listener, Name}, undefined) of
                undefined ->
                    {error, not_found};
                #{transport := h2, server_ref := ServerRef} ->
                    h2:stop_server(ServerRef),
                    persistent_term:erase({webtransport_listener, Name}),
                    exit(Pid, shutdown),
                    ok;
                #{transport := h3, server_ref := ServerRef} ->
                    quic_h3:stop_server(ServerRef),
                    persistent_term:erase({webtransport_listener, Name}),
                    exit(Pid, shutdown),
                    ok
            end
    end.

%% @doc List all active listeners.
-spec listeners() -> [listener_name()].
listeners() ->
    %% Get all persistent terms that match our pattern
    [Name || {{webtransport_listener, Name}, _} <- persistent_term:get()].

%% @doc Get information about a listener.
-spec listener_info(listener_name()) -> {ok, map()} | {error, not_found}.
listener_info(Name) ->
    case persistent_term:get({webtransport_listener, Name}, undefined) of
        undefined -> {error, not_found};
        Info -> {ok, maps:without([server_ref], Info)}
    end.

%% ============================================================================
%% Client API
%% ============================================================================

%% @doc Connect to a WebTransport server.
%%
%% Options:
%% - `transport' - Optional. Either `h2' or `h3' (default: h3)
%% - `cacertfile' - Optional. CA certificate file for verification
%% - `verify' - Optional. `verify_none' or `verify_peer' (default: verify_peer)
%% - `headers' - Optional. Extra headers to send in CONNECT request
%% - `timeout' - Optional. Connection timeout in ms (default: 30000)
%%
-spec connect(Host, Port, Path, Opts) -> {ok, session()} | {error, term()} when
    Host :: string() | binary(),
    Port :: inet:port_number(),
    Path :: binary(),
    Opts :: connect_opts().
connect(Host, Port, Path, Opts) ->
    connect(Host, Port, Path, Opts, undefined).

%% @doc Connect to a WebTransport server with a custom handler.
-spec connect(Host, Port, Path, Opts, Handler) -> {ok, session()} | {error, term()} when
    Host :: string() | binary(),
    Port :: inet:port_number(),
    Path :: binary(),
    Opts :: connect_opts(),
    Handler :: module() | undefined.
connect(Host, Port, Path, Opts, Handler) ->
    Transport = maps:get(transport, Opts, h3),
    case Transport of
        h2 -> connect_h2(Host, Port, Path, Opts, Handler);
        h3 -> connect_h3(Host, Port, Path, Opts, Handler)
    end.

%% ============================================================================
%% Session API
%% ============================================================================

%% @doc Open a new stream on the session.
-spec open_stream(session(), bidi | uni) -> {ok, stream()} | {error, term()}.
open_stream(Session, Type) when Type =:= bidi; Type =:= uni ->
    webtransport_session:open_stream(Session, Type).

%% @doc Send data on a stream.
-spec send(session(), stream(), iodata()) -> ok | {error, term()}.
send(Session, Stream, Data) ->
    webtransport_session:send(Session, Stream, Data, false).

%% @doc Send data on a stream with optional FIN flag.
-spec send(session(), stream(), iodata(), fin | nofin) -> ok | {error, term()}.
send(Session, Stream, Data, fin) ->
    webtransport_session:send(Session, Stream, Data, true);
send(Session, Stream, Data, nofin) ->
    webtransport_session:send(Session, Stream, Data, false).

%% @doc Send an unreliable datagram.
-spec send_datagram(session(), iodata()) -> ok | {error, term()}.
send_datagram(Session, Data) ->
    webtransport_session:send_datagram(Session, Data).

%% @doc Close a stream gracefully (send FIN).
-spec close_stream(session(), stream()) -> ok | {error, term()}.
close_stream(Session, Stream) ->
    webtransport_session:close_stream(Session, Stream).

%% @doc Abruptly terminate a stream with an error code.
-spec reset_stream(session(), stream(), non_neg_integer()) -> ok | {error, term()}.
reset_stream(Session, Stream, ErrorCode) ->
    webtransport_session:reset_stream(Session, Stream, ErrorCode).

%% @doc Request that the peer stop sending on a stream.
-spec stop_sending(session(), stream(), non_neg_integer()) -> ok | {error, term()}.
stop_sending(Session, Stream, ErrorCode) ->
    webtransport_session:stop_sending(Session, Stream, ErrorCode).

%% @doc Signal that no new streams will be created.
-spec drain_session(session()) -> ok.
drain_session(Session) ->
    webtransport_session:drain(Session).

%% @doc Close the session gracefully.
-spec close_session(session()) -> ok.
close_session(Session) ->
    close_session(Session, 0, <<>>).

%% @doc Close the session with an error code.
-spec close_session(session(), non_neg_integer()) -> ok.
close_session(Session, ErrorCode) ->
    close_session(Session, ErrorCode, <<>>).

%% @doc Close the session with an error code and reason.
%% draft-14 §4.6 / draft-15 §5: Reason must be at most 1024 UTF-8 bytes.
-spec close_session(session(), non_neg_integer(), binary()) -> ok | {error, reason_too_long}.
close_session(_Session, _ErrorCode, Reason) when byte_size(Reason) > 1024 ->
    {error, reason_too_long};
close_session(Session, ErrorCode, Reason) ->
    webtransport_session:close(Session, ErrorCode, Reason).

%% @doc Get session information.
-spec session_info(session()) -> {ok, map()} | {error, term()}.
session_info(Session) ->
    webtransport_session:get_info(Session).

%% ============================================================================
%% Internal Functions - Listener Setup
%% ============================================================================

validate_listener_opts(Opts) ->
    Required = [transport, port, certfile, keyfile, handler],
    Missing = [K || K <- Required, not maps:is_key(K, Opts)],
    case Missing of
        [] -> ok;
        _ -> {error, {missing_options, Missing}}
    end.

start_h2_listener(Name, Opts) ->
    #{
        port := Port,
        certfile := CertFile,
        keyfile := KeyFile,
        handler := Handler
    } = Opts,

    HandlerOpts = maps:get(handler_opts, Opts, #{}),

    %% Create wrapper handler for H2
    H2Handler = make_h2_handler(Handler, HandlerOpts, Opts),

    WtSettings = #{
        enable_connect_protocol => 1,
        wt_initial_max_data =>
            maps:get(max_data, Opts, ?DEFAULT_MAX_DATA),
        wt_initial_max_streams_bidi =>
            maps:get(max_streams_bidi, Opts, ?DEFAULT_MAX_STREAMS_BIDI),
        wt_initial_max_streams_uni =>
            maps:get(max_streams_uni, Opts, ?DEFAULT_MAX_STREAMS_UNI)
    },
    ServerOpts = #{
        cert => CertFile,
        key => KeyFile,
        handler => H2Handler,
        enable_connect_protocol => true,
        settings => WtSettings
    },

    %% h2:start_server spawn_link's acceptor/manager processes to its caller.
    %% Delegating to a persistent owner process keeps those links on a
    %% long-lived pid so the listener survives after the caller returns.
    case start_h2_server_owner(Name, Port, Handler, ServerOpts) of
        {ok, OwnerPid, ServerRef} ->
            persistent_term:put({webtransport_listener, Name}, #{
                transport => h2,
                port => Port,
                handler => Handler,
                server_ref => ServerRef
            }),
            {ok, OwnerPid};
        {error, Reason} ->
            {error, Reason}
    end.

start_h2_server_owner(Name, Port, _Handler, ServerOpts) ->
    Parent = self(),
    Ref = make_ref(),
    OwnerPid = spawn(fun() -> h2_owner_init(Parent, Ref, Name, Port, ServerOpts) end),
    receive
        {Ref, {ok, ServerRef}} ->
            {ok, OwnerPid, ServerRef};
        {Ref, {error, Reason}} ->
            {error, Reason}
    after 5000 ->
        exit(OwnerPid, kill),
        {error, h2_listener_start_timeout}
    end.

h2_owner_init(Parent, Ref, Name, Port, ServerOpts) ->
    case h2:start_server(Port, ServerOpts) of
        {ok, ServerRef} ->
            try register(Name, self()) catch _:_ -> ok end,
            Parent ! {Ref, {ok, ServerRef}},
            h2_owner_loop(Name, ServerRef);
        {error, Reason} ->
            Parent ! {Ref, {error, Reason}}
    end.

h2_owner_loop(Name, ServerRef) ->
    receive
        {stop, _From} ->
            h2:stop_server(ServerRef),
            persistent_term:erase({webtransport_listener, Name}),
            ok;
        _ ->
            h2_owner_loop(Name, ServerRef)
    end.

start_h3_listener(Name, Opts) ->
    #{
        port := Port,
        certfile := CertFile,
        keyfile := KeyFile,
        handler := Handler
    } = Opts,

    HandlerOpts = maps:get(handler_opts, Opts, #{}),

    %% Read and decode certificate and key
    case read_cert_and_key(CertFile, KeyFile) of
        {ok, CertDer, PrivateKey} ->
            Claim = wt_stream_type_handler(),

            ConnectionHandler = fun(_QuicConnPid) ->
                {ok, Router} = webtransport_h3_router:start(undefined),
                #{
                    owner => Router,
                    handler => make_h3_handler(Handler, HandlerOpts, Opts, Router),
                    stream_type_handler => Claim,
                    h3_datagram_enabled => true
                }
            end,

            ServerCompatMode = maps:get(compat_mode, Opts, auto),
            %% `auto' advertises both latest (draft-15) AND legacy
            %% (draft-02) setting keys. H3 SETTINGS are connection-
            %% scoped and sent before any CONNECT, so the server cannot
            %% know the client's draft at SETTINGS time. Peers MUST
            %% ignore unknown settings (RFC 9114 §7.2.4.1), so the
            %% merged shape is wire-safe. Pinning to a specific mode
            %% sends only that mode's settings.
            AdvertisedSettings = case ServerCompatMode of
                auto ->
                    maps:merge(wt_h3:default_settings(latest),
                               wt_h3:default_settings(legacy_browser_compat));
                Other ->
                    wt_h3:default_settings(Other)
            end,
            ServerOpts = #{
                cert => CertDer,
                key => PrivateKey,
                handler => make_h3_handler(Handler, HandlerOpts, Opts, undefined),
                settings => AdvertisedSettings,
                quic_opts => #{
                    max_datagram_frame_size => 65535,
                    reset_stream_at => true
                },
                stream_type_handler => Claim,
                connection_handler => ConnectionHandler
            },

            case quic_h3:start_server(Name, Port, ServerOpts) of
                {ok, ServerRef} ->
                    Pid = spawn(?MODULE, listener_loop, [Name]),
                    register(Name, Pid),
                    persistent_term:put({webtransport_listener, Name}, #{
                        transport => h3,
                        port => Port,
                        handler => Handler,
                        server_ref => ServerRef
                    }),
                    {ok, Pid};
                {error, Reason} ->
                    {error, Reason}
            end;
        {error, Reason} ->
            {error, Reason}
    end.

listener_loop(Name) ->
    receive
        {stop, _From} ->
            persistent_term:erase({webtransport_listener, Name}),
            ok;
        _ ->
            listener_loop(Name)
    end.

%% ============================================================================
%% Internal Functions - Request Handling
%% ============================================================================

handle_h2_request(Conn, StreamId, <<"CONNECT">>, Path, Headers, Handler, HandlerOpts, Opts) ->
    case is_webtransport_request(Headers) of
        true ->
            case run_origin_check(Handler, Headers, HandlerOpts) of
                accept ->
                    accept_h2_session(Conn, StreamId, Path, Headers,
                                      Handler, HandlerOpts, Opts);
                {reject, Status, Reason} ->
                    h2:send_response(Conn, StreamId, Status, []),
                    h2:send_data(Conn, StreamId, Reason, true)
            end;
        false ->
            h2:send_response(Conn, StreamId, 400, []),
            h2:send_data(Conn, StreamId, <<"Bad Request">>, true)
    end;

handle_h2_request(Conn, StreamId, _Method, _Path, _Headers, _Handler, _HandlerOpts, _Opts) ->
    h2:send_response(Conn, StreamId, 405, []),
    h2:send_data(Conn, StreamId, <<"Method Not Allowed">>, true).

handle_h3_request(H3Conn, StreamId, <<"CONNECT">>, Path, Headers,
                  Handler, HandlerOpts, Opts, Router) ->
    ListenerMode = maps:get(compat_mode, Opts, auto),
    case classify_h3_connect(Headers, ListenerMode) of
        {ok, ClientMode} ->
            case run_origin_check(Handler, Headers, HandlerOpts) of
                accept ->
                    Opts1 = Opts#{compat_mode => ClientMode},
                    accept_h3_session(H3Conn, StreamId, Path, Headers,
                                      Handler, HandlerOpts, Opts1, Router);
                {reject, Status, Reason} ->
                    quic_h3:send_response(H3Conn, StreamId, Status, []),
                    quic_h3:send_data(H3Conn, StreamId, Reason, true)
            end;
        {error, Reason} ->
            Body = iolist_to_binary(io_lib:format("Bad Request: ~p", [Reason])),
            quic_h3:send_response(H3Conn, StreamId, 400, []),
            quic_h3:send_data(H3Conn, StreamId, Body, true)
    end;

handle_h3_request(H3Conn, StreamId, _Method, _Path, _Headers,
                  _Handler, _HandlerOpts, _Opts, _Router) ->
    quic_h3:send_response(H3Conn, StreamId, 405, []),
    quic_h3:send_data(H3Conn, StreamId, <<"Method Not Allowed">>, true).

accept_h2_session(Conn, StreamId, Path, Headers, Handler, HandlerOpts, Opts) ->
    h2:send_response(Conn, StreamId, 200, []),
    TransportState = webtransport_h2:new(Conn, StreamId),
    Authority = proplists:get_value(<<":authority">>, Headers, <<>>),
    Request = #{
        path => Path,
        authority => Authority,
        headers => Headers
    },
    %% Parse WebTransport-Init from the client's CONNECT request and
    %% apply the greater-of rule (draft-14 §4.3.2) against our defaults.
    ServerDefaults = #{
        u  => maps:get(max_streams_uni, Opts, ?DEFAULT_MAX_STREAMS_UNI),
        bl => maps:get(max_streams_bidi, Opts, ?DEFAULT_MAX_STREAMS_BIDI),
        br => maps:get(max_data, Opts, ?DEFAULT_MAX_DATA)
    },
    Negotiated = case proplists:get_value(<<"webtransport-init">>, Headers) of
        undefined -> ServerDefaults;
        InitBin ->
            case wt_h2_init:parse(InitBin) of
                {ok, Init} -> wt_h2_init:apply_greater_of(Init, ServerDefaults);
                {error, _} -> ServerDefaults
            end
    end,
    SessionOpts = maps:merge(HandlerOpts, #{
        request => Request,
        is_server => true,
        handler_opts => HandlerOpts,
        max_data => maps:get(br, Negotiated, ?DEFAULT_MAX_DATA),
        max_streams_bidi => maps:get(bl, Negotiated, ?DEFAULT_MAX_STREAMS_BIDI),
        max_streams_uni => maps:get(u, Negotiated, ?DEFAULT_MAX_STREAMS_UNI)
    }),
    case webtransport_session:start_link(h2, TransportState, Handler, SessionOpts) of
        {ok, Session} ->
            LoopPid = spawn(fun() -> h2_data_loop(Conn, StreamId, Session) end),
            _ = h2:set_stream_handler(Conn, StreamId, LoopPid);
        {error, Reason} ->
            h2:send_response(Conn, StreamId, 500, []),
            h2:send_data(Conn, StreamId,
                         iolist_to_binary(io_lib:format("~p", [Reason])), true)
    end.

accept_h3_session(H3Conn, StreamId, Path, Headers, Handler, HandlerOpts, Opts, Router) ->
    TransportState = webtransport_h3:new(H3Conn, StreamId, Router),
    Authority = proplists:get_value(<<":authority">>, Headers, <<>>),
    Request = #{
        path => Path,
        authority => Authority,
        headers => Headers
    },
    SessionOpts = maps:merge(HandlerOpts, #{
        request => Request,
        is_server => true,
        handler_opts => HandlerOpts,
        max_data => maps:get(max_data, Opts, ?DEFAULT_MAX_DATA),
        max_streams_bidi => maps:get(max_streams_bidi, Opts, ?DEFAULT_MAX_STREAMS_BIDI),
        max_streams_uni => maps:get(max_streams_uni, Opts, ?DEFAULT_MAX_STREAMS_UNI)
    }),
    %% Register the session with the router BEFORE the 200 goes out.
    %% Otherwise the client sees 200, opens extension streams, and the
    %% stream_type_data arrives at the router while sessions is still
    %% empty. The router then silently drops the data.
    case webtransport_session:start_link(h3, TransportState, Handler, SessionOpts) of
        {ok, Session} ->
            quic_h3:set_stream_handler(H3Conn, StreamId, Session),
            case Router of
                undefined -> ok;
                _ -> webtransport_h3_router:register_session(Router, StreamId, Session)
            end,
            quic_h3:send_response(H3Conn, StreamId, 200, []),
            ok;
        {error, Reason} ->
            quic_h3:send_response(H3Conn, StreamId, 500, []),
            quic_h3:send_data(H3Conn, StreamId,
                              iolist_to_binary(io_lib:format("~p", [Reason])), true)
    end.

%% Defaults to accept when the handler does not export origin_check/2.
run_origin_check(Handler, Headers, Opts) ->
    _ = code:ensure_loaded(Handler),
    case erlang:function_exported(Handler, origin_check, 2) of
        true ->
            try Handler:origin_check(Headers, Opts) of
                accept -> accept;
                {reject, Status, Reason}
                  when is_integer(Status), Status >= 400, Status < 600,
                       is_binary(Reason) ->
                    {reject, Status, Reason};
                Other ->
                    logger:warning("invalid origin_check/2 result: ~p", [Other]),
                    accept
            catch
                Kind:Err:Stk ->
                    logger:warning("origin_check/2 crashed ~p:~p ~p", [Kind, Err, Stk]),
                    {reject, 500, <<"origin check failed">>}
            end;
        false ->
            %% No custom origin_check/2 callback. The drafts (h3 §3.2,
            %% h2 §3.2) require: "the server MUST verify the Origin header
            %% to ensure that the specified origin is allowed". If the
            %% request carries an Origin header (browser client), reject
            %% by default so servers cannot accidentally skip verification.
            %% Non-browser requests (no Origin header) are accepted.
            case proplists:get_value(<<"origin">>, Headers) of
                undefined -> accept;
                _ -> {reject, 403, <<"origin not allowed">>}
            end
    end.

is_webtransport_request(Headers) ->
    case proplists:get_value(<<":protocol">>, Headers) of
        <<"webtransport">> -> true;
        _ -> false
    end.

%% Classify an incoming h3 CONNECT request against the listener's
%% compat_mode. Returns `{ok, ClientMode}' when the request is well-formed
%% and allowed, or `{error, Reason}' when it is malformed or disallowed.
classify_h3_connect(Headers, ListenerMode) ->
    case wt_h3:detect_compat_mode(Headers) of
        {ok, ClientMode} ->
            case allowed_by_listener(ClientMode, ListenerMode) of
                true -> {ok, ClientMode};
                false -> {error, {compat_mode_refused, ClientMode}}
            end;
        {error, _} = Err ->
            Err
    end.

allowed_by_listener(_, auto) -> true;
allowed_by_listener(Mode, Mode) -> true;
allowed_by_listener(_, _) -> false.

%% ============================================================================
%% Internal Functions - Data Loops
%% ============================================================================

h2_data_loop(Conn, StreamId, Session) ->
    receive
        {h2, Conn, {data, StreamId, Data, _IsFin}} ->
            %% Decode capsules and dispatch
            case webtransport_h2:decode_capsules(Data) of
                {ok, Capsules, _Rest} ->
                    lists:foreach(fun(Capsule) ->
                        dispatch_h2_capsule(Session, Capsule)
                    end, Capsules);
                {error, Reason} ->
                    %% Malformed capsule framing on the CONNECT stream is
                    %% a protocol violation. Close the session.
                    logger:warning("h2 capsule decode error: ~p", [Reason]),
                    webtransport_session:close(Session, 0, <<"malformed capsule">>)
            end,
            h2_data_loop(Conn, StreamId, Session);
        {h2, Conn, {stream_reset, StreamId, _ErrorCode}} ->
            webtransport_session:close(Session, 0, <<"stream reset">>);
        {h2, Conn, closed} ->
            webtransport_session:close(Session, 0, <<"connection closed">>);
        _ ->
            h2_data_loop(Conn, StreamId, Session)
    end.

dispatch_h2_capsule(Session, {wt_stream, WtStreamId, Data}) ->
    %% h2 multiplexes WT streams over CONNECT; peer-opened streams don't get
    %% a separate open signal, so we seed the session's stream map on first
    %% data (handle_stream_opened is idempotent). Direction follows the QUIC
    %% stream-id rule (bit 1 = 0 bidi, 1 = uni) per draft-14.
    webtransport_session:handle_stream_opened(Session, WtStreamId,
                                              webtransport_stream:stream_type(WtStreamId)),
    webtransport_session:handle_stream_data(Session, WtStreamId, Data, false);
dispatch_h2_capsule(Session, {wt_stream_fin, WtStreamId, Data}) ->
    webtransport_session:handle_stream_opened(Session, WtStreamId,
                                              webtransport_stream:stream_type(WtStreamId)),
    webtransport_session:handle_stream_data(Session, WtStreamId, Data, true);
dispatch_h2_capsule(Session, {datagram, Data}) ->
    webtransport_session:handle_datagram_data(Session, Data);
dispatch_h2_capsule(Session, {reset_stream, WtStreamId, ErrorCode}) ->
    webtransport_session:handle_stream_closed(Session, WtStreamId, {reset, ErrorCode});
dispatch_h2_capsule(Session, {stop_sending, WtStreamId, ErrorCode}) ->
    webtransport_session:handle_stream_closed(Session, WtStreamId, {stop_sending, ErrorCode});
dispatch_h2_capsule(Session, Capsule) ->
    webtransport_session:handle_capsule(Session, Capsule).

%% ============================================================================
%% Internal Functions - Client Connection
%% ============================================================================

connect_h2(Host, Port, Path, Opts, Handler) ->
    Caller = self(),
    UserHandlerOpts = maps:get(handler_opts, Opts, #{}),
    HandlerOpts = maps:merge(#{owner => Caller}, UserHandlerOpts),
    Opts1 = Opts#{ssl_opts => build_h2_ssl_opts(Opts)},
    case webtransport_h2:connect(Host, Port, Path, Opts1) of
        {ok, TransportState} ->
            %% Default handler for client
            ActualHandler = case Handler of
                undefined -> webtransport_client_handler;
                _ -> Handler
            end,
            Request = #{
                path => Path,
                authority => iolist_to_binary([Host, ":", integer_to_list(Port)]),
                headers => maps:get(headers, Opts, [])
            },
            SessionOpts = #{
                request => Request,
                is_server => false,
                handler_opts => HandlerOpts
            },
            case webtransport_session:start_link(h2, TransportState, ActualHandler, SessionOpts) of
                {ok, Session} ->
                    H2Conn = webtransport_h2:h2_conn(TransportState),
                    StreamId = webtransport_h2:connect_stream_id(TransportState),
                    LoopPid = spawn_link(fun() -> h2_data_loop(H2Conn, StreamId, Session) end),
                    _ = h2:set_stream_handler(H2Conn, StreamId, LoopPid),
                    {ok, Session};
                {error, _} = Err ->
                    Err
            end;
        {error, Reason} ->
            {error, Reason}
    end.

connect_h3(Host, Port, Path, Opts0, Handler) ->
    Caller = self(),
    UserHandlerOpts = maps:get(handler_opts, Opts0, #{}),
    HandlerOpts = maps:merge(#{owner => Caller}, UserHandlerOpts),
    Opts = Opts0#{handler_opts => HandlerOpts},
    HostBin = if is_list(Host) -> list_to_binary(Host); true -> Host end,
    {ok, Router} = webtransport_h3_router:start_link(Caller),
    H3ConnOpts = build_h3_connect_opts(Opts),
    case webtransport_h3_router:client_connect(Router, Host, Port, H3ConnOpts) of
        {ok, H3Conn} ->
            h3_validate_and_request(H3Conn, HostBin, Port, Path, Opts, Handler, Router);
        {error, Reason} ->
            {error, Reason}
    end.

build_h2_ssl_opts(Opts) ->
    Base = maps:get(ssl_opts, Opts, []),
    WithVerify =
        case maps:get(verify, Opts, verify_peer) of
            verify_none -> [{verify, verify_none} | Base];
            verify_peer -> [{verify, verify_peer} | Base]
        end,
    WithCACerts =
        case {maps:find(cacertfile, Opts), maps:find(cacerts, Opts)} of
            {{ok, CAFile}, _} ->
                case read_cacerts_file(CAFile) of
                    {ok, CACerts} -> [{cacerts, CACerts} | WithVerify];
                    _ -> WithVerify
                end;
            {_, {ok, CACerts}} -> [{cacerts, CACerts} | WithVerify];
            _ -> WithVerify
        end,
    WithCert =
        case {maps:find(certfile, Opts), maps:find(cert, Opts)} of
            {{ok, CertFile}, _} -> [{certfile, CertFile} | WithCACerts];
            {_, {ok, CertDer}} -> [{cert, CertDer} | WithCACerts];
            _ -> WithCACerts
        end,
    case {maps:find(keyfile, Opts), maps:find(key, Opts)} of
        {{ok, KeyFile}, _} -> [{keyfile, KeyFile} | WithCert];
        {_, {ok, KeyTerm}} -> [{key, KeyTerm} | WithCert];
        _ -> WithCert
    end.

build_h3_connect_opts(Opts) ->
    %% Client picks a compat_mode explicitly; default `latest' (draft-15).
    %% Clients must not auto-probe the server, so the default is a single
    %% disjoint handshake shape.
    CompatMode = case maps:get(compat_mode, Opts, latest) of
        legacy_browser_compat -> legacy_browser_compat;
        _ -> latest
    end,
    WTSettings = wt_h3:default_settings(CompatMode),
    BaseOpts = #{
        settings => WTSettings,
        sync => true,
        connect_timeout => maps:get(timeout, Opts, 30000),
        verify => maps:get(verify, Opts, verify_peer),
        quic_opts => #{
            max_datagram_frame_size => 65535,
            reset_stream_at => true
        },
        h3_datagram_enabled => true,
        stream_type_handler => fun
            (uni,  _, 16#54) -> claim;
            (bidi, _, 16#41) -> claim;
            (_, _, _) -> ignore
        end
    },
    %% Handle client certificate if provided
    WithCert = case {maps:find(certfile, Opts), maps:find(cert, Opts)} of
        {{ok, CertFile}, _} ->
            case read_cert_file(CertFile) of
                {ok, CertDer} -> BaseOpts#{cert => CertDer};
                _ -> BaseOpts
            end;
        {_, {ok, CertDer}} ->
            BaseOpts#{cert => CertDer};
        _ ->
            BaseOpts
    end,
    %% Handle client key if provided
    WithKey = case {maps:find(keyfile, Opts), maps:find(key, Opts)} of
        {{ok, KeyFile}, _} ->
            case read_key_file(KeyFile) of
                {ok, KeyTerm} -> WithCert#{key => KeyTerm};
                _ -> WithCert
            end;
        {_, {ok, KeyTerm}} ->
            WithCert#{key => KeyTerm};
        _ ->
            WithCert
    end,
    %% Handle CA certificates if provided
    case {maps:find(cacertfile, Opts), maps:find(cacerts, Opts)} of
        {{ok, CAFile}, _} ->
            case read_cacerts_file(CAFile) of
                {ok, CACerts} -> WithKey#{cacerts => CACerts};
                _ -> WithKey
            end;
        {_, {ok, CACerts}} ->
            WithKey#{cacerts => CACerts};
        _ ->
            WithKey
    end.

h3_validate_and_request(H3Conn, HostBin, Port, Path, Opts, Handler, Router) ->
    CompatMode = maps:get(compat_mode, Opts, latest),
    QuicConn = quic_h3:get_quic_conn(H3Conn),
    case wt_h3:validate_wt_support(H3Conn, QuicConn, CompatMode) of
        ok ->
            Authority = <<HostBin/binary, ":", (integer_to_binary(Port))/binary>>,
            h3_send_connect(H3Conn, Authority, Path, Opts, Handler, Router, CompatMode);
        {error, Reason} ->
            quic_h3:close(H3Conn),
            {error, Reason}
    end.

h3_send_connect(H3Conn, Authority, Path, Opts, Handler, Router, CompatMode) ->
    Headers = maps:get(headers, Opts, []),
    case wt_h3:request_session(H3Conn, Authority, Path, Headers, CompatMode) of
        {ok, SessionId} ->
            h3_await_response(H3Conn, SessionId, Authority, Path, Opts, Handler, Router);
        {error, Reason} ->
            quic_h3:close(H3Conn),
            {error, Reason}
    end.

h3_await_response(H3Conn, SessionId, Authority, Path, Opts, Handler, Router) ->
    Timeout = maps:get(timeout, Opts, 30000),
    receive
        {quic_h3, H3Conn, {response, SessionId, Status, Headers}} when Status >= 200, Status < 300 ->
            h3_start_session(H3Conn, SessionId, Authority, Path, Headers, Opts, Handler, Router);
        {quic_h3, H3Conn, {response, SessionId, Status, _Headers}} ->
            quic_h3:close(H3Conn),
            {error, {http_error, Status}};
        {quic_h3, H3Conn, closed} ->
            {error, connection_closed}
    after Timeout ->
        quic_h3:close(H3Conn),
        {error, timeout}
    end.

h3_start_session(H3Conn, SessionId, Authority, Path, Headers, Opts, Handler, Router) ->
    TransportState = webtransport_h3:new(H3Conn, SessionId, Router),
    ActualHandler = case Handler of
        undefined -> webtransport_client_handler;
        _ -> Handler
    end,
    Request = #{
        path => Path,
        authority => Authority,
        headers => Headers
    },
    HandlerOpts = maps:get(handler_opts, Opts, #{}),
    SessionOpts = #{
        request => Request,
        is_server => false,
        handler_opts => HandlerOpts
    },
    case webtransport_session:start_link(h3, TransportState, ActualHandler, SessionOpts) of
        {ok, Session} ->
            webtransport_h3_router:register_session(Router, SessionId, Session),
            {ok, Session};
        {error, _} = Err ->
            Err
    end.

%% ============================================================================
%% Handler Factories
%% ============================================================================

make_h2_handler(Handler, HandlerOpts, Opts) ->
    fun(Conn, StreamId, Method, Path, Headers) ->
        handle_h2_request(Conn, StreamId, Method, Path, Headers, Handler, HandlerOpts, Opts)
    end.

make_h3_handler(Handler, HandlerOpts, Opts, Router) ->
    fun(H3Conn, StreamId, Method, Path, Headers) ->
        handle_h3_request(H3Conn, StreamId, Method, Path, Headers,
                          Handler, HandlerOpts, Opts, Router)
    end.

wt_stream_type_handler() ->
    fun
        (uni,  _StreamId, 16#54) -> claim;
        (bidi, _StreamId, 16#41) -> claim;
        (_, _, _) -> ignore
    end.

%% ============================================================================
%% Certificate/Key Helpers
%% ============================================================================

%% @private Read and decode certificate and private key files.
read_cert_and_key(CertFile, KeyFile) ->
    case {file:read_file(CertFile), file:read_file(KeyFile)} of
        {{ok, CertPem}, {ok, KeyPem}} ->
            case public_key:pem_decode(CertPem) of
                [{_, CertDer, _} | _] ->
                    case decode_private_key(KeyPem) of
                        {ok, PrivateKey} ->
                            {ok, CertDer, PrivateKey};
                        {error, Reason} ->
                            {error, {key_decode_failed, Reason}}
                    end;
                [] ->
                    {error, invalid_certificate}
            end;
        {{error, CertErr}, _} ->
            {error, {cert_read_failed, CertErr}};
        {_, {error, KeyErr}} ->
            {error, {key_read_failed, KeyErr}}
    end.

%% @private Decode a PEM-encoded private key.
decode_private_key(PemData) ->
    case public_key:pem_decode(PemData) of
        [{Type, Der, not_encrypted}] ->
            decode_key_entry(Type, Der);
        [{Type, Der, _Cipher}] ->
            decode_key_entry(Type, Der);
        _ ->
            {error, invalid_private_key}
    end.

decode_key_entry('RSAPrivateKey', Der) ->
    {ok, public_key:der_decode('RSAPrivateKey', Der)};
decode_key_entry('ECPrivateKey', Der) ->
    {ok, public_key:der_decode('ECPrivateKey', Der)};
decode_key_entry('PrivateKeyInfo', Der) ->
    {ok, public_key:der_decode('PrivateKeyInfo', Der)};
decode_key_entry(Type, _Der) ->
    {error, {unsupported_key_type, Type}}.

%% @private Read a single certificate from a PEM file.
read_cert_file(CertFile) ->
    case file:read_file(CertFile) of
        {ok, PemData} ->
            case public_key:pem_decode(PemData) of
                [{_, CertDer, _} | _] -> {ok, CertDer};
                [] -> {error, invalid_certificate}
            end;
        {error, Reason} ->
            {error, {cert_read_failed, Reason}}
    end.

%% @private Read a private key from a PEM file.
read_key_file(KeyFile) ->
    case file:read_file(KeyFile) of
        {ok, PemData} ->
            decode_private_key(PemData);
        {error, Reason} ->
            {error, {key_read_failed, Reason}}
    end.

%% @private Read CA certificates from a PEM file.
read_cacerts_file(CAFile) ->
    case file:read_file(CAFile) of
        {ok, PemData} ->
            Certs = [Der || {_, Der, _} <- public_key:pem_decode(PemData)],
            case Certs of
                [] -> {error, no_certificates};
                _ -> {ok, Certs}
            end;
        {error, Reason} ->
            {error, {cacerts_read_failed, Reason}}
    end.