warpdrive_proxy/server/

mod.rs

//! Server module - Pingora HTTP server initialization and lifecycle
//!
//! This module handles:
//! - Pingora server initialization
//! - HTTP/HTTPS listener configuration
//! - Server lifecycle (startup, shutdown)
//! - Thread pool configuration
//!
//! Based on Pingora's quick start guide and examples.

use anyhow::{Context, Result, anyhow};
use pingora::apps::HttpServerOptions;
use pingora::prelude::*;
use pingora::server::Server;
use pingora::server::configuration::Opt;
use std::net::TcpListener;
use std::sync::Arc;
use tracing::info;

use crate::acme::{AcmeProvisioner, ChallengeStore};
use crate::config::Config;
use crate::proxy::WarpDriveProxy;

/// Validate that a port can be bound
///
/// This performs a pre-flight check to ensure the port is available before Pingora tries to bind.
/// If binding fails, we provide helpful error messages with common causes and solutions.
fn validate_port_binding(port: u16, protocol: &str) -> Result<()> {
    let addr = format!("0.0.0.0:{}", port);

    match TcpListener::bind(&addr) {
        Ok(_listener) => {
            // Port is available (listener will be dropped immediately)
            Ok(())
        }
        Err(err) => {
            // Build helpful error message based on error kind
            let mut error_msg = format!("Failed to bind {} listener to {}", protocol, addr);
            let mut suggestions = Vec::new();

            match err.kind() {
                std::io::ErrorKind::PermissionDenied => {
                    error_msg.push_str("\n\n❌ Permission denied");

                    if port < 1024 {
                        suggestions.push(format!(
                            "Port {} requires root privileges or CAP_NET_BIND_SERVICE capability",
                            port
                        ));
                        suggestions.push("Solutions:".to_string());
                        suggestions.push("  1. Use an unprivileged port (>= 1024), e.g., 8080 for HTTP, 8443 for HTTPS".to_string());
                        suggestions.push(format!(
                            "     Set WARPDRIVE_HTTP_PORT={} or WARPDRIVE_HTTPS_PORT=8443",
                            if port == 80 { "8080" } else { "8443" }
                        ));
                        suggestions.push(
                            "  2. Run with sudo (NOT recommended for production)".to_string(),
                        );
                        suggestions.push("  3. Grant capability: sudo setcap 'cap_net_bind_service=+ep' $(which warpdrive)".to_string());
                    } else {
                        suggestions
                            .push("This port should not require elevated privileges".to_string());
                        suggestions.push("Check file system permissions or security policies (SELinux, AppArmor)".to_string());
                    }
                }
                std::io::ErrorKind::AddrInUse => {
                    error_msg.push_str("\n\n❌ Address already in use");
                    suggestions.push(format!(
                        "Another process is already listening on port {}",
                        port
                    ));
                    suggestions.push("Solutions:".to_string());
                    suggestions.push("  1. Stop the conflicting process".to_string());
                    suggestions.push(format!(
                        "     Check what's using the port: lsof -i :{} or netstat -tulpn | grep {}",
                        port, port
                    ));
                    suggestions.push(format!(
                        "  2. Use a different port (set WARPDRIVE_{}_PORT=<port>)",
                        if protocol == "HTTP" { "HTTP" } else { "HTTPS" }
                    ));
                    suggestions.push(
                        "  3. In Docker/Kubernetes, ensure no other container is using this port"
                            .to_string(),
                    );
                }
                std::io::ErrorKind::AddrNotAvailable => {
                    error_msg.push_str("\n\n❌ Address not available");
                    suggestions.push(
                        "The IP address 0.0.0.0 cannot be bound (network configuration issue)"
                            .to_string(),
                    );
                    suggestions.push("Check network interface configuration".to_string());
                }
                _ => {
                    error_msg.push_str(&format!("\n\n❌ {}", err));
                    suggestions
                        .push("This may be a network configuration or OS-level issue".to_string());
                }
            }

            // Add environment-specific suggestions
            suggestions.push("".to_string());
            suggestions.push("Environment-specific tips:".to_string());
            suggestions.push(
                "  • Docker: Ensure -p flag matches container port (e.g., -p 8080:8080)"
                    .to_string(),
            );
            suggestions.push("  • AWS ECS: Check port mappings in task definition".to_string());
            suggestions.push(
                "  • Kubernetes: Verify containerPort in pod spec matches config".to_string(),
            );

            for suggestion in suggestions {
                error_msg.push('\n');
                error_msg.push_str(&suggestion);
            }

            Err(anyhow!(error_msg))
        }
    }
}

/// Start the WarpDrive proxy server
///
/// This initializes a Pingora server with the WarpDrive proxy handler,
/// configures HTTP listeners based on the config, and runs the server.
///
/// This function blocks forever (until a shutdown signal is received).
/// Pingora creates its own runtime internally, so this must be called
/// from a non-async context.
///
/// # Errors
///
/// Returns an error if:
/// - Server initialization fails
/// - Port binding fails
/// - Server runtime encounters an error
pub fn start_server(config: Config) -> Result<()> {
    info!("Starting WarpDrive server");

    // Create shared config reference
    let config = Arc::new(config);

    // Create Pingora server configuration
    // Using default options for now (can be customized via Opt struct)
    let mut server = Server::new(Some(Opt {
        upgrade: false,
        daemon: false,
        nocapture: false,
        test: false,
        conf: None,
    }))
    .context("Failed to create Pingora server")?;

    // Bootstrap the server (sets up signal handlers, etc.)
    server.bootstrap();

    info!("Server bootstrapped successfully");

    // Validate port bindings before proceeding
    // This provides clear error messages if ports are unavailable
    info!("Validating port bindings...");
    validate_port_binding(config.http_port, "HTTP").context("HTTP port validation failed")?;

    if config.has_manual_tls() || config.has_acme_domains() {
        validate_port_binding(config.https_port, "HTTPS")
            .context("HTTPS port validation failed")?;
    }

    info!("Port validation successful");

    // Create router if TOML config is present
    let router = if let Some(ref toml_config) = config.toml_config {
        Some(
            crate::router::Router::from_config(toml_config)
                .map_err(|e| anyhow::anyhow!("Failed to create router: {}", e))?,
        )
    } else {
        None
    };

    // Initialize ACME challenge store
    let challenge_store = ChallengeStore::default();

    // Initialize ACME provisioner if ACME domains are configured
    if config.has_acme_domains() {
        info!("Initializing ACME certificate provisioner");

        // Create a runtime for async ACME operations
        let runtime =
            tokio::runtime::Runtime::new().context("Failed to create Tokio runtime for ACME")?;

        runtime.block_on(async {
            let provisioner = AcmeProvisioner::new(config.clone(), challenge_store.clone())
                .context("Failed to create ACME provisioner")?;

            // Provision certificates for all ACME domains
            for domain in &config.tls_domains {
                info!("Provisioning certificate for domain: {}", domain);
                match provisioner.provision_certificate(domain).await {
                    Ok((cert_path, key_path)) => {
                        info!("Certificate provisioned successfully for {}", domain);
                        info!("  Certificate: {}", cert_path);
                        info!("  Private key: {}", key_path);
                    }
                    Err(e) => {
                        return Err(anyhow::anyhow!(
                            "Failed to provision certificate for {}: {}",
                            domain,
                            e
                        ));
                    }
                }
            }
            Ok::<(), anyhow::Error>(())
        })?;
    }

    // Create proxy service
    let proxy = WarpDriveProxy::new(config.clone(), router, challenge_store);

    // Create the proxy service
    // This wraps our ProxyHttp implementation in Pingora's service infrastructure
    let mut proxy_service = http_proxy_service(&server.configuration, proxy);

    // Enable HTTP/2 cleartext (h2c) support if configured
    // This allows WarpDrive to accept HTTP/2 over plain TCP (no TLS) for AWS ALB and Cloud Run
    if config.h2c_enabled {
        info!("Enabling HTTP/2 cleartext (h2c) support");
        if let Some(app_logic) = proxy_service.app_logic_mut() {
            let mut server_opts = HttpServerOptions::default();
            server_opts.h2c = true;
            app_logic.server_options = Some(server_opts);
        }
    }

    // Add HTTP listener
    let http_addr = format!("0.0.0.0:{}", config.http_port);
    info!("Adding HTTP listener on {}", http_addr);
    proxy_service.add_tcp(&http_addr);

    // Add HTTPS listener when TLS is configured (manual cert/key paths)
    if config.has_manual_tls() {
        let https_addr = format!("0.0.0.0:{}", config.https_port);
        let cert_path = config.tls_cert_path.as_ref().unwrap();
        let key_path = config.tls_key_path.as_ref().unwrap();

        info!("Adding HTTPS listener on {}", https_addr);
        info!("  Certificate: {}", cert_path);
        info!("  Private key: {}", key_path);

        proxy_service
            .add_tls(&https_addr, cert_path, key_path)
            .context("Failed to add TLS listener")?;
    }

    // Add the proxy service to the server
    server.add_service(proxy_service);

    info!("WarpDrive proxy server ready");
    info!("  HTTP listening on: {}", http_addr);
    if config.has_manual_tls() {
        info!("  HTTPS listening on: 0.0.0.0:{}", config.https_port);
    }
    info!(
        "  Proxying to: {}:{}",
        config.target_host, config.target_port
    );

    // Run the server forever
    // This blocks until a shutdown signal is received (SIGTERM, SIGINT)
    info!("Server started successfully");
    server.run_forever()
}

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

    #[test]
    fn test_server_config() {
        let config = Config::default();
        assert_eq!(config.http_port, 8080);
        assert_eq!(config.target_port, 3000);
    }

    #[test]
    fn test_port_validation_success() {
        // Find an available port by binding and immediately dropping
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        drop(listener);

        // Validation should succeed on available port
        let result = validate_port_binding(port, "HTTP");
        assert!(
            result.is_ok(),
            "Validation should succeed for available port"
        );
    }

    #[test]
    fn test_port_validation_address_in_use() {
        // Bind to 0.0.0.0 (same as validation function) and keep it occupied
        let listener = TcpListener::bind("0.0.0.0:0").unwrap();
        let port = listener.local_addr().unwrap().port();

        // Validation should fail with "Address already in use"
        let result = validate_port_binding(port, "HTTP");
        assert!(result.is_err(), "Validation should fail for occupied port");

        let err = result.unwrap_err();
        let err_str = format!("{:?}", err);
        assert!(
            err_str.contains("Address already in use"),
            "Error should mention 'Address already in use'"
        );
        assert!(
            err_str.contains("lsof -i"),
            "Error should suggest using lsof"
        );
        assert!(
            err_str.contains("WARPDRIVE_HTTP_PORT"),
            "Error should mention WARPDRIVE_HTTP_PORT"
        );
    }
}