// Glaze Library
// For the license information refer to glaze.hpp

#pragma once

#include <atomic>
#include <chrono>
#include <concepts>
#include <cstdlib>
#include <expected>
#include <functional>
#include <future>
#include <glaze/glaze.hpp>
#include <iostream>
#include <limits>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <source_location>
#include <thread>
#include <unordered_map>
#include <vector>

#include "glaze/ext/glaze_asio.hpp"
#include "glaze/net/http_router.hpp"
#include "glaze/util/env.hpp"
#include "glaze/util/itoa.hpp"
#include "glaze/util/key_transformers.hpp"

#ifdef GLZ_ENABLE_SSL
#include <openssl/ssl.h> // For SSL_set_tlsext_host_name
#endif

namespace glz
{
   inline int strncasecmp(const char* s1, const char* s2, size_t n)
   {
      for (size_t i = 0; i < n; ++i) {
         unsigned char c1 = static_cast<unsigned char>(s1[i]);
         unsigned char c2 = static_cast<unsigned char>(s2[i]);
         if (c1 == '\0' || c2 == '\0') {
            return c1 - c2;
         }
         int diff = std::tolower(c1) - std::tolower(c2);
         if (diff != 0) {
            return diff;
         }
      }
      return 0;
   }

   // Streaming strategy options
   enum class stream_read_strategy {
      bulk_transfer, // Deliver larger chunks, better throughput (default)
      immediate_delivery // Deliver smaller chunks immediately, lower latency
   };

   struct url_parts
   {
      std::string protocol;
      std::string host;
      uint16_t port;
      std::string path;
   };

   // Socket type aliases for HTTP client
   using tcp_socket = asio::ip::tcp::socket;
#ifdef GLZ_ENABLE_SSL
   using ssl_socket = asio::ssl::stream<asio::ip::tcp::socket>;
   using socket_variant = std::variant<std::shared_ptr<tcp_socket>, std::shared_ptr<ssl_socket>>;
#else
   using socket_variant = std::variant<std::shared_ptr<tcp_socket>>;
#endif

   // SSL error codes for detailed error reporting
   enum class ssl_error {
      success = 0,
      ssl_not_supported, // HTTPS requested but SSL support not compiled in
      sni_hostname_failed // Failed to set SNI hostname (SSL_set_tlsext_host_name)
      // Note: Handshake and certificate errors propagate as native ASIO/OpenSSL error codes
      // for more detailed error information
   };

   // SSL error category for std::error_code integration
   class ssl_error_category : public std::error_category
   {
     public:
      const char* name() const noexcept override { return "glaze.ssl"; }

      std::string message(int ev) const override
      {
         switch (static_cast<ssl_error>(ev)) {
         case ssl_error::success:
            return "Success";
         case ssl_error::ssl_not_supported:
            return "SSL/TLS not supported: GLZ_ENABLE_SSL not defined";
         case ssl_error::sni_hostname_failed:
            return "Failed to set SNI hostname for TLS connection";
         default:
            return "Unknown SSL error";
         }
      }

      // Map to equivalent standard error conditions where applicable
      std::error_condition default_error_condition(int ev) const noexcept override
      {
         switch (static_cast<ssl_error>(ev)) {
         case ssl_error::ssl_not_supported:
            return std::errc::protocol_not_supported;
         case ssl_error::sni_hostname_failed:
            return std::errc::protocol_error;
         default:
            return std::error_condition(ev, *this);
         }
      }
   };

   // Get the singleton instance of the SSL error category
   inline const ssl_error_category& get_ssl_error_category() noexcept
   {
      static ssl_error_category instance;
      return instance;
   }

   // Create std::error_code from ssl_error
   inline std::error_code make_error_code(ssl_error e) noexcept
   {
      return {static_cast<int>(e), get_ssl_error_category()};
   }
   // HTTP client error codes
   enum class http_client_error {
      success = 0,
      response_too_large, // Response body exceeds max_response_body_size
   };

   // HTTP client error category for std::error_code integration
   class http_client_error_category : public std::error_category
   {
     public:
      const char* name() const noexcept override { return "glaze.http_client"; }

      std::string message(int ev) const override
      {
         switch (static_cast<http_client_error>(ev)) {
         case http_client_error::success:
            return "Success";
         case http_client_error::response_too_large:
            return "Response body size exceeds configured maximum";
         default:
            return "Unknown HTTP client error";
         }
      }
   };

   inline const http_client_error_category& get_http_client_error_category() noexcept
   {
      static http_client_error_category instance;
      return instance;
   }

   inline std::error_code make_error_code(http_client_error e) noexcept
   {
      return {static_cast<int>(e), get_http_client_error_category()};
   }
} // namespace glz

// Enable automatic conversion from glz::ssl_error to std::error_code
template <>
struct std::is_error_code_enum<glz::ssl_error> : std::true_type
{};

template <>
struct std::is_error_code_enum<glz::http_client_error> : std::true_type
{};

namespace glz
{
   namespace detail
   {
      // Helper to check if a socket variant is open
      inline bool socket_is_open(const socket_variant& v)
      {
         return std::visit(
            [](const auto& sock) {
               if (!sock) return false;
               if constexpr (requires { sock->next_layer(); }) {
                  return sock->next_layer().is_open();
               }
               else {
                  return sock->is_open();
               }
            },
            v);
      }

      // Active liveness check for a pooled TCP socket: peek 1 byte. Returns false if the
      // peer has already closed (FIN seen, RST surfaces as a recv error) or anything else
      // would make the socket unusable. Returns true only when the socket is healthy and
      // idle (no data pending), which is the only state where a kept-alive connection
      // should be reused.
      //
      // Caller contracts:
      //   1. Plain TCP sockets only. SSL sockets are NOT a fit here: a TCP-layer peek
      //      only sees ciphertext, OpenSSL may have already pulled the server's
      //      close_notify into its internal read buffer (so peek returns would_block
      //      even though the connection is dead), and any byte we'd see at the TCP
      //      layer is an encrypted record we can't decode. acquire() bypasses this
      //      function for SSL sockets and relies on timestamp eviction plus the
      //      transparent-retry path instead.
      //   2. A return value of false means "do not reuse this socket"; the caller is
      //      expected to discard (close) it. We make a best-effort attempt to restore
      //      the original blocking-mode flag before any false return, but if either
      //      toggle or restore fails the socket may be left in an unspecified blocking
      //      state, so retaining it for a sync read/write would be a footgun.
      inline bool tcp_socket_is_pooled_alive(asio::ip::tcp::socket& tcp_sock)
      {
         if (!tcp_sock.is_open()) return false;

         asio::error_code ec;
         const bool was_non_blocking = tcp_sock.non_blocking();
         tcp_sock.non_blocking(true, ec);
         if (ec) {
            // Toggle failed. The state may or may not have changed; attempt to restore
            // it so the caller's invariants are preserved if they decide not to close
            // the socket. The "discard on false" contract still applies.
            asio::error_code restore_ec;
            tcp_sock.non_blocking(was_non_blocking, restore_ec);
            return false;
         }

         char buf;
         const std::size_t n = tcp_sock.receive(asio::buffer(&buf, 1), asio::ip::tcp::socket::message_peek, ec);

         asio::error_code restore_ec;
         tcp_sock.non_blocking(was_non_blocking, restore_ec);
         if (restore_ec) {
            // Failed to restore blocking mode; the socket is now unusable for the sync
            // code paths. Caller must discard.
            return false;
         }

         if (ec == asio::error::would_block || ec == asio::error::try_again) {
            return true; // healthy idle: nothing pending
         }
         if (ec) {
            return false; // any other error means the socket is unusable
         }
         if (n == 0) {
            return false; // peer sent FIN
         }
         // Unexpected bytes on an idle pooled connection (most often a TLS close_notify
         // alert at the TCP layer, occasionally an unsolicited 408 Request Timeout from
         // servers that emit one before closing per RFC 7230 §6.5). Don't reuse.
         return false;
      }

      // Assemble the HTTP/1.1 request line + headers + body into a single contiguous
      // buffer once, so the async chain (and retry) can share it without re-copying the
      // body or re-iterating the headers map. For large idempotent PUT bodies this is
      // the difference between O(1) and O(N) body copies per request through the chain.
      inline std::string build_http_request_bytes(const std::string& method, const url_parts& url, bool use_https,
                                                  const std::string& body,
                                                  const std::unordered_map<std::string, std::string>& headers)
      {
         const bool is_default_port = (!use_https && url.port == 80) || (use_https && url.port == 443);

         std::string request_str;
         request_str.reserve(512 + body.size());
         request_str.append(method);
         request_str.append(" ");
         request_str.append(url.path);
         request_str.append(" HTTP/1.1\r\n");
         request_str.append("Host: ");
         request_str.append(url.host);
         if (!is_default_port) {
            char
               port_buf[8]; // a uint16_t port is at most 5 digits; pad so the sizing does not depend on itoa internals
            auto* end = glz::to_chars(port_buf, url.port);
            request_str.push_back(':');
            request_str.append(port_buf, static_cast<size_t>(end - port_buf));
         }
         request_str.append("\r\n");
         request_str.append("Connection: keep-alive\r\n");
         if (!body.empty()) {
            request_str.append("Content-Length: ");
            request_str.append(std::to_string(body.size()));
            request_str.append("\r\n");
         }
         for (const auto& [name, value] : headers) {
            if (header_field_has_crlf(name, value)) [[unlikely]] {
               continue;
            }
            request_str.append(name);
            request_str.append(": ");
            request_str.append(value);
            request_str.append("\r\n");
         }
         request_str.append("\r\n");
         request_str.append(body);
         return request_str;
      }

#ifdef GLZ_ENABLE_SSL
      // Configure SNI and hostname verification for client TLS connections.
      inline bool configure_ssl_client_hostname(ssl_socket& sock, const std::string& host)
      {
         if (!SSL_set_tlsext_host_name(sock.native_handle(), host.c_str())) {
            return false;
         }

         sock.set_verify_callback(asio::ssl::host_name_verification(host));
         return true;
      }

      enum class ssl_ca_source { explicit_file, env_ssl_cert_file, env_ssl_cert_dir, default_verify_paths };

      inline std::optional<std::string_view> non_empty_path(std::optional<std::string_view> value)
      {
         if (value && !value->empty()) {
            return value;
         }
         return std::nullopt;
      }

      inline std::optional<std::string> env_path(const char* name) { return getenv_nonempty(name); }

      inline std::optional<std::string_view> to_sv_opt(const std::optional<std::string>& value)
      {
         if (value && !value->empty()) {
            return std::string_view{*value};
         }
         return std::nullopt;
      }

      std::optional<std::string_view> to_sv_opt(std::optional<std::string>&&) = delete;

      template <typename Loader>
      concept ssl_ca_path_loader = requires(Loader&& loader, std::string_view path) {
         { std::forward<Loader>(loader)(path) } -> std::convertible_to<std::error_code>;
      };

      template <typename Loader>
      concept ssl_ca_default_loader = requires(Loader&& loader) {
         { std::forward<Loader>(loader)() } -> std::convertible_to<std::error_code>;
      };

      template <ssl_ca_path_loader LoadFile, ssl_ca_path_loader LoadDir, ssl_ca_default_loader LoadDefault>
      inline std::expected<ssl_ca_source, std::error_code> configure_ssl_ca_fallback(
         std::optional<std::string_view> explicit_file, std::optional<std::string_view> env_cert_file,
         std::optional<std::string_view> env_cert_dir, LoadFile&& load_file, LoadDir&& load_dir,
         LoadDefault&& load_default)
      {
         std::optional<std::error_code> last_error{};

         const auto try_file = [&](std::optional<std::string_view> path,
                                   ssl_ca_source source) -> std::optional<ssl_ca_source> {
            if (auto non_empty = non_empty_path(path)) {
               if (const std::error_code ec = load_file(*non_empty); ec) {
                  last_error = ec;
               }
               else {
                  return source;
               }
            }
            return std::nullopt;
         };

         if (auto source = try_file(explicit_file, ssl_ca_source::explicit_file)) {
            return *source;
         }

         if (auto source = try_file(env_cert_file, ssl_ca_source::env_ssl_cert_file)) {
            return *source;
         }

         if (auto non_empty = non_empty_path(env_cert_dir)) {
            if (const std::error_code ec = load_dir(*non_empty); ec) {
               last_error = ec;
            }
            else {
               return ssl_ca_source::env_ssl_cert_dir;
            }
         }

         if (const std::error_code ec = load_default(); ec) {
            if (last_error) {
               return std::unexpected(*last_error);
            }
            return std::unexpected(ec);
         }

         return ssl_ca_source::default_verify_paths;
      }
#endif

      // Helper to close a socket variant with optional SSL shutdown
      // graceful_shutdown: if true, performs SSL shutdown before closing (recommended for proper TLS termination)
      inline void close_socket(socket_variant& v, bool graceful_shutdown = true)
      {
         std::visit(
            [graceful_shutdown](auto& sock) {
               if (!sock) return;
               asio::error_code ec;
               if constexpr (requires { sock->next_layer(); }) {
                  // SSL socket - perform graceful shutdown if requested
                  if (graceful_shutdown && sock->next_layer().is_open()) {
                     // Perform SSL shutdown to send close_notify alert
                     // This is best-effort - we ignore errors since we're closing anyway
                     sock->shutdown(ec);
                     // Note: We don't wait for the peer's close_notify response
                     // as that could block indefinitely. The TCP close will
                     // terminate the connection cleanly.
                  }
                  sock->next_layer().close(ec);
               }
               else {
                  sock->close(ec);
               }
               (void)ec; // Errors during close are intentionally ignored
            },
            v);
      }
   }

   inline std::expected<url_parts, std::error_code> parse_url(std::string_view url)
   {
      // Check minimum length
      if (url.size() < 8) { // Minimum for "http://" + 1 char host
         return std::unexpected(std::make_error_code(std::errc::invalid_argument));
      }

      // Check for control characters to prevent CRLF injection
      for (unsigned char c : url) {
         if (c < 32 || c == 127) {
            return std::unexpected(std::make_error_code(std::errc::invalid_argument));
         }
      }

      // Find protocol
      size_t protocol_end = url.find("://");
      if (protocol_end == std::string_view::npos) {
         return std::unexpected(std::make_error_code(std::errc::invalid_argument));
      }

      std::string protocol(url.substr(0, protocol_end));
      if (protocol != "http" && protocol != "https" && protocol != "ws" && protocol != "wss") {
         return std::unexpected(std::make_error_code(std::errc::invalid_argument));
      }

      // Process host, port and path
      size_t host_start = protocol_end + 3;
      if (host_start >= url.size()) {
         return std::unexpected(std::make_error_code(std::errc::invalid_argument));
      }

      size_t host_end = url.find_first_of("/:", host_start);
      std::string host;
      std::string port_str;
      std::string path = "/";

      if (host_end == std::string_view::npos) {
         host = std::string(url.substr(host_start));
      }
      else if (url[host_end] == ':') {
         host = std::string(url.substr(host_start, host_end - host_start));
         size_t port_start = host_end + 1;
         size_t port_end = url.find('/', port_start);

         if (port_end == std::string_view::npos) {
            port_str = std::string(url.substr(port_start));
         }
         else {
            port_str = std::string(url.substr(port_start, port_end - port_start));
            path = std::string(url.substr(port_end));
         }

         if (!port_str.empty() &&
             !std::all_of(port_str.begin(), port_str.end(), [](unsigned char c) { return std::isdigit(c); })) {
            return std::unexpected(std::make_error_code(std::errc::invalid_argument));
         }
      }
      else if (url[host_end] == '/') {
         host = std::string(url.substr(host_start, host_end - host_start));
         path = std::string(url.substr(host_end));
      }

      if (host.empty() || host.find_first_of(":/") != std::string::npos) {
         return std::unexpected(std::make_error_code(std::errc::invalid_argument));
      }

      uint16_t port = 0;
      if (port_str.empty()) {
         port = (protocol == "https" || protocol == "wss") ? 443 : 80;
      }
      else {
         try {
            long port_long = std::stol(port_str);
            if (port_long <= 0 || port_long > 65535) {
               return std::unexpected(std::make_error_code(std::errc::invalid_argument));
            }
            port = static_cast<uint16_t>(port_long);
         }
         catch (const std::exception&) {
            return std::unexpected(std::make_error_code(std::errc::invalid_argument));
         }
      }

      return url_parts{std::move(protocol), std::move(host), port, std::move(path)};
   }

   // Idempotent HTTP methods are safe to retry transparently when a pooled connection
   // turns out to be stale. POST/PATCH are intentionally excluded by default because
   // a write that succeeded locally may have already been processed by the server.
   inline bool is_idempotent_method(std::string_view method) noexcept
   {
      return method == "GET" || method == "HEAD" || method == "OPTIONS" || method == "PUT" || method == "DELETE" ||
             method == "TRACE";
   }

   // Errors that almost always indicate "the connection failed before the server could
   // process the request" rather than a genuine application failure. When the request is
   // idempotent and no response bytes have been received yet, these warrant exactly one
   // transparent retry on a fresh connection. connection_aborted covers ECONNABORTED on
   // macOS writes to a half-closed peer.
   //
   // Implementation note: the make_error_code() calls force conversion to std::error_code
   // before comparison. Without this, the asio::error::* enums resolve to
   // boost::asio::error::misc_errors when glaze is built against boost.asio, and there is
   // no operator== between std::error_code and a boost-side enum. By materializing both
   // sides as std::error_code, the comparison works under standalone asio (where
   // std::error_code == asio::error_code) and boost.asio (via boost.system's implicit
   // conversion to std::error_code) without #ifdef branches.
   inline bool is_retriable_pool_error(std::error_code ec) noexcept
   {
      static const std::error_code eof_ec = make_error_code(asio::error::eof);
      static const std::error_code reset_ec = make_error_code(asio::error::connection_reset);
      static const std::error_code pipe_ec = make_error_code(asio::error::broken_pipe);
      static const std::error_code not_connected_ec = make_error_code(asio::error::not_connected);
      static const std::error_code shutdown_ec = make_error_code(asio::error::shut_down);
      static const std::error_code aborted_ec = make_error_code(asio::error::connection_aborted);
      return ec == eof_ec || ec == reset_ec || ec == pipe_ec || ec == not_connected_ec || ec == shutdown_ec ||
             ec == aborted_ec;
   }

   // Per-attempt state for the async request chain. response_started is set to true the
   // moment we read the first bytes of the response (the header read completes); after
   // that point, a subsequent failure means the server already processed the request and
   // we MUST NOT retry, even on an idempotent method, even if the error code looks
   // retriable. This is the precise gate for safe transparent retry.
   //
   // shared_ptr because the chain can fan out across async callbacks across worker
   // threads; atomic because the read-side (set in read_response's success callback) and
   // the retry handler's check (in the user-handler dispatch) are in distinct callbacks.
   struct async_attempt_state
   {
      std::atomic<bool> response_started{false};
   };

   // Synchronous variant of async_attempt_state's signal. Returned alongside the
   // request outcome so perform_sync_request can decide whether retry is safe.
   struct sync_attempt_outcome
   {
      std::expected<response, std::error_code> outcome;
      bool response_started = false;
   };

   // HTTP connection pool for reusing sockets
   struct http_connection_pool
   {
      struct connection_key
      {
         std::string host;
         uint16_t port;
         bool is_https;

         bool operator==(const connection_key& other) const
         {
            return host == other.host && port == other.port && is_https == other.is_https;
         }
      };

      struct connection_key_hash
      {
         size_t operator()(const connection_key& key) const
         {
            return std::hash<std::string>{}(key.host) ^ (std::hash<uint16_t>{}(key.port) << 1) ^
                   (std::hash<bool>{}(key.is_https) << 2);
         }
      };

      // A pooled connection plus the time it was returned to the pool. The timestamp is
      // used to evict entries older than idle_timeout_ before they're handed back out.
      // This is the primary defense against stale-on-arrival connections (e.g. uvicorn
      // closing kept-alive connections after its 5s timeout).
      struct pool_entry
      {
         socket_variant socket;
         std::chrono::steady_clock::time_point returned_at;
      };

      // Result of acquiring a connection. from_pool=true means the socket was reused
      // from the pool, false means it was just created. This is no longer the gate for
      // retry decisions (the chain now uses the precise "no response bytes received
      // yet" signal via async_attempt_state / sync_attempt_outcome), but is retained as
      // an informational signal callers may use for diagnostics or stats.
      struct [[nodiscard]] acquire_result
      {
         socket_variant socket;
         bool from_pool;
      };

      mutable std::mutex pool_mtx; // Protects available_connections
      std::unordered_map<connection_key, std::vector<pool_entry>, connection_key_hash> available_connections;
      asio::any_io_executor io_executor;
      std::atomic<bool> graceful_ssl_shutdown_{true}; // Whether to perform SSL shutdown when closing connections
      // Default to 4 seconds: just under uvicorn's 5s default keep-alive timeout. Any
      // entry idle longer than this is evicted on next acquire().
      //
      // Portability note on the atomic: steady_clock::duration is typically a 64-bit
      // integer, and std::atomic<duration> is lock-free on 64-bit platforms and on
      // most modern 32-bit ones (ARMv7+ via ldrexd/strexd, x86 32-bit via cmpxchg8b).
      // On older 32-bit ARM or restricted embedded targets it may fall back to a
      // library-provided lock; behavior stays correct, latency on this load gets
      // worse. The same acquire() path takes pool_mtx for each candidate pop, so this
      // load is dominated by the mutex on any contention level worth worrying about.
      // Users on a lock-fallback platform can verify via
      // std::atomic<std::chrono::steady_clock::duration>::is_always_lock_free.
      std::atomic<std::chrono::steady_clock::duration> idle_timeout_{std::chrono::seconds(4)};
      std::atomic<size_t> max_per_host_{10};
      // Whether to actively peek a pooled TCP socket on acquire. When true (default), a
      // dead-on-arrival socket is detected before any request bytes go on the wire.
      // Disabling this is mainly useful for tests that want to deterministically exercise
      // the transparent-retry path, but can also trade a syscall per acquire for slightly
      // higher latency to handle stale connections (which then surface as a request-time
      // error and trigger the retry path on idempotent methods).
      std::atomic<bool> active_liveness_check_{true};
#ifdef GLZ_ENABLE_SSL
      mutable std::shared_mutex ssl_mtx; // Protects ssl_context - shared for reads, exclusive for writes
      std::shared_ptr<asio::ssl::context> ssl_context;
#endif

      http_connection_pool(asio::any_io_executor executor) : io_executor(executor)
      {
#ifdef GLZ_ENABLE_SSL
         // Use tls_client to allow negotiation of TLS 1.2/1.3 (highest mutually supported version)
         // This automatically disables insecure protocols (SSLv3, TLS 1.0, TLS 1.1)
         ssl_context = std::make_shared<asio::ssl::context>(asio::ssl::context::tls_client);
         // Best effort: if default trust paths are not available on this platform,
         // callers can still configure explicit trust roots later.
         asio::error_code ec;
         ssl_context->set_default_verify_paths(ec);
         ssl_context->set_verify_mode(asio::ssl::verify_peer);
#endif
      }

      socket_variant create_fresh_connection(bool is_https)
      {
         if (is_https) {
#ifdef GLZ_ENABLE_SSL
            // Hold shared lock while creating SSL socket to prevent concurrent context modification
            std::shared_lock<std::shared_mutex> ssl_lock(ssl_mtx);
            return socket_variant{std::make_shared<ssl_socket>(io_executor, *ssl_context)};
#else
            // Return TCP socket if SSL not available (will fail at handshake check)
            return socket_variant{std::make_shared<tcp_socket>(io_executor)};
#endif
         }
         return socket_variant{std::make_shared<tcp_socket>(io_executor)};
      }

      // Acquire a connection. Tries to reuse a non-expired, alive pooled entry first
      // (LIFO: most recently returned is tried first); falls back to creating a fresh
      // socket. The from_pool flag tells the caller whether a failed request can be
      // safely retried on a fresh connection.
      //
      // Lock discipline: pool_mtx is held only long enough to pop one candidate, then
      // released before any potentially blocking work (peek, close_socket -> SSL
      // shutdown). This is essential under concurrency: otherwise every acquire/return
      // would serialize behind the slowest socket teardown.
      acquire_result acquire(const std::string& host, uint16_t port, bool is_https)
      {
         connection_key key{host, port, is_https};
         const auto now = std::chrono::steady_clock::now();
         const auto idle_timeout = idle_timeout_.load();
         const bool active_check = active_liveness_check_.load();
         const bool graceful = graceful_ssl_shutdown_.load();

         while (true) {
            std::optional<pool_entry> entry;
            {
               std::lock_guard<std::mutex> lock(pool_mtx);
               auto it = available_connections.find(key);
               if (it != available_connections.end() && !it->second.empty()) {
                  entry.emplace(std::move(it->second.back()));
                  it->second.pop_back();
               }
            }
            if (!entry) break;

            // Timestamp eviction: anything older than the idle timeout is presumed dead;
            // skip the peek (which can race with FIN arriving) and just close it.
            if (idle_timeout.count() > 0 && now - entry->returned_at > idle_timeout) {
               detail::close_socket(entry->socket, graceful);
               continue;
            }

            if (!detail::socket_is_open(entry->socket)) {
               detail::close_socket(entry->socket, graceful);
               continue;
            }

            // Active peek check applies only to plain TCP sockets. For SSL, peeking the
            // TCP layer would only see ciphertext (see tcp_socket_is_pooled_alive
            // contract); SSL pooled sockets rely on timestamp eviction and the
            // transparent-retry path to handle stale connections.
            if (active_check) {
               const bool alive = std::visit(
                  [](auto& sock) -> bool {
                     if constexpr (requires { sock->next_layer(); }) {
                        return sock->next_layer().is_open();
                     }
                     else {
                        return detail::tcp_socket_is_pooled_alive(*sock);
                     }
                  },
                  entry->socket);
               if (!alive) {
                  detail::close_socket(entry->socket, graceful);
                  continue;
               }
            }

            return acquire_result{std::move(entry->socket), true};
         }

         return acquire_result{create_fresh_connection(is_https), false};
      }

      // Backwards-compatible wrapper: returns just the socket and discards the from_pool
      // flag. New internal code should call acquire() and use the from_pool flag to drive
      // retry behavior.
      socket_variant get_connection(const std::string& host, uint16_t port, bool is_https)
      {
         return acquire(host, port, is_https).socket;
      }

      void return_connection(const std::string& host, uint16_t port, bool is_https, socket_variant socket)
      {
         if (!detail::socket_is_open(socket)) {
            return;
         }

         connection_key key{host, port, is_https};
         std::lock_guard<std::mutex> lock(pool_mtx);

         auto& connections = available_connections[key];
         if (connections.size() < max_per_host_.load()) {
            connections.push_back(pool_entry{std::move(socket), std::chrono::steady_clock::now()});
         }
         // else: Pool is full - socket destructor will close it when it goes out of scope
      }

      // Configure how long a pooled connection may sit idle before it is evicted on next
      // acquire. Default is 4s, just under uvicorn's 5s keep-alive timeout. Set to zero
      // to disable timestamp-based eviction (peek-based liveness check still applies).
      void set_idle_timeout(std::chrono::steady_clock::duration timeout) { idle_timeout_.store(timeout); }
      std::chrono::steady_clock::duration idle_timeout() const { return idle_timeout_.load(); }

      // Maximum number of idle connections kept per host. Excess connections are closed
      // when returned. Default is 10.
      void set_max_connections_per_host(size_t n) { max_per_host_.store(n); }
      size_t max_connections_per_host() const { return max_per_host_.load(); }

      // Whether to actively peek pooled TCP sockets on acquire to detect server-side
      // close before sending a request. Default is true. Disabling forces stale
      // connections to surface as request-time errors and (for idempotent methods)
      // exercises the transparent-retry path. SSL sockets are unaffected: the active
      // check is never applied to them.
      void set_active_liveness_check(bool enabled) { active_liveness_check_.store(enabled); }
      bool active_liveness_check() const { return active_liveness_check_.load(); }

      // Drop and close every pooled connection. Useful in tests and when a host has
      // signalled (e.g. via 503) that all current sessions should be abandoned.
      void clear()
      {
         std::unordered_map<connection_key, std::vector<pool_entry>, connection_key_hash> drained;
         {
            std::lock_guard<std::mutex> lock(pool_mtx);
            drained.swap(available_connections);
         }
         const bool graceful = graceful_ssl_shutdown_.load();
         for (auto& [_, bucket] : drained) {
            for (auto& entry : bucket) {
               detail::close_socket(entry.socket, graceful);
            }
         }
      }

      // Set whether to perform graceful SSL shutdown when closing connections
      // Disabling this can improve performance but may cause TLS session issues
      void set_graceful_ssl_shutdown(bool enabled) { graceful_ssl_shutdown_.store(enabled); }

      bool graceful_ssl_shutdown() const { return graceful_ssl_shutdown_.load(); }

#ifdef GLZ_ENABLE_SSL
      // Thread-safe SSL context configuration
      // Takes exclusive lock to prevent concurrent socket creation during modification
      // WARNING: Using asio::ssl::verify_none disables certificate verification,
      // making the connection vulnerable to man-in-the-middle attacks.
      // Only use verify_none for testing with self-signed certificates.
      void set_ssl_verify_mode(asio::ssl::verify_mode mode)
      {
         std::unique_lock<std::shared_mutex> lock(ssl_mtx);
         ssl_context->set_verify_mode(mode);
      }

      // Thread-safe access to SSL context for advanced configuration
      // The provided callable is executed while holding an exclusive lock
      // Example: pool.configure_ssl_context([](auto& ctx) { ctx.set_options(...); });
      template <typename Func>
      void configure_ssl_context(Func&& func)
      {
         std::unique_lock<std::shared_mutex> lock(ssl_mtx);
         func(*ssl_context);
      }

      // Configure CA trust roots for server certificate verification.
      // Fallback order:
      // 1) explicit cert bundle path (if provided)
      // 2) SSL_CERT_FILE environment variable
      // 3) SSL_CERT_DIR environment variable
      // 4) OpenSSL default verify paths
      std::expected<void, std::error_code> configure_system_ca_certificates(
         std::optional<std::string_view> cert_bundle_file = std::nullopt)
      {
         std::unique_lock<std::shared_mutex> lock(ssl_mtx);

         const auto env_cert_file = detail::env_path("SSL_CERT_FILE");
         const auto env_cert_dir = detail::env_path("SSL_CERT_DIR");

         auto result = detail::configure_ssl_ca_fallback(
            cert_bundle_file, detail::to_sv_opt(env_cert_file), detail::to_sv_opt(env_cert_dir),
            [&](std::string_view path) -> std::error_code {
               asio::error_code ec;
               ssl_context->load_verify_file(std::string(path), ec);
               return ec;
            },
            [&](std::string_view path) -> std::error_code {
               asio::error_code ec;
               ssl_context->add_verify_path(std::string(path), ec);
               return ec;
            },
            [&]() -> std::error_code {
               asio::error_code ec;
               ssl_context->set_default_verify_paths(ec);
               return ec;
            });

         if (!result) {
            return std::unexpected(result.error());
         }

         ssl_context->set_verify_mode(asio::ssl::verify_peer);
         return {};
      }

      // Direct access to SSL context (NOT thread-safe)
      // WARNING: Use configure_ssl_context() for thread-safe modifications
      // This is provided for advanced use cases where the caller manages synchronization
      asio::ssl::context& ssl_context_unsafe() { return *ssl_context; }
#endif
   };

   // Forward declaration for streaming connection
   struct http_stream_connection;

   // Handler function types for streaming
   using http_data_handler = std::function<void(std::string_view data)>;
   using http_error_handler =
      std::function<void(std::error_code ec)>; // May carry HTTP statuses via http_status_category()
   using http_connect_handler = std::function<void(const response& headers)>;
   using http_disconnect_handler = std::function<void()>;

   // Streaming HTTP connection handle
   struct http_stream_connection
   {
      std::shared_ptr<socket_variant> socket;
      std::shared_ptr<asio::steady_timer> timer;
      std::shared_ptr<asio::streambuf> buffer; // Use unified streambuf for all reads
      bool is_connected{false};
      std::atomic<bool> should_stop{false};
      stream_read_strategy strategy{stream_read_strategy::bulk_transfer}; // Default strategy
      std::function<bool(int)> status_is_error{}; // Evaluated before treating status as failure
      bool is_https{false}; // Track if this is an HTTPS connection

      // Constructor with optional buffer size limit and strategy
      http_stream_connection(size_t max_buffer_size = 1024 * 1024,
                             stream_read_strategy read_strategy = stream_read_strategy::bulk_transfer)
         : buffer(std::make_shared<asio::streambuf>(max_buffer_size)), strategy(read_strategy)
      {}

      // User-facing disconnect. Signals the internal loops to stop.
      // The actual socket closing/pooling is handled by the internal disconnect handler.
      void disconnect()
      {
         bool expected = false;
         if (should_stop.compare_exchange_strong(expected, true)) {
            if (socket && detail::socket_is_open(*socket)) {
               asio::error_code ec;
               // This cancels pending async operations on the socket, triggering their handlers
               // with asio::error::operation_aborted.
               std::visit(
                  [&](auto& sock) {
                     if constexpr (requires { sock->next_layer(); }) {
                        sock->next_layer().cancel(ec);
                     }
                     else {
                        sock->cancel(ec);
                     }
                  },
                  *socket);
            }
            if (timer) {
               timer->cancel();
            }
         }
      }

      ~http_stream_connection() { disconnect(); }
   };
   // deprecated, use stream_request_params_v2 instead
   struct stream_request_params
   {
      std::string url;
      http_data_handler on_data;
      http_error_handler on_error;
      std::string method{"GET"};
      std::string body{};
      std::unordered_map<std::string, std::string> headers{};
      http_connect_handler on_connect{};
      http_disconnect_handler on_disconnect{};
      std::chrono::seconds timeout{std::chrono::seconds{30}};
      stream_read_strategy strategy{stream_read_strategy::bulk_transfer};
      std::function<bool(int)> status_is_error{[](int status) { return status >= 400; }};
      size_t max_buffer_size{1024 * 1024};
   };

   // Stream request parameters struct
   struct stream_request_params_v2
   {
      std::string method{"GET"};
      std::string url;
      std::chrono::seconds timeout{std::chrono::seconds{30}};
      stream_read_strategy strategy{stream_read_strategy::bulk_transfer};
      size_t max_buffer_size{1024 * 1024};
      std::string body;
      std::unordered_map<std::string, std::string> headers;
      http_connect_handler on_connect;
      http_disconnect_handler on_disconnect;
      http_data_handler on_data;
      http_error_handler on_error;
      std::function<bool(int)> status_is_error{[](int status) { return status >= 400; }};
   };

   struct http_client
   {
      /**
       * @brief Construct an HTTP client
       *
       * @param executor Optional external asio executor to use. If using a
       *                 default-constructed asio::any_io_executor() without target
       *                 executor, it creates its own io_context.  Using an external
       *                 executor allows sharing the io_context and its event loop with
       *                 other asio-based components or managing the io_context lifecycle
       *                 externally.
       *
       * @note When using an external executor, no worker threads are created by start_workers().
       *       The caller is responsible for running the executor (e.g., calling io_context::run()
       *       from their own threads) and to maintain its lifecycle.
       *
       *
       * Example with internal io_context (handler is called from a different thread created by http_client):
       * @code
       *   glz::http_client client{};
       *   client.get_async("http://example.com", {},
       * 				   [](std::expected<glz::response, std::error_code> resp){
       * 				     if (resp.has_value()) {
       * 				       std::cout << resp->response_body << std::endl;
       * 				     } else {
       *				          std::cerr << "Error: " << resp.error().message() << std::endl;
       *				        }
       *				   });
       * @endcode
       *
       * Example with external provided io_executor (handler runs in same thread which is calling io_ctx.run()).
       * io_context can be shared for other async operations:
       * @code
       *   // Create a shared io_context
       *   asio::io_context io_ctx;
       *   glz::http_client client{io_ctx.get_executor()};
       *   client.get_async("http://example.com", {},
       * 				   [](std::expected<glz::response, std::error_code> resp){
       * 				     if (resp.has_value()) {
       * 				       std::cout << resp->response_body << std::endl;
       * 				     } else {
       *				          std::cerr << "Error: " << resp.error().message() << std::endl;
       *				        }
       *				   });
       *   io_ctx.run();
       * @endcode
       */
      http_client(asio::any_io_executor executor = asio::any_io_executor())
         : async_io_context(executor ? std::shared_ptr<asio::io_context>() : std::make_shared<asio::io_context>()),
           io_executor(executor ? executor : async_io_context->get_executor()),
           connection_pool(std::make_shared<http_connection_pool>(io_executor))
      {
         if (async_io_context) {
            work_guard.emplace(asio::make_work_guard(*async_io_context));
         }
         start_workers();
      }

      ~http_client() { stop_workers(); }

#ifdef GLZ_ENABLE_SSL
      // Set SSL certificate verification mode (thread-safe)
      // WARNING: Using asio::ssl::verify_none disables certificate verification,
      // making the connection vulnerable to man-in-the-middle attacks.
      // Only use verify_none for testing with self-signed certificates.
      void set_ssl_verify_mode(asio::ssl::verify_mode mode) { connection_pool->set_ssl_verify_mode(mode); }

      // Thread-safe SSL context configuration
      // The provided callable is executed while holding an exclusive lock
      // Example: client.configure_ssl_context([](auto& ctx) { ctx.set_options(...); });
      template <typename Func>
      void configure_ssl_context(Func&& func)
      {
         connection_pool->configure_ssl_context(std::forward<Func>(func));
      }

      // Configure CA trust roots with explicit/env/default fallback order.
      std::expected<void, std::error_code> configure_system_ca_certificates(
         std::optional<std::string_view> cert_bundle_file = std::nullopt)
      {
         return connection_pool->configure_system_ca_certificates(cert_bundle_file);
      }

      // Direct access to SSL context (NOT thread-safe)
      // WARNING: Use configure_ssl_context() for thread-safe modifications
      // This is provided for advanced use cases where the caller manages synchronization
      asio::ssl::context& ssl_context_unsafe() { return connection_pool->ssl_context_unsafe(); }
#endif

      // Set whether to perform graceful SSL shutdown when closing connections
      // When enabled (default), sends TLS close_notify alert before closing
      // Disabling can improve performance but may cause:
      // - Peer to interpret closure as truncation attack
      // - TLS session resumption issues
      void set_graceful_ssl_shutdown(bool enabled) { connection_pool->set_graceful_ssl_shutdown(enabled); }

      // Check if graceful SSL shutdown is enabled
      bool graceful_ssl_shutdown() const { return connection_pool->graceful_ssl_shutdown(); }

      // How long an idle pooled connection may sit before it is evicted on the next request.
      // Default is 4 seconds, chosen to be just under uvicorn's 5s default keep-alive timeout.
      // Tune this to slightly less than the server's keep-alive idle timeout. Set to zero
      // to disable timestamp-based eviction (the active peek check still discards dead
      // connections, but is a smaller window).
      void set_pool_idle_timeout(std::chrono::steady_clock::duration timeout)
      {
         connection_pool->set_idle_timeout(timeout);
      }
      std::chrono::steady_clock::duration pool_idle_timeout() const { return connection_pool->idle_timeout(); }

      // Maximum number of idle connections kept per host. Default is 10, in line with
      // urllib3/requests and similar to OkHttp's 5 and Java HttpClient's 6 (per the
      // HTTP/1.1 RFC recommendation of 6 per origin).
      //
      // Performance note: returns above the cap are closed rather than pooled. For a
      // workload whose peak per-host concurrency is K, every request beyond the K-th
      // active connection pays the full connect + TLS handshake cost on each request.
      // For sustained concurrency above the default, raise this to at least your
      // observed steady-state per-host concurrency.
      void set_pool_max_connections_per_host(size_t n) { connection_pool->set_max_connections_per_host(n); }
      size_t pool_max_connections_per_host() const { return connection_pool->max_connections_per_host(); }

      // Whether to actively peek pooled TCP sockets on acquire (default true). Disabling
      // skips the syscall and forces stale connections to surface as request errors;
      // idempotent methods will then exercise the transparent retry path. SSL sockets
      // are unaffected (peek is never applied to ciphertext).
      void set_pool_active_liveness_check(bool enabled) { connection_pool->set_active_liveness_check(enabled); }
      bool pool_active_liveness_check() const { return connection_pool->active_liveness_check(); }

      // Drop and close every pooled connection. Useful in tests and when a host has
      // signalled (e.g. via 503) that all current sessions should be abandoned.
      void clear_connection_pool() { connection_pool->clear(); }

      // Synchronous GET request - truly synchronous, no promises/futures
      std::expected<response, std::error_code> get(std::string_view url,
                                                   const std::unordered_map<std::string, std::string>& headers = {})
      {
         auto url_result = parse_url(url);
         if (!url_result) {
            return std::unexpected(url_result.error());
         }

         return perform_sync_request("GET", *url_result, "", headers);
      }

      // Synchronous POST request - truly synchronous, no promises/futures
      std::expected<response, std::error_code> post(std::string_view url, const std::string& body,
                                                    const std::unordered_map<std::string, std::string>& headers = {})
      {
         auto url_result = parse_url(url);
         if (!url_result) {
            return std::unexpected(url_result.error());
         }

         return perform_sync_request("POST", *url_result, body, headers);
      }

      // Synchronous PUT request - truly synchronous, no promises/futures
      std::expected<response, std::error_code> put(std::string_view url, const std::string& body,
                                                   const std::unordered_map<std::string, std::string>& headers = {})
      {
         auto url_result = parse_url(url);
         if (!url_result) {
            return std::unexpected(url_result.error());
         }

         return perform_sync_request("PUT", *url_result, body, headers);
      }

      // Synchronous JSON POST request
      template <class T>
      std::expected<response, std::error_code> post_json(
         std::string_view url, const T& data, const std::unordered_map<std::string, std::string>& headers = {})
      {
         std::string json_str;
         auto ec = glz::write_json(data, json_str);
         if (ec) {
            return std::unexpected(std::make_error_code(std::errc::invalid_argument));
         }

         auto merged_headers = headers;
         merged_headers["content-type"] = "application/json";

         return post(url, json_str, merged_headers);
      }

      // Synchronous JSON PUT request
      template <class T>
      std::expected<response, std::error_code> put_json(
         std::string_view url, const T& data, const std::unordered_map<std::string, std::string>& headers = {})
      {
         std::string json_str;
         auto ec = glz::write_json(data, json_str);
         if (ec) {
            return std::unexpected(std::make_error_code(std::errc::invalid_argument));
         }

         auto merged_headers = headers;
         merged_headers["content-type"] = "application/json";

         return put(url, json_str, merged_headers);
      }

      [[deprecated("use stream_request_v2 instead")]]
      std::shared_ptr<http_stream_connection> stream_request(const stream_request_params& params)
      {
         auto url_result = parse_url(params.url);
         if (!url_result) {
            asio::post(io_executor, [on_error = params.on_error, error = url_result.error()]() { on_error(error); });
            return nullptr;
         }

         return perform_stream_request(params.method, *url_result, params.body, params.max_buffer_size, params.headers,
                                       params.timeout, params.strategy, params.status_is_error, params.on_data,
                                       params.on_error, params.on_connect, params.on_disconnect);
      }

      std::shared_ptr<http_stream_connection> stream_request_v2(const stream_request_params_v2& params)
      {
         auto url_result = parse_url(params.url);
         if (!url_result) {
            asio::post(io_executor, [on_error = params.on_error, error = url_result.error()]() { on_error(error); });
            return nullptr;
         }

         return perform_stream_request(params.method, *url_result, params.body, params.max_buffer_size, params.headers,
                                       params.timeout, params.strategy, params.status_is_error, params.on_data,
                                       params.on_error, params.on_connect, params.on_disconnect);
      }

      // Asynchronous GET request
      template <typename CompletionHandler>
      void get_async(std::string_view url, const std::unordered_map<std::string, std::string>& headers,
                     CompletionHandler&& handler)
      {
         auto url_result = parse_url(url);
         if (!url_result) {
            asio::post(io_executor, [handler = std::forward<CompletionHandler>(handler),
                                     error = url_result.error()] mutable { handler(std::unexpected(error)); });
            return;
         }

         perform_request_async("GET", *url_result, "", headers, std::forward<CompletionHandler>(handler));
      }

      // Overload for get_async without completion handler (returns future)
      std::future<std::expected<response, std::error_code>> get_async(
         std::string_view url, const std::unordered_map<std::string, std::string>& headers = {})
      {
         std::promise<std::expected<response, std::error_code>> promise;
         auto future = promise.get_future();

         get_async(url, headers,
                   [promise = std::move(promise)](std::expected<response, std::error_code> result) mutable {
                      promise.set_value(std::move(result));
                   });

         return future;
      }

      // Asynchronous POST request
      template <typename CompletionHandler>
      void post_async(std::string_view url, const std::string& body,
                      const std::unordered_map<std::string, std::string>& headers, CompletionHandler&& handler)
      {
         auto url_result = parse_url(url);
         if (!url_result) {
            asio::post(io_executor, [handler = std::forward<CompletionHandler>(handler),
                                     error = url_result.error()] mutable { handler(std::unexpected(error)); });
            return;
         }

         perform_request_async("POST", *url_result, body, headers, std::forward<CompletionHandler>(handler));
      }

      // Overload for post_async without completion handler (returns future)
      std::future<std::expected<response, std::error_code>> post_async(
         std::string_view url, const std::string& body,
         const std::unordered_map<std::string, std::string>& headers = {})
      {
         std::promise<std::expected<response, std::error_code>> promise;
         auto future = promise.get_future();

         post_async(url, body, headers,
                    [promise = std::move(promise)](std::expected<response, std::error_code> result) mutable {
                       promise.set_value(std::move(result));
                    });

         return future;
      }

      // Async JSON POST request
      template <class T, typename CompletionHandler>
      void post_json_async(std::string_view url, const T& data,
                           const std::unordered_map<std::string, std::string>& headers, CompletionHandler&& handler)
      {
         std::string json_str;
         auto ec = glz::write_json(data, json_str);
         if (ec) {
            asio::post(io_executor, [handler = std::forward<CompletionHandler>(handler)]() {
               handler(std::unexpected(std::make_error_code(std::errc::invalid_argument)));
            });
            return;
         }

         auto merged_headers = headers;
         merged_headers["content-type"] = "application/json";

         post_async(url, json_str, merged_headers, std::forward<CompletionHandler>(handler));
      }

      // Overload for post_json_async without completion handler (returns future)
      template <class T>
      std::future<std::expected<response, std::error_code>> post_json_async(
         std::string_view url, const T& data, const std::unordered_map<std::string, std::string>& headers = {})
      {
         std::promise<std::expected<response, std::error_code>> promise;
         auto future = promise.get_future();

         post_json_async(url, data, headers,
                         [promise = std::move(promise)](std::expected<response, std::error_code> result) mutable {
                            promise.set_value(std::move(result));
                         });

         return future;
      }

      // Get executor for async operations (timers, etc.)
      asio::any_io_executor get_executor() const { return io_executor; }

      // Set maximum response body size in bytes (0 = unlimited).
      // Responses exceeding this return http_client_error::response_too_large.
      // Must be configured before issuing requests; not safe to change while requests are in flight.
      http_client& max_response_body_size(size_t max_size)
      {
         max_response_body_size_ = max_size;
         return *this;
      }

      size_t max_response_body_size() const { return max_response_body_size_; }

     private:
      size_t max_response_body_size_ = http_default_max_body_size;
      // For async operations only when no io_executor is provided
      std::shared_ptr<asio::io_context> async_io_context;
      asio::any_io_executor io_executor;
      std::shared_ptr<http_connection_pool> connection_pool;
      std::vector<std::thread> worker_threads;
      // Engaged only when we own async_io_context; disengaged for external executors.
      std::optional<asio::executor_work_guard<asio::io_context::executor_type>> work_guard;

      void start_workers()
      {
         // don't start worker threads when an io_executor was provided
         if (!async_io_context) return;

         size_t num_threads = (std::max)(2u, std::thread::hardware_concurrency());
         worker_threads.reserve(num_threads);

         for (size_t i = 0; i < num_threads; ++i) {
            worker_threads.emplace_back([this]() {
               while (true) {
                  try {
                     async_io_context->run();
                     break;
                  }
                  catch (const std::exception& e) {
                     std::cerr << "HTTP client worker error: " << e.what() << std::endl;
                  }
               }
            });
         }
      }

      void stop_workers()
      {
         // don't stop worker threads when an io_executor was provided
         if (!async_io_context) return;

         work_guard.reset();
         async_io_context->stop();

         for (auto& thread : worker_threads) {
            if (thread.joinable()) {
               thread.join();
            }
         }
      }

      std::shared_ptr<http_stream_connection> perform_stream_request(
         const std::string& method, const url_parts& url, const std::string& body, size_t max_buffer_size,
         const std::unordered_map<std::string, std::string>& headers, std::chrono::seconds timeout,
         stream_read_strategy strategy, std::function<bool(int)> status_is_error, http_data_handler on_data,
         http_error_handler on_error, http_connect_handler on_connect, http_disconnect_handler on_disconnect)
      {
         const bool use_https = (url.protocol == "https");

#ifndef GLZ_ENABLE_SSL
         if (use_https) {
            asio::post(io_executor,
                       [on_error = std::move(on_error)]() { on_error(make_error_code(ssl_error::ssl_not_supported)); });
            return nullptr;
         }
#endif

         auto connection = std::make_shared<http_stream_connection>(max_buffer_size, strategy);
         connection->socket =
            std::make_shared<socket_variant>(connection_pool->get_connection(url.host, url.port, use_https));
         connection->timer = std::make_shared<asio::steady_timer>(io_executor);
         connection->is_https = use_https;

         connection->status_is_error = std::move(status_is_error);

         // Wrap the disconnect handler to return the socket to the pool
         auto internal_on_disconnect = [this, user_on_disconnect = std::move(on_disconnect), connection, url,
                                        use_https]() {
            connection->is_connected = false;
            // Call the user's handler if provided
            if (user_on_disconnect) {
               user_on_disconnect();
            }
            // Return the connection to the pool for reuse
            if (connection->socket) {
               connection_pool->return_connection(url.host, url.port, use_https, std::move(*connection->socket));
            }
         };

         // Set connection timeout
         connection->timer->expires_after(timeout);
         connection->timer->async_wait([connection, on_error, internal_on_disconnect](asio::error_code ec) {
            if (!ec && !connection->is_connected && !connection->should_stop) {
               // Mark for stop to prevent race conditions
               connection->disconnect();
               on_error(std::make_error_code(std::errc::timed_out));
               internal_on_disconnect();
            }
         });

         // Check if the socket is already open
         if (detail::socket_is_open(*connection->socket)) {
            // If already connected, skip resolve and connect, and just send the request
            asio::post(io_executor, [this, url, method, body, headers, connection, on_data = std::move(on_data),
                                     on_error = std::move(on_error), on_connect = std::move(on_connect),
                                     internal_on_disconnect = std::move(internal_on_disconnect)]() mutable {
               send_stream_request(url, method, body, headers, connection, std::move(on_data), std::move(on_error),
                                   std::move(on_connect), std::move(internal_on_disconnect));
            });
         }
         else {
            // If not connected, resolve and connect as before
            auto resolver = std::make_shared<asio::ip::tcp::resolver>(io_executor);
            resolver->async_resolve(
               url.host, std::to_string(url.port),
               [this, url, method, body, headers, connection, resolver, use_https, on_data = std::move(on_data),
                on_error = std::move(on_error), on_connect = std::move(on_connect),
                internal_on_disconnect = std::move(internal_on_disconnect)](
                  asio::error_code ec, asio::ip::tcp::resolver::results_type results) mutable {
                  if (ec || connection->should_stop) {
                     on_error(ec);
                     internal_on_disconnect(); // Ensure cleanup on resolve error
                     return;
                  }

                  // Connect the underlying TCP socket
                  std::visit(
                     [&, this](auto& sock) {
                        asio::ip::tcp::socket* tcp_sock;
                        if constexpr (requires { sock->next_layer(); }) {
                           tcp_sock = &sock->next_layer();
                        }
                        else {
                           tcp_sock = sock.get();
                        }

                        asio::async_connect(
                           *tcp_sock, results,
                           [this, url, method, body, headers, connection, use_https, on_data = std::move(on_data),
                            on_error = std::move(on_error), on_connect = std::move(on_connect),
                            internal_on_disconnect = std::move(internal_on_disconnect)](
                              asio::error_code ec, const asio::ip::tcp::endpoint&) mutable {
                              if (ec || connection->should_stop) {
                                 on_error(ec);
                                 internal_on_disconnect(); // Ensure cleanup on connect error
                                 return;
                              }

#ifndef GLZ_ENABLE_SSL
                              (void)use_https;
#endif

#ifdef GLZ_ENABLE_SSL
                              if (use_https) {
                                 perform_stream_ssl_handshake(url, method, body, headers, connection,
                                                              std::move(on_data), std::move(on_error),
                                                              std::move(on_connect), std::move(internal_on_disconnect));
                              }
                              else
#endif
                              {
                                 send_stream_request(url, method, body, headers, connection, std::move(on_data),
                                                     std::move(on_error), std::move(on_connect),
                                                     std::move(internal_on_disconnect));
                              }
                           });
                     },
                     *connection->socket);
               });
         }

         return connection;
      }

#ifdef GLZ_ENABLE_SSL
      void perform_stream_ssl_handshake(const url_parts& url, const std::string& method, const std::string& body,
                                        const std::unordered_map<std::string, std::string>& headers,
                                        std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                                        http_error_handler on_error, http_connect_handler on_connect,
                                        http_disconnect_handler on_disconnect)
      {
         auto& ssl_sock = std::get<std::shared_ptr<ssl_socket>>(*connection->socket);

         // Set SNI hostname for virtual hosting support
         if (!detail::configure_ssl_client_hostname(*ssl_sock, url.host)) {
            on_error(make_error_code(ssl_error::sni_hostname_failed));
            on_disconnect();
            return;
         }

         ssl_sock->async_handshake(
            asio::ssl::stream_base::client, [this, url, method, body, headers, connection, on_data = std::move(on_data),
                                             on_error = std::move(on_error), on_connect = std::move(on_connect),
                                             on_disconnect = std::move(on_disconnect)](asio::error_code ec) mutable {
               if (ec || connection->should_stop) {
                  on_error(ec);
                  on_disconnect();
                  return;
               }

               send_stream_request(url, method, body, headers, connection, std::move(on_data), std::move(on_error),
                                   std::move(on_connect), std::move(on_disconnect));
            });
      }
#endif

      // Needs to take the wrapped disconnect handler and use keep-alive
      void send_stream_request(const url_parts& url, const std::string& method, const std::string& body,
                               const std::unordered_map<std::string, std::string>& headers,
                               std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                               http_error_handler on_error, http_connect_handler on_connect,
                               http_disconnect_handler on_disconnect)
      {
         const bool use_https = url.protocol == "https";

         auto request_buffer =
            std::make_shared<std::string>(detail::build_http_request_bytes(method, url, use_https, body, headers));

         std::visit(
            [&, this](auto& sock) {
               asio::async_write(*sock, asio::buffer(*request_buffer),
                                 [this, connection, request_buffer, on_data = std::move(on_data),
                                  on_error = std::move(on_error), on_connect = std::move(on_connect),
                                  on_disconnect = std::move(on_disconnect)](asio::error_code ec, std::size_t) mutable {
                                    if (ec || connection->should_stop) {
                                       on_error(ec);
                                       if (on_disconnect) on_disconnect();
                                       return;
                                    }

                                    read_stream_response(connection, std::move(on_data), std::move(on_error),
                                                         std::move(on_connect), std::move(on_disconnect));
                                 });
            },
            *connection->socket);
      }

      void read_stream_response(std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                                http_error_handler on_error, http_connect_handler on_connect,
                                http_disconnect_handler on_disconnect)
      {
         // Use the connection's unified buffer
         std::visit(
            [&, this](auto& sock) {
               asio::async_read_until(
                  *sock, *connection->buffer, "\r\n\r\n",
                  [this, connection, on_data = std::move(on_data), on_error = std::move(on_error),
                   on_connect = std::move(on_connect), on_disconnect = std::move(on_disconnect)](
                     asio::error_code ec, std::size_t bytes_transferred) mutable {
                     if (ec || connection->should_stop) {
                        on_error(ec);
                        if (on_disconnect) on_disconnect();
                        return;
                     }

                     // Create a zero-copy string_view of the received headers
                     std::string_view header_data{static_cast<const char*>(connection->buffer->data().data()),
                                                  bytes_transferred};

                     // Parse status line
                     auto line_end = header_data.find("\r\n");
                     if (line_end == std::string_view::npos) {
                        on_error(std::make_error_code(std::errc::protocol_error));
                        if (on_disconnect) on_disconnect();
                        return;
                     }
                     std::string_view status_line = header_data.substr(0, line_end);
                     header_data.remove_prefix(line_end + 2);

                     auto parsed_status = parse_http_status_line(status_line);
                     if (!parsed_status) {
                        on_error(parsed_status.error());
                        if (on_disconnect) on_disconnect();
                        return;
                     }

                     response response_headers;
                     response_headers.status_code = parsed_status->status_code;

                     // Parse header fields
                     while (!header_data.starts_with("\r\n")) {
                        line_end = header_data.find("\r\n");
                        if (line_end == std::string_view::npos) {
                           on_error(std::make_error_code(std::errc::protocol_error));
                           if (on_disconnect) on_disconnect();
                           return;
                        }

                        std::string_view header_line = header_data.substr(0, line_end);
                        header_data.remove_prefix(line_end + 2);

                        auto colon_pos = header_line.find(':');
                        if (colon_pos != std::string::npos) {
                           std::string_view name = header_line.substr(0, colon_pos);
                           // Skip past ':' and any whitespace
                           size_t value_start = header_line.find_first_not_of(" \t", colon_pos + 1);
                           std::string_view value =
                              (value_start != std::string::npos) ? header_line.substr(value_start) : "";

                           // Convert header name to lowercase for case-insensitive lookups (RFC 7230)
                           response_headers.response_headers[to_lower_case(name)] = std::string(value);
                        }
                     }

                     // Consume all processed header data from the streambuf
                     connection->buffer->consume(bytes_transferred);

                     connection->is_connected = true;
                     connection->timer->cancel();

                     if (on_connect) {
                        on_connect(response_headers);
                     }

                     const bool status_is_error = connection->status_is_error
                                                     ? connection->status_is_error(parsed_status->status_code)
                                                     : parsed_status->status_code >= 400;

                     if (status_is_error) {
                        // Propagate the precise HTTP status via a dedicated error category.
                        on_error(make_http_status_error(parsed_status->status_code));
                        if (on_disconnect) on_disconnect();
                        return;
                     }

                     bool is_chunked = false;
                     auto it = response_headers.response_headers.find("transfer-encoding");
                     if (it != response_headers.response_headers.end()) {
                        if (it->second.find("chunked") != std::string::npos) {
                           is_chunked = true;
                        }
                     }

                     if (is_chunked) {
                        start_chunked_reading(connection, std::move(on_data), std::move(on_error),
                                              std::move(on_disconnect));
                     }
                     else {
                        start_stream_reading(connection, std::move(on_data), std::move(on_error),
                                             std::move(on_disconnect));
                     }
                  });
            },
            *connection->socket);
      }

      void start_chunked_reading(std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                                 http_error_handler on_error, http_disconnect_handler on_disconnect)
      {
         // Begin the chunk parsing loop.
         // The buffer may already contain the first chunk size line or more data.
         read_chunk_size(connection, std::move(on_data), std::move(on_error), std::move(on_disconnect));
      }

      void read_chunk_size(std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                           http_error_handler on_error, http_disconnect_handler on_disconnect)
      {
         std::visit(
            [&, this](auto& sock) {
               asio::async_read_until(
                  *sock, *connection->buffer, "\r\n",
                  [this, connection, on_data = std::move(on_data), on_error = std::move(on_error),
                   on_disconnect = std::move(on_disconnect)](asio::error_code ec,
                                                             std::size_t bytes_transferred) mutable {
                     if (ec || connection->should_stop) {
                        if (ec != asio::error::eof && ec != asio::error::operation_aborted && !connection->should_stop)
                           on_error(ec);
                        if (on_disconnect) on_disconnect();
                        return;
                     }

                     std::string_view line_view{static_cast<const char*>(connection->buffer->data().data()),
                                                bytes_transferred - 2}; // -2 to exclude CRLF

                     // Ignore chunk extensions
                     auto semi_pos = line_view.find(';');
                     if (semi_pos != std::string_view::npos) {
                        line_view = line_view.substr(0, semi_pos);
                     }

                     size_t chunk_size;
                     auto [ptr, parse_ec] =
                        std::from_chars(line_view.data(), line_view.data() + line_view.size(), chunk_size, 16);

                     if (parse_ec != std::errc{}) {
                        on_error(std::make_error_code(std::errc::protocol_error));
                        if (on_disconnect) on_disconnect();
                        return;
                     }

                     // Consume the size line and CRLF from the buffer
                     connection->buffer->consume(bytes_transferred);

                     if (chunk_size == 0) {
                        // Last chunk
                        if (on_disconnect) on_disconnect();
                        return;
                     }

                     read_chunk_body(connection, chunk_size, std::move(on_data), std::move(on_error),
                                     std::move(on_disconnect));
                  });
            },
            *connection->socket);
      }

      void read_chunk_body(std::shared_ptr<http_stream_connection> connection, size_t chunk_size,
                           http_data_handler on_data, http_error_handler on_error,
                           http_disconnect_handler on_disconnect)
      {
         // A chunk size with no room for the trailing CRLF wraps chunk_size + 2 below, passes the
         // buffered-size check with a tiny length, and hands on_data a view of chunk_size bytes
         // over a few-byte buffer. Reject it as a malformed chunk.
         if (chunk_size > (std::numeric_limits<size_t>::max)() - 2) [[unlikely]] {
            on_error(std::make_error_code(std::errc::protocol_error));
            if (on_disconnect) on_disconnect();
            return;
         }

         // We need to read 'chunk_size' bytes of data, plus 2 bytes for the trailing CRLF.
         size_t total_to_read = chunk_size + 2;

         // Check if we have enough data in the buffer already.
         if (connection->buffer->size() >= total_to_read) {
            std::string_view data{static_cast<const char*>(connection->buffer->data().data()), chunk_size};
            on_data(data);
            connection->buffer->consume(total_to_read);

            // Post the next read to avoid deep recursion
            std::visit(
               [&, this](auto& sock) {
                  asio::post(sock->get_executor(), [this, connection, on_data = std::move(on_data),
                                                    on_error = std::move(on_error),
                                                    on_disconnect = std::move(on_disconnect)]() mutable {
                     read_chunk_size(connection, std::move(on_data), std::move(on_error), std::move(on_disconnect));
                  });
               },
               *connection->socket);
            return;
         }

         // We need to read more from the socket.
         std::visit(
            [&, this](auto& sock) {
               asio::async_read(
                  *sock, *connection->buffer, asio::transfer_exactly(total_to_read - connection->buffer->size()),
                  [this, connection, chunk_size, on_data = std::move(on_data), on_error = std::move(on_error),
                   on_disconnect = std::move(on_disconnect)](asio::error_code ec, std::size_t) mutable {
                     if (ec || connection->should_stop) {
                        if (ec != asio::error::eof && ec != asio::error::operation_aborted && !connection->should_stop)
                           on_error(ec);
                        if (on_disconnect) on_disconnect();
                        return;
                     }

                     std::string_view data{static_cast<const char*>(connection->buffer->data().data()), chunk_size};
                     on_data(data);
                     connection->buffer->consume(chunk_size + 2); // Consume data + trailing CRLF

                     read_chunk_size(connection, std::move(on_data), std::move(on_error), std::move(on_disconnect));
                  });
            },
            *connection->socket);
      }

      void start_stream_reading(std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                                http_error_handler on_error, http_disconnect_handler on_disconnect)
      {
         // Dispatch to the appropriate strategy
         switch (connection->strategy) {
         case stream_read_strategy::bulk_transfer:
            start_stream_reading_bulk(connection, std::move(on_data), std::move(on_error), std::move(on_disconnect));
            break;
         case stream_read_strategy::immediate_delivery:
            start_stream_reading_immediate(connection, std::move(on_data), std::move(on_error),
                                           std::move(on_disconnect));
            break;
         }
      }

      void start_stream_reading_bulk(std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                                     http_error_handler on_error, http_disconnect_handler on_disconnect)
      {
         // Process any existing data in buffer first
         if (connection->buffer->size() > 0) {
            std::string_view data{static_cast<const char*>(connection->buffer->data().data()),
                                  connection->buffer->size()};
            on_data(data);
            connection->buffer->consume(connection->buffer->size());
         }

         if (connection->should_stop) {
            if (on_disconnect) on_disconnect();
            return;
         }

         // Use async_read with transfer_at_least(1) - may read more data for efficiency
         std::visit(
            [&, this](auto& sock) {
               asio::async_read(*sock, *connection->buffer, asio::transfer_at_least(1),
                                [this, connection, on_data, on_error, on_disconnect](
                                   asio::error_code ec, std::size_t /*bytes_transferred*/) {
                                   if (ec || connection->should_stop) {
                                      if (ec != asio::error::eof && ec != asio::error::operation_aborted &&
                                          !connection->should_stop) {
                                         on_error(ec);
                                      }
                                      if (on_disconnect) on_disconnect();
                                      return;
                                   }

                                   // Recurse to process the new data
                                   start_stream_reading_bulk(connection, on_data, on_error, on_disconnect);
                                });
            },
            *connection->socket);
      }

      void start_stream_reading_immediate(std::shared_ptr<http_stream_connection> connection, http_data_handler on_data,
                                          http_error_handler on_error, http_disconnect_handler on_disconnect)
      {
         // Process existing buffer content first
         if (connection->buffer->size() > 0) {
            std::string_view data{static_cast<const char*>(connection->buffer->data().data()),
                                  connection->buffer->size()};
            on_data(data);
            connection->buffer->consume(connection->buffer->size());
         }

         if (connection->should_stop) {
            if (on_disconnect) on_disconnect();
            return;
         }

         // Use async_read_some for immediate delivery of available data
         constexpr size_t read_size = 8192;
         std::visit(
            [&, this](auto& sock) {
               sock->async_read_some(connection->buffer->prepare(read_size), [this, connection, on_data, on_error,
                                                                              on_disconnect](
                                                                                asio::error_code ec,
                                                                                std::size_t bytes_transferred) {
                  if (ec || connection->should_stop) {
                     if (ec != asio::error::eof && ec != asio::error::operation_aborted && !connection->should_stop) {
                        on_error(ec);
                     }
                     if (on_disconnect) on_disconnect();
                     return;
                  }

                  // Commit the received data and deliver immediately
                  connection->buffer->commit(bytes_transferred);

                  std::string_view data{static_cast<const char*>(connection->buffer->data().data()), bytes_transferred};
                  on_data(data);
                  connection->buffer->consume(bytes_transferred);

                  // Continue reading
                  start_stream_reading_immediate(connection, on_data, on_error, on_disconnect);
               });
            },
            *connection->socket);
      }

      std::expected<response, std::error_code> perform_sync_request(
         const std::string& method, const url_parts& url, const std::string& body,
         const std::unordered_map<std::string, std::string>& headers)
      {
         const bool use_https = (url.protocol == "https");

#ifndef GLZ_ENABLE_SSL
         if (use_https) {
            return std::unexpected(make_error_code(ssl_error::ssl_not_supported));
         }
#endif

         auto acquired = connection_pool->acquire(url.host, url.port, use_https);
         const bool idempotent = is_idempotent_method(method);

         // Build the wire bytes once. If retry fires, the same bytes are written on
         // the fresh socket: no rebuild, no extra body copy.
         const std::string request_str = detail::build_http_request_bytes(method, url, use_https, body, headers);

         auto attempt = perform_sync_request_attempt(url, request_str, use_https, std::move(acquired.socket));

         // Retry once if and only if: idempotent method, server has not yet started
         // sending a response (so the request bytes were not processed), and the error
         // is the kind that indicates a connection problem rather than an application
         // failure. This catches both stale pooled connections and fresh connections
         // that fail before the server can read (listener overload, server-side close
         // immediately after accept, mid-write RST, etc.).
         if (idempotent && !attempt.outcome && !attempt.response_started &&
             is_retriable_pool_error(attempt.outcome.error())) {
            auto fresh = connection_pool->create_fresh_connection(use_https);
            attempt = perform_sync_request_attempt(url, request_str, use_https, std::move(fresh));
         }

         return std::move(attempt.outcome);
      }

      sync_attempt_outcome perform_sync_request_attempt(const url_parts& url, const std::string& request_str,
                                                        bool use_https, socket_variant socket_var)
      {
         sync_attempt_outcome r{};

         try {
            // If socket is not connected, connect it synchronously
            if (!detail::socket_is_open(socket_var)) {
               asio::ip::tcp::resolver resolver(io_executor);
               auto endpoints = resolver.resolve(url.host, std::to_string(url.port));

               // Connect the underlying TCP socket
               std::visit(
                  [&](auto& sock) {
                     if constexpr (requires { sock->next_layer(); }) {
                        asio::connect(sock->next_layer(), endpoints);
                     }
                     else {
                        asio::connect(*sock, endpoints);
                     }
                  },
                  socket_var);

#ifdef GLZ_ENABLE_SSL
               // Perform SSL handshake for HTTPS connections
               if (use_https) {
                  auto& ssl_sock = std::get<std::shared_ptr<ssl_socket>>(socket_var);

                  // Set SNI hostname for virtual hosting support
                  if (!detail::configure_ssl_client_hostname(*ssl_sock, url.host)) {
                     detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                     r.outcome = std::unexpected(make_error_code(ssl_error::sni_hostname_failed));
                     return r;
                  }

                  // Perform the SSL handshake
                  asio::error_code handshake_ec;
                  ssl_sock->handshake(asio::ssl::stream_base::client, handshake_ec);
                  if (handshake_ec) {
                     detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                     r.outcome = std::unexpected(handshake_ec);
                     return r;
                  }
               }
#endif
            }

            // Send pre-built request bytes synchronously
            std::visit([&](auto& sock) { asio::write(*sock, asio::buffer(request_str)); }, socket_var);

            // Read response headers synchronously
            asio::streambuf response_buffer;
            asio::error_code ec;
            size_t header_bytes = std::visit(
               [&](auto& sock) { return asio::read_until(*sock, response_buffer, "\r\n\r\n", ec); }, socket_var);
            if (ec) {
               detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
               r.outcome = std::unexpected(ec);
               return r;
            }

            // Headers received: from this point on, the server has begun responding and
            // a retry of this exact request would re-execute server-side work that has
            // already happened. Mark response_started so perform_sync_request will not
            // retry on subsequent failures.
            r.response_started = true;

            // Create a zero-copy view of the header data
            std::string_view header_data{static_cast<const char*>(response_buffer.data().data()), header_bytes};

            // Parse status line from the view
            auto line_end = header_data.find("\r\n");
            if (line_end == std::string_view::npos) {
               r.outcome = std::unexpected(std::make_error_code(std::errc::protocol_error));
               return r;
            }
            std::string_view status_line = header_data.substr(0, line_end);
            header_data.remove_prefix(line_end + 2); // Advance past status line

            auto parsed_status = parse_http_status_line(status_line);
            if (!parsed_status) {
               r.outcome = std::unexpected(parsed_status.error());
               return r;
            }

            // Parse headers from the view
            std::unordered_map<std::string, std::string> response_headers;
            size_t content_length = 0;
            bool has_content_length = false;
            bool connection_close = false;
            bool is_chunked = false;

            while (!header_data.starts_with("\r\n")) {
               line_end = header_data.find("\r\n");
               if (line_end == std::string_view::npos) {
                  r.outcome = std::unexpected(std::make_error_code(std::errc::protocol_error));
                  return r;
               }
               std::string_view header_line = header_data.substr(0, line_end);
               header_data.remove_prefix(line_end + 2);

               auto colon_pos = header_line.find(':');
               if (colon_pos != std::string::npos) {
                  std::string_view name = header_line.substr(0, colon_pos);
                  size_t value_start = header_line.find_first_not_of(" \t", colon_pos + 1);
                  std::string_view value = (value_start != std::string::npos) ? header_line.substr(value_start) : "";

                  if (name.length() == 14 && glz::strncasecmp(name.data(), "Content-Length", 14) == 0) {
                     std::from_chars(value.data(), value.data() + value.size(), content_length);
                     has_content_length = true;
                  }
                  else if (name.length() == 17 && glz::strncasecmp(name.data(), "Transfer-Encoding", 17) == 0) {
                     if (value.find("chunked") != std::string_view::npos) {
                        is_chunked = true;
                     }
                  }
                  else if (name.length() == 10 && glz::strncasecmp(name.data(), "Connection", 10) == 0) {
                     if (value.find("close") != std::string_view::npos) {
                        connection_close = true;
                     }
                  }

                  // Convert header name to lowercase for case-insensitive lookups (RFC 7230)
                  response_headers.emplace(to_lower_case(name), value);
               }
            }

            // Consume header data, leaving only the over-read body part.
            response_buffer.consume(header_bytes);

            // Reject responses that exceed the configured body size limit
            if (max_response_body_size_ > 0 && !is_chunked && content_length > max_response_body_size_) {
               detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
               r.outcome = std::unexpected(make_error_code(http_client_error::response_too_large));
               return r;
            }

            std::string response_body;

            if (is_chunked) {
               // Read chunked transfer-encoded body
               while (true) {
                  // Read chunk size line (hex digits followed by CRLF)
                  asio::error_code read_ec;
                  std::visit([&](auto& sock) { asio::read_until(*sock, response_buffer, "\r\n", read_ec); },
                             socket_var);
                  if (read_ec) {
                     detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                     r.outcome = std::unexpected(read_ec);
                     return r;
                  }

                  // Parse chunk size from the buffer
                  std::string_view size_line{static_cast<const char*>(response_buffer.data().data()),
                                             response_buffer.size()};
                  auto crlf_pos = size_line.find("\r\n");
                  if (crlf_pos == std::string_view::npos) {
                     r.outcome = std::unexpected(std::make_error_code(std::errc::protocol_error));
                     return r;
                  }
                  std::string_view chunk_size_str = size_line.substr(0, crlf_pos);

                  // Ignore chunk extensions
                  auto semi_pos = chunk_size_str.find(';');
                  if (semi_pos != std::string_view::npos) {
                     chunk_size_str = chunk_size_str.substr(0, semi_pos);
                  }

                  size_t chunk_size = 0;
                  auto [ptr, parse_ec] = std::from_chars(chunk_size_str.data(),
                                                         chunk_size_str.data() + chunk_size_str.size(), chunk_size, 16);
                  if (parse_ec != std::errc{}) {
                     r.outcome = std::unexpected(std::make_error_code(std::errc::protocol_error));
                     return r;
                  }

                  // Consume the chunk size line + CRLF
                  response_buffer.consume(crlf_pos + 2);

                  if (chunk_size == 0) {
                     // Terminal chunk; skip optional trailers and final CRLF (RFC 7230 §4.1)
                     while (true) {
                        std::visit([&](auto& sock) { asio::read_until(*sock, response_buffer, "\r\n", read_ec); },
                                   socket_var);
                        if (read_ec) {
                           connection_close = true;
                           break;
                        }
                        std::string_view line_view{static_cast<const char*>(response_buffer.data().data()),
                                                   response_buffer.size()};
                        auto pos = line_view.find("\r\n");
                        bool empty_line = (pos == 0);
                        response_buffer.consume(pos + 2);
                        if (empty_line) break; // end of trailers
                     }
                     break;
                  }

                  // Check accumulated body size against limit before reading chunk data
                  if (max_response_body_size_ > 0 && response_body.size() + chunk_size > max_response_body_size_) {
                     detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                     r.outcome = std::unexpected(make_error_code(http_client_error::response_too_large));
                     return r;
                  }

                  // A chunk size that leaves no room for the trailing CRLF wraps chunk_size + 2
                  // below and appends with a bogus length, so reject it as malformed.
                  if (chunk_size > (std::numeric_limits<size_t>::max)() - 2) {
                     detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                     r.outcome = std::unexpected(std::make_error_code(std::errc::protocol_error));
                     return r;
                  }

                  // Read chunk data + trailing CRLF
                  size_t total_needed = chunk_size + 2; // data + CRLF
                  if (response_buffer.size() < total_needed) {
                     size_t to_read = total_needed - response_buffer.size();
                     std::visit(
                        [&](auto& sock) {
                           asio::read(*sock, response_buffer, asio::transfer_exactly(to_read), read_ec);
                        },
                        socket_var);
                     if (read_ec) {
                        detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                        r.outcome = std::unexpected(read_ec);
                        return r;
                     }
                  }

                  // Append chunk data to response body
                  const char* chunk_data = static_cast<const char*>(response_buffer.data().data());
                  response_body.append(chunk_data, chunk_size);

                  // Consume chunk data + trailing CRLF
                  response_buffer.consume(total_needed);
               }
            }
            else if (!has_content_length && connection_close) {
               // No Content-Length and not chunked: read until connection close (RFC 7230 §3.3.3)
               // Read incrementally to enforce max_response_body_size_ without unbounded allocation
               asio::error_code read_ec;
               while (true) {
                  std::visit(
                     [&](auto& sock) { asio::read(*sock, response_buffer, asio::transfer_at_least(1), read_ec); },
                     socket_var);
                  if (read_ec) break;
                  if (max_response_body_size_ > 0 && response_buffer.size() > max_response_body_size_) {
                     detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                     r.outcome = std::unexpected(make_error_code(http_client_error::response_too_large));
                     return r;
                  }
               }

               if (read_ec != asio::error::eof) {
                  detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                  r.outcome = std::unexpected(read_ec);
                  return r;
               }

               response_body.assign(static_cast<const char*>(response_buffer.data().data()), response_buffer.size());
               connection_close = true;
            }
            else {
               // Read body based on content-length
               size_t body_in_buffer = response_buffer.size();
               if (content_length > body_in_buffer) {
                  size_t remaining_to_read = content_length - body_in_buffer;
                  asio::error_code read_ec;
                  std::visit(
                     [&](auto& sock) {
                        asio::read(*sock, response_buffer, asio::transfer_exactly(remaining_to_read), read_ec);
                     },
                     socket_var);
                  if (read_ec) {
                     detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
                     r.outcome = std::unexpected(read_ec);
                     return r;
                  }
               }

               response_body.assign(static_cast<const char*>(response_buffer.data().data()),
                                    (std::min)(content_length, response_buffer.size()));
            }

            response resp;
            resp.status_code = parsed_status->status_code;
            resp.response_headers = std::move(response_headers);
            resp.response_body = std::move(response_body);

            if (!connection_close) {
               connection_pool->return_connection(url.host, url.port, use_https, socket_var);
            }
            // else, the server will close the connection, so we don't return it to the pool.

            r.outcome = std::move(resp);
            return r;
         }
         catch (const asio::system_error& e) {
            detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
            r.outcome = std::unexpected(e.code());
            return r;
         }
         catch (...) {
            detail::close_socket(socket_var, connection_pool->graceful_ssl_shutdown());
            r.outcome = std::unexpected(std::make_error_code(std::errc::connection_refused));
            return r;
         }
      }

      template <typename CompletionHandler>
      void perform_request_async(const std::string& method, const url_parts& url, const std::string& body,
                                 const std::unordered_map<std::string, std::string>& headers,
                                 CompletionHandler&& handler)
      {
         const bool use_https = (url.protocol == "https");

#ifndef GLZ_ENABLE_SSL
         if (use_https) {
            asio::post(io_executor, [handler = std::forward<CompletionHandler>(handler)]() mutable {
               handler(std::unexpected(make_error_code(ssl_error::ssl_not_supported)));
            });
            return;
         }
#endif

         auto acquired = connection_pool->acquire(url.host, url.port, use_https);
         auto socket_var = std::make_shared<socket_variant>(std::move(acquired.socket));

         // Build the wire bytes once. This is shared by the original attempt and any
         // retry, and by every async lambda capture in the chain - those just bump the
         // shared_ptr refcount instead of copying the request body and header map.
         auto request_bytes =
            std::make_shared<std::string>(detail::build_http_request_bytes(method, url, use_https, body, headers));

         // Non-idempotent methods (POST, PATCH) are never retried because the client
         // cannot tell whether a write+read failure means the server already processed
         // the request. Forward directly without wrapping; no per-attempt state needed.
         if (!is_idempotent_method(method)) {
            auto state = std::make_shared<async_attempt_state>();
            do_perform_request_async_attempt(socket_var, request_bytes, url, use_https, state,
                                             std::forward<CompletionHandler>(handler));
            return;
         }

         // Idempotent: wrap the user handler so a retriable failure with no response
         // bytes received yet triggers exactly one retry on a fresh connection. This
         // catches both stale pooled connections AND fresh connections that fail before
         // the server can read (listener overload, immediate close after accept,
         // mid-write RST, etc.).
         auto state = std::make_shared<async_attempt_state>();

         using handler_t = std::decay_t<CompletionHandler>;
         auto user_handler_ptr = std::make_shared<handler_t>(std::forward<CompletionHandler>(handler));

         auto retry_handler = [this, request_bytes, url, use_https, user_handler_ptr,
                               state](std::expected<response, std::error_code> result) mutable {
            if (!result && !state->response_started.load() && is_retriable_pool_error(result.error())) {
               auto fresh = std::make_shared<socket_variant>(connection_pool->create_fresh_connection(use_https));
               // Fresh attempt gets its own state so the retry's response_started signal
               // is tracked independently of the failed first attempt.
               auto new_state = std::make_shared<async_attempt_state>();
               do_perform_request_async_attempt(fresh, request_bytes, url, use_https, new_state,
                                                [user_handler_ptr](std::expected<response, std::error_code> r) {
                                                   (*user_handler_ptr)(std::move(r));
                                                });
               return;
            }
            (*user_handler_ptr)(std::move(result));
         };

         do_perform_request_async_attempt(socket_var, request_bytes, url, use_https, state, std::move(retry_handler));
      }

      template <typename CompletionHandler>
      void do_perform_request_async_attempt(std::shared_ptr<socket_variant> socket_var,
                                            std::shared_ptr<std::string> request_bytes, const url_parts& url,
                                            bool use_https, std::shared_ptr<async_attempt_state> state,
                                            CompletionHandler&& handler)
      {
         if (detail::socket_is_open(*socket_var)) {
            send_request(socket_var, request_bytes, url, use_https, state, std::forward<CompletionHandler>(handler));
         }
         else {
            auto resolver = std::make_shared<asio::ip::tcp::resolver>(io_executor);
            resolver->async_resolve(url.host, std::to_string(url.port),
                                    [this, socket_var, resolver, request_bytes, url, use_https, state,
                                     handler = std::forward<CompletionHandler>(handler)](
                                       asio::error_code ec, asio::ip::tcp::resolver::results_type results) mutable {
                                       if (ec) {
                                          handler(std::unexpected(ec));
                                          return;
                                       }

                                       connect_and_send(socket_var, results, request_bytes, url, use_https, state,
                                                        std::move(handler));
                                    });
         }
      }

      template <typename CompletionHandler>
      void connect_and_send(std::shared_ptr<socket_variant> socket_var, asio::ip::tcp::resolver::results_type results,
                            std::shared_ptr<std::string> request_bytes, const url_parts& url, bool use_https,
                            std::shared_ptr<async_attempt_state> state, CompletionHandler&& handler)
      {
         // Get the underlying TCP socket for connection
         std::visit(
            [&, this](auto& sock) {
               asio::ip::tcp::socket* tcp_sock;
               if constexpr (requires { sock->next_layer(); }) {
                  tcp_sock = &sock->next_layer();
               }
               else {
                  tcp_sock = sock.get();
               }

               asio::async_connect(
                  *tcp_sock, results,
                  [this, socket_var, request_bytes, url, use_https, state,
                   handler = std::forward<CompletionHandler>(handler)](asio::error_code ec,
                                                                       const asio::ip::tcp::endpoint&) mutable {
                     if (ec) {
                        handler(std::unexpected(ec));
                        return;
                     }

#ifdef GLZ_ENABLE_SSL
                     if (use_https) {
                        perform_ssl_handshake(socket_var, request_bytes, url, state, std::move(handler));
                     }
                     else
#endif
                     {
                        send_request(socket_var, request_bytes, url, use_https, state, std::move(handler));
                     }
                  });
            },
            *socket_var);
      }

#ifdef GLZ_ENABLE_SSL
      template <typename CompletionHandler>
      void perform_ssl_handshake(std::shared_ptr<socket_variant> socket_var, std::shared_ptr<std::string> request_bytes,
                                 const url_parts& url, std::shared_ptr<async_attempt_state> state,
                                 CompletionHandler&& handler)
      {
         auto& ssl_sock = std::get<std::shared_ptr<ssl_socket>>(*socket_var);

         // Set SNI hostname for virtual hosting support
         if (!detail::configure_ssl_client_hostname(*ssl_sock, url.host)) {
            asio::post(io_executor, [handler = std::forward<CompletionHandler>(handler)]() mutable {
               handler(std::unexpected(make_error_code(ssl_error::sni_hostname_failed)));
            });
            return;
         }

         ssl_sock->async_handshake(asio::ssl::stream_base::client,
                                   [this, socket_var, request_bytes, url, state,
                                    handler = std::forward<CompletionHandler>(handler)](asio::error_code ec) mutable {
                                      if (ec) {
                                         handler(std::unexpected(ec));
                                         return;
                                      }

                                      send_request(socket_var, request_bytes, url, true, state, std::move(handler));
                                   });
      }
#endif

      template <typename CompletionHandler>
      void send_request(std::shared_ptr<socket_variant> socket_var, std::shared_ptr<std::string> request_bytes,
                        const url_parts& url, bool use_https, std::shared_ptr<async_attempt_state> state,
                        CompletionHandler&& handler)
      {
         std::visit(
            [&, this](auto& sock) {
               asio::async_write(
                  *sock, asio::buffer(*request_bytes),
                  [this, socket_var, request_bytes, url, use_https, state,
                   handler = std::forward<CompletionHandler>(handler)](asio::error_code ec, std::size_t) mutable {
                     if (ec) {
                        handler(std::unexpected(ec));
                        return;
                     }

                     read_response(socket_var, url, use_https, state, std::move(handler));
                  });
            },
            *socket_var);
      }

      template <typename CompletionHandler>
      void read_response(std::shared_ptr<socket_variant> socket_var, const url_parts& url, bool use_https,
                         std::shared_ptr<async_attempt_state> state, CompletionHandler&& handler)
      {
         auto buffer = std::make_shared<asio::streambuf>();

         std::visit(
            [&, this](auto& sock) {
               asio::async_read_until(
                  *sock, *buffer, "\r\n\r\n",
                  [this, socket_var, buffer, url, use_https, state, handler = std::forward<CompletionHandler>(handler)](
                     asio::error_code ec, std::size_t bytes_transferred) mutable {
                     if (ec) {
                        handler(std::unexpected(ec));
                        return;
                     }
                     // Headers received: signal that the server has started responding.
                     // Subsequent failures (mid-body, trailers) must NOT trigger a retry,
                     // as the server has already processed the request. The retry handler
                     // checks state->response_started before deciding to retry.
                     state->response_started.store(true);
                     // Pass the known header size to the parsing function
                     parse_and_read_body(socket_var, buffer, bytes_transferred, url, use_https, std::move(handler));
                  });
            },
            *socket_var);
      }

      // After the terminal chunk (0\r\n), skip optional trailers and the final CRLF (RFC 7230 §4.1),
      // then build and deliver the response.
      template <typename CompletionHandler>
      void async_consume_trailers(std::shared_ptr<socket_variant> socket_var, std::shared_ptr<asio::streambuf> buffer,
                                  std::shared_ptr<std::string> body, const url_parts& url, bool use_https,
                                  int status_code, std::unordered_map<std::string, std::string> response_headers,
                                  CompletionHandler&& handler)
      {
         std::visit(
            [&, this](auto& sock) {
               asio::async_read_until(
                  *sock, *buffer, "\r\n",
                  [this, socket_var, buffer, body = std::move(body), url, use_https, status_code,
                   response_headers = std::move(response_headers), handler = std::forward<CompletionHandler>(handler)](
                     asio::error_code ec, std::size_t bytes_transferred) mutable {
                     if (ec) {
                        // Body is complete; deliver what we have but don't pool the connection
                        response resp;
                        resp.status_code = status_code;
                        resp.response_headers = std::move(response_headers);
                        resp.response_body = std::move(*body);
                        handler(std::move(resp));
                        return;
                     }

                     bool empty_line = (bytes_transferred == 2); // just "\r\n"
                     buffer->consume(bytes_transferred);

                     if (empty_line) {
                        // End of trailers — build response
                        response resp;
                        resp.status_code = status_code;
                        resp.response_headers = std::move(response_headers);
                        resp.response_body = std::move(*body);

                        auto connection_header = resp.response_headers.find("connection");
                        if (connection_header == resp.response_headers.end() ||
                            connection_header->second.find("close") == std::string::npos) {
                           connection_pool->return_connection(url.host, url.port, use_https, std::move(*socket_var));
                        }

                        handler(std::move(resp));
                     }
                     else {
                        // Trailer line — skip it and read the next line
                        async_consume_trailers(socket_var, buffer, std::move(body), url, use_https, status_code,
                                               std::move(response_headers), std::move(handler));
                     }
                  });
            },
            *socket_var);
      }

      // Async chunked body reading: reads chunk size line, then chunk data, repeating until terminal chunk.
      template <typename CompletionHandler>
      void async_read_chunked_body(std::shared_ptr<socket_variant> socket_var, std::shared_ptr<asio::streambuf> buffer,
                                   std::shared_ptr<std::string> body, const url_parts& url, bool use_https,
                                   int status_code, std::unordered_map<std::string, std::string> response_headers,
                                   CompletionHandler&& handler)
      {
         // Read until we have the chunk size line
         std::visit(
            [&, this](auto& sock) {
               asio::async_read_until(
                  *sock, *buffer, "\r\n",
                  [this, socket_var, buffer, body, url, use_https, status_code,
                   response_headers = std::move(response_headers), handler = std::forward<CompletionHandler>(handler)](
                     asio::error_code ec, std::size_t bytes_transferred) mutable {
                     if (ec) {
                        handler(std::unexpected(ec));
                        return;
                     }

                     // Parse chunk size (hex)
                     std::string_view size_line{static_cast<const char*>(buffer->data().data()),
                                                bytes_transferred - 2}; // exclude CRLF

                     // Ignore chunk extensions
                     auto semi_pos = size_line.find(';');
                     if (semi_pos != std::string_view::npos) {
                        size_line = size_line.substr(0, semi_pos);
                     }

                     size_t chunk_size = 0;
                     auto [ptr, parse_ec] =
                        std::from_chars(size_line.data(), size_line.data() + size_line.size(), chunk_size, 16);
                     if (parse_ec != std::errc{}) {
                        handler(std::unexpected(std::make_error_code(std::errc::protocol_error)));
                        return;
                     }

                     // Consume the chunk size line + CRLF
                     buffer->consume(bytes_transferred);

                     if (chunk_size == 0) {
                        // Terminal chunk — skip optional trailers and final CRLF (RFC 7230 §4.1),
                        // then build the response.
                        async_consume_trailers(socket_var, buffer, std::move(body), url, use_https, status_code,
                                               std::move(response_headers), std::move(handler));
                        return;
                     }

                     // Read chunk data + trailing CRLF
                     async_read_chunk_data(socket_var, buffer, body, chunk_size, url, use_https, status_code,
                                           std::move(response_headers), std::move(handler));
                  });
            },
            *socket_var);
      }

      template <typename CompletionHandler>
      void async_read_chunk_data(std::shared_ptr<socket_variant> socket_var, std::shared_ptr<asio::streambuf> buffer,
                                 std::shared_ptr<std::string> body, size_t chunk_size, const url_parts& url,
                                 bool use_https, int status_code,
                                 std::unordered_map<std::string, std::string> response_headers,
                                 CompletionHandler&& handler)
      {
         // Check accumulated body size against limit before reading chunk data
         if (max_response_body_size_ > 0 && body->size() + chunk_size > max_response_body_size_) {
            detail::close_socket(*socket_var, connection_pool->graceful_ssl_shutdown());
            handler(std::unexpected(make_error_code(http_client_error::response_too_large)));
            return;
         }

         // A chunk size that leaves no room for the trailing CRLF wraps chunk_size + 2 below
         // and appends with a bogus length, so reject it as malformed.
         if (chunk_size > (std::numeric_limits<size_t>::max)() - 2) {
            detail::close_socket(*socket_var, connection_pool->graceful_ssl_shutdown());
            handler(std::unexpected(std::make_error_code(std::errc::protocol_error)));
            return;
         }

         size_t total_needed = chunk_size + 2; // chunk data + trailing CRLF

         if (buffer->size() >= total_needed) {
            // Data already in buffer
            const char* chunk_data = static_cast<const char*>(buffer->data().data());
            body->append(chunk_data, chunk_size);
            buffer->consume(total_needed);

            // Continue reading next chunk
            async_read_chunked_body(socket_var, buffer, body, url, use_https, status_code, std::move(response_headers),
                                    std::forward<CompletionHandler>(handler));
            return;
         }

         size_t to_read = total_needed - buffer->size();
         std::visit(
            [&, this](auto& sock) {
               asio::async_read(
                  *sock, *buffer, asio::transfer_exactly(to_read),
                  [this, socket_var, buffer, body, chunk_size, total_needed, url, use_https, status_code,
                   response_headers = std::move(response_headers),
                   handler = std::forward<CompletionHandler>(handler)](asio::error_code ec, std::size_t) mutable {
                     if (ec) {
                        handler(std::unexpected(ec));
                        return;
                     }

                     const char* chunk_data = static_cast<const char*>(buffer->data().data());
                     body->append(chunk_data, chunk_size);
                     buffer->consume(total_needed);

                     // Continue reading next chunk
                     async_read_chunked_body(socket_var, buffer, body, url, use_https, status_code,
                                             std::move(response_headers), std::move(handler));
                  });
            },
            *socket_var);
      }

      // Async EOF-delimited body reading: reads incrementally until connection close,
      // checking max_response_body_size_ after each read to prevent unbounded allocation.
      template <typename CompletionHandler>
      void async_read_eof_body(std::shared_ptr<socket_variant> socket_var, std::shared_ptr<asio::streambuf> buffer,
                               const url_parts& url, bool use_https, int status_code,
                               std::unordered_map<std::string, std::string> response_headers,
                               CompletionHandler&& handler)
      {
         std::visit(
            [&, this](auto& sock) {
               asio::async_read(
                  *sock, *buffer, asio::transfer_at_least(1),
                  [this, socket_var, buffer, url, use_https, status_code,
                   response_headers = std::move(response_headers),
                   handler = std::forward<CompletionHandler>(handler)](asio::error_code ec, std::size_t) mutable {
                     if (ec && ec != asio::error::eof) {
                        handler(std::unexpected(ec));
                        return;
                     }

                     if (max_response_body_size_ > 0 && buffer->size() > max_response_body_size_) {
                        detail::close_socket(*socket_var, connection_pool->graceful_ssl_shutdown());
                        handler(std::unexpected(make_error_code(http_client_error::response_too_large)));
                        return;
                     }

                     if (ec == asio::error::eof) {
                        // Connection closed — body is complete
                        std::string body(static_cast<const char*>(buffer->data().data()), buffer->size());

                        response resp;
                        resp.status_code = status_code;
                        resp.response_headers = std::move(response_headers);
                        resp.response_body = std::move(body);

                        handler(std::move(resp));
                        return;
                     }

                     // More data available — continue reading
                     async_read_eof_body(socket_var, buffer, url, use_https, status_code, std::move(response_headers),
                                         std::move(handler));
                  });
            },
            *socket_var);
      }

      template <typename CompletionHandler>
      void parse_and_read_body(std::shared_ptr<socket_variant> socket_var, std::shared_ptr<asio::streambuf> buffer,
                               size_t header_size, const url_parts& url, bool use_https, CompletionHandler&& handler)
      {
         std::string_view header_section{static_cast<const char*>(buffer->data().data()), header_size};

         // Parse the status line from the view.
         auto line_end = header_section.find("\r\n");
         if (line_end == std::string_view::npos) {
            handler(std::unexpected(std::make_error_code(std::errc::protocol_error)));
            return;
         }
         std::string_view status_line = header_section.substr(0, line_end);
         header_section.remove_prefix(line_end + 2); // Move the view past the status line and its CRLF

         auto parsed_status = parse_http_status_line(status_line);
         if (!parsed_status) {
            handler(std::unexpected(parsed_status.error()));
            return;
         }

         // Parse all header fields from the view.
         std::unordered_map<std::string, std::string> response_headers;
         size_t content_length = 0;
         bool has_content_length = false;
         bool connection_close = false;
         bool is_chunked = false;
         // The header section ends with an empty line ("\r\n"), which means our view will start with it.
         while (!header_section.starts_with("\r\n")) {
            line_end = header_section.find("\r\n");
            if (line_end == std::string_view::npos) {
               handler(std::unexpected(std::make_error_code(std::errc::protocol_error)));
               return;
            }
            std::string_view header_line = header_section.substr(0, line_end);
            header_section.remove_prefix(line_end + 2);

            auto colon_pos = header_line.find(':');
            if (colon_pos != std::string::npos) {
               std::string_view name = header_line.substr(0, colon_pos);
               // Skip past ':' and any leading whitespace on the value.
               size_t value_start = header_line.find_first_not_of(" \t", colon_pos + 1);
               std::string_view value = (value_start != std::string::npos) ? header_line.substr(value_start) : "";

               // A case-insensitive comparison is more robust for header names.
               if (name.size() == 14 && (name[0] == 'C' || name[0] == 'c') &&
                   glz::strncasecmp(name.data(), "Content-Length", 14) == 0) {
                  std::from_chars(value.data(), value.data() + value.size(), content_length);
                  has_content_length = true;
               }
               else if (name.size() == 17 && (name[0] == 'T' || name[0] == 't') &&
                        glz::strncasecmp(name.data(), "Transfer-Encoding", 17) == 0) {
                  if (value.find("chunked") != std::string_view::npos) {
                     is_chunked = true;
                  }
               }
               else if (name.size() == 10 && (name[0] == 'C' || name[0] == 'c') &&
                        glz::strncasecmp(name.data(), "Connection", 10) == 0) {
                  if (value.find("close") != std::string_view::npos) {
                     connection_close = true;
                  }
               }
               // Convert header name to lowercase for case-insensitive lookups (RFC 7230)
               response_headers.emplace(to_lower_case(name), value);
            }
         }

         // Consume the entire header block from the streambuf.
         // This efficiently discards the header data we've just parsed, leaving only body data.
         buffer->consume(header_size);

         // Reject responses that exceed the configured body size limit
         if (max_response_body_size_ > 0 && !is_chunked && content_length > max_response_body_size_) {
            detail::close_socket(*socket_var, connection_pool->graceful_ssl_shutdown());
            handler(std::unexpected(make_error_code(http_client_error::response_too_large)));
            return;
         }

         if (is_chunked) {
            auto body = std::make_shared<std::string>();
            async_read_chunked_body(socket_var, buffer, body, url, use_https, parsed_status->status_code,
                                    std::move(response_headers), std::forward<CompletionHandler>(handler));
         }
         else if (!has_content_length && connection_close) {
            // No Content-Length and not chunked: read until connection close (RFC 7230 §3.3.3)
            async_read_eof_body(socket_var, buffer, url, use_https, parsed_status->status_code,
                                std::move(response_headers), std::forward<CompletionHandler>(handler));
         }
         else {
            // Read the rest of the body based on content-length.
            size_t body_already_in_buffer = buffer->size();
            size_t remaining_to_read =
               (content_length > body_already_in_buffer) ? (content_length - body_already_in_buffer) : 0;

            std::visit(
               [&, this](auto& sock) {
                  asio::async_read(
                     *sock, *buffer, asio::transfer_exactly(remaining_to_read),
                     [this, socket_var, buffer, url, use_https, content_length,
                      status_code = parsed_status->status_code, response_headers = std::move(response_headers),
                      handler = std::forward<CompletionHandler>(handler)](asio::error_code ec, std::size_t) mutable {
                        // transfer_exactly only completes without error once the full
                        // Content-Length has been read, so any error here - including EOF - means
                        // the peer closed before delivering the declared body. That is an
                        // incomplete message (RFC 7230 §3.3.3), not a short success: close the
                        // connection (with graceful SSL shutdown if configured; never pool a socket
                        // the peer just closed) and surface the error. Matches the synchronous path.
                        if (ec) {
                           detail::close_socket(*socket_var, connection_pool->graceful_ssl_shutdown());
                           handler(std::unexpected(ec));
                           return;
                        }

                        // Full body received. A server can deliver more than content_length octets
                        // in the same segment (a pipelined next response or trailing junk); clamp
                        // so those bytes never leak into response_body. Matches the synchronous path.
                        std::string body(static_cast<const char*>(buffer->data().data()),
                                         (std::min)(content_length, buffer->size()));

                        response resp;
                        resp.status_code = status_code;
                        resp.response_headers = std::move(response_headers);
                        resp.response_body = std::move(body);

                        // Pool the connection unless the server asked to close it. A truncated or
                        // peer-closed connection already returned above, so anything reaching here
                        // is a complete response on a still-open socket.
                        auto connection_header = resp.response_headers.find("connection");
                        const bool server_wants_close = connection_header != resp.response_headers.end() &&
                                                        connection_header->second.find("close") != std::string::npos;
                        if (server_wants_close) {
                           detail::close_socket(*socket_var, connection_pool->graceful_ssl_shutdown());
                        }
                        else {
                           connection_pool->return_connection(url.host, url.port, use_https, std::move(*socket_var));
                        }

                        handler(std::move(resp));
                     });
               },
               *socket_var);
         }
      }
   };
}