//! HTTP server implementation
//!
//! This module contains the main HTTP/HTTPS server that handles incoming
//! connections, routes requests, and dispatches to appropriate handlers.

pub mod static_files;

use crate::acme::AcmeManager;
use crate::config::{HeaderRules, ResolvedConfig, RouteAction};
use crate::middleware::{
    apply_expires, apply_response_headers, basic_auth_challenge_response, default_security_headers,
    error_response, is_basic_auth_authorized, redirect_response, status_response,
};
use crate::proxy::{
    ProxyConfig, ProxyError, ReverseProxy, WebSocketUpgradeResult, bidirectional_copy,
    is_websocket_upgrade,
};
use crate::routing::{MatchResult, Router};
use crate::tls::{NegotiatedProtocol, TlsManager};
use bytes::Bytes;
use http::{HeaderValue, Request, Response, StatusCode, header};
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use hyper::body::Incoming;
use hyper::server::conn::{http1, http2};
use hyper::service::service_fn;
use hyper_util::rt::{TokioExecutor, TokioIo};
use lru::LruCache;
use static_files::{StaticFileConfig, serve_static};
use std::convert::Infallible;
use std::net::{IpAddr, SocketAddr};
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Semaphore;
use tokio_rustls::TlsAcceptor;
use tracing::{debug, error, info, trace, warn};

/// Server statistics
#[derive(Default)]
pub struct ServerStats {
    /// Total requests handled
    pub requests: AtomicU64,
    /// Active connections
    pub active_connections: AtomicUsize,
    /// Request errors
    pub errors: AtomicU64,
}

/// Per-IP rate limiting state
struct IpRateLimit {
    connections: AtomicUsize,
}

/// Normalize IP address for rate limiting
/// For IPv6, use /64 prefix to prevent bypass via multiple addresses
fn normalize_ip_for_rate_limiting(ip: IpAddr) -> IpAddr {
    match ip {
        IpAddr::V4(_) => ip, // IPv4 uses full address
        IpAddr::V6(ipv6) => {
            // Extract first 64 bits (8 bytes) and zero out the rest
            let segments = ipv6.segments();
            let normalized = std::net::Ipv6Addr::new(
                segments[0],
                segments[1],
                segments[2],
                segments[3],
                0,
                0,
                0,
                0,
            );
            IpAddr::V6(normalized)
        }
    }
}

/// Main HTTP server
pub struct Server {
    config: Arc<ResolvedConfig>,
    raw_config: Arc<crate::config::Config>,
    router: Arc<Router>,
    proxy: Arc<ReverseProxy>,
    tls_manager: Arc<TlsManager>,
    acme_manager: Option<Arc<AcmeManager>>,
    stats: Arc<ServerStats>,
    global_headers: HeaderRules,
    /// Connection limiter for DoS protection
    connection_semaphore: Arc<Semaphore>,
    /// Per-IP rate limiting (bounded LRU to prevent memory exhaustion)
    ip_rate_limits: Arc<parking_lot::Mutex<LruCache<IpAddr, Arc<IpRateLimit>>>>,
    /// Maximum request body size
    #[allow(dead_code)]
    max_body_size: u64,
}

impl Server {
    /// Create a new server from configuration
    pub fn new(config: ResolvedConfig, raw_config: crate::config::Config) -> Arc<Self> {
        let router = Arc::new(Router::new(&config.hosts));

        let proxy_config = ProxyConfig {
            io_timeout: config.proxy_timeout_io,
            max_body_size: config.limit_request_body,
        };
        let proxy = ReverseProxy::new(proxy_config);

        let tls_manager = Arc::new(TlsManager::with_default_fallback(config.sni_fallback));
        let acme_manager = AcmeManager::from_config(&raw_config, Arc::clone(&tls_manager));

        // Merge global headers with security defaults
        let global_headers = default_security_headers().merge_with(&config.global_headers);

        // Connection limit: reduce from 65536 to more reasonable default
        let max_connections = 10000;

        Arc::new(Self {
            max_body_size: config.limit_request_body,
            config: Arc::new(config),
            raw_config: Arc::new(raw_config),
            router,
            proxy,
            tls_manager,
            acme_manager,
            stats: Arc::new(ServerStats::default()),
            global_headers,
            connection_semaphore: Arc::new(Semaphore::new(max_connections)),
            // Bounded LRU cache: max 100k IPs to prevent memory exhaustion
            ip_rate_limits: Arc::new(parking_lot::Mutex::new(LruCache::new(
                NonZeroUsize::new(100_000).unwrap(),
            ))),
        })
    }

    /// Get a reference to the router (for HTTP/3 server)
    pub fn router(&self) -> Arc<Router> {
        Arc::clone(&self.router)
    }

    /// Get a reference to the proxy (for HTTP/3 server)
    pub fn proxy(&self) -> Arc<ReverseProxy> {
        Arc::clone(&self.proxy)
    }

    /// Run the server
    pub async fn run(self: Arc<Self>) -> anyhow::Result<()> {
        info!("Starting pyx reverse proxy");

        let mut handles = Vec::new();

        if let Some(acme_manager) = &self.acme_manager {
            acme_manager.load_cached_certificates();
        }

        // Note: LRU cache automatically evicts old entries, no cleanup task needed

        // Start listeners
        for listener_config in &self.config.listeners {
            let server = Arc::clone(&self);
            let addr = listener_config.addr;
            let tls_config = listener_config.tls_config.clone();

            let handle = tokio::spawn(async move {
                if let Err(e) = server.run_listener(addr, tls_config).await {
                    error!("Listener {} failed: {}", addr, e);
                }
            });

            handles.push(handle);
        }

        if let Some(acme_manager) = &self.acme_manager {
            Arc::clone(acme_manager).start();
        }

        // Wait for all listeners
        for handle in handles {
            handle.await?;
        }

        Ok(())
    }

    /// Run a single listener
    async fn run_listener(
        self: Arc<Self>,
        addr: SocketAddr,
        tls_config: Option<Arc<crate::config::TlsListenerConfig>>,
    ) -> anyhow::Result<()> {
        let listener = TcpListener::bind(addr).await?;
        let tls_label = match tls_config.as_ref() {
            Some(cfg) if cfg.letsencrypt => " (TLS, Let's Encrypt)",
            Some(_) => " (TLS)",
            None => "",
        };
        info!("Listening on {}{}", addr, tls_label);

        // Build TLS acceptor if needed
        let tls_acceptor = if tls_config.is_some() {
            // Load certificates for ALL hostnames that have SSL configured on this listener
            // This supports multiple certificates on the same port via SNI
            for (host_name, host_value) in &self.raw_config.hosts {
                // Check if this host has SSL configuration
                if let Some(listen_value) = &host_value.listen {
                    if let Ok(listen_config) =
                        serde_yaml::from_value::<crate::config::ListenConfig>(listen_value.clone())
                    {
                        // Check if this listen config matches our listener's address
                        if listen_config.socket_addr() == addr {
                            // This host uses this listener - load its certificate if it has SSL
                            if let Some(ssl) = &listen_config.ssl {
                                // Parse hostname from "hostname:port" format
                                if let Some(colon_pos) = host_name.rfind(':') {
                                    let hostname = &host_name[..colon_pos];
                                    if let (Some(cert_file), Some(key_file)) =
                                        (&ssl.certificate_file, &ssl.key_file)
                                    {
                                        if let Err(err) = self
                                            .tls_manager
                                            .load_cert(hostname, cert_file, key_file)
                                        {
                                            warn!(
                                                "Failed to load TLS certificate for {}: {}",
                                                hostname, err
                                            );
                                        }
                                    } else if ssl.letsencrypt.is_on() {
                                        debug!(
                                            "TLS certificate for {} will be provided by Let's Encrypt",
                                            hostname
                                        );
                                    } else {
                                        warn!(
                                            "TLS certificate paths missing for {}; handshakes for this host will fail",
                                            hostname
                                        );
                                    }
                                }
                            }
                        }
                    }
                }
            }

            // Use HTTP/1-only config if HTTP/2 is disabled
            let server_config = if self.config.http2.enabled {
                self.tls_manager.build_server_config()?
            } else {
                self.tls_manager.build_server_config_http1_only()?
            };
            Some(TlsAcceptor::from(server_config))
        } else {
            None
        };

        loop {
            let (stream, peer_addr) = match listener.accept().await {
                Ok(conn) => conn,
                Err(e) => {
                    warn!("Accept error: {}", e);
                    continue;
                }
            };

            // Check per-IP rate limit
            // For IPv6, rate limit by /64 prefix to prevent bypass
            const MAX_CONNECTIONS_PER_IP: usize = 100;
            let ip_addr = normalize_ip_for_rate_limiting(peer_addr.ip());

            let ip_limit = {
                let mut cache = self.ip_rate_limits.lock();
                cache
                    .get_or_insert(ip_addr, || {
                        Arc::new(IpRateLimit {
                            connections: AtomicUsize::new(0),
                        })
                    })
                    .clone()
            };

            let ip_conns = ip_limit.connections.fetch_add(1, Ordering::Relaxed);
            if ip_conns >= MAX_CONNECTIONS_PER_IP {
                warn!(
                    "Per-IP connection limit reached for {}, rejecting",
                    peer_addr
                );
                ip_limit.connections.fetch_sub(1, Ordering::Relaxed);
                drop(stream);
                continue;
            }

            // Acquire global connection permit
            let permit = match self.connection_semaphore.clone().try_acquire_owned() {
                Ok(p) => p,
                Err(_) => {
                    warn!("Global connection limit reached, rejecting {}", peer_addr);
                    ip_limit.connections.fetch_sub(1, Ordering::Relaxed);
                    drop(stream);
                    continue;
                }
            };

            self.stats
                .active_connections
                .fetch_add(1, Ordering::Relaxed);

            let server = Arc::clone(&self);
            let tls_acceptor = tls_acceptor.clone();

            tokio::spawn(async move {
                let result = if let Some(acceptor) = tls_acceptor {
                    server
                        .clone()
                        .handle_tls_connection(stream, peer_addr, acceptor)
                        .await
                } else {
                    server.clone().handle_connection(stream, peer_addr).await
                };

                if let Err(e) = result {
                    trace!("Connection {} error: {}", peer_addr, e);
                    server.stats.errors.fetch_add(1, Ordering::Relaxed);
                }

                server
                    .stats
                    .active_connections
                    .fetch_sub(1, Ordering::Relaxed);
                ip_limit.connections.fetch_sub(1, Ordering::Relaxed);
                drop(permit);
            });
        }
    }

    /// Handle a plain HTTP connection (with WebSocket upgrade support)
    async fn handle_connection(
        self: Arc<Self>,
        stream: TcpStream,
        peer_addr: SocketAddr,
    ) -> anyhow::Result<()> {
        stream.set_nodelay(true)?;

        let io = TokioIo::new(stream);
        let server = Arc::clone(&self);

        let mut builder = http1::Builder::new();
        builder.keep_alive(true);
        builder.timer(hyper_util::rt::TokioTimer::new());

        // Serve connection
        // Use with_upgrades() to support WebSocket upgrades
        builder
            .serve_connection(
                io,
                service_fn(move |req| {
                    let server = Arc::clone(&server);
                    // Per-request timeout (2 minutes) - but WebSocket upgrades bypass this
                    async move {
                        tokio::time::timeout(
                            Duration::from_secs(120),
                            server.handle_request(req, peer_addr, false),
                        )
                        .await
                        .unwrap_or_else(|_| {
                            Ok(
                                error_response(StatusCode::REQUEST_TIMEOUT, "Request timeout")
                                    .map(|body| body.map_err(|e| match e {}).boxed()),
                            )
                        })
                    }
                }),
            )
            .with_upgrades()
            .await?;

        Ok(())
    }

    /// Handle a TLS connection with HTTP/2 support via ALPN
    async fn handle_tls_connection(
        self: Arc<Self>,
        stream: TcpStream,
        peer_addr: SocketAddr,
        acceptor: TlsAcceptor,
    ) -> anyhow::Result<()> {
        stream.set_nodelay(true)?;

        self.tls_manager.record_handshake_start();

        // TLS handshake with timeout
        let tls_stream =
            match tokio::time::timeout(Duration::from_secs(10), acceptor.accept(stream)).await {
                Ok(Ok(stream)) => {
                    self.tls_manager.record_handshake_complete();
                    stream
                }
                Ok(Err(e)) => {
                    self.tls_manager.record_handshake_failed();
                    return Err(anyhow::anyhow!("TLS handshake failed: {}", e));
                }
                Err(_) => {
                    self.tls_manager.record_handshake_failed();
                    return Err(anyhow::anyhow!("TLS handshake timeout"));
                }
            };

        // Check if HTTP/2 is enabled in config
        if !self.config.http2.enabled {
            // HTTP/2 disabled, always use HTTP/1.1
            return self
                .handle_http1_tls_connection(tls_stream, peer_addr)
                .await;
        }

        // Determine negotiated protocol from ALPN
        let protocol = {
            let (_, conn) = tls_stream.get_ref();
            NegotiatedProtocol::from_alpn(conn.alpn_protocol())
        };

        trace!("TLS connection from {} using {:?}", peer_addr, protocol);

        match protocol {
            NegotiatedProtocol::H2 => self.handle_http2_connection(tls_stream, peer_addr).await,
            NegotiatedProtocol::Http1 => {
                self.handle_http1_tls_connection(tls_stream, peer_addr)
                    .await
            }
        }
    }

    /// Handle HTTP/1.1 over TLS (with WebSocket upgrade support)
    async fn handle_http1_tls_connection<S>(
        self: Arc<Self>,
        tls_stream: S,
        peer_addr: SocketAddr,
    ) -> anyhow::Result<()>
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
    {
        let io = TokioIo::new(tls_stream);
        let server = Arc::clone(&self);

        let mut builder = http1::Builder::new();
        builder.keep_alive(true);
        builder.timer(hyper_util::rt::TokioTimer::new());

        // Serve connection
        // Use with_upgrades() to support WebSocket upgrades (wss://)
        builder
            .serve_connection(
                io,
                service_fn(move |req| {
                    let server = Arc::clone(&server);
                    // Per-request timeout (2 minutes) - but WebSocket upgrades bypass this
                    async move {
                        tokio::time::timeout(
                            Duration::from_secs(120),
                            server.handle_request(req, peer_addr, true),
                        )
                        .await
                        .unwrap_or_else(|_| {
                            Ok(
                                error_response(StatusCode::REQUEST_TIMEOUT, "Request timeout")
                                    .map(|body| body.map_err(|e| match e {}).boxed()),
                            )
                        })
                    }
                }),
            )
            .with_upgrades()
            .await?;

        Ok(())
    }

    /// Handle HTTP/2 connection
    async fn handle_http2_connection<S>(
        self: Arc<Self>,
        stream: S,
        peer_addr: SocketAddr,
    ) -> anyhow::Result<()>
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
    {
        let io = TokioIo::new(stream);
        let server = Arc::clone(&self);

        let mut builder = http2::Builder::new(TokioExecutor::new());

        // HTTP/2 specific settings from config
        let h2_config = &self.config.http2;
        builder
            .timer(hyper_util::rt::TokioTimer::new())
            .keep_alive_interval(Duration::from_secs(h2_config.idle_timeout))
            .keep_alive_timeout(Duration::from_secs(20))
            .max_concurrent_streams(h2_config.max_concurrent_streams)
            .initial_stream_window_size(h2_config.initial_stream_window)
            .initial_connection_window_size(h2_config.initial_connection_window)
            .max_frame_size(h2_config.max_frame_size);

        // Serve connection
        builder
            .serve_connection(
                io,
                service_fn(move |req| {
                    let server = Arc::clone(&server);
                    // Per-request timeout (2 minutes)
                    async move {
                        tokio::time::timeout(
                            Duration::from_secs(120),
                            server.handle_request(req, peer_addr, true),
                        )
                        .await
                        .unwrap_or_else(|_| {
                            Ok(
                                error_response(StatusCode::REQUEST_TIMEOUT, "Request timeout")
                                    .map(|body| body.map_err(|e| match e {}).boxed()),
                            )
                        })
                    }
                }),
            )
            .await?;

        Ok(())
    }

    /// Handle a single HTTP request (streaming)
    async fn handle_request(
        self: Arc<Self>,
        mut request: Request<Incoming>,
        peer_addr: SocketAddr,
        is_tls: bool,
    ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, Infallible> {
        self.stats.requests.fetch_add(1, Ordering::Relaxed);

        // HTTP/2 uses :authority pseudo-header, HTTP/1.1 uses Host header
        let raw_host = request
            .uri()
            .authority()
            .map(|a| a.as_str())
            .or_else(|| {
                request
                    .headers()
                    .get(header::HOST)
                    .and_then(|h: &HeaderValue| h.to_str().ok())
            })
            .unwrap_or("");

        // SECURITY: Normalize host by appending default port if missing.
        // This ensures HTTP requests route to :80 configs and HTTPS to :443 configs,
        // preventing HTTP requests from accessing HTTPS-only paths.
        //
        // Example: Request to http://example.com/secret should route to example.com:80,
        // not fall through to example.com:443 via hostname-only lookup.
        let host: std::borrow::Cow<'_, str> = if raw_host.contains(':') {
            // Host already has a port
            std::borrow::Cow::Borrowed(raw_host)
        } else if raw_host.is_empty() {
            std::borrow::Cow::Borrowed(raw_host)
        } else {
            // Append default port based on TLS status
            let default_port = if is_tls { 443 } else { 80 };
            std::borrow::Cow::Owned(format!("{}:{}", raw_host, default_port))
        };

        let path = request.uri().path();

        if let Some(acme_manager) = &self.acme_manager {
            if let Some(body) = acme_manager.challenge_response(path) {
                let response = Response::builder()
                    .status(StatusCode::OK)
                    .header(header::CONTENT_TYPE, "text/plain")
                    .body(Full::new(Bytes::from(body)).map_err(|e| match e {}).boxed())
                    .unwrap();
                return Ok(response);
            }
        }

        trace!(
            "Request: {} {} {} from {}",
            request.method(),
            host,
            path,
            peer_addr
        );

        // Route the request
        let result = match self.router.route(&host, path) {
            Some(r) => r,
            None => {
                debug!("No route for {} {}", host, path);
                let resp = error_response(StatusCode::NOT_FOUND, "Not Found");
                return Ok(resp.map(|body| body.map_err(|e| match e {}).boxed()));
            }
        };

        if let Some(auth) = &result.auth {
            if !is_basic_auth_authorized(request.headers(), auth) {
                debug!("Basic auth failed for {} {}", host, path);
                let resp = basic_auth_challenge_response(&auth.realm);
                return Ok(resp.map(|body| body.map_err(|e| match e {}).boxed()));
            }

            request.headers_mut().remove(header::AUTHORIZATION);
        }

        // Execute the action
        let client_scheme = if is_tls { "https" } else { "http" };
        let response = match self
            .execute_action(request, &result, peer_addr, client_scheme)
            .await
        {
            Ok(r) => r,
            Err(e) => {
                warn!("Request error: {}", e);
                self.stats.errors.fetch_add(1, Ordering::Relaxed);
                let resp = error_response(StatusCode::BAD_GATEWAY, &format!("Proxy error: {}", e));
                resp.map(|body| body.map_err(|e| match e {}).boxed())
            }
        };

        Ok(response)
    }

    /// Execute a route action (streaming)
    async fn execute_action(
        &self,
        request: Request<Incoming>,
        route: &MatchResult,
        peer_addr: SocketAddr,
        client_scheme: &str,
    ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, ProxyError> {
        match &route.action {
            RouteAction::Redirect { url, status } => {
                // Apply header rules to redirect response
                let mut response = redirect_response(*status, url);
                let merged_headers = self.global_headers.merge_with(&route.headers);
                apply_response_headers(&mut response, &merged_headers);
                apply_expires(&mut response, route.expires);
                Ok(response.map(|body| body.map_err(|e| match e {}).boxed()))
            }

            RouteAction::Status => {
                let mut response = status_response();
                let merged_headers = self.global_headers.merge_with(&route.headers);
                apply_response_headers(&mut response, &merged_headers);
                apply_expires(&mut response, route.expires);
                Ok(response.map(|body| body.map_err(|e| match e {}).boxed()))
            }

            RouteAction::StaticFiles {
                dir,
                index,
                send_gzip,
                dirlisting,
            } => {
                let config = StaticFileConfig {
                    root: dir.clone(),
                    index: index.clone(),
                    send_gzip: *send_gzip,
                    dirlisting: *dirlisting,
                    prefix: route.matched_path.clone(),
                };

                match serve_static(&request, &config).await {
                    Ok(mut response) => {
                        let merged_headers = self.global_headers.merge_with(&route.headers);
                        apply_response_headers(&mut response, &merged_headers);
                        apply_expires(&mut response, route.expires);
                        // Map Box<dyn Error> to hyper::Error
                        Ok(response.map(|body| {
                            body.map_err(|e| {
                                // Log the streaming error
                                tracing::warn!("Static file streaming error: {}", e);
                                // Convert Box<dyn Error> to hyper::Error
                                // We do this by creating a simple wrapper since hyper::Error
                                // doesn't have public constructors for custom errors
                                // The stream will just terminate on error which is acceptable behavior
                                match e.downcast::<std::io::Error>() {
                                    Ok(io_err) => {
                                        // For io errors, we can't construct hyper::Error directly
                                        // So we'll have the stream fail silently by returning
                                        // a placeholder error. In practice, the connection will close.
                                        // This is a limitation of hyper's error handling.
                                        panic!("IO error in stream: {}", io_err)
                                    }
                                    Err(other) => {
                                        panic!("Stream error: {}", other)
                                    }
                                }
                            })
                            .boxed()
                        }))
                    }
                    Err(e) => {
                        let status = e.status_code();
                        let msg = e.to_string();
                        Ok(error_response(status, &msg)
                            .map(|body| body.map_err(|e| match e {}).boxed()))
                    }
                }
            }

            RouteAction::Proxy {
                upstream,
                preserve_host,
            } => {
                let merged_response_headers = self.global_headers.merge_with(&route.headers);

                // Check if this is a WebSocket upgrade request
                if is_websocket_upgrade(&request) {
                    trace!(
                        "WebSocket upgrade request from {} to {}",
                        peer_addr, upstream
                    );

                    // Extract info we need from the request BEFORE getting the OnUpgrade
                    let method = request.method().clone();
                    let uri = request.uri().clone();
                    let request_headers = request.headers().clone();

                    // Get the OnUpgrade from the request - this consumes the request
                    let on_upgrade = hyper::upgrade::on(request);

                    // Handle WebSocket upgrade with the extracted info
                    let result = self
                        .proxy
                        .proxy_websocket_with_info(
                            method,
                            &uri,
                            &request_headers,
                            upstream,
                            *preserve_host,
                            &route.proxy_headers,
                            Some(peer_addr.ip()),
                            client_scheme,
                        )
                        .await?;

                    return match result {
                        WebSocketUpgradeResult::Upgraded {
                            upstream: mut upstream_conn,
                        } => {
                            // Build a 101 Switching Protocols response
                            let response = Response::builder()
                                .status(StatusCode::SWITCHING_PROTOCOLS)
                                .header(header::CONNECTION, "Upgrade")
                                .header(header::UPGRADE, "websocket")
                                .body(Full::new(Bytes::new()).map_err(|e| match e {}).boxed())
                                .unwrap();

                            debug!(
                                "WebSocket upgrade to {} successful, starting tunnel",
                                upstream
                            );

                            // Spawn a task to handle the bidirectional copy after upgrade completes
                            let upstream_str = upstream.clone();
                            tokio::spawn(async move {
                                // Wait for the client upgrade to complete
                                match on_upgrade.await {
                                    Ok(upgraded) => {
                                        let client = TokioIo::new(upgraded);
                                        // Bidirectionally copy data between client and upstream
                                        match bidirectional_copy(client, &mut *upstream_conn).await
                                        {
                                            Ok((sent, recv)) => {
                                                debug!(
                                                    "WebSocket tunnel to {} closed (sent: {}, recv: {})",
                                                    upstream_str, sent, recv
                                                );
                                            }
                                            Err(e) => {
                                                debug!(
                                                    "WebSocket tunnel to {} error: {}",
                                                    upstream_str, e
                                                );
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        warn!("WebSocket client upgrade failed: {}", e);
                                    }
                                }
                            });

                            Ok(response)
                        }
                        WebSocketUpgradeResult::Rejected {
                            status,
                            headers,
                            body,
                        } => {
                            // Upstream rejected the upgrade - return their response
                            debug!(
                                "WebSocket upgrade to {} rejected with status {}",
                                upstream, status
                            );
                            let mut builder = Response::builder().status(status);
                            for (name, value) in headers.iter() {
                                builder = builder.header(name.clone(), value.clone());
                            }
                            let response = builder
                                .body(Full::new(body).map_err(|e| match e {}).boxed())
                                .unwrap();
                            Ok(response)
                        }
                    };
                }

                // Normal HTTP proxy - no body buffering
                let mut response: Response<Incoming> = self
                    .proxy
                    .proxy(
                        request,
                        upstream,
                        *preserve_host,
                        &route.proxy_headers,
                        &merged_response_headers,
                        Some(peer_addr.ip()), // Pass client IP for X-Forwarded-For
                        client_scheme,        // Pass client scheme for X-Forwarded-Proto
                    )
                    .await?;

                // Apply expires header
                apply_expires(&mut response, route.expires);

                // Box the body to unify types (streaming, no conversion)
                Ok(response.map(|body| body.boxed()))
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_server_stats() {
        let stats = ServerStats::default();
        assert_eq!(stats.requests.load(Ordering::Relaxed), 0);
        stats.requests.fetch_add(1, Ordering::Relaxed);
        assert_eq!(stats.requests.load(Ordering::Relaxed), 1);
    }
}
