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

#pragma once

#include <algorithm>
#include <array>
#include <atomic>
#include <cctype>
#include <charconv>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <expected>
#include <functional>
#include <glaze/glaze.hpp>
#include <iostream>
#include <mutex>
#include <optional>
#include <source_location>
#include <thread>
#include <unordered_map>

#include "glaze/ext/glaze_asio.hpp"
#include "glaze/net/cors.hpp"
#include "glaze/net/http.hpp"
#include "glaze/net/http_router.hpp"
#include "glaze/net/openapi.hpp"
#include "glaze/net/websocket_connection.hpp"
#include "glaze/util/compare.hpp"
#include "glaze/util/itoa.hpp"
#include "glaze/util/key_transformers.hpp"

// To deconflict Windows.h
#ifdef DELETE
#undef DELETE
#endif

namespace glz
{
   namespace detail
   {
      // Produces an RFC 7231 §7.1.1.1 IMF-fixdate formatted HTTP date string
      [[nodiscard]] inline std::string format_http_date(const std::chrono::sys_seconds time)
      {
         static constexpr std::array<std::string_view, 7> weekdays = {"Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"};
         static constexpr std::array<std::string_view, 12> months = {"Jan", "Feb", "Mar", "Apr", "May", "Jun",
                                                                     "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"};

         const auto day_point = std::chrono::floor<std::chrono::days>(time);
         const std::chrono::year_month_day date{day_point};
         const std::chrono::weekday weekday{day_point};
         const std::chrono::hh_mm_ss time_of_day{time - day_point};

         char buffer[64];
         const auto size = std::snprintf(
            buffer, sizeof(buffer), "%.*s, %02u %.*s %04d %02u:%02u:%02u GMT",
            int32_t(weekdays[weekday.c_encoding()].size()), weekdays[weekday.c_encoding()].data(), uint32_t(date.day()),
            int32_t(months[uint32_t(date.month()) - 1].size()), months[uint32_t(date.month()) - 1].data(),
            int32_t(date.year()), uint32_t(time_of_day.hours().count()), uint32_t(time_of_day.minutes().count()),
            uint32_t(time_of_day.seconds().count()));
         if (size < 0 || size_t(size) >= sizeof(buffer)) {
            return {};
         }
         return {buffer, size_t(size)};
      }

      // Type trait to detect SSL streams
      template <typename T>
      inline constexpr bool is_ssl_stream = false;

#ifdef GLZ_ENABLE_SSL
      template <typename T>
      inline constexpr bool is_ssl_stream<asio::ssl::stream<T>> = true;
#endif

      inline bool is_valid_ipv4_address(std::string_view address) noexcept
      {
         // RFC 3986, Section 3.2.2:
         // A host identified by an IPv4 literal address is represented in
         // dotted-decimal notation (a sequence of four decimal numbers in
         // the range 0 to 255, separated by ".")
         // IPv4address = dec-octet "." dec-octet "." dec-octet "." dec-octet

         constexpr size_t min_ipv4_length = 7; // "x.x.x.x"
         constexpr size_t max_ipv4_length = 15; // "xxx.xxx.xxx.xxx"

         if (address.size() < min_ipv4_length || address.size() > max_ipv4_length) {
            return false;
         }

         size_t pos = 0;

         // An IPv4 address contains exactly 4 octets separated by '.' (ignoring legacy forms)
         for (int octet_index = 0; octet_index < 4; ++octet_index) {
            if (pos == address.size()) {
               return false;
            }

            size_t octet_start = pos;
            uint32_t octet_value = 0;
            uint32_t digit_count = 0;

            while (pos < address.size() && address[pos] != '.') {
               char ch = address[pos];

               // IPv4 allows only numeric (0-9) and '.' chars
               if (ch < '0' || ch > '9') [[unlikely]] {
                  return false;
               }

               octet_value = (octet_value * 10) + static_cast<uint32_t>(ch - '0');
               ++digit_count;
               ++pos;

               // An octet cannot be longer than 3 digits (0-255)
               if (digit_count > 3) {
                  return false;
               }
            }

            // An octet should contain at least 1 digit
            if (digit_count == 0) {
               return false;
            }

            // IPv4 multi-digit octet cannot have leading zeros
            if (digit_count > 1 && address[octet_start] == '0') {
               return false;
            }

            // An octet must be in the range 0-255
            if (octet_value > 255) {
               return false;
            }

            if (octet_index < 3) {
               // Parsed fewer than 4 octets, but the address already ended
               if (pos == address.size()) {
                  return false;
               }

               if (address[pos] != '.') {
                  return false;
               }

               ++pos;
            }
            // All 4 octets have been parsed, so any remaining tail is invalid
            else if (pos != address.size()) {
               return false;
            }
         }

         return true;
      }

      inline bool is_ipv6_hex_digit(char ch) noexcept
      {
         return (ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f') || (ch >= 'A' && ch <= 'F');
      }

      inline bool count_ipv6_pieces(std::string_view text, size_t& piece_count) noexcept
      {
         piece_count = 0;

         // Empty side is valid after splitting around "::".
         // Examples: "::1", "1::", "::".
         if (text.empty()) {
            return true;
         }

         size_t pos = 0;

         while (pos < text.size()) {
            size_t digit_count = 0;

            // Parse one explicit IPv6 piece.
            // RFC 3986 grammar name:
            // h16 = 1*4HEXDIG
            while (pos < text.size() && text[pos] != ':') {
               char ch = text[pos];

               if (!is_ipv6_hex_digit(ch)) [[unlikely]] {
                  return false;
               }

               ++digit_count;
               ++pos;

               if (digit_count > 4) [[unlikely]] {
                  return false;
               }
            }

            if (digit_count == 0) [[unlikely]] {
               return false;
            }

            ++piece_count;

            if (pos == text.size()) {
               return true;
            }

            // Parser stopped before the end, so it must be at ':'
            if (text[pos] != ':') [[unlikely]] {
               return false;
            }

            ++pos;

            // After a single ':' there must be another explicit piece.
            // Valid "::" compression is handled before this function is called
            if (pos == text.size()) [[unlikely]] {
               return false;
            }
         }

         return true;
      }

      inline bool is_valid_ipv6_address(std::string_view address) noexcept
      {
         constexpr size_t min_ipv6_length = 2; // "::"
         constexpr size_t max_ipv6_length = 45; // "xxxx:xxxx:xxxx:xxxx:xxxx:xxxx:255.255.255.255"

         if (address.size() < min_ipv6_length || address.size() > max_ipv6_length) {
            return false;
         }

         std::string_view ipv6_part = address;
         size_t ipv4_tail_piece_count = 0;

         // RFC 3986 describes IPv6 format in Section 3.2.2, stating that the last
         // 32 bits of an IPv6 address (2 last parts) can be an IPv4 address
         bool has_ipv4_tail = address.find('.') != std::string_view::npos;
         if (has_ipv4_tail) {
            // Find the last IPv6 separator before the IPv4 tail
            size_t tail_separator_pos = address.rfind(':');
            if (tail_separator_pos == std::string_view::npos) {
               return false;
            }

            // Assume that everything after the last IPv6 separator is the
            // IPv4 tail, the IPv4 validator will catch any issues
            std::string_view ipv4_tail = address.substr(tail_separator_pos + 1); // +1 to skip ':'
            if (!is_valid_ipv4_address(ipv4_tail)) {
               return false;
            }

            size_t ipv6_part_size = tail_separator_pos;

            // For cases such as "::xxx.x.x.x" we check if there is a compression
            // token and do not count it as a part of IPv6
            if (tail_separator_pos > 0 && address[tail_separator_pos - 1] == ':') {
               ipv6_part_size = tail_separator_pos + 1;
            }

            // When an IPv4 tail is present, it takes 2 last pieces of IPv6 address (32 bits)
            ipv4_tail_piece_count = 2;
            ipv6_part = address.substr(0, ipv6_part_size);
         }

         size_t compression_pos = ipv6_part.find("::");
         if (compression_pos != std::string_view::npos) {
            std::string_view left_part = ipv6_part.substr(0, compression_pos);
            std::string_view right_part = ipv6_part.substr(compression_pos + 2);

            // Only one compression token is allowed in an IPv6 literal.
            // Only the right side needs to be scanned because the first
            // compression token has already been found
            if (right_part.find("::") != std::string_view::npos) {
               return false;
            }

            size_t left_piece_count = 0;
            size_t right_piece_count = 0;

            if (!count_ipv6_pieces(left_part, left_piece_count)) {
               return false;
            }

            if (!count_ipv6_pieces(right_part, right_piece_count)) {
               return false;
            }

            size_t piece_count = left_piece_count + right_piece_count + ipv4_tail_piece_count;

            // Address must not have 8 pieces if compression token is present
            return piece_count < 8;
         }

         // No compression token present, just count all of the IPv6 pieces
         size_t piece_count = 0;
         if (!count_ipv6_pieces(ipv6_part, piece_count)) {
            return false;
         }

         // IPv6 pieces + optional 2 pieces from IPv4 tail
         return piece_count + ipv4_tail_piece_count == 8;
      }

      inline bool is_ipv4_address_like(std::string_view text) noexcept
      {
         if (text.empty()) {
            return false;
         }

         bool has_ipv4_separator = false;

         for (char ch : text) {
            if (ch == '.') {
               has_ipv4_separator = true;
               continue;
            }

            // IPv4 address can consist only of digits and '.'
            bool is_digit = ch >= '0' && ch <= '9';
            if (!is_digit) {
               return false;
            }
         }

         return has_ipv4_separator;
      }

      inline bool is_valid_hostname(std::string_view hostname) noexcept
      {
         // The current implementation accepts strict ASCII DNS-like hostnames, allows
         // numeric-only DNS labels, and additionally permits the underscore '_',
         // while rejecting raw Unicode and percent-encoding.

         // The underscore is not a valid DNS label character under RFC 1035, but it
         // appears routinely in real-world Host values: internal service names
         // (e.g. "my_service.internal") and underscore-prefixed labels such as "_dmarc"
         // or the SRV form "_sip._tcp". Common servers (nginx, Go net/http, Node.js)
         // tolerate it, and since this server does not route on the Host value, rejecting
         // '_' is pure interop cost with no security benefit (it is inert as a delimiter).
         // It is therefore treated as an ordinary label character valid at any position,
         // including leading and trailing.
         //
         // We deliberately do NOT extend this to '~', the other character in RFC 3986's
         // reg-name "unreserved" set that is likewise not a DNS label character: unlike
         // '_', tilde does not occur in practice as a hostname character, so accepting it
         // would widen the input surface with no interop gain. Unreserved-set membership
         // alone is thus not the criterion here; real-world usage is.

         // Before implementing a percent-encoding parser, we should weigh the
         // pros and cons and accept the fact that it may create an additional
         // attack surface for HTTP server routing.

         // RFC 1035 limits a DNS name to 255 octets in DNS message form.
         // For dotted text form used in Host, the maximum length without
         // a trailing root dot is 253 characters:
         // 63 "." 63 "." 63 "." 61
         constexpr size_t max_host_name_length = 253;

         // RFC 1035, Section 2.3.1:
         // Labels must be 63 characters or less.
         constexpr size_t max_dns_label_length = 63;

         if (hostname.empty()) {
            return false;
         }

         // Accept and normalize absolute hostnames that end with '.'
         if (hostname.ends_with('.')) {
            hostname.remove_suffix(1);
            if (hostname.empty()) {
               return false;
            }
         }

         if (hostname.empty() || hostname.size() > max_host_name_length) {
            return false;
         }

         size_t dns_label_length = 0;
         char prev_char = '\0';

         for (char ch : hostname) {
            if (ch == '.') {
               // DNS label must not be empty
               if (dns_label_length == 0) {
                  return false;
               }

               // RFC 1035, Section 2.3.1:
               // They must ... end with a letter or digit...
               if (prev_char == '-') {
                  return false;
               }

               dns_label_length = 0;
               prev_char = ch;
               continue;
            }

            // RFC 1035, Section 2.3.1:
            // The labels must follow the rules for ARPANET host names. They must
            // start with a letter, end with a letter or digit, and have as
            // interior characters only letters, digits, and hyphen.
            // The underscore is permitted as a pragmatic superset (see note above);
            // unlike hyphen it carries no start/end position restriction.
            bool is_dns_label_char = (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') ||
                                     ch == '-' || ch == '_';
            if (!is_dns_label_char) {
               return false;
            }

            // RFC 1123, Section 2.1:
            // One aspect of host name syntax is hereby changed: the restriction on the
            // first character is relaxed to allow either a letter or a digit.
            if (dns_label_length == 0 && ch == '-') {
               return false;
            }

            ++dns_label_length;

            if (dns_label_length > max_dns_label_length) {
               return false;
            }

            prev_char = ch;
         }

         if (dns_label_length == 0) {
            return false;
         }

         // RFC 1035, Section 2.3.1:
         // They must ... end with a letter or digit...
         if (prev_char == '-') {
            return false;
         }

         return true;
      }

      inline bool is_valid_port(std::string_view port_str) noexcept
      {
         // RFC 3986, Section 3.2.3:
         // port = *DIGIT
         //
         // RFC 9110, Section 4.2.1 and Section 4.2.2:
         // If the port subcomponent is empty or not given, the default port is used.

         if (port_str.empty()) {
            return true;
         }

         constexpr uint32_t min_port = 1;
         constexpr uint32_t max_port = 65535;
         uint32_t port_value = 0;

         for (char ch : port_str) {
            if (ch < '0' || ch > '9') [[unlikely]] {
               return false;
            }

            port_value = (port_value * 10) + static_cast<uint32_t>(ch - '0');

            if (port_value > max_port) [[unlikely]] {
               return false;
            }
         }

         return port_value >= min_port;
      }

      // Deliberately stricter than RFC 3986 authority:
      //  - userinfo is rejected per RFC 9110, Section 4.2.4 (deprecated in http/https URIs)
      //  - IPvFuture literals are rejected by internal policy
      inline bool is_valid_authority(std::string_view authority) noexcept
      {
         // RFC 9110, Section 7.2:
         // Host = uri-host [ ":" port ]
         std::string_view uri_host = authority;

         // RFC 9110, Section 4.2.1 and Section 4.2.2:
         // A sender MUST NOT generate an "http" or "https" URI with an empty host
         // identifier. A recipient that processes such a URI reference MUST reject
         // it as invalid.
         if (authority.empty()) {
            return false;
         }

         // RFC 3986, Section 3.2.2:
         // host = IP-literal / IPv4address / reg-name
         // IP-literal = "[" ( IPv6address / IPvFuture  ) "]"

         // URI Host inside brackets must be interpreted as an IPv6 literal
         if (uri_host.starts_with('[')) {
            size_t closing_bracket_pos = uri_host.find(']');
            if (closing_bracket_pos == std::string_view::npos) {
               return false;
            }

            // Bracketed IPv6 literal must not be empty
            if (closing_bracket_pos == 1) {
               return false;
            }

            std::string_view bracketless_ipv6 = uri_host.substr(1, closing_bracket_pos - 1);
            if (!is_valid_ipv6_address(bracketless_ipv6)) {
               return false;
            }

            // No port is present and IPv6 is validated - host is valid
            if (closing_bracket_pos + 1 == uri_host.size()) {
               return true;
            }

            std::string_view tail_after_ipv6 = uri_host.substr(closing_bracket_pos + 1);

            // After the IPv6 brackets, only a port is allowed
            if (!tail_after_ipv6.starts_with(':')) {
               return false;
            }

            std::string_view port_str = tail_after_ipv6.substr(1);

            return is_valid_port(port_str);
         }

         // RFC 9110, Section 4.2.3:
         // If the port is equal to the default port for a scheme, the normal
         // form is to omit the port subcomponent.
         size_t port_delimiter_pos = authority.find(':');
         if (port_delimiter_pos != std::string_view::npos) {
            // Only one port delimiter is allowed
            if (authority.find(':', port_delimiter_pos + 1) != std::string_view::npos) {
               return false;
            }

            uri_host = authority.substr(0, port_delimiter_pos);
            std::string_view port_str = authority.substr(port_delimiter_pos + 1);

            if (uri_host.empty()) {
               return false;
            }

            if (!is_valid_port(port_str)) {
               return false;
            }
         }

         // RFC 3986, Section 3.2.2:
         // The syntax rule for host is ambiguous because it does not completely
         // distinguish between an IPv4address and a reg-name.
         // In order to disambiguate the syntax, we apply the "first-match-wins"
         // algorithm: If host matches the rule for IPv4address, then it should
         // be considered an IPv4 address literal and not a reg-name.
         if (is_valid_ipv4_address(uri_host)) {
            return true;
         }

         // A host that looks like an IPv4 address but failed IPv4 validation is invalid
         if (is_ipv4_address_like(uri_host)) {
            return false;
         }

         return is_valid_hostname(uri_host);
      }

      inline bool is_tchar(char ch) noexcept
      {
         // RFC 9110, Section 5.6.2:
         // token = 1*tchar
         // tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*"
         //       / "+" / "-" / "." / "^" / "_" / "`" / "|" / "~"
         //       / DIGIT / ALPHA
         return (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || (ch >= '0' && ch <= '9') || ch == '!' ||
                ch == '#' || ch == '$' || ch == '%' || ch == '&' || ch == '\'' || ch == '*' || ch == '+' || ch == '-' ||
                ch == '.' || ch == '^' || ch == '_' || ch == '`' || ch == '|' || ch == '~';
      }

      inline bool is_pct_encoded(std::string_view str) noexcept
      {
         // RFC 3986, Appendix A:
         // pct-encoded = "%" HEXDIG HEXDIG

         if (str.size() != 3) {
            return false;
         }

         auto is_pct_encoded_hex_digit = [](char ch) -> bool {
            return ('0' <= ch && ch <= '9') || ('A' <= ch && ch <= 'F') || ('a' <= ch && ch <= 'f');
         };

         return str.starts_with('%') && is_pct_encoded_hex_digit(str[1]) && is_pct_encoded_hex_digit(str[2]);
      }

      inline constexpr bool is_unreserved(char value) noexcept
      {
         // RFC 3986, Appendix A:
         // unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"
         return ('A' <= value && value <= 'Z') || ('a' <= value && value <= 'z') || ('0' <= value && value <= '9') ||
                value == '-' || value == '.' || value == '_' || value == '~';
      }

      inline constexpr bool is_sub_delim(char value) noexcept
      {
         // RFC 3986, Appendix A:
         // sub-delims = "!" / "$" / "&" / "'" / "(" / ")" / "*" / "+" / "," / ";" / "="
         return value == '!' || value == '$' || value == '&' || value == '\'' || value == '(' || value == ')' ||
                value == '*' || value == '+' || value == ',' || value == ';' || value == '=';
      }

      inline bool is_valid_absolute_path(std::string_view path) noexcept
      {
         // RFC 9112, Section 3.2.1:
         // absolute-path  = 1*( "/" segment )
         //
         // RFC 3986, Appendix A:
         // segment     = *pchar
         // pchar       = unreserved / pct-encoded / sub-delims / ":" / "@"
         // unreserved  = ALPHA / DIGIT / "-" / "." / "_" / "~"
         // pct-encoded = "%" HEXDIG HEXDIG
         // sub-delims  = "!" / "$" / "&" / "'" / "(" / ")"
         //             / "*" / "+" / "," / ";" / "="

         if (path.empty() || !path.starts_with('/')) {
            return false;
         }

         size_t pos = 0;
         while (pos != path.size()) {
            char ch = path[pos];

            // Skip segment delimiters
            if (ch == '/') {
               pos++;
               continue;
            }

            // Handle a percent-encoded token, which must be exactly 3 bytes
            if (ch == '%') {
               if (!is_pct_encoded(path.substr(pos, 3))) {
                  return false;
               }

               pos += 3;
               continue;
            }

            // Percent encoding has already been validated above, so all that
            // remains is to check whether the character is one of the allowed
            // pchar, unreserved, or sub-delimiter characters.
            if (ch != ':' && ch != '@' && !is_unreserved(ch) && !is_sub_delim(ch)) {
               return false;
            }

            pos++;
         }

         return true;
      }

      inline bool is_valid_uri_query(std::string_view query) noexcept
      {
         // RFC 3986, Appendix A:
         // query       = *( pchar / "/" / "?" )
         // pchar       = unreserved / pct-encoded / sub-delims / ":" / "@"
         // unreserved  = ALPHA / DIGIT / "-" / "." / "_" / "~"
         // pct-encoded = "%" HEXDIG HEXDIG
         // sub-delims  = "!" / "$" / "&" / "'" / "(" / ")"
         //             / "*" / "+" / "," / ";" / "="

         if (query.empty()) {
            return true;
         }

         size_t pos = 0;
         while (pos < query.size()) {
            char ch = query[pos];

            // Handle a percent-encoded token, which must be exactly 3 bytes
            if (ch == '%') {
               if (!is_pct_encoded(query.substr(pos, 3))) {
                  return false;
               }

               pos += 3;
               continue;
            }

            // Percent encoding has already been validated above, so all that
            // remains is to check whether the character is one of the allowed
            // query, pchar, unreserved, or sub-delimiter characters.
            if (ch != '/' && ch != '?' && ch != ':' && ch != '@' && !is_unreserved(ch) && !is_sub_delim(ch))
               [[unlikely]] {
               return false;
            }

            pos++;
         }

         return true;
      }

      inline bool is_valid_origin_form(std::string_view origin) noexcept
      {
         if (!origin.starts_with('/')) {
            return false;
         }

         if (origin.size() == 1) {
            return true; // "/" is a valid origin-form
         }

         std::string_view absolute_path = origin;
         std::string_view query;

         size_t query_separator_pos = origin.find('?', 1);
         bool has_query_params = query_separator_pos != std::string_view::npos;

         if (has_query_params) {
            absolute_path = origin.substr(0, query_separator_pos);
            query = origin.substr(query_separator_pos + 1);

            if (!is_valid_uri_query(query)) {
               return false;
            }
         }

         return is_valid_absolute_path(absolute_path);
      }

      inline bool is_valid_request_target(http_method method, std::string_view target) noexcept
      {
         // RFC 9112, Section 3.2:
         // There are four distinct formats for the request-target, depending on both
         // the method being requested and whether the request is to a proxy.
         // request-target = origin-form
         //                / absolute-form
         //                / authority-form
         //                / asterisk-form

         // We do not validate the authority-form, since this form is intended
         // exclusively for CONNECT requests, and Glaze does not currently
         // support this method. It is expected that the CONNECT method has been
         // rejected before this function is called.

         // RFC 3986 "scheme" + "://" separator
         static constexpr std::string_view http_scheme = "http://";
         static constexpr std::string_view https_scheme = "https://";

         // RFC 9112, Section 3.2.4:
         // When a client wishes to request OPTIONS for the server as a whole, as
         // opposed to a specific named resource of that server, the client MUST
         // send only "*" (%x2A) as the request-target.
         if (target == "*") {
            return method == http_method::OPTIONS ? true : false;
         }

         // RFC 9112, 9110 and 3986:
         // origin-form    = absolute-path [ "?" query ]
         // absolute-path  = 1*( "/" segment )
         // segment        = *pchar
         bool is_origin_form = target.starts_with('/');
         if (is_origin_form) {
            return is_valid_origin_form(target);
         }

         // RFC 9112, Section 3.2.2:
         // absolute-form = absolute-URI
         // ...
         // A server MUST accept the absolute-form in requests even though most
         // HTTP/1.1 clients will only send the absolute-form to a proxy.
         //
         // RFC 9110, Section 4.2.1 and Section 4.2.2:
         // http-URI  = "http"  "://" authority path-abempty [ "?" query ]
         // https-URI = "https" "://" authority path-abempty [ "?" query ]

         // RFC 3986, Section 3.1: An implementation should accept uppercase
         // letters as equivalent to lowercase in scheme names (e.g., allow
         // "HTTP" as well as "http") for the sake of robustness but should only
         // produce lowercase scheme names for consistency.

         // We need to determine exactly which prefix the string has if the
         // request target is specified in absolute-form, but we don't want to
         // validate the prefix again unnecessarily
         bool starts_with_http = glz::striequal(target.substr(0, http_scheme.size()), http_scheme);
         bool starts_with_https = false;
         if (!starts_with_http) {
            starts_with_https = glz::striequal(target.substr(0, https_scheme.size()), https_scheme);
         }

         bool is_absolute_form = starts_with_http || starts_with_https;
         if (is_absolute_form) {
            std::string_view scheme_prefix = starts_with_http ? http_scheme : https_scheme;
            target.remove_prefix(scheme_prefix.size());

            std::string_view authority = target;

            // RFC 3986, Section 3.2:
            // The authority component is preceded by a double slash ("//") and is
            // terminated by the next slash ("/"), question mark ("?"), or number
            // sign ("#") character, or by the end of the URI.
            //
            // '#' is deliberately not searched for here. RFC 3986, Section 4.3
            // defines absolute-URI = scheme ":" hier-part [ "?" query ], with
            // no fragment allowed, so a '#' anywhere in the target must fail
            // validation, which it does in the authority/path/query checks
            size_t authority_end = target.find_first_of("/?");
            bool ends_with_authority = authority_end == std::string_view::npos;

            // RFC 3986, Section 3.3:
            // path-abempty ... begins with "/" or is empty
            //
            // In cases such as "http://example.com" no path or query is present
            // and the entire remainder is the authority
            if (ends_with_authority) {
               return is_valid_authority(authority);
            }

            // Otherwise, a path or query is present, and we validate the authority first
            authority = target.substr(0, authority_end);
            if (!is_valid_authority(authority)) {
               return false;
            }

            target.remove_prefix(authority.size());

            // After the authority is validated and removed from the string, all
            // that should be left is the origin-form, but the authority can be
            // followed either by a path or a query, so we need to decide which
            // one to validate
            bool is_path_after_authority = target.starts_with('/');
            if (is_path_after_authority) {
               return is_valid_origin_form(target);
            }

            // If the authority is not followed by a path, it MUST be followed by a query
            return is_valid_uri_query(target);
         }

         return false;
      }

      inline std::expected<detail::request_line, int> parse_request_line(std::string_view request_line)
      {
         static constexpr std::string_view http_version_prefix = "HTTP/";

         // RFC 9112, Section 3:
         // request-line = method SP request-target SP HTTP-version

         if (request_line.empty()) {
            return std::unexpected(400);
         }

         size_t first_space = request_line.find(' ');
         bool starts_with_space = first_space == 0;

         if (first_space == std::string_view::npos || starts_with_space) {
            return std::unexpected(400);
         }

         std::string_view method_sv = request_line.substr(0, first_space);

         // from_string validates by matching against known methods; rejects all unknown input
         auto method_opt = glz::from_string(method_sv);
         if (!method_opt) {
            // RFC 9112:
            // method = token
            // token = 1*tchar
            if (!std::ranges::all_of(method_sv, is_tchar)) {
               return std::unexpected(400);
            }

            // RFC 9110, Section 15.6.2:
            // The 501 (Not Implemented) status code ... is the appropriate response
            // when the server does not recognize the request method...
            return std::unexpected(501);
         }

         size_t second_space = request_line.find(' ', first_space + 1);
         if (second_space == std::string_view::npos) {
            return std::unexpected(400);
         }

         // The RFC 9112, Section 3 grammar prohibits two consecutive spaces
         bool has_extra_space_after_method = second_space == first_space + 1;
         if (has_extra_space_after_method) {
            return std::unexpected(400);
         }

         std::string_view target = request_line.substr(first_space + 1, second_space - first_space - 1);
         if (target.empty()) {
            return std::unexpected(400);
         }

         if (!is_valid_request_target(*method_opt, target)) {
            return std::unexpected(400);
         }

         // The RFC 9112 grammar states that only 2 SP characters are allowed inside the request-line
         bool has_third_space = request_line.find(' ', second_space + 1) != std::string_view::npos;
         if (has_third_space) {
            return std::unexpected(400);
         }

         std::string_view http_version = request_line.substr(second_space + 1);
         if (!http_version.starts_with(http_version_prefix)) {
            return std::unexpected(400);
         }

         http_version.remove_prefix(http_version_prefix.size());

         // After the "HTTP/" prefix is removed, exactly 3 characters of the
         // version should remain
         if (http_version.size() != 3) {
            return std::unexpected(400);
         }

         // RFC 9112, Section 2.3:
         // HTTP-version = HTTP-name "/" DIGIT "." DIGIT
         char major_v = http_version[0];
         char version_separator = http_version[1];
         char minor_v = http_version[2];
         if (!glz::is_digit(major_v) || version_separator != '.' || !glz::is_digit(minor_v)) {
            return std::unexpected(400);
         }

         // RFC 9110, Section 6.2:
         // A recipient that receives a message with a major version number that it
         // implements and a minor version number higher than what it implements
         // SHOULD process the message as if it were in the highest minor version
         // within that major version to which the recipient is conformant.
         bool is_http11 = major_v == '1' && minor_v >= '1';

         return detail::request_line{
            .method = *method_opt,
            .target = target,
            .is_http11 = is_http11,
         };
      }
   }

   // Interface for streaming connections (type-erased for HTTP/HTTPS compatibility)
   struct streaming_connection_interface
   {
      using data_sent_handler = std::function<void(std::error_code)>;
      using disconnect_handler = std::function<void()>;

      virtual ~streaming_connection_interface() = default;

      // Send initial headers for streaming response
      virtual void send_headers(int status_code, const std::unordered_map<std::string, std::string>& headers = {},
                                data_sent_handler handler = {}) = 0;

      // Send a chunk of data
      virtual void send_chunk(std::string_view data, data_sent_handler handler = {}) = 0;

      // Send Server-Sent Event
      virtual void send_event(std::string_view event_type, std::string_view data, std::string_view id = {},
                              data_sent_handler handler = {}) = 0;

      // Close the streaming connection
      virtual void close(disconnect_handler handler = {}) = 0;

      // Set disconnect handler for client disconnection
      virtual void on_disconnect(disconnect_handler handler) = 0;

      // Check if connection is still alive
      virtual bool is_open() const = 0;

      // Get remote endpoint info
      virtual std::string remote_address() const = 0;
      virtual uint16_t remote_port() const = 0;

      // Check if headers have been sent
      virtual bool is_headers_sent() const = 0;

      // Get executor for async operations (timers, etc.)
      virtual asio::any_io_executor get_executor() const = 0;

      // Send JSON as Server-Sent Event (convenience method using send_event)
      template <class T>
      void send_json_event(const T& data, std::string_view event_type = "message", std::string_view id = {},
                           data_sent_handler handler = {})
      {
         std::string json_str;
         auto ec = glz::write_json(data, json_str);
         if (!ec) {
            send_event(event_type, json_str, id, handler);
         }
         else if (handler) {
            handler(std::make_error_code(std::errc::invalid_argument));
         }
      }
   };

   // Streaming connection handle for server-side streaming
   template <typename SocketType = asio::ip::tcp::socket>
   struct streaming_connection : public streaming_connection_interface,
                                 public std::enable_shared_from_this<streaming_connection<SocketType>>
   {
      using data_sent_handler = std::function<void(std::error_code)>;
      using disconnect_handler = std::function<void()>;

      streaming_connection(std::shared_ptr<SocketType> socket)
         : socket_(socket), is_headers_sent_(false), is_closed_(false)
      {}

      // Send initial headers for streaming response
      void send_headers(int status_code, const std::unordered_map<std::string, std::string>& headers = {},
                        data_sent_handler handler = {}) override
      {
         if (is_headers_sent_) return;
         is_headers_sent_ = true;

         std::string response_str;
         response_str.reserve(512);

         response_str.append("HTTP/1.1 ");
         response_str.append(std::to_string(status_code));
         response_str.append(" ");
         response_str.append(get_status_message(status_code));
         response_str.append("\r\n");

         // Add custom headers
         for (const auto& [name, value] : headers) {
            if (header_field_has_crlf(name, value)) [[unlikely]] {
               continue;
            }
            response_str.append(name);
            response_str.append(": ");
            response_str.append(value);
            response_str.append("\r\n");
         }

         // Default headers for streaming. A control header that was dropped above
         // for containing CR/LF must not suppress its auto-generated default: a
         // dropped Transfer-Encoding/Connection would otherwise leave the stream
         // unframed (no chunked framing, no keep-alive signal). Treat a
         // present-but-invalid header as absent, matching what was written.
         const auto present_and_valid = [&](const char* key) {
            const auto it = headers.find(key);
            return it != headers.end() && !header_field_has_crlf(it->first, it->second);
         };

         if (!present_and_valid("transfer-encoding")) {
            response_str.append("Transfer-Encoding: chunked\r\n");
            chunked_encoding_ = true;
         }
         if (!present_and_valid("connection")) {
            response_str.append("Connection: keep-alive\r\n");
         }
         if (!present_and_valid("cache-control")) {
            response_str.append("Cache-Control: no-cache\r\n");
         }

         response_str.append("\r\n");

         auto buffer = std::make_shared<std::string>(std::move(response_str));
         auto self = this->shared_from_this();

         asio::async_write(*socket_, asio::buffer(*buffer), [self, buffer, handler](std::error_code ec, std::size_t) {
            if (handler) handler(ec);
         });
      }

      // Send a chunk of data
      void send_chunk(std::string_view data, data_sent_handler handler = {}) override
      {
         if (is_closed_) return;

         auto self = this->shared_from_this();

         if (chunked_encoding_) {
            // Format as HTTP chunk
            std::string chunk;
            chunk.reserve(data.size() + 20); // 20 is a good estimate for hex size + CRLFs

            // Chunk size in hex, converted efficiently with std::to_chars
            std::array<char, 16> size_buf{}; // 64-bit size_t in hex is at most 16 chars
            if (auto [ptr, ec] = std::to_chars(size_buf.data(), size_buf.data() + size_buf.size(), data.size(), 16);
                ec == std::errc()) {
               chunk.append(std::string_view(size_buf.data(), ptr - size_buf.data()));
            }
            chunk.append("\r\n");
            chunk.append(data);
            chunk.append("\r\n");

            auto buffer = std::make_shared<std::string>(std::move(chunk));
            asio::async_write(*socket_, asio::buffer(*buffer),
                              [self, buffer, handler](std::error_code ec, std::size_t) {
                                 if (handler) handler(ec);
                              });
         }
         else {
            // Send raw data
            auto buffer = std::make_shared<std::string>(data);
            asio::async_write(*socket_, asio::buffer(*buffer),
                              [self, buffer, handler](std::error_code ec, std::size_t) {
                                 if (handler) handler(ec);
                              });
         }
      }

      // Send Server-Sent Event
      void send_event(std::string_view event_type, std::string_view data, std::string_view id = {},
                      data_sent_handler handler = {}) override
      {
         std::string sse_data;
         sse_data.reserve(data.size() + 50);

         if (!id.empty()) {
            sse_data.append("id: ");
            sse_data.append(id);
            sse_data.append("\n");
         }

         if (!event_type.empty()) {
            sse_data.append("event: ");
            sse_data.append(event_type);
            sse_data.append("\n");
         }

         sse_data.append("data: ");
         sse_data.append(data);
         sse_data.append("\n\n");

         send_chunk(sse_data, handler);
      }

      // Close the streaming connection
      void close(disconnect_handler handler = {}) override
      {
         if (is_closed_) return;
         is_closed_ = true;

         auto self = this->shared_from_this();

         if (chunked_encoding_) {
            // Send final chunk
            std::string final_chunk = "0\r\n\r\n";
            auto buffer = std::make_shared<std::string>(std::move(final_chunk));

            asio::async_write(*socket_, asio::buffer(*buffer), [self, buffer, handler](asio::error_code, std::size_t) {
               if (handler) handler();
               asio::error_code close_ec;
               self->socket_->lowest_layer().close(close_ec);
            });
         }
         else {
            if (handler) handler();
            asio::error_code ec;
            socket_->lowest_layer().close(ec);
         }
      }

      // Set disconnect handler for client disconnection
      void on_disconnect(disconnect_handler handler) override
      {
         disconnect_handler_ = handler;
         start_disconnect_detection();
      }

      // Check if connection is still alive
      bool is_open() const override { return socket_ && socket_->lowest_layer().is_open() && !is_closed_; }

      // Get remote endpoint info
      std::string remote_address() const override
      {
         if (socket_) {
            try {
               return socket_->lowest_layer().remote_endpoint().address().to_string();
            }
            catch (...) {
            }
         }
         return "";
      }

      uint16_t remote_port() const override
      {
         if (socket_) {
            try {
               return socket_->lowest_layer().remote_endpoint().port();
            }
            catch (...) {
            }
         }
         return 0;
      }

      bool is_headers_sent() const override { return is_headers_sent_; }

      asio::any_io_executor get_executor() const override { return socket_->lowest_layer().get_executor(); }

      std::shared_ptr<SocketType> socket_;

     private:
      disconnect_handler disconnect_handler_;
      bool is_headers_sent_;
      bool is_closed_;
      bool chunked_encoding_ = false;

      void start_disconnect_detection()
      {
         if (!socket_ || is_closed_) return;

         auto self = this->shared_from_this();

         // For SSL streams, we can't use async_receive with message_peek on lowest_layer()
         // because it returns basic_socket which doesn't support async_receive.
         // Use a timer-based check instead that verifies the socket is still open.
         // Trade-off: SSL disconnect detection has up to 1-second latency vs TCP's immediate
         // detection. This is acceptable for most streaming use cases where cleanup timing
         // is not critical. For applications requiring faster detection, consider implementing
         // application-level heartbeats.
         if constexpr (detail::is_ssl_stream<SocketType>) {
            auto timer = std::make_shared<asio::steady_timer>(socket_->lowest_layer().get_executor());
            timer->expires_after(std::chrono::seconds(1));
            timer->async_wait([self, timer](std::error_code ec) {
               if (ec) return; // Timer cancelled
               if (!self->socket_->lowest_layer().is_open()) {
                  if (self->disconnect_handler_) {
                     self->is_closed_ = true;
                     self->disconnect_handler_();
                  }
               }
               else if (!self->is_closed_) {
                  self->start_disconnect_detection();
               }
            });
         }
         else {
            // For TCP sockets, use async_receive with message_peek for more responsive detection
            auto buffer = std::make_shared<std::array<uint8_t, 1>>();
            socket_->async_receive(
               asio::buffer(*buffer), asio::socket_base::message_peek, [self, buffer](std::error_code ec, std::size_t) {
                  if (ec && self->disconnect_handler_) {
                     self->is_closed_ = true;
                     self->disconnect_handler_();
                  }
                  else if (!ec && !self->is_closed_) {
                     // Continue monitoring
                     auto timer = std::make_shared<asio::steady_timer>(self->socket_->lowest_layer().get_executor());
                     timer->expires_after(std::chrono::seconds(1));
                     timer->async_wait([self, timer](std::error_code) { self->start_disconnect_detection(); });
                  }
               });
         }
      }

      std::string_view get_status_message(int status_code)
      {
         switch (status_code) {
         case 200:
            return "OK";
         case 201:
            return "Created";
         case 204:
            return "No Content";
         case 400:
            return "Bad Request";
         case 401:
            return "Unauthorized";
         case 403:
            return "Forbidden";
         case 404:
            return "Not Found";
         case 500:
            return "Internal Server Error";
         default:
            return "Unknown";
         }
      }
   };

   // Enhanced response class with streaming support
   struct streaming_response
   {
      std::shared_ptr<streaming_connection_interface> stream;

      streaming_response(std::shared_ptr<streaming_connection_interface> conn) : stream(std::move(conn)) {}

      // Send headers and start streaming
      streaming_response& start_stream(int status_code = 200,
                                       const std::unordered_map<std::string, std::string>& headers = {})
      {
         if (stream) {
            stream->send_headers(status_code, headers);
         }
         return *this;
      }

      // Send a chunk of data
      streaming_response& send(std::string_view data)
      {
         if (stream) {
            stream->send_chunk(data);
         }
         return *this;
      }

      // Send JSON data
      template <class T>
      streaming_response& send_json(const T& data)
      {
         if (stream) {
            std::string json_str;
            auto ec = glz::write_json(data, json_str);
            if (!ec) {
               stream->send_chunk(json_str);
            }
         }
         return *this;
      }

      // Send Server-Sent Event
      streaming_response& send_event(std::string_view event_type, std::string_view data, std::string_view id = {})
      {
         if (stream) {
            stream->send_event(event_type, data, id);
         }
         return *this;
      }

      // Helper for SSE setup
      streaming_response& as_event_stream()
      {
         return start_stream(200, {{"Content-Type", "text/event-stream"},
                                   {"Cache-Control", "no-cache"},
                                   {"Access-Control-Allow-Origin", "*"}});
      }

      // Close the stream
      void close()
      {
         if (stream) {
            stream->close();
         }
      }
   };

   // streaming_handler is declared in glaze/net/http_router.hpp so the router can
   // store streaming routes uniformly with normal routes.

   // Wrapping middleware types
   // Non-copyable, non-movable wrapper to enforce synchronous execution
   // This prevents accidental storage and async invocation which would cause dangling references
   struct next_handler final
   {
      explicit next_handler(std::function<void()> fn) : fn_(std::move(fn)) {}

      // Prevent copying and moving to enforce synchronous usage
      next_handler(const next_handler&) = delete;
      next_handler(next_handler&&) = delete;
      next_handler& operator=(const next_handler&) = delete;
      next_handler& operator=(next_handler&&) = delete;

      void operator()() const { fn_(); }

     private:
      std::function<void()> fn_;
   };

   // C++20 concept for wrapping middleware callables
   template <typename F>
   concept wrapping_middleware_callable = requires(F&& f, const request& req, response& res, const next_handler& next) {
      { std::forward<F>(f)(req, res, next) } -> std::same_as<void>;
   };

   using wrapping_middleware = std::function<void(const request&, response&, const next_handler&)>;

   // Request/Response lifecycle hooks for metrics and logging
   using request_hook = std::function<void(const request&, response&)>;
   using response_hook = std::function<void(const request&, const response&)>;

   /**
    * @brief Configuration for HTTP connection behavior
    *
    * Controls keep-alive (persistent connection) settings for the HTTP server.
    * HTTP/1.1 defaults to keep-alive; this configuration allows tuning or disabling it.
    */
   struct connection_config
   {
      /**
       * @brief Enable HTTP/1.1 persistent connections (keep-alive)
       *
       * When enabled, the server will reuse connections for multiple requests.
       * When disabled, each request/response closes the connection.
       * Default: true (HTTP/1.1 compliant behavior)
       */
      bool keep_alive = true;

      /**
       * @brief Timeout for idle connections in seconds
       *
       * After sending a response, if no new request arrives within this timeout,
       * the connection is closed. Set to 0 to disable timeout (not recommended).
       * Default: 60 seconds
       */
      uint32_t keep_alive_timeout = 60;

      /**
       * @brief Maximum requests allowed per connection
       *
       * After this many requests on a single connection, the server sends
       * Connection: close and closes the connection. Set to 0 for unlimited.
       * Default: 0 (unlimited)
       */
      uint32_t max_requests_per_connection = 0;

      /**
       * @brief Maximum allowed request body size in bytes
       *
       * Requests whose Content-Length exceeds this limit are rejected with
       * HTTP 413 (Payload Too Large) before the body is allocated or read.
       * Only applies to Content-Length-based requests (the server does not
       * currently support chunked request Transfer-Encoding).
       * Set to 0 for no limit.
       * Default: 100 MB
       */
      size_t max_request_body_size = http_default_max_body_size;
   };

   // Server implementation using non-blocking asio with WebSocket support
   template <bool EnableTLS = false>
   struct http_server
   {
      // Socket type abstraction
      using socket_type = std::conditional_t<EnableTLS,
#ifdef GLZ_ENABLE_SSL
                                             asio::ssl::stream<asio::ip::tcp::socket>,
#else
                                             asio::ip::tcp::socket,
#endif
                                             asio::ip::tcp::socket>;

      /**
       * @brief Construct an HTTP server
       *
       * @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().
       *       The caller is responsible for running the executor (e.g., calling io_context::run()
       *       from their own threads).
       *
       * @param custom_error_handler Optional error handler for server errors.
       *                             If not provided, uses a default handler that prints
       *                             to stderr.
       *
       * Example with internal io_context (default):
       * @code
       * glz::http_server server;
       * server.bind(8080);
       * server.start();  // Creates worker threads using hardware_concurrency()
       * @endcode
       *
       * Example with external io_context:
       * @code
       * auto io_ctx = std::make_shared<asio::io_context>();
       * glz::http_server server(io_ctx->get_executor());
       * server.bind(8080);
       * server.start(0);  // Pass 0 to skip creating worker threads
       *
       * // Run io_context in your own thread(s)
       * std::thread io_thread([io_ctx]() {
       *    io_ctx->run();
       * });
       *
       * // ... later ...
       * server.stop();
       * io_ctx->stop();
       * io_thread.join();
       * @endcode
       *
       * Example with custom error handler:
       * @code
       * auto error_handler = [](std::error_code ec, std::source_location loc) {
       *    my_logger.error("Server error at {}:{}: {}",
       *                    loc.file_name(), loc.line(), ec.message());
       * };
       * glz::http_server server({}, error_handler);
       * @endcode
       */
      inline http_server(asio::any_io_executor executor = asio::any_io_executor(),
                         glz::error_handler custom_error_handler = {})
         : io_executor(executor)
      {
         if (!io_executor) {
            io_context = std::make_shared<asio::io_context>();
            io_executor = io_context->get_executor();
         }
         if (custom_error_handler) {
            error_handler = std::move(custom_error_handler);
         }
         else {
            error_handler = [](std::error_code ec, std::source_location loc) {
               std::fprintf(stderr, "Error at %s:%d: %s\n", loc.file_name(), static_cast<int>(loc.line()),
                            ec.message().c_str());
            };
         }

         // Initialize SSL context for TLS-enabled servers
         if constexpr (EnableTLS) {
#ifdef GLZ_ENABLE_SSL
            ssl_context = std::make_unique<asio::ssl::context>(asio::ssl::context::tlsv12);
            ssl_context->set_default_verify_paths();
#else
            static_assert(!EnableTLS, "TLS support requires GLZ_ENABLE_SSL to be defined and OpenSSL to be available");
#endif
         }
      }

      /**
       * @brief Construct an HTTP server (Legacy Constructor, construct using external io_context)
       *
       * @param context Optional external io_context to use. If nullptr, creates its own.
       *                Using an external io_context allows sharing the event loop with
       *                other asio-based components or managing the io_context lifecycle
       *                externally.
       * @param custom_error_handler Optional error handler for server errors.
       *                             If not provided, uses a default handler that prints
       *                             to stderr.
       *
       * Example with internal io_context (default):
       * @code
       * glz::http_server server;
       * server.bind(8080);
       * server.start();  // Creates worker threads using hardware_concurrency()
       * @endcode
       *
       * Example with external io_context:
       * @code
       * auto io_ctx = std::make_shared<asio::io_context>();
       * glz::http_server server(io_ctx);
       * server.bind(8080);
       * server.start(0);  // Pass 0 to skip creating worker threads
       *
       * // Run io_context in your own thread(s)
       * std::thread io_thread([io_ctx]() {
       *    io_ctx->run();
       * });
       *
       * // ... later ...
       * server.stop();
       * io_ctx->stop();
       * io_thread.join();
       * @endcode
       *
       * Example with custom error handler:
       * @code
       * auto error_handler = [](std::error_code ec, std::source_location loc) {
       *    my_logger.error("Server error at {}:{}: {}",
       *                    loc.file_name(), loc.line(), ec.message());
       * };
       * glz::http_server server(nullptr, error_handler);
       * @endcode
       */
      inline http_server(std::shared_ptr<asio::io_context> context, glz::error_handler custom_error_handler = {})
         : http_server(context ? context->get_executor() : asio::any_io_executor(), custom_error_handler)
      {
         // Preserve shared ownership to prevent use-after-free. The delegated constructor
         // only stores io_context when no executor is provided, but the executor alone
         // doesn't keep the io_context alive.
         if (context) {
            io_context = context;
         }
      }

      inline ~http_server()
      {
         if (running) {
            stop();
         }
         // Clean up any remaining threads
         for (auto& thread : threads) {
            if (thread.joinable()) {
               thread.join();
            }
         }
         threads.clear();

         // Destruction sequence (reverse declaration order of http_server members):
         //   1. threads are joined above (worker threads stop running io_context).
         //   2. root_router is destroyed below, which releases the websocket_server
         //      shared_ptrs it owns; ~websocket_server calls close_all_connections()
         //      and force-closes the sockets.
         //   3. io_context is destroyed last.
         // The invariant that matters: sockets must be deregistered from the reactor
         // before io_context dies. Because root_router (declared after io_context)
         // is destroyed first, that holds.
      }

      inline http_server& bind(std::string_view address, uint16_t port)
      {
         try {
            asio::ip::tcp::endpoint endpoint(asio::ip::make_address(address), port);
            acceptor = std::make_unique<asio::ip::tcp::acceptor>(io_executor, endpoint);
         }
         catch (...) {
            error_handler(std::make_error_code(std::errc::address_in_use), std::source_location::current());
         }
         return *this;
      }

      inline http_server& bind(uint16_t port) { return bind("0.0.0.0", port); }

      inline uint16_t port(asio::error_code& ec) const
      {
         if (!acceptor) {
            ec = asio::error::not_connected;
            return 0;
         }
         auto endpoint = acceptor->local_endpoint(ec);
         if (ec) {
            return 0;
         }
         return endpoint.port();
      }

      inline uint16_t port() const
      {
         if (!acceptor) {
            throw std::runtime_error("Server not bound");
         }
         return acceptor->local_endpoint().port();
      }

      /**
       * @brief Start the HTTP server
       *
       * @param num_threads Number of worker threads to create:
       *                    - std::nullopt (default): Uses std::thread::hardware_concurrency(),
       *                      with a minimum of 1 thread if hardware_concurrency() returns 0
       *                    - 0: No worker threads created (caller manages io_context::run())
       *                    - N: Creates exactly N worker threads
       *
       * When using an external io_context, pass 0 to avoid creating threads,
       * then call io_context::run() in your own thread(s).
       *
       * Example with default behavior:
       * @code
       * glz::http_server server;
       * server.bind(8080);
       * server.start();  // Uses hardware_concurrency() threads
       * @endcode
       *
       * Example with specific thread count:
       * @code
       * glz::http_server server;
       * server.bind(8080);
       * server.start(4);  // Creates exactly 4 worker threads
       * @endcode
       *
       * Example with external io_context:
       * @code
       * auto io_ctx = std::make_shared<asio::io_context>();
       * glz::http_server server(io_ctx);
       * server.bind(8080);
       * server.start(0);  // No worker threads - caller manages io_context
       *
       * // Run io_context in your own thread
       * std::thread io_thread([io_ctx]() {
       *    io_ctx->run();
       * });
       *
       * // ... later ...
       * server.stop();
       * io_ctx->stop();
       * io_thread.join();
       * @endcode
       */
      inline void start(std::optional<size_t> num_threads = std::nullopt)
      {
         if (running || !acceptor) {
            return;
         }

         running = true;

         // Determine number of threads
         size_t actual_threads;
         if (num_threads.has_value()) {
            // Use the explicitly specified value (including 0)
            actual_threads = *num_threads;
         }
         else {
            // Use hardware concurrency, but ensure at least 1 thread
            actual_threads = std::thread::hardware_concurrency();
            if (actual_threads == 0) {
               actual_threads = 1;
            }
         }

         // Start the acceptor
         do_accept();
         // Start worker threads (unless explicitly set to 0)
         if (io_context && actual_threads > 0) {
            threads.reserve(actual_threads);
            for (size_t i = 0; i < actual_threads; ++i) {
               threads.emplace_back([this] {
                  io_context->run();
                  // Don't report errors during shutdown
               });
            }
         }
      }

      inline void stop()
      {
         {
            std::lock_guard<std::mutex> lock(shutdown_mutex);
            if (!running) {
               return;
            }
            running = false;
         }

         // Stop accepting new connections
         if (acceptor) {
            try {
               acceptor->close();
            }
            catch (...) {
               // Ignore errors on close
            }
         }

         // Cancel the signal handler
         if (signals_) {
            try {
               signals_->cancel();
            }
            catch (...) {
               // Ignore errors on cancel
            }
         }

         // Stop the io_context
         if (io_context) io_context->stop();

         // Only join threads if we're not in one of the worker threads
         auto current_thread_id = std::this_thread::get_id();
         bool is_worker_thread = false;
         for (const auto& thread : threads) {
            if (thread.get_id() == current_thread_id) {
               is_worker_thread = true;
               break;
            }
         }

         if (!is_worker_thread) {
            for (auto& thread : threads) {
               if (thread.joinable()) {
                  thread.join();
               }
            }
            threads.clear();
         }

         // Notify any threads waiting for shutdown
         shutdown_cv.notify_all();
      }

      inline http_server& mount(std::string_view base_path, const http_router& router)
      {
         auto join = [&](const std::string& path) {
            std::string full_path = std::string(base_path);
            if (!full_path.empty() && full_path.back() == '/') {
               full_path.pop_back();
            }
            full_path += path;
            return full_path;
         };

         // Mount normal routes
         for (const auto& [path, method_handlers] : router.normal_routes.routes) {
            const auto full_path = join(path);
            for (const auto& [method, entry] : method_handlers) {
               root_router.route(method, full_path, entry.handle, entry.spec);
            }
         }

         // Mount streaming routes
         for (const auto& [path, method_handlers] : router.streaming_routes.routes) {
            const auto full_path = join(path);
            for (const auto& [method, entry] : method_handlers) {
               root_router.stream(method, full_path, entry.handle, entry.spec);
            }
         }

         // Mount WebSocket routes. Entries are keyed by http_method::GET (the
         // upgrade method), so look that slot up directly rather than iterating.
         for (const auto& [path, method_handlers] : router.websocket_routes.routes) {
            if (auto it = method_handlers.find(http_method::GET); it != method_handlers.end()) {
               root_router.websocket(join(path), it->second.handle, it->second.spec);
            }
         }

         // Add middleware
         for (const auto& middleware : router.middlewares) {
            root_router.use(middleware);
         }

         return *this;
      }

      inline http_server& use(handler middleware)
      {
         root_router.use(std::move(middleware));
         return *this;
      }

      /**
       * @brief Register wrapping middleware that executes around route handlers
       *
       * Wrapping middleware can execute code before AND after the handler by wrapping
       * the next() function. This enables natural timing measurements, response
       * transformation, and comprehensive error handling.
       *
       * Use cases:
       * - Request/response timing and metrics
       * - Logging with full context (including response status and timing)
       * - Response transformation (compression, encryption, header injection)
       * - Error handling around handlers (try-catch wrapping)
       * - Any cross-cutting concerns that need both before and after logic
       *
       * CRITICAL SAFETY REQUIREMENT:
       * ============================
       * The next() handler MUST be called synchronously within the middleware function.
       * The next_handler type is intentionally non-copyable and non-movable to prevent
       * accidental storage and asynchronous invocation.
       *
       * Storing next() or calling it asynchronously will result in:
       * - Compile-time errors (if you try to copy/move it)
       * - Undefined behavior due to dangling references (if you work around the safety)
       *
       * The request, response, and handler objects are only valid during the synchronous
       * execution of the middleware chain.
       *
       * CORRECT Usage:
       * ```cpp
       * server.wrap([](const request& req, response& res, auto next) {
       *     auto start = std::chrono::steady_clock::now();
       *     next(); // ✓ Synchronous call - correct
       *     auto duration = std::chrono::steady_clock::now() - start;
       *     std::cout << "Request took " << duration.count() << "ns\n";
       * });
       * ```
       *
       * INCORRECT Usage (will not compile):
       * ```cpp
       * server.wrap([](const request& req, response& res, auto next) {
       *     // ✗ COMPILE ERROR: next_handler is non-copyable
       *     std::thread([next]() {
       *         std::this_thread::sleep_for(std::chrono::seconds(1));
       *         next(); // Would cause dangling references
       *     }).detach();
       * });
       * ```
       *
       * For asynchronous operations, complete them BEFORE or AFTER calling next():
       * ```cpp
       * server.wrap([](const request& req, response& res, auto next) {
       *     // Async work BEFORE handler
       *     auto data = fetch_from_cache_async().get();
       *
       *     next(); // Synchronous handler execution
       *
       *     // Async work AFTER handler (fire and forget logging)
       *     std::async(std::launch::async, [status = res.status_code]() {
       *         log_to_remote_server(status);
       *     });
       * });
       * ```
       *
       * Execution order with multiple middleware:
       * ```cpp
       * server.wrap([](auto&, auto&, auto next) {
       *     std::cout << "A before\n";
       *     next();
       *     std::cout << "A after\n";
       * });
       * server.wrap([](auto&, auto&, auto next) {
       *     std::cout << "B before\n";
       *     next();
       *     std::cout << "B after\n";
       * });
       * // Output: A before -> B before -> handler -> B after -> A after
       * ```
       *
       * @tparam F Callable type (must satisfy wrapping_middleware_callable concept)
       * @param middleware The wrapping middleware function
       * @return Reference to this server for method chaining
       */
      template <wrapping_middleware_callable F>
      inline http_server& wrap(F&& middleware)
      {
         wrapping_middlewares_.push_back(std::forward<F>(middleware));
         return *this;
      }

      inline http_router& route(http_method method, std::string_view path, handler handle, const route_spec& spec = {})
      {
         return root_router.route(method, path, handle, spec);
      }

      inline http_router& get(std::string_view path, handler handle, const route_spec& spec = {})
      {
         return root_router.get(path, handle, spec);
      }

      inline http_router& post(std::string_view path, handler handle, const route_spec& spec = {})
      {
         return root_router.post(path, handle, spec);
      }

      inline http_router& put(std::string_view path, handler handle, const route_spec& spec = {})
      {
         return root_router.put(path, handle, spec);
      }

      inline http_router& del(std::string_view path, handler handle, const route_spec& spec = {})
      {
         return root_router.del(path, handle, spec);
      }

      inline http_router& patch(std::string_view path, handler handle, const route_spec& spec = {})
      {
         return root_router.patch(path, handle, spec);
      }

      // Register a streaming route. Delegates to root_router so streaming routes
      // share the radix-tree matcher with normal routes (and support ":param" paths).
      inline http_server& stream(http_method method, std::string_view path, streaming_handler handle,
                                 const route_spec& spec = {})
      {
         root_router.stream(method, path, std::move(handle), spec);
         return *this;
      }

      // Convenience methods for streaming
      inline http_server& stream_get(std::string_view path, streaming_handler handle, const route_spec& spec = {})
      {
         return stream(http_method::GET, path, std::move(handle), spec);
      }

      inline http_server& stream_post(std::string_view path, streaming_handler handle, const route_spec& spec = {})
      {
         return stream(http_method::POST, path, std::move(handle), spec);
      }

      inline http_server& on_error(error_handler handle)
      {
         this->error_handler = std::move(handle);
         return *this;
      }

      /**
       * @brief Enable API inspection by exposing an OpenAPI 3.0 specification.
       *
       * This provides a standard, machine-readable way for clients to discover the API.
       *
       * @param path The path to expose the OpenAPI JSON specification on.
       * @param title The title of the API.
       * @param version The version of the API.
       * @return Reference to this server for method chaining.
       */
      http_server& enable_openapi_spec(std::string_view path = "/openapi.json",
                                       std::string_view title = "API Specification", std::string_view version = "1.0.0")
      {
         get(path, [this, title = std::string(title), version = std::string(version)](const request&, response& res) {
            open_api spec{};
            spec.info.title = title;
            spec.info.version = version;

            for (const auto& [route_path, method_handlers] : root_router.normal_routes.routes) {
               // Convert router path /:param to OpenAPI path /{param}
               std::string openapi_path = route_path;
               size_t pos = 0;
               while ((pos = openapi_path.find(':', pos)) != std::string::npos) {
                  openapi_path.replace(pos, 1, "{");
                  size_t end_pos = openapi_path.find('/', pos);
                  if (end_pos == std::string::npos) {
                     openapi_path.push_back('}');
                  }
                  else {
                     openapi_path.insert(end_pos, "}");
                  }
               }

               auto& path_item = spec.paths[openapi_path];

               for (const auto& [method, route_entry] : method_handlers) {
                  openapi_operation op;
                  op.summary = route_entry.spec.description;
                  if (!route_entry.spec.tags.empty()) {
                     op.tags = route_entry.spec.tags;
                  }
                  op.operationId = std::string(to_string(method)) + route_path;
                  op.responses["200"].description = "OK";

                  // Add request body schema
                  if (route_entry.spec.request_body_schema) {
                     openapi_request_body req_body;
                     req_body.required = true;
                     if (auto schema_val = glz::read_json<glz::schema>(*route_entry.spec.request_body_schema)) {
                        req_body.content["application/json"].schema = *schema_val;
                     }
                     op.requestBody = req_body;

                     // Add schema to components
                     if (!spec.components) spec.components.emplace();
                     if (!spec.components->schemas) spec.components->schemas.emplace();
                     if (auto schema_val = glz::read_json<glz::schema>(*route_entry.spec.request_body_schema)) {
                        spec.components->schemas->operator[](*route_entry.spec.request_body_type_name) = *schema_val;
                     }
                  }

                  // Add response schema
                  if (route_entry.spec.response_schema) {
                     openapi_response res_obj;
                     res_obj.description = "Successful response";
                     res_obj.content.emplace(); // Emplace the unordered_map
                     if (auto schema_val = glz::read_json<glz::schema>(*route_entry.spec.response_schema)) {
                        res_obj.content.value()["application/json"].schema = *schema_val;
                     }
                     op.responses["200"] = res_obj;

                     // Add schema to components
                     if (!spec.components) spec.components.emplace();
                     if (!spec.components->schemas) spec.components->schemas.emplace();
                     if (auto schema_val = glz::read_json<glz::schema>(*route_entry.spec.response_schema)) {
                        spec.components->schemas->operator[](*route_entry.spec.response_type_name) = *schema_val;
                     }
                  }

                  // Extract path parameters
                  auto segments = http_router::split_path(route_path);
                  for (const auto& segment : segments) {
                     if (!segment.empty() && segment.front() == ':') {
                        if (!op.parameters) op.parameters.emplace();
                        auto& param = op.parameters->emplace_back();
                        param.name = segment.substr(1);
                        param.in = "path";
                        param.required = true;
                        if (auto it = route_entry.spec.constraints.find(param.name);
                            it != route_entry.spec.constraints.end()) {
                           param.description = it->second.description;
                        }
                     }
                  }

                  switch (method) {
                  case http_method::GET:
                     path_item.get = op;
                     break;
                  case http_method::POST:
                     path_item.post = op;
                     break;
                  case http_method::PUT:
                     path_item.put = op;
                     break;
                  case http_method::DELETE:
                     path_item.del = op;
                     break;
                  case http_method::PATCH:
                     path_item.patch = op;
                     break;
                  default:
                     break;
                  }
               }
            }
            res.json(spec);
         });
         return *this;
      }

      /**
       * @brief Enable CORS with default configuration (allows all origins)
       *
       * This is suitable for development environments
       *
       * @return Reference to this server for method chaining
       */
      http_server& enable_cors()
      {
         root_router.use(glz::simple_cors());
         return *this;
      }

      /**
       * @brief Enable CORS with custom configuration
       *
       * @param config The CORS configuration to use
       * @return Reference to this server for method chaining
       */
      http_server& enable_cors(const glz::cors_config& config)
      {
         root_router.use(glz::create_cors_middleware(config));
         return *this;
      }

      /**
       * @brief Enable CORS for specific origins
       *
       * This is suitable for production environments where you want to restrict
       * which origins can access your API
       *
       * @param origins Vector of allowed origins (e.g., {"https://example.com", "https://app.example.com"})
       * @param allow_credentials Whether to allow credentials (cookies, auth headers)
       * @return Reference to this server for method chaining
       */
      http_server& enable_cors(const std::vector<std::string>& origins, bool allow_credentials = false)
      {
         root_router.use(glz::restrictive_cors(origins, allow_credentials));
         return *this;
      }

      /**
       * @brief Register a WebSocket handler for a specific path
       *
       * @param path The path to handle WebSocket connections on
       * @param server The WebSocket server instance to handle connections
       * @return Reference to this http_server for method chaining
       */
      inline http_server& websocket(std::string_view path, std::shared_ptr<websocket_server> server,
                                    const route_spec& spec = {})
      {
         root_router.websocket(path, std::move(server), spec);
         return *this;
      }

      /**
       * @brief Register a hook to be called when a request is received
       *
       * Request hooks are called after the request is fully parsed, before middleware
       * and route handlers are executed. This is useful for metrics tracking, logging,
       * or request preprocessing.
       *
       * For timing measurements, consider using wrapping middleware instead, which can
       * naturally measure duration by wrapping the next() function.
       *
       * @param hook The function to call for each request
       * @return Reference to this server for method chaining
       */
      inline http_server& on_request(request_hook hook)
      {
         request_hooks_.push_back(std::move(hook));
         return *this;
      }

      /**
       * @brief Register a hook to be called before a response is sent
       *
       * Response hooks are called after the route handler completes, but before the
       * response is sent to the client. This is useful for metrics collection,
       * response logging, or adding final response headers.
       *
       * For timing measurements, consider using wrapping middleware instead, which can
       * naturally measure duration by wrapping the next() function.
       *
       * @param hook The function to call before each response
       * @return Reference to this server for method chaining
       */
      inline http_server& on_response(response_hook hook)
      {
         response_hooks_.push_back(std::move(hook));
         return *this;
      }

      /**
       * @brief Load SSL certificate and private key for HTTPS servers
       *
       * @param cert_file Path to the certificate file (PEM format)
       * @param key_file Path to the private key file (PEM format)
       * @return Reference to this server for method chaining
       */
      inline http_server& load_certificate(const std::string& cert_file, const std::string& key_file)
      {
         if constexpr (EnableTLS) {
#ifdef GLZ_ENABLE_SSL
            ssl_context->use_certificate_chain_file(cert_file);
            ssl_context->use_private_key_file(key_file, asio::ssl::context::pem);
#endif
         }
         else {
            (void)cert_file;
            (void)key_file;
         }
         return *this;
      }

      /**
       * @brief Set SSL verification mode
       *
       * @param mode SSL verification mode
       * @return Reference to this server for method chaining
       */
      inline http_server& set_ssl_verify_mode([[maybe_unused]] int mode)
      {
         if constexpr (EnableTLS) {
#ifdef GLZ_ENABLE_SSL
            ssl_context->set_verify_mode(mode);
#endif
         }
         return *this;
      }

      /**
       * @brief Enable signal handling for graceful shutdown
       *
       * Registers signal handlers for SIGINT (Ctrl+C) and SIGTERM.
       * When these signals are received, the server will stop gracefully.
       *
       * @return Reference to this server for method chaining
       */
      inline http_server& with_signals()
      {
         signal_handling_enabled = true;

         // Create signal_set that will handle SIGINT and SIGTERM
         signals_ = std::make_unique<asio::signal_set>(io_executor, SIGINT, SIGTERM);

         // Set up async handler - this properly captures 'this' and integrates with ASIO
         signals_->async_wait([this](std::error_code ec, int signal_number) {
            if (!ec) {
               std::cout << "\nReceived signal " << signal_number << ", shutting down..." << std::endl;
               std::cout.flush();
               this->stop();
            }
         });

         return *this;
      }

      /**
       * @brief Configure connection behavior (keep-alive settings)
       *
       * @param config The connection configuration to use
       * @return Reference to this server for method chaining
       *
       * Example:
       * @code
       * server.connection_settings({
       *    .keep_alive = true,           // Enable persistent connections
       *    .keep_alive_timeout = 30,     // 30 second idle timeout
       *    .max_requests_per_connection = 100  // Close after 100 requests
       * });
       * @endcode
       */
      inline http_server& connection_settings(const connection_config& config)
      {
         conn_config_ = config;
         return *this;
      }

      /**
       * @brief Enable or disable HTTP keep-alive (persistent connections)
       *
       * When disabled, the server sends "Connection: close" with every response
       * and closes the connection after each request/response cycle.
       *
       * @param enable true to enable keep-alive (default), false to disable
       * @return Reference to this server for method chaining
       */
      inline http_server& keep_alive(bool enable)
      {
         conn_config_.keep_alive = enable;
         return *this;
      }

      /**
       * @brief Set the keep-alive timeout for idle connections
       *
       * @param timeout_seconds Seconds to wait for a new request before closing
       * @return Reference to this server for method chaining
       */
      inline http_server& keep_alive_timeout(uint32_t timeout_seconds)
      {
         conn_config_.keep_alive_timeout = timeout_seconds;
         return *this;
      }

      /**
       * @brief Set maximum requests per connection
       *
       * @param max_requests Maximum requests before closing (0 = unlimited)
       * @return Reference to this server for method chaining
       */
      inline http_server& max_requests_per_connection(uint32_t max_requests)
      {
         conn_config_.max_requests_per_connection = max_requests;
         return *this;
      }

      // Set maximum request body size in bytes (0 = unlimited).
      // Requests exceeding this are rejected with HTTP 413.
      // Must be configured before starting the server; not safe to change while serving.
      inline http_server& max_request_body_size(size_t max_size)
      {
         conn_config_.max_request_body_size = max_size;
         return *this;
      }

      size_t max_request_body_size() const { return conn_config_.max_request_body_size; }

      /**
       * @brief Wait for a shutdown signal
       *
       * This method blocks until the server is stopped, either by calling stop()
       * or by receiving a signal if signal handling is enabled with with_signals().
       *
       * @return void
       */
      inline void wait_for_signal()
      {
         std::unique_lock<std::mutex> lock(shutdown_mutex);
         shutdown_cv.wait(lock, [this] { return !running; });

         // After shutdown is signaled, wait for all threads to finish
         lock.unlock();
         for (auto& thread : threads) {
            if (thread.joinable()) {
               thread.join();
            }
         }
         threads.clear();
      }

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

     private:
      asio::any_io_executor io_executor;
      std::shared_ptr<asio::io_context> io_context;
      std::unique_ptr<asio::ip::tcp::acceptor> acceptor;
      std::vector<std::thread> threads;
      http_router root_router;
      std::atomic<bool> running{false};
      glz::error_handler error_handler;

      // Wrapping middleware (executes around handlers)
      std::vector<wrapping_middleware> wrapping_middlewares_;

      // Lifecycle hooks for metrics and logging
      std::vector<request_hook> request_hooks_;
      std::vector<response_hook> response_hooks_;

      // Connection configuration
      connection_config conn_config_;

      // Connection state for managing keep-alive connections
      // Uses socket_type which is either tcp::socket (HTTP) or ssl::stream<tcp::socket> (HTTPS)
      struct connection_state : public std::enable_shared_from_this<connection_state>
      {
         socket_type socket;
         std::string read_buf; // Flat read buffer (capacity reused across keep-alive)
         size_t buf_len = 0; // Valid data in read_buf[0..buf_len)
         size_t buf_consumed = 0; // Bytes consumed for current request (shifted on next request)
         request request_;
         asio::ip::tcp::endpoint remote_endpoint;
         asio::steady_timer idle_timer;
         uint32_t request_count = 0;
         bool should_close = false;
         response response_; // Persists across keep-alive requests to reuse capacity
         std::string header_buf; // Persists across keep-alive requests to reuse capacity

         connection_state(socket_type sock, asio::ip::tcp::endpoint endpoint)
            : socket(std::move(sock)),
              remote_endpoint(std::move(endpoint)),
              idle_timer(socket.lowest_layer().get_executor())
         {}
      };

      // Helper for executing wrapping middleware chain
      void execute_middleware_chain(size_t index, request& req, response& res, const handler& h)
      {
         if (index >= wrapping_middlewares_.size()) {
            h(req, res);
            return;
         }

         // Wrap the continuation in next_handler to enforce synchronous execution
         next_handler next([this, index, &req, &res, &h]() { execute_middleware_chain(index + 1, req, res, h); });

         wrapping_middlewares_[index](req, res, next);
      }

      // Signal handling members
      bool signal_handling_enabled = false;
      std::unique_ptr<asio::signal_set> signals_;
      std::condition_variable shutdown_cv;
      std::mutex shutdown_mutex;

#ifdef GLZ_ENABLE_SSL
      std::conditional_t<EnableTLS, std::unique_ptr<asio::ssl::context>, std::monostate> ssl_context;
#endif

      inline void do_accept()
      {
         acceptor->async_accept([this](std::error_code ec, asio::ip::tcp::socket socket) {
            if (!ec) {
               asio::error_code endpoint_ec;
               auto remote_endpoint = socket.remote_endpoint(endpoint_ec);
               if (endpoint_ec) {
                  error_handler(endpoint_ec, std::source_location::current());
               }
               else {
                  if constexpr (EnableTLS) {
#ifdef GLZ_ENABLE_SSL
                     // For HTTPS: create connection eagerly, then perform SSL handshake
                     auto conn = std::make_shared<connection_state>(socket_type(std::move(socket), *ssl_context),
                                                                    remote_endpoint);
                     conn->socket.async_handshake(asio::ssl::stream_base::server,
                                                  [this, conn](std::error_code handshake_ec) {
                                                     if (!handshake_ec) {
                                                        start_connection(conn);
                                                     }
                                                     else {
                                                        // SSL handshake failed - explicitly close the socket
                                                        error_handler(handshake_ec, std::source_location::current());
                                                        asio::error_code close_ec;
                                                        conn->socket.lowest_layer().close(close_ec);
                                                     }
                                                  });
#endif
                  }
                  else {
                     // For HTTP: use TCP socket directly
                     auto conn = std::make_shared<connection_state>(std::move(socket), remote_endpoint);
                     start_connection(conn);
                  }
               }
            }
            else if (running) {
               // Only report errors if we're still running (not shutting down)
               error_handler(ec, std::source_location::current());
            }

            // Continue accepting if still running
            if (running) {
               do_accept();
            }
         });
      }

      // Start handling a new connection
      inline void start_connection(std::shared_ptr<connection_state> conn) { read_request(conn); }

      // Start or reset the idle timer for keep-alive connections
      inline void start_idle_timer(std::shared_ptr<connection_state> conn)
      {
         if (conn_config_.keep_alive_timeout == 0) {
            return; // No timeout configured
         }

         conn->idle_timer.expires_after(std::chrono::seconds(conn_config_.keep_alive_timeout));
         conn->idle_timer.async_wait([conn](asio::error_code ec) {
            if (!ec) {
               // Timer expired - send FIN; socket closed via RAII when conn is destroyed
               asio::error_code close_ec;
               conn->socket.lowest_layer().shutdown(asio::ip::tcp::socket::shutdown_send, close_ec);
            }
            // If ec is operation_aborted, the timer was cancelled (new request arrived)
         });
      }

      // Cancel the idle timer (called when a new request starts)
      inline void cancel_idle_timer(std::shared_ptr<connection_state> conn) { conn->idle_timer.cancel(); }

      // Parse result from try_parse_request
      enum struct parse_status { complete, incomplete, error };
      struct parse_result
      {
         parse_status status = parse_status::incomplete;
         size_t headers_end = 0; // Byte offset past the final \r\n\r\n
         bool is_http_11 = false;
      };

      // Read the next HTTP request from the connection
      inline void read_request(std::shared_ptr<connection_state> conn)
      {
         // Start idle timer for keep-alive connections (except for first request)
         if (conn->request_count > 0 && conn_config_.keep_alive) {
            start_idle_timer(conn);
         }

         // Check for pipelined data already in buffer
         if (conn->buf_len > 0) {
            auto result = try_parse_request(conn);
            if (result.status == parse_status::complete) {
               cancel_idle_timer(conn);
               conn->request_count++;
               finish_request(conn, result);
               return;
            }
            else if (result.status == parse_status::error) {
               cancel_idle_timer(conn);
               return; // error response already sent by try_parse_request
            }
         }

         do_read_headers(conn);
      }

      // Read headers using async_read_until with dynamic_buffer wrapping our flat string
      inline void do_read_headers(std::shared_ptr<connection_state> conn)
      {
         // Trim read_buf to buf_len so dynamic_buffer starts from the right position
         conn->read_buf.resize(conn->buf_len);

         asio::async_read_until(conn->socket, asio::dynamic_buffer(conn->read_buf), "\r\n\r\n",
                                [this, conn](asio::error_code ec, std::size_t /*bytes_transferred*/) {
                                   cancel_idle_timer(conn);

                                   if (ec) {
                                      if (ec != asio::error::eof && ec != asio::error::operation_aborted) {
                                         error_handler(ec, std::source_location::current());
                                      }
                                      return;
                                   }

                                   // dynamic_buffer grew read_buf; update buf_len to match
                                   conn->buf_len = conn->read_buf.size();

                                   auto result = try_parse_request(conn);

                                   if (result.status == parse_status::complete) {
                                      conn->request_count++;

                                      finish_request(conn, result);
                                   }
                                   else if (result.status == parse_status::error) {
                                      // error response already sent by try_parse_request
                                   }
                                   else {
                                      // incomplete — shouldn't happen since async_read_until found \r\n\r\n
                                      send_error_response_with_close(conn, 400, "Bad Request");
                                   }
                                });
      }

      // Single-pass HTTP request parser: scans data line by line, parsing semantics as it goes.
      // Returns complete when all headers are found, incomplete if more data needed.
      inline parse_result try_parse_request(std::shared_ptr<connection_state> conn)
      {
         std::string_view data(conn->read_buf.data(), conn->buf_len);
         parse_result result;

         // Clear headers for re-parse safety (preserves bucket allocation)
         conn->request_.headers.clear();

         // --- Parse request line ---
         size_t request_line_end = data.find("\r\n");
         if (request_line_end == std::string_view::npos) return result; // incomplete

         std::string_view request_line_str = data.substr(0, request_line_end);

         auto request_line = glz::detail::parse_request_line(request_line_str);
         if (!request_line.has_value()) {
            int status_code = request_line.error();
            result.status = parse_status::error;
            send_error_response_with_close(conn, status_code, get_status_message(status_code));
            return result;
         }

         conn->request_.method = request_line->method;
         conn->request_.target = request_line->target;
         result.is_http_11 = request_line->is_http11;

         // --- Parse headers line by line (single pass) ---
         size_t pos = request_line_end + 2;
         auto& headers = conn->request_.headers;
         bool host_header_parsed = false;

         // RFC 9112, Section 3.2:
         // A server MUST respond with a 400 (Bad Request) status code to any
         // HTTP/1.1 request message that lacks a Host header field and to any
         // request message that contains more than one Host header field line
         // or a Host header field with an invalid field value.
         //
         // The "HTTP/1.1" qualifier applies only to a missing Host field. The
         // words "any request message" make the duplicate and invalid Host
         // rules independent of the request's HTTP version. This means that we
         // MUST also enforce those rules for HTTP/1.0 requests.

         while (true) {
            // Need at least 2 bytes to check for empty line
            if (pos + 1 >= conn->buf_len) return result; // incomplete

            // Empty line = end of headers
            if (data[pos] == '\r' && data[pos + 1] == '\n') {
               // A missing Host field is an error only for HTTP/1.1 requests
               if (result.is_http_11 && !host_header_parsed) {
                  result.status = parse_status::error;
                  send_error_response_with_close(conn, 400, "Bad Request");
                  return result;
               }

               result.headers_end = pos + 2;
               result.status = parse_status::complete;
               return result;
            }

            // RFC 7230 3.2.4: a header line beginning with SP/HTAB is obsolete line
            // folding (obs-fold). Accepting it would store a continuation under a
            // mangled key (e.g. "\ttransfer-encoding"), letting an obfuscated
            // Transfer-Encoding/Content-Length slip past the header lookups that frame
            // the body and desync this server from a folding-aware proxy (request
            // smuggling). Reject with 400 and close.
            if (data[pos] == ' ' || data[pos] == '\t') {
               result.status = parse_status::error;
               send_error_response_with_close(conn, 400, "Bad Request");
               return result;
            }

            // Find end of this header line
            size_t line_end = data.find("\r\n", pos);
            if (line_end == std::string_view::npos) return result; // incomplete

            // Parse header name:value
            std::string_view line = data.substr(pos, line_end - pos);
            auto colon_pos = line.find(':');
            if (colon_pos != std::string_view::npos) {
               std::string_view name_sv = line.substr(0, colon_pos);
               // RFC 7230 3.2.4: no whitespace is allowed between the field-name and
               // the colon. Tolerating it would store the name under a key with a
               // trailing space/tab (e.g. "transfer-encoding ") that header lookups
               // miss, so an obfuscated "Transfer-Encoding : chunked" would desync body
               // framing (request smuggling). Reject with 400 and close.
               if (!name_sv.empty() && (name_sv.back() == ' ' || name_sv.back() == '\t')) {
                  result.status = parse_status::error;
                  send_error_response_with_close(conn, 400, "Bad Request");
                  return result;
               }
               std::string_view value_sv = line.substr(colon_pos + 1);
               value_sv.remove_prefix((std::min)(value_sv.find_first_not_of(" \t"), value_sv.size()));
               // RFC 9110 §5.5 / RFC 9112 §5: a field value excludes trailing OWS as well;
               // recipients MUST strip it before evaluating (e.g. "Host: example.com " is legal
               // and must not be rejected by the Host reg-name check). Leading OWS is already
               // gone above, so an all-whitespace value is empty here and the guard no-ops.
               if (const auto last = value_sv.find_last_not_of(" \t"); last != std::string_view::npos) {
                  value_sv.remove_suffix(value_sv.size() - (last + 1));
               }
               std::string key(name_sv);
               for (auto& c : key) c = ascii_tolower(c);
               // RFC 7230 3.3.2: multiple Content-Length fields with differing values make the
               // body framing unrecoverable. The header map keeps last-wins, so a second
               // Content-Length would silently override the first; a proxy that frames the body
               // by the first value then desyncs from this server (CL.CL request smuggling).
               // Reject with 400 and close. Identical repeats are tolerated.
               if (key == "content-length") {
                  if (auto existing = headers.find(key); existing != headers.end() && existing->second != value_sv) {
                     result.status = parse_status::error;
                     send_error_response_with_close(conn, 400, "Bad Request");
                     return result;
                  }
               }

               // Duplicate or invalid Host fields are errors for any HTTP/1.x request
               if (key == "host") {
                  if (host_header_parsed || !detail::is_valid_authority(value_sv)) {
                     result.status = parse_status::error;
                     send_error_response_with_close(conn, 400, "Bad Request");
                     return result;
                  }
                  host_header_parsed = true;
               }

               headers[std::move(key)] = std::string(value_sv);
            }

            pos = line_end + 2;
         }
      }

      // Post-parse dispatch: keep-alive, body reading, handler invocation.
      // Does NOT shift the read buffer — header string_views remain valid through the handler.
      // Buffer compaction happens in the keep-alive reset (send_response_with_conn completion).
      inline void finish_request(std::shared_ptr<connection_state> conn, const parse_result& result)
      {
         auto& headers = conn->request_.headers;

         bool should_keep_alive = determine_keep_alive(headers, result.is_http_11, conn);
         conn->should_close = !should_keep_alive;

         if (is_websocket_upgrade(headers)) {
            handle_websocket_upgrade_with_conn(std::move(conn));
            return;
         }

         // Chunked request bodies are unsupported, so a request that carries
         // Transfer-Encoding cannot be framed by Content-Length. Treating it as a
         // zero-length body leaves the encoded payload in the buffer, where the
         // keep-alive loop parses it as a second, smuggled request (TE/CL desync).
         // RFC 7230 3.3.1 requires answering an undecodable Transfer-Encoding with
         // 501 and closing the connection.
         if (headers.find("transfer-encoding") != headers.end()) {
            send_error_response_with_close(conn, 501, "Not Implemented");
            return;
         }

         std::size_t content_length = 0;
         if (auto it = headers.find("content-length"); it != headers.end()) {
            const auto& cl = it->second;
            auto [ptr, ec] = std::from_chars(cl.data(), cl.data() + cl.size(), content_length);
            if (ec != std::errc{} || ptr != cl.data() + cl.size()) {
               send_error_response_with_close(conn, 400, "Bad Request");
               return;
            }
         }

         // Reject oversized request bodies before allocating memory
         if (conn_config_.max_request_body_size > 0 && content_length > conn_config_.max_request_body_size) {
            send_error_response_with_close(conn, 413, "Payload Too Large");
            return;
         }

         // Body bytes in the buffer start at headers_end
         const size_t body_offset = result.headers_end;
         const size_t available_body = conn->buf_len - body_offset;

         if (content_length > 0) {
            conn->request_.body.resize(content_length);
            const size_t initial_body_size = (std::min)(content_length, available_body);
            if (initial_body_size > 0) {
               std::memcpy(conn->request_.body.data(), &conn->read_buf[body_offset], initial_body_size);
            }
            // Track consumed bytes (headers + body taken from buffer)
            conn->buf_consumed = body_offset + initial_body_size;

            const size_t missing_bytes = content_length - initial_body_size;
            if (missing_bytes > 0) {
               asio::async_read(conn->socket, asio::buffer(&conn->request_.body[initial_body_size], missing_bytes),
                                asio::transfer_exactly(missing_bytes), [this, conn](std::error_code ec, size_t) {
                                   if (ec) {
                                      error_handler(ec, std::source_location::current());
                                      return;
                                   }
                                   process_full_request_with_conn(conn);
                                });
            }
            else {
               process_full_request_with_conn(conn);
            }
         }
         else {
            conn->buf_consumed = body_offset;
            process_full_request_with_conn(conn);
         }
      }

      // Case-insensitive substring search
      static bool ci_contains(std::string_view haystack, std::string_view needle)
      {
         if (needle.size() > haystack.size()) return false;
         for (size_t i = 0; i <= haystack.size() - needle.size(); ++i) {
            if (glz::striequal(haystack.substr(i, needle.size()), needle)) {
               return true;
            }
         }
         return false;
      }

      inline bool determine_keep_alive(const std::unordered_map<std::string, std::string>& headers, bool is_http_11,
                                       std::shared_ptr<connection_state> conn)
      {
         // If server has keep-alive disabled, always close
         if (!conn_config_.keep_alive) {
            return false;
         }

         // Check max requests per connection limit
         if (conn_config_.max_requests_per_connection > 0 &&
             conn->request_count >= conn_config_.max_requests_per_connection) {
            return false;
         }

         // Check client's Connection header (case-insensitive)
         auto conn_header_it = headers.find("connection");
         if (conn_header_it != headers.end()) {
            if (ci_contains(conn_header_it->second, "close")) {
               return false;
            }
            if (ci_contains(conn_header_it->second, "keep-alive")) {
               return true;
            }
         }

         // Default behavior based on HTTP version
         // HTTP/1.1: keep-alive by default
         // HTTP/1.0: close by default
         return is_http_11;
      }

      // Process a full request using connection state (new keep-alive aware version)
      inline void process_full_request_with_conn(std::shared_ptr<connection_state> conn)
      {
         // Create the request object
         request& request = conn->request_;

         // Lazy remote IP resolution (deferred from accept to avoid syscall + allocation per connection)
         if (request.remote_ip.empty()) {
            request.remote_ip = conn->remote_endpoint.address().to_string();
            request.remote_port = conn->remote_endpoint.port();
         }

         // Parse path and query string from target
         const auto [path_view, query_string] = split_target(conn->request_.target);
         request.path = std::string(path_view);
         request.query = parse_urlencoded(query_string);

         // Check for a streaming handler first. Streaming routes share the same
         // radix-tree matcher as normal routes, so they support ":param" paths.
         //
         // Fast path: if no streaming routes are registered (the common case),
         // skip the match entirely. match_streaming() would otherwise allocate
         // a params map, split the path into a vector of segment strings, and
         // run an empty-tree walk on every request. The empty-check is O(1).
         if (!root_router.streaming_routes.routes.empty()) {
            auto [stream_handle, stream_params] = root_router.match_streaming(conn->request_.method, request.path);
            if (stream_handle) {
               request.params = std::move(stream_params);
               // Streaming handlers take over the connection (no keep-alive loop)
               handle_streaming_request_with_conn(std::move(conn), stream_handle);
               return;
            }
         }

         // Find a matching route
         auto [handle, params] = root_router.match(conn->request_.method, request.path);

         // Reuse the connection-persistent response object (preserves capacity).
         // Cleared in the keep-alive reset path after the response is written.
         response& response = conn->response_;

         // Call request hooks for metrics/logging
         try {
            for (const auto& hook : request_hooks_) {
               hook(request, response);
            }
         }
         catch (const std::exception&) {
            error_handler(std::make_error_code(std::errc::invalid_argument), std::source_location::current());
            send_error_response_with_close(conn, 500, "Internal Server Error");
            return;
         }

         if (!handle) {
            if (conn->request_.method == http_method::OPTIONS) {
               std::vector<http_method> allowed_methods;
               std::unordered_map<std::string, std::string> preflight_params;

               auto capture_first_params = [&](std::unordered_map<std::string, std::string>&& candidate_params) {
                  if (preflight_params.empty()) {
                     preflight_params = std::move(candidate_params);
                  }
               };

               auto try_method = [&](http_method m) {
                  if (m == http_method::OPTIONS) return;
                  auto [candidate_handle, candidate_params] = root_router.match(m, conn->request_.target);
                  if (candidate_handle) {
                     allowed_methods.push_back(m);
                     capture_first_params(std::move(candidate_params));
                  }
               };

               try_method(http_method::GET);
               try_method(http_method::POST);
               try_method(http_method::PUT);
               try_method(http_method::DELETE);
               try_method(http_method::PATCH);
               try_method(http_method::HEAD);

               if (!allowed_methods.empty()) {
                  request.params = std::move(preflight_params);

                  bool has_request_method_header = false;
                  bool requested_method_known = false;
                  http_method requested_method{};

                  if (auto request_method_it = request.headers.find("access-control-request-method");
                      request_method_it != request.headers.end()) {
                     has_request_method_header = true;
                     std::string method_token{request_method_it->second};
                     auto trim_pos = method_token.find_last_not_of(" \t\r\n");
                     if (trim_pos != std::string::npos) {
                        method_token.erase(trim_pos + 1);
                     }
                     else {
                        method_token.clear();
                     }

                     if (!method_token.empty()) {
                        std::transform(method_token.begin(), method_token.end(), method_token.begin(),
                                       [](unsigned char c) { return static_cast<char>(std::toupper(c)); });

                        if (auto parsed = from_string(method_token)) {
                           requested_method = *parsed;
                           requested_method_known = true;
                        }
                     }
                  }

                  if (std::none_of(allowed_methods.begin(), allowed_methods.end(),
                                   [](http_method m) { return m == http_method::OPTIONS; })) {
                     allowed_methods.push_back(http_method::OPTIONS);
                  }

                  std::string allow_header;
                  allow_header.reserve(allowed_methods.size() * 8);
                  for (size_t i = 0; i < allowed_methods.size(); ++i) {
                     if (i > 0) allow_header.append(", ");
                     allow_header.append(std::string{to_string(allowed_methods[i])});
                  }
                  response.header("Allow", allow_header);

                  bool requested_method_allowed = true;
                  if (has_request_method_header) {
                     requested_method_allowed =
                        requested_method_known && std::find(allowed_methods.begin(), allowed_methods.end(),
                                                            requested_method) != allowed_methods.end();
                  }

                  try {
                     for (const auto& middleware : root_router.middlewares) {
                        middleware(request, response);
                     }
                  }
                  catch (const std::exception&) {
                     error_handler(std::make_error_code(std::errc::invalid_argument), std::source_location::current());
                     send_error_response_with_close(conn, 500, "Internal Server Error");
                     return;
                  }

                  if (!requested_method_allowed) {
                     response.status(405);
                     for (const auto& hook : response_hooks_) {
                        try {
                           hook(request, response);
                        }
                        catch (const std::exception&) {
                           error_handler(std::make_error_code(std::errc::invalid_argument),
                                         std::source_location::current());
                        }
                     }
                     send_response_with_conn(conn, response);
                     return;
                  }

                  for (const auto& hook : response_hooks_) {
                     try {
                        hook(request, response);
                     }
                     catch (const std::exception&) {
                        error_handler(std::make_error_code(std::errc::invalid_argument),
                                      std::source_location::current());
                     }
                  }
                  send_response_with_conn(conn, response);
                  return;
               }
            }

            // No matching route found
            send_error_response_with_conn(conn, 404, "Not Found");
            return;
         }

         request.params = std::move(params);

         try {
            for (const auto& middleware : root_router.middlewares) {
               middleware(request, response);
            }

            if (wrapping_middlewares_.empty()) {
               handle(request, response);
            }
            else {
               execute_middleware_chain(0, request, response, handle);
            }

            for (const auto& hook : response_hooks_) {
               try {
                  hook(request, response);
               }
               catch (const std::exception&) {
                  error_handler(std::make_error_code(std::errc::invalid_argument), std::source_location::current());
               }
            }

            send_response_with_conn(conn, response);
         }
         catch (const std::exception&) {
            error_handler(std::make_error_code(std::errc::invalid_argument), std::source_location::current());
            send_error_response_with_close(conn, 500, "Internal Server Error");
         }
      }

      // Pre-cached status lines for common HTTP status codes
      inline static std::string_view get_status_line(int code)
      {
         switch (code) {
         case 200:
            return "HTTP/1.1 200 OK\r\n";
         case 201:
            return "HTTP/1.1 201 Created\r\n";
         case 204:
            return "HTTP/1.1 204 No Content\r\n";
         case 301:
            return "HTTP/1.1 301 Moved Permanently\r\n";
         case 400:
            return "HTTP/1.1 400 Bad Request\r\n";
         case 401:
            return "HTTP/1.1 401 Unauthorized\r\n";
         case 403:
            return "HTTP/1.1 403 Forbidden\r\n";
         case 404:
            return "HTTP/1.1 404 Not Found\r\n";
         case 405:
            return "HTTP/1.1 405 Method Not Allowed\r\n";
         case 500:
            return "HTTP/1.1 500 Internal Server Error\r\n";
         case 501:
            return "HTTP/1.1 501 Not Implemented\r\n";
         case 503:
            return "HTTP/1.1 503 Service Unavailable\r\n";
         default:
            return {};
         }
      }

      // Send response with keep-alive support
      inline void send_response_with_conn(std::shared_ptr<connection_state> conn, response& response)
      {
         // Reuse connection-persistent header buffer (preserves capacity across requests)
         conn->header_buf.clear();
         auto& h = conn->header_buf;
         char num_buf[20];

         // Status line (pre-cached for common codes)
         auto status_line = get_status_line(response.status_code);
         if (!status_line.empty()) {
            h.append(status_line);
         }
         else {
            h.append("HTTP/1.1 ");
            auto* end = glz::to_chars(num_buf, static_cast<int32_t>(response.status_code));
            h.append(num_buf, size_t(end - num_buf));
            h.append(" ");
            h.append(get_status_message(response.status_code));
            h.append("\r\n");
         }

         for (const auto& [name, value] : response.response_headers) {
            if (header_field_has_crlf(name, value)) [[unlikely]] {
               continue;
            }
            h.append(name);
            h.append(": ");
            h.append(value);
            h.append("\r\n");
         }

         if (!(response.user_headers_set & response::has_content_length)) {
            h.append("Content-Length: ");
            auto* end = glz::to_chars(num_buf, static_cast<uint64_t>(response.response_body.size()));
            h.append(num_buf, size_t(end - num_buf));
            h.append("\r\n");
         }

         if (!(response.user_headers_set & response::has_date)) {
            h.append("Date: ");
            h.append(get_current_date());
            h.append("\r\n");
         }

         if (!(response.user_headers_set & response::has_server)) {
            h.append("Server: Glaze/1.0\r\n");
         }

         // Add Connection header based on keep-alive decision
         if (!(response.user_headers_set & response::has_connection)) {
            if (conn->should_close) {
               h.append("Connection: close\r\n");
            }
            else {
               h.append("Connection: keep-alive\r\n");
               // Add Keep-Alive header with timeout info
               if (conn_config_.keep_alive_timeout > 0) {
                  h.append("Keep-Alive: timeout=");
                  auto* end = glz::to_chars(num_buf, conn_config_.keep_alive_timeout);
                  h.append(num_buf, size_t(end - num_buf));
                  if (conn_config_.max_requests_per_connection > 0) {
                     h.append(", max=");
                     end = glz::to_chars(num_buf, conn_config_.max_requests_per_connection);
                     h.append(num_buf, size_t(end - num_buf));
                  }
                  h.append("\r\n");
               }
            }
         }

         h.append("\r\n");

         // Scatter-gather write: send headers and body as separate buffers.
         // Zero-copy for the body — write directly from conn->response_.response_body.
         // conn (shared_ptr) keeps header_buf and response_body alive during the async write.
         std::array<asio::const_buffer, 2> bufs = {asio::buffer(h), asio::buffer(response.response_body)};

         // Custom completion condition: removes the default 64KB-per-chunk cap
         // in asio::async_write (transfer_all uses 65536). Without this, a 960KB
         // response requires ~15 event loop iterations instead of ~8.
         auto unlimited = [](const asio::error_code& ec, std::size_t) -> std::size_t {
            return ec ? 0 : (std::numeric_limits<std::size_t>::max)();
         };
         asio::async_write(conn->socket, bufs, unlimited,
                           [this, conn](asio::error_code ec, std::size_t /*bytes_transferred*/) {
                              if (ec) {
                                 // Write error - connection is dead
                                 return;
                              }

                              if (conn->should_close) {
                                 // Send FIN; socket closed via RAII when conn is destroyed
                                 asio::error_code close_ec;
                                 conn->socket.lowest_layer().shutdown(asio::ip::tcp::socket::shutdown_send, close_ec);
                              }
                              else {
                                 // Keep-alive: compact buffer (shift leftover pipelined data to front)
                                 conn->request_.params.clear();
                                 conn->request_.query.clear();
                                 conn->request_.headers.clear();
                                 conn->request_.body.clear();
                                 conn->response_.clear();

                                 size_t leftover = conn->buf_len - conn->buf_consumed;
                                 if (leftover > 0) {
                                    std::memmove(conn->read_buf.data(), &conn->read_buf[conn->buf_consumed], leftover);
                                 }
                                 conn->buf_len = leftover;
                                 conn->buf_consumed = 0;

                                 read_request(conn);
                              }
                           });
      }

      // Send error response with keep-alive support (for normal errors)
      inline void send_error_response_with_conn(std::shared_ptr<connection_state> conn, int status_code,
                                                std::string_view message)
      {
         conn->response_.clear();
         conn->response_.status(status_code).content_type("text/plain").body(message);
         send_response_with_conn(conn, conn->response_);
      }

      // Send error response and close connection (for protocol errors)
      inline void send_error_response_with_close(std::shared_ptr<connection_state> conn, int status_code,
                                                 std::string_view message)
      {
         conn->should_close = true; // Force close after error
         send_error_response_with_conn(conn, status_code, message);
      }

      inline bool is_websocket_upgrade(const std::unordered_map<std::string, std::string>& headers)
      {
         auto upgrade_it = headers.find("upgrade");
         if (upgrade_it == headers.end()) return false;

         auto connection_it = headers.find("connection");
         if (connection_it == headers.end()) return false;

         if (!ci_contains(upgrade_it->second, "websocket")) return false;
         return ci_contains(connection_it->second, "upgrade");
      }

      inline void handle_websocket_upgrade_with_conn(std::shared_ptr<connection_state> conn)
      {
         request& request = conn->request_;

         // Parse path and query string from target so query parameters don't break the lookup
         // and so the WebSocket handler receives them like a regular handler would.
         const auto [path_view, query_string] = split_target(request.target);
         request.path = std::string(path_view);
         request.query = parse_urlencoded(query_string);

         // WebSocket routes share the same radix-tree matcher, so ":param" path
         // parameters are extracted into request.params just like normal routes.
         auto [ws_server, ws_params] = root_router.match_websocket(request.path);
         if (!ws_server) {
            send_error_response_with_close(conn, 404, "Not Found");
            return;
         }
         request.params = std::move(ws_params);

         // Copy request for WebSocket handler (request body is typically empty for upgrades)
         glz::request req{request};

         // Create WebSocket connection and start it
         // Uses socket_type which is either tcp::socket (ws://) or ssl::stream (wss://)
         auto socket_ptr = std::make_shared<socket_type>(std::move(conn->socket));
         auto ws_conn = std::make_shared<websocket_connection<socket_type>>(socket_ptr, ws_server);
         ws_conn->start(req);
      }

      inline void handle_streaming_request_with_conn(std::shared_ptr<connection_state> conn,
                                                     const streaming_handler& handler)
      {
         // Create streaming connection (works for both HTTP and HTTPS via interface)
         auto socket_ptr = std::make_shared<socket_type>(std::move(conn->socket));
         auto stream_conn = std::make_shared<streaming_connection<socket_type>>(socket_ptr);
         streaming_response stream_res(stream_conn);

         try {
            // Call the streaming handler
            handler(conn->request_, stream_res);
         }
         catch (const std::exception&) {
            // If handler throws immediately, try to send an error response.
            if (!stream_conn->is_headers_sent()) {
               stream_conn->send_headers(500, {{"Content-Type", "text/plain"}});
               stream_conn->send_chunk("Internal Server Error");
            }
            stream_conn->close();
            // Log the error
            error_handler(std::make_error_code(std::errc::invalid_argument), std::source_location::current());
         }
      }

      inline std::string_view get_status_message(int status_code)
      {
         switch (status_code) {
         case 200:
            return "OK";
         case 201:
            return "Created";
         case 204:
            return "No Content";
         case 400:
            return "Bad Request";
         case 401:
            return "Unauthorized";
         case 403:
            return "Forbidden";
         case 404:
            return "Not Found";
         case 405:
            return "Method Not Allowed";
         case 409:
            return "Conflict";
         case 418:
            return "I'm a teapot";
         case 500:
            return "Internal Server Error";
         case 501:
            return "Not Implemented";
         case 503:
            return "Service Unavailable";
         default:
            return "Unknown";
         }
      }

      inline const std::string& get_current_date()
      {
         static thread_local std::string cached;
         static thread_local std::chrono::sys_seconds cached_time{};
         const auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
         if (cached.empty() || now != cached_time) {
            cached_time = now;
            cached = detail::format_http_date(now);
         }
         return cached;
      }
   };

   // Utility functions for common streaming patterns
   namespace streaming_utils
   {
      // Create a periodic data sender (works with both HTTP and HTTPS)
      // Uses enable_shared_from_this to avoid circular reference memory leaks
      template <typename T>
      void send_periodic_data(std::shared_ptr<streaming_connection_interface> conn, std::function<T()> data_generator,
                              std::chrono::milliseconds interval, size_t max_events = 0)
      {
         if (!conn || !conn->is_open()) return;

         struct periodic_sender : std::enable_shared_from_this<periodic_sender>
         {
            std::shared_ptr<streaming_connection_interface> conn;
            std::shared_ptr<asio::steady_timer> timer;
            std::function<T()> data_generator;
            std::chrono::milliseconds interval;
            size_t max_events;
            size_t counter = 0;

            periodic_sender(std::shared_ptr<streaming_connection_interface> c, std::function<T()> gen,
                            std::chrono::milliseconds intv, size_t max_ev)
               : conn(std::move(c)),
                 timer(std::make_shared<asio::steady_timer>(conn->get_executor())),
                 data_generator(std::move(gen)),
                 interval(intv),
                 max_events(max_ev)
            {}

            void send_next()
            {
               if (!conn->is_open() || (max_events > 0 && counter >= max_events)) {
                  conn->close();
                  return;
               }

               try {
                  T data = data_generator();
                  conn->send_json_event(data, "data", std::to_string(counter),
                                        [self = this->shared_from_this()](std::error_code ec) {
                                           if (!ec) {
                                              self->counter++;
                                              self->timer->expires_after(self->interval);
                                              self->timer->async_wait([self](std::error_code) { self->send_next(); });
                                           }
                                           else {
                                              self->conn->close();
                                           }
                                        });
               }
               catch (const std::exception&) {
                  conn->close();
               }
            }
         };

         auto sender = std::make_shared<periodic_sender>(conn, std::move(data_generator), interval, max_events);
         sender->send_next();
      }

      // Create a data stream from a collection (works with both HTTP and HTTPS)
      // Uses enable_shared_from_this to avoid circular reference memory leaks
      // Note: Makes a copy of the container to ensure data outlives the async operation
      template <typename Container>
      void stream_collection(std::shared_ptr<streaming_connection_interface> conn, const Container& data,
                             std::chrono::milliseconds delay_between_items = std::chrono::milliseconds(10))
      {
         if (!conn || !conn->is_open()) return;

         struct collection_sender : std::enable_shared_from_this<collection_sender>
         {
            std::shared_ptr<streaming_connection_interface> conn;
            std::shared_ptr<asio::steady_timer> timer;
            Container data;
            typename Container::const_iterator it;
            std::chrono::milliseconds delay;

            collection_sender(std::shared_ptr<streaming_connection_interface> c, Container d,
                              std::chrono::milliseconds dl)
               : conn(std::move(c)),
                 timer(std::make_shared<asio::steady_timer>(conn->get_executor())),
                 data(std::move(d)),
                 it(data.begin()),
                 delay(dl)
            {}

            void send_next()
            {
               if (!conn->is_open() || it == data.end()) {
                  conn->close();
                  return;
               }

               conn->send_json_event(*it, "item", "", [self = this->shared_from_this()](std::error_code ec) {
                  if (!ec) {
                     ++self->it;
                     self->timer->expires_after(self->delay);
                     self->timer->async_wait([self](std::error_code) { self->send_next(); });
                  }
                  else {
                     self->conn->close();
                  }
               });
            }
         };

         auto sender = std::make_shared<collection_sender>(conn, data, delay_between_items);
         sender->send_next();
      }
   } // namespace streaming_utils

   // Alias for HTTPS server
   using https_server = http_server<true>;
}