Crafter.Network/implementations/Crafter.Network-ClientQUIC.cpp

//SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

module;
#include <msquic.h>
#include <cstring>
module Crafter.Network:ClientQUIC_impl;
import :ClientQUIC;
import Crafter.Thread;
import std;

using namespace Crafter;

namespace {
    // Process-wide msquic API table + registration. Initialised lazily on
    // first ClientQUIC/ListenerQUIC construction; tear-down happens at
    // process exit via the destructor of the static object.
    struct MsQuicRuntime {
        const QUIC_API_TABLE* api = nullptr;
        HQUIC registration = nullptr;
        std::mutex initMutex;
        bool initialised = false;

        void Ensure() {
            std::lock_guard lock(initMutex);
            if (initialised) return;
            QUIC_STATUS s = MsQuicOpen2(&api);
            if (QUIC_FAILED(s)) {
                throw QUICException(std::format("MsQuicOpen2 failed: 0x{:x}", static_cast<unsigned>(s)));
            }
            QUIC_REGISTRATION_CONFIG regConfig{ "crafter.network", QUIC_EXECUTION_PROFILE_LOW_LATENCY };
            s = api->RegistrationOpen(&regConfig, &registration);
            if (QUIC_FAILED(s)) {
                MsQuicClose(api);
                api = nullptr;
                throw QUICException(std::format("RegistrationOpen failed: 0x{:x}", static_cast<unsigned>(s)));
            }
            initialised = true;
        }

        ~MsQuicRuntime() {
            if (registration) api->RegistrationClose(registration);
            if (api) MsQuicClose(api);
        }
    };

    MsQuicRuntime& Runtime() {
        static MsQuicRuntime r;
        r.Ensure();
        return r;
    }

    // Open-stream bookkeeping, shared by a connection and every stream on it.
    // A stream counts as open from the moment msquic has a callback handler
    // for it until its handle has been StreamClose'd. ~ClientQUIC drains this
    // before closing the connection: QUICStream::Stop only *initiates* a
    // shutdown whose async completion does the actual close, so without the
    // wait a connection can be closed with streams still open on the msquic
    // side.
    //
    // Reached through a shared_ptr rather than through the owning ClientQUIC*
    // so that a stream finalising after the (bounded) wait gave up decrements
    // a live object instead of a freed connection.
    struct StreamRegistry {
        std::mutex mtx;
        std::condition_variable cv;
        int open = 0;

        void Add() {
            std::lock_guard lk(mtx);
            ++open;
        }

        void Remove() {
            {
                std::lock_guard lk(mtx);
                --open;
            }
            cv.notify_all();
        }

        // False if the timeout expired with streams still open.
        bool WaitForDrain(std::chrono::milliseconds timeout) {
            std::unique_lock lk(mtx);
            return cv.wait_for(lk, timeout, [&]{ return open == 0; });
        }
    };

    // Bound on how long ~ClientQUIC waits for the connection's streams to
    // finish closing. The decrements arrive on msquic worker threads, so an
    // unbounded wait would turn a dropped peer into a hung destructor.
    //
    // Expiring is a bug, not a slow path: ConnectionShutdown makes msquic
    // deliver SHUTDOWN_COMPLETE for every stream it knows about, OpenStream
    // passes SHUTDOWN_ON_FAIL so a stream whose start lost the race to that
    // shutdown gets one too, and QUICStream::Stop aborts the receive direction
    // so a bidi stream does not sit waiting on a peer that will never close
    // its send side. With those three the count always reaches zero; the bound
    // only stops an msquic-side surprise from wedging the destructor outright.
    constexpr std::chrono::milliseconds streamDrainTimeout{ 5000 };

    // Encode an ALPN string into the wire format msquic expects: a length
    // byte followed by the ASCII characters. Lifetime of the returned buffer
    // matches the caller's storage in `out`.
    QUIC_BUFFER MakeAlpn(const std::string& alpn, std::vector<std::uint8_t>& out) {
        if (alpn.size() > 255) throw QUICException("ALPN string too long (max 255)");
        out.assign(alpn.begin(), alpn.end());
        QUIC_BUFFER b{};
        b.Length = static_cast<std::uint32_t>(out.size());
        b.Buffer = out.data();
        return b;
    }
}

// ---------------- QUICStream::Impl ----------------
struct QUICStream::Impl {
    ClientQUIC* connection = nullptr;

    std::mutex mtx;
    std::condition_variable cv;
    std::deque<std::vector<char>> pending;
    bool peerSendClosed = false;
    bool shutdownComplete = false;
    bool sendInFlight = false;

    // Up to `count` received bytes, waiting for them; fewer once the peer
    // has closed its side and the rest is taken.
    std::vector<char> TakeUpTo(std::uint32_t count) {
        std::vector<char> out;
        out.reserve(count);
        std::unique_lock lk(mtx);
        while (out.size() < count) {
            cv.wait(lk, [&]{ return !pending.empty() || peerSendClosed || shutdownComplete; });
            Take(out, count);
            if (pending.empty() && (peerSendClosed || shutdownComplete)) break;
        }
        return out;
    }

    // Exactly `count` received bytes, once they have all arrived; nothing,
    // leaving what did arrive where it is, once the peer has closed its side
    // short of them. One hold of the lock throughout, so no other receive
    // takes bytes from among them.
    std::optional<std::vector<char>> TakeExactly(std::uint32_t count) {
        std::unique_lock lk(mtx);
        auto arrived = [&] {
            std::size_t bytes = 0;
            for (const std::vector<char>& chunk : pending) bytes += chunk.size();
            return bytes;
        };
        cv.wait(lk, [&]{ return arrived() >= count || peerSendClosed || shutdownComplete; });
        if (arrived() < count) return std::nullopt;
        std::vector<char> out;
        out.reserve(count);
        Take(out, count);
        return out;
    }

    // Moves pending bytes onto `out` until it holds `count`. Under mtx.
    void Take(std::vector<char>& out, std::size_t count) {
        while (!pending.empty() && out.size() < count) {
            auto& chunk = pending.front();
            std::size_t want = std::min<std::size_t>(chunk.size(), count - out.size());
            out.insert(out.end(), chunk.begin(), chunk.begin() + want);
            if (want == chunk.size()) {
                pending.pop_front();
            } else {
                chunk.erase(chunk.begin(), chunk.begin() + want);
            }
        }
    }

    // Lifetime: msquic delivers stream callbacks (RECEIVE, SHUTDOWN_COMPLETE, …)
    // possibly AFTER the owning QUICStream wrapper is destroyed (the wrapper only
    // initiates a graceful shutdown; completion is async). To make that safe, the
    // callback context is a heap `shared_ptr<Impl>*` (selfRef) that holds one
    // strong ref; every callback copies it locally so the Impl stays alive for
    // the callback's duration. The terminal SHUTDOWN_COMPLETE callback closes the
    // stream and releases selfRef, after which the Impl is freed once the last
    // ref (wrapper / in-flight callback copy) drops.
    std::shared_ptr<Impl>* selfRef = nullptr;

    // The owning connection's open-stream count. Incremented where the msquic
    // handler is installed (QUICStream's handle constructor / OpenStream),
    // decremented once the handle has been closed.
    std::shared_ptr<StreamRegistry> registry;

    // The connection's registry. Defined below ClientQUIC::Impl (which owns
    // it); a member of QUICStream::Impl rather than a free function so it
    // inherits QUICStream's friendship with ClientQUIC.
    static std::shared_ptr<StreamRegistry> RegistryOf(ClientQUIC* connection);

    // ---- msquic handle ownership ----------------------------------------
    // StreamClose is "equivalent to free": any other msquic call on the handle
    // that overlaps it is a use-after-free, which msquic reports as a bugcheck
    // on whichever thread trips over the wreckage. The handle therefore has a
    // refcount, and whoever drops the last ref closes it. There are two kinds
    // of holder:
    //
    //   - The msquic callback holds exactly one ref for the stream's lifetime,
    //     released by Retire() once SHUTDOWN_COMPLETE has been delivered (or by
    //     the OpenStream error paths, where no callback will ever run).
    //   - Every app thread inside an msquic call on the handle holds one for
    //     the duration of that call (AcquireHandle / ReleaseHandle).
    //
    // Without the second kind, a thread sending on a stream races the worker
    // thread that closes it when the connection is torn down underneath —
    // the shape any client with periodic acks or keepalives has.
    std::mutex handleMtx;
    HQUIC handle = nullptr;    // cleared when the last ref lets go
    int handleRefs = 1;        // starts with the callback's ref
    bool retired = false;      // the callback's ref has been released

    // The handle, valid until the matching ReleaseHandle; nullptr once the
    // stream is retired, in which case do not call ReleaseHandle.
    HQUIC AcquireHandle() {
        std::lock_guard lk(handleMtx);
        if (retired || !handle) return nullptr;
        ++handleRefs;
        return handle;
    }

    void ReleaseHandle() {
        HQUIC toClose = nullptr;
        {
            std::lock_guard lk(handleMtx);
            if (--handleRefs > 0) return;
            toClose = handle;
            handle = nullptr;
        }
        // Last ref: nobody else can be inside an msquic call on this handle,
        // and AcquireHandle will hand it out no more.
        if (toClose) Runtime().api->StreamClose(toClose);
        std::shared_ptr<StreamRegistry> reg = std::move(registry);
        std::shared_ptr<Impl>* ref = selfRef;
        selfRef = nullptr;
        delete ref;   // may destroy *this — touch no members past here
        // Last of all: a destructor blocked in WaitForDrain may run
        // ConnectionClose the moment this hits zero, and msquic wants every
        // StreamClose on the connection to have happened first.
        if (reg) reg->Remove();
    }

    // Publish the msquic handle. Called once, while the stream is still local
    // to its constructor and no callback can run on it.
    void SetHandle(HQUIC h) {
        std::lock_guard lk(handleMtx);
        handle = h;
    }

    // Release the callback's ref. Idempotent, so the SHUTDOWN_COMPLETE handler
    // and an OpenStream error path can both call it without racing.
    void Retire() {
        {
            std::lock_guard lk(handleMtx);
            if (retired) return;
            retired = true;
        }
        ReleaseHandle();
    }

    static QUIC_STATUS QUIC_API Callback(HQUIC stream, void* ctx, QUIC_STREAM_EVENT* ev) {
        // Keep the Impl alive for the whole callback, even if the wrapper (and
        // its ref) go away concurrently.
        std::shared_ptr<Impl> keepAlive = *static_cast<std::shared_ptr<Impl>*>(ctx);
        Impl* self = keepAlive.get();
        switch (ev->Type) {
            case QUIC_STREAM_EVENT_RECEIVE: {
                std::vector<char> chunk;
                std::uint64_t total = 0;
                for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) {
                    total += ev->RECEIVE.Buffers[i].Length;
                }
                chunk.reserve(static_cast<std::size_t>(total));
                for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) {
                    const QUIC_BUFFER& b = ev->RECEIVE.Buffers[i];
                    chunk.insert(chunk.end(), b.Buffer, b.Buffer + b.Length);
                }
                {
                    std::lock_guard lk(self->mtx);
                    if (!chunk.empty()) self->pending.push_back(std::move(chunk));
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_STREAM_EVENT_SEND_COMPLETE: {
                {
                    std::lock_guard lk(self->mtx);
                    self->sendInFlight = false;
                }
                if (ev->SEND_COMPLETE.ClientContext) {
                    delete[] static_cast<char*>(ev->SEND_COMPLETE.ClientContext);
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN:
            case QUIC_STREAM_EVENT_PEER_SEND_ABORTED: {
                {
                    std::lock_guard lk(self->mtx);
                    self->peerSendClosed = true;
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE: {
                {
                    std::lock_guard lk(self->mtx);
                    self->peerSendClosed = true;
                    self->shutdownComplete = true;
                    // No further SEND_COMPLETE can arrive, so release anyone
                    // waiting on one — they throw on shutdownComplete instead.
                    self->sendInFlight = false;
                }
                self->cv.notify_all();
                self->Retire();
                return QUIC_STATUS_SUCCESS;
            }
            default:
                return QUIC_STATUS_SUCCESS;
        }
    }
};

QUICStream::QUICStream() = default;

QUICStream::QUICStream(HQUIC handle, ClientQUIC* connection)
    : handle(handle), connection(connection), impl(std::make_shared<Impl>())
{
    impl->SetHandle(handle);   // handleRefs already accounts for the callback
    impl->connection = connection;
    impl->registry = Impl::RegistryOf(connection);
    if (impl->registry) impl->registry->Add();
    impl->selfRef = new std::shared_ptr<Impl>(impl);   // strong ref owned by the callback
    Runtime().api->SetCallbackHandler(handle, reinterpret_cast<void*>(&Impl::Callback), impl->selfRef);
}

QUICStream::QUICStream(QUICStream&& other) noexcept
    : handle(other.handle), connection(other.connection),
      canSend(other.canSend), canReceive(other.canReceive),
      impl(std::move(other.impl))
{
    other.handle = nullptr;
    other.connection = nullptr;
}

QUICStream& QUICStream::operator=(QUICStream&& other) noexcept {
    if (this != &other) {
        Stop();
        handle = other.handle;
        connection = other.connection;
        canSend = other.canSend;
        canReceive = other.canReceive;
        impl = std::move(other.impl);
        other.handle = nullptr;
        other.connection = nullptr;
    }
    return *this;
}

QUICStream::~QUICStream() {
    Stop();
}

void QUICStream::Stop() {
    if (!handle) return;
    handle = nullptr;   // the wrapper is done with the stream either way
    if (!impl) return;
    // Only INITIATE the shutdown here; the async SHUTDOWN_COMPLETE callback
    // releases the handle, and the Impl is a shared_ptr held by the callback's
    // selfRef, so it safely outlives this wrapper until that callback runs.
    // A null handle means SHUTDOWN_COMPLETE already fired and msquic has closed
    // the stream — shutting it down again would be a use-after-free.
    HQUIC live = impl->AcquireHandle();
    if (!live) return;
    // GRACEFUL closes our send direction only, and msquic rejects it combined
    // with any other flag, so the receive direction needs a call of its own.
    // Without it a bidirectional stream whose peer keeps its send side open
    // never reaches SHUTDOWN_COMPLETE, so its handle is never closed and the
    // connection can never finish tearing down.
    Runtime().api->StreamShutdown(live, QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0);
    if (canReceive) {
        Runtime().api->StreamShutdown(live, QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE, 0);
    }
    impl->ReleaseHandle();
}

void QUICStream::SendSync(const void* buffer, std::uint32_t size, bool finish) {
    if (!handle || !canSend) throw QUICClosedException();
    // Hold the handle across StreamSend: without this a connection torn down
    // on another thread can close the stream between the check above and the
    // call below, and msquic bugchecks on the freed handle.
    HQUIC live = impl->AcquireHandle();
    if (!live) throw QUICClosedException();
    auto* copy = new char[size];
    std::memcpy(copy, buffer, size);
    QUIC_BUFFER quicBuf{};
    quicBuf.Buffer = reinterpret_cast<std::uint8_t*>(copy);
    quicBuf.Length = size;
    {
        std::lock_guard lk(impl->mtx);
        impl->sendInFlight = true;
    }
    QUIC_SEND_FLAGS flags = finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE;
    QUIC_STATUS s = Runtime().api->StreamSend(live, &quicBuf, 1, flags, copy);
    impl->ReleaseHandle();
    if (QUIC_FAILED(s)) {
        {
            // No SEND_COMPLETE will arrive for a send msquic never took.
            std::lock_guard lk(impl->mtx);
            impl->sendInFlight = false;
        }
        impl->cv.notify_all();
        delete[] copy;
        throw QUICException(std::format("StreamSend failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    std::unique_lock lk(impl->mtx);
    impl->cv.wait(lk, [&]{ return !impl->sendInFlight || impl->shutdownComplete; });
    if (impl->shutdownComplete) throw QUICClosedException();
}

std::vector<char> QUICStream::RecieveSync() {
    if (!handle || !canReceive) throw QUICClosedException();
    std::unique_lock lk(impl->mtx);
    impl->cv.wait(lk, [&]{ return !impl->pending.empty() || impl->peerSendClosed || impl->shutdownComplete; });
    if (!impl->pending.empty()) {
        auto out = std::move(impl->pending.front());
        impl->pending.pop_front();
        return out;
    }
    throw QUICClosedException();
}

std::vector<char> QUICStream::RecieveUntilCloseSync() {
    if (!handle || !canReceive) throw QUICClosedException();
    std::vector<char> out;
    while (true) {
        std::unique_lock lk(impl->mtx);
        impl->cv.wait(lk, [&]{ return !impl->pending.empty() || impl->peerSendClosed || impl->shutdownComplete; });
        while (!impl->pending.empty()) {
            auto& chunk = impl->pending.front();
            out.insert(out.end(), chunk.begin(), chunk.end());
            impl->pending.pop_front();
        }
        if (impl->peerSendClosed || impl->shutdownComplete) return out;
    }
}

std::vector<char> QUICStream::RecieveUntilFullSync(std::uint32_t bufferSize) {
    if (!handle || !canReceive) throw QUICClosedException();
    // Closed early: what did arrive stays for the next receive.
    std::optional<std::vector<char>> out = impl->TakeExactly(bufferSize);
    if (!out) throw QUICClosedException();
    return std::move(*out);
}

void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish,
                           std::function<void(bool sent)> onSent) {
    // Copy now: the caller's buffer may not outlive the enqueued task.
    std::vector<char> copy(static_cast<const char*>(buffer),
                           static_cast<const char*>(buffer) + size);
    ThreadPool::Enqueue([this, copy = std::move(copy), finish, onSent = std::move(onSent)]() mutable {
        // Nothing may leave a pool task; the callback hears how it went.
        bool sent = true;
        try {
            this->SendSync(copy.data(), static_cast<std::uint32_t>(copy.size()), finish);
        } catch (...) {
            sent = false;
        }
        if (onSent) onSent(sent);
    });
}

// The async receives deliver a closed stream as the browser build does: an
// empty chunk, as nothing may leave a pool task.
void QUICStream::RecieveAsync(std::function<void(std::vector<char>)> cb) {
    ThreadPool::Enqueue([this, cb]{
        std::vector<char> chunk;
        try {
            chunk = this->RecieveSync();
        } catch (const QUICClosedException&) {}
        cb(std::move(chunk));
    });
}
void QUICStream::RecieveUntilCloseAsync(std::function<void(std::vector<char>)> cb) {
    ThreadPool::Enqueue([this, cb]{
        std::vector<char> data;
        try {
            data = this->RecieveUntilCloseSync();
        } catch (const QUICClosedException&) {}
        cb(std::move(data));
    });
}
void QUICStream::RecieveUntilFullAsync(std::uint32_t bufferSize, std::function<void(std::vector<char>)> cb) {
    ThreadPool::Enqueue([this, bufferSize, cb]{
        cb(handle && canReceive ? impl->TakeUpTo(bufferSize) : std::vector<char>{});
    });
}

void QUICStream::PrependReceived(std::vector<char> bytes) {
    if (bytes.empty() || !impl) return;
    {
        std::lock_guard lk(impl->mtx);
        impl->pending.push_front(std::move(bytes));
    }
    impl->cv.notify_all();
}

std::uint64_t QUICStream::GetStreamId() const {
    if (!handle) throw QUICException("GetStreamId: stream is not open");
    // GetParam blocks on the connection's msquic worker, so it must not be the
    // worker that waits for us — hence a refcount rather than an exclusion
    // lock: SHUTDOWN_COMPLETE just drops its ref and returns.
    HQUIC live = impl->AcquireHandle();
    if (!live) throw QUICException("GetStreamId: stream is not open");
    QUIC_UINT62 id = 0;
    std::uint32_t size = sizeof(id);
    QUIC_STATUS s = Runtime().api->GetParam(live, QUIC_PARAM_STREAM_ID, &size, &id);
    impl->ReleaseHandle();
    if (QUIC_FAILED(s)) {
        throw QUICException(std::format("GetParam(QUIC_PARAM_STREAM_ID) failed: 0x{:x}",
                                        static_cast<unsigned>(s)));
    }
    return static_cast<std::uint64_t>(id);
}

// ---------------- ClientQUIC::Impl ----------------
struct ClientQUIC::Impl {
    HQUIC connection = nullptr;
    HQUIC configuration = nullptr;
    bool ownsConfiguration = true;

    std::mutex mtx;
    std::condition_variable cv;
    bool connected = false;
    bool closed = false;
    QUIC_STATUS shutdownStatus = QUIC_STATUS_SUCCESS;

    std::function<void(QUICStream)> onStream;
    std::function<void(std::vector<char>)> onDatagram;
    std::deque<std::vector<char>> datagramQueue;
    // Streams the peer started before the user installed an OnStream
    // handler. Without this backlog the early streams (e.g. an h3 server's
    // control stream right after handshake) would be aborted in the
    // PEER_STREAM_STARTED branch and the connection would die with
    // H3_MISSING_SETTINGS on the peer side.
    std::deque<QUICStream> pendingStreams;

    // Streams currently open on this connection. Shared with each QUICStream
    // so the count survives a stream that outlives us.
    std::shared_ptr<StreamRegistry> streams = std::make_shared<StreamRegistry>();

    ClientQUIC* outer = nullptr;

    // Take the connection handle away from whoever else might close it.
    // `connection` is the single owner token: exactly one of ~ClientQUIC and
    // the SHUTDOWN_COMPLETE callback wins the claim, and only the winner calls
    // ConnectionClose. Without this the two race and msquic bugchecks on the
    // second close (CXPLAT_TEL_ASSERT(!Connection->State.HandleClosed)).
    HQUIC ClaimConnection() {
        std::lock_guard lk(mtx);
        HQUIC claimed = connection;
        connection = nullptr;
        return claimed;
    }

    // Shut the connection down and close it, waiting (bounded) for every
    // stream on it to reach StreamClose first. No-op if someone else already
    // claimed the handle. Never holds `mtx` across an msquic call: the
    // connection callback takes it too.
    void CloseConnection() {
        HQUIC claimed = ClaimConnection();
        if (!claimed) return;
        // Aborts any stream still open; each one's SHUTDOWN_COMPLETE runs
        // Retire, which closes it and drops the count below.
        Runtime().api->ConnectionShutdown(claimed, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
        streams->WaitForDrain(streamDrainTimeout);
        Runtime().api->ConnectionClose(claimed);
    }

    static QUIC_STATUS QUIC_API Callback(HQUIC conn, void* ctx, QUIC_CONNECTION_EVENT* ev) {
        auto* self = static_cast<Impl*>(ctx);
        switch (ev->Type) {
            case QUIC_CONNECTION_EVENT_CONNECTED: {
                {
                    std::lock_guard lk(self->mtx);
                    self->connected = true;
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_TRANSPORT:
            case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_PEER: {
                {
                    std::lock_guard lk(self->mtx);
                    self->closed = true;
                    if (ev->Type == QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_TRANSPORT) {
                        self->shutdownStatus = ev->SHUTDOWN_INITIATED_BY_TRANSPORT.Status;
                    }
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE: {
                {
                    std::lock_guard lk(self->mtx);
                    self->closed = true;
                }
                self->cv.notify_all();
                // AppCloseInProgress means ~ClientQUIC is already inside
                // ConnectionClose; otherwise close the handle here so a
                // connection dropped by the peer does not leak while the
                // owning ClientQUIC lives on. ClaimConnection settles which
                // of the two actually does it. Every stream on the connection
                // has already been shut down (and so closed) by the time
                // msquic reports the connection complete.
                if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress == 0
                    && self->ClaimConnection() != nullptr) {
                    Runtime().api->ConnectionClose(conn);
                }
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_PEER_STREAM_STARTED: {
                HQUIC streamHandle = ev->PEER_STREAM_STARTED.Stream;
                bool unidirectional = (ev->PEER_STREAM_STARTED.Flags
                                       & QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL) != 0;
                QUICStream stream(streamHandle, self->outer);
                if (unidirectional) {
                    // Peer-initiated unidi: peer sends, we read; we cannot send.
                    stream.canSend = false;
                    stream.canReceive = true;
                }
                std::function<void(QUICStream)> cb;
                {
                    std::lock_guard lk(self->mtx);
                    cb = self->onStream;
                    if (!cb) {
                        // Buffer until OnStream is installed; OnStream's
                        // setter drains this queue.
                        self->pendingStreams.push_back(std::move(stream));
                        return QUIC_STATUS_SUCCESS;
                    }
                }
                auto* shared = new QUICStream(std::move(stream));
                ThreadPool::Enqueue([cb, shared]{
                    cb(std::move(*shared));
                    delete shared;
                });
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_DATAGRAM_RECEIVED: {
                std::vector<char> chunk(ev->DATAGRAM_RECEIVED.Buffer->Buffer,
                                       ev->DATAGRAM_RECEIVED.Buffer->Buffer + ev->DATAGRAM_RECEIVED.Buffer->Length);
                if (self->onDatagram) {
                    auto cb = self->onDatagram;
                    ThreadPool::Enqueue([cb, chunk = std::move(chunk)]() mutable { cb(std::move(chunk)); });
                } else {
                    std::lock_guard lk(self->mtx);
                    self->datagramQueue.push_back(std::move(chunk));
                    self->cv.notify_all();
                }
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_DATAGRAM_SEND_STATE_CHANGED: {
                // msquic fires this event multiple times per datagram (e.g.
                // SENT -> ACKNOWLEDGED). Free the combined QUIC_BUFFER+payload
                // allocation only on a terminal state — SENT and LOST_SUSPECT
                // are intermediate and may be followed by another transition.
                auto state = ev->DATAGRAM_SEND_STATE_CHANGED.State;
                if (ev->DATAGRAM_SEND_STATE_CHANGED.ClientContext &&
                    (state == QUIC_DATAGRAM_SEND_LOST_DISCARDED
                     || state == QUIC_DATAGRAM_SEND_ACKNOWLEDGED
                     || state == QUIC_DATAGRAM_SEND_ACKNOWLEDGED_SPURIOUS
                     || state == QUIC_DATAGRAM_SEND_CANCELED)) {
                    ::operator delete(ev->DATAGRAM_SEND_STATE_CHANGED.ClientContext);
                }
                return QUIC_STATUS_SUCCESS;
            }
            default:
                return QUIC_STATUS_SUCCESS;
        }
    }
};

static HQUIC OpenClientConfiguration(const std::string& alpn, const QUICClientCredentials& creds) {
    std::vector<std::uint8_t> alpnBuf;
    QUIC_BUFFER alpnBuffer = MakeAlpn(alpn, alpnBuf);

    QUIC_SETTINGS settings{};
    settings.IsSet.IdleTimeoutMs = 1;
    settings.IdleTimeoutMs = creds.settings.idleTimeoutMs;
    settings.IsSet.HandshakeIdleTimeoutMs = 1;
    settings.HandshakeIdleTimeoutMs = creds.settings.handshakeIdleTimeoutMs;
    // Keep the connection alive across long idle gaps. msquic sends PING frames
    // on its own timer thread (independent of the app), so a request/response
    // connection survives even while the app is blocked for a long time between
    // requests (e.g. a client waiting on slow LLM inference). Interval < idle
    // timeout so the peer's idle timer never expires.
    settings.IsSet.KeepAliveIntervalMs = 1;
    settings.KeepAliveIntervalMs = creds.settings.keepAliveIntervalMs;
    settings.IsSet.DatagramReceiveEnabled = 1;
    settings.DatagramReceiveEnabled = creds.settings.datagramReceiveEnabled ? 1 : 0;
    // Allow the server to open unidi/bidi streams to us. msquic defaults
    // both peer-stream-count limits to 0; with that, the server's HTTP/3
    // control stream + QPACK encoder/decoder streams can't be created and
    // most h3 servers will close the connection after handshake. We don't
    // currently use server push (h3 pushes ride on unidi 0x01 streams) but
    // the bidi cap is harmless to grant.
    settings.IsSet.PeerUnidiStreamCount = 1;
    settings.PeerUnidiStreamCount = creds.settings.peerUnidiStreamCount;
    settings.IsSet.PeerBidiStreamCount = 1;
    settings.PeerBidiStreamCount = creds.settings.peerBidiStreamCount;

    HQUIC cfg = nullptr;
    QUIC_STATUS s = Runtime().api->ConfigurationOpen(Runtime().registration, &alpnBuffer, 1,
                                                     &settings, sizeof(settings), nullptr, &cfg);
    if (QUIC_FAILED(s)) throw QUICException(std::format("ConfigurationOpen failed: 0x{:x}", static_cast<unsigned>(s)));

    QUIC_CREDENTIAL_CONFIG cc{};
    cc.Type = QUIC_CREDENTIAL_TYPE_NONE;
    cc.Flags = QUIC_CREDENTIAL_FLAG_CLIENT;
    if (creds.insecureNoServerValidation) {
        cc.Flags |= QUIC_CREDENTIAL_FLAG_NO_CERTIFICATE_VALIDATION;
    }
    s = Runtime().api->ConfigurationLoadCredential(cfg, &cc);
    if (QUIC_FAILED(s)) {
        Runtime().api->ConfigurationClose(cfg);
        throw QUICException(std::format("ConfigurationLoadCredential failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    return cfg;
}

ClientQUIC::ClientQUIC(const char* host, std::uint16_t port, std::string alpnIn,
                       QUICClientCredentials creds)
    : alpn(std::move(alpnIn)), impl(std::make_unique<Impl>())
{
    impl->outer = this;
    impl->configuration = OpenClientConfiguration(alpn, creds);

    QUIC_STATUS s = Runtime().api->ConnectionOpen(Runtime().registration,
                                                  reinterpret_cast<QUIC_CONNECTION_CALLBACK_HANDLER>(&Impl::Callback),
                                                  impl.get(), &impl->connection);
    if (QUIC_FAILED(s)) {
        Runtime().api->ConfigurationClose(impl->configuration);
        throw QUICException(std::format("ConnectionOpen failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    s = Runtime().api->ConnectionStart(impl->connection, impl->configuration, QUIC_ADDRESS_FAMILY_UNSPEC, host, port);
    if (QUIC_FAILED(s)) {
        Runtime().api->ConnectionClose(impl->connection);
        Runtime().api->ConfigurationClose(impl->configuration);
        throw QUICException(std::format("ConnectionStart failed: 0x{:x}", static_cast<unsigned>(s)));
    }

    QUIC_STATUS handshakeStatus = QUIC_STATUS_SUCCESS;
    {
        std::unique_lock lk(impl->mtx);
        // Bounded by HandshakeIdleTimeoutMs above: an unreachable peer ends in
        // SHUTDOWN_INITIATED_BY_TRANSPORT rather than waiting here forever.
        impl->cv.wait(lk, [&]{ return impl->connected || impl->closed; });
        if (impl->connected) return;
        handshakeStatus = impl->shutdownStatus;
    }
    // Throwing from here means ~ClientQUIC never runs, so tear the msquic
    // handles down by hand — the connection callback holds `impl`, which is
    // about to be destroyed with the half-built object.
    impl->CloseConnection();
    Runtime().api->ConfigurationClose(impl->configuration);
    impl->configuration = nullptr;
    throw QUICException(std::format("QUIC handshake failed: 0x{:x}",
                                    static_cast<unsigned>(handshakeStatus)));
}

ClientQUIC::ClientQUIC(std::string host, std::uint16_t port, std::string alpnIn, QUICClientCredentials creds)
    : ClientQUIC(host.c_str(), port, std::move(alpnIn), std::move(creds)) {}

ClientQUIC::ClientQUIC(HQUIC connectionHandle, HQUIC serverConfiguration, std::string alpnIn)
    : alpn(std::move(alpnIn)), impl(std::make_unique<Impl>())
{
    impl->outer = this;
    impl->connection = connectionHandle;
    impl->configuration = serverConfiguration;
    impl->ownsConfiguration = false;
    impl->connected = true;
    Runtime().api->SetCallbackHandler(connectionHandle,
                                      reinterpret_cast<void*>(&Impl::Callback),
                                      impl.get());
}

ClientQUIC::ClientQUIC(ClientQUIC&& other) noexcept
    : alpn(std::move(other.alpn)), impl(std::move(other.impl))
{
    if (impl) impl->outer = this;
}

ClientQUIC::~ClientQUIC() {
    if (!impl) return;
    // Before the connection goes: streams the peer started that no OnStream
    // handler ever claimed. Destroying them here rather than with the rest of
    // `impl` keeps every stream teardown ahead of ConnectionClose, so the
    // drain below actually covers them.
    {
        std::deque<QUICStream> backlog;
        {
            std::lock_guard lk(impl->mtx);
            std::swap(backlog, impl->pendingStreams);
        }
    }
    impl->CloseConnection();
    if (impl->configuration && impl->ownsConfiguration) {
        Runtime().api->ConfigurationClose(impl->configuration);
        impl->configuration = nullptr;
    }
}

void ClientQUIC::Stop() {
    if (!impl) return;
    // Read the handle under the lock: the SHUTDOWN_COMPLETE callback may be
    // closing the connection concurrently on an msquic worker.
    HQUIC conn = nullptr;
    {
        std::lock_guard lk(impl->mtx);
        conn = impl->connection;
    }
    if (conn) Runtime().api->ConnectionShutdown(conn, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
}

QUICStream ClientQUIC::OpenStream(bool unidirectional) {
    HQUIC streamHandle = nullptr;
    QUICStream stream;
    stream.impl = std::make_shared<QUICStream::Impl>();
    stream.impl->connection = this;
    // Count the stream as open before StreamOpen registers the callback, so a
    // callback can never observe a half-initialised Impl. The two failure
    // paths below undo it.
    stream.impl->registry = impl->streams;
    stream.impl->registry->Add();
    stream.impl->selfRef = new std::shared_ptr<QUICStream::Impl>(stream.impl);
    QUIC_STREAM_OPEN_FLAGS openFlags = unidirectional
        ? QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL
        : QUIC_STREAM_OPEN_FLAG_NONE;
    QUIC_STATUS s = Runtime().api->StreamOpen(impl->connection, openFlags,
                                              reinterpret_cast<QUIC_STREAM_CALLBACK_HANDLER>(&QUICStream::Impl::Callback),
                                              stream.impl->selfRef, &streamHandle);
    if (QUIC_FAILED(s)) {
        delete stream.impl->selfRef;
        stream.impl->selfRef = nullptr;
        // No msquic stream exists, so there is no handle to release.
        std::shared_ptr<StreamRegistry> registry = std::move(stream.impl->registry);
        registry->Remove();
        throw QUICException(std::format("StreamOpen failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    stream.handle = streamHandle;
    stream.connection = this;
    stream.impl->SetHandle(streamHandle);
    if (unidirectional) {
        // We initiated the unidi stream: we send, peer reads.
        stream.canSend = true;
        stream.canReceive = false;
    }
    // SHUTDOWN_ON_FAIL, not NONE. StreamStart is asynchronous — it returns
    // QUIC_STATUS_PENDING and queues the start onto the connection — so it can
    // still fail later, and it does exactly that (QUIC_STATUS_INVALID_STATE)
    // whenever the connection is shut down before the queued start runs. That
    // is an everyday race for a client that sends periodically: opening a
    // stream just as the connection goes away. Without this flag msquic leaves
    // such a stream neither started nor shut down, so SHUTDOWN_COMPLETE never
    // arrives, the handle is never closed, and the registration's rundown at
    // exit() blocks forever waiting for it. With it, the failed start is
    // followed by SHUTDOWN_COMPLETE like any other stream.
    s = Runtime().api->StreamStart(streamHandle, QUIC_STREAM_START_FLAG_SHUTDOWN_ON_FAIL);
    if (QUIC_FAILED(s)) {
        // Synchronous failure: the stream never started, so no callback will
        // ever run. Release the callback's ref by hand, which closes the
        // handle, and clear the wrapper's so ~QUICStream is a no-op.
        stream.handle = nullptr;
        stream.impl->Retire();
        throw QUICException(std::format("StreamStart failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    return stream;
}

void ClientQUIC::SendDatagram(const void* buffer, std::uint32_t size) {
    // msquic stores the QUIC_BUFFER pointer (not a copy) on the send queue
    // and serialises async on a worker thread. Both the QUIC_BUFFER and the
    // payload it points at must outlive the call until DATAGRAM_SEND_STATE
    // reports a terminal state. Pack them together in a single allocation.
    auto* mem = static_cast<std::uint8_t*>(::operator new(sizeof(QUIC_BUFFER) + size));
    auto* hdr = reinterpret_cast<QUIC_BUFFER*>(mem);
    auto* payload = mem + sizeof(QUIC_BUFFER);
    std::memcpy(payload, buffer, size);
    hdr->Buffer = payload;
    hdr->Length = size;
    QUIC_STATUS s = Runtime().api->DatagramSend(impl->connection, hdr, 1,
                                                QUIC_SEND_FLAG_NONE, mem);
    if (QUIC_FAILED(s)) {
        ::operator delete(mem);
        throw QUICException(std::format("DatagramSend failed: 0x{:x}", static_cast<unsigned>(s)));
    }
}

void ClientQUIC::OnStream(std::function<void(QUICStream)> cb) {
    std::deque<QUICStream> backlog;
    {
        std::lock_guard lk(impl->mtx);
        impl->onStream = cb;
        std::swap(backlog, impl->pendingStreams);
    }
    while (!backlog.empty()) {
        auto* shared = new QUICStream(std::move(backlog.front()));
        backlog.pop_front();
        auto handler = cb;
        ThreadPool::Enqueue([handler, shared]{
            handler(std::move(*shared));
            delete shared;
        });
    }
}

void ClientQUIC::OnDatagram(std::function<void(std::vector<char>)> cb) {
    impl->onDatagram = std::move(cb);
}

std::vector<char> ClientQUIC::RecieveDatagramSync() {
    std::unique_lock lk(impl->mtx);
    impl->cv.wait(lk, [&]{ return !impl->datagramQueue.empty() || impl->closed; });
    if (!impl->datagramQueue.empty()) {
        auto out = std::move(impl->datagramQueue.front());
        impl->datagramQueue.pop_front();
        return out;
    }
    throw QUICClosedException();
}

HQUIC ClientQUIC::GetHandle() const { return impl ? impl->connection : nullptr; }

std::shared_ptr<StreamRegistry> QUICStream::Impl::RegistryOf(ClientQUIC* connection) {
    if (!connection || !connection->impl) return nullptr;
    return connection->impl->streams;
}