//! HTTP reverse proxy using Rama
//!
//! Routes external traffic to ships based on configured route patterns.
//! Supports multiple named binds (http, https, ws, etc.).
//! Integrates WASM modules for request/response processing.
//! Uses circuit breakers to protect against slow/failing backends.

use std::collections::HashMap;
use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;

use breaker_machines::CircuitBreaker;
use rama::{
    Context, Layer, Service,
    http::{
        Body, Request, Response, StatusCode, Uri,
        client::EasyHttpWebClient,
        layer::trace::TraceLayer,
        server::HttpServer,
    },
    net::stream::layer::http::BodyLimitLayer,
    rt::Executor,
    service::service_fn,
    tcp::server::TcpListener,
};
use regex::Regex;
use tokio::sync::{Mutex, RwLock};
use tracing::{error, info, warn};

use crate::config::{Bind, Manifest};
use crate::modules::{ModuleAction, ModuleRuntime, RequestInfo, ResponseInfo};

/// Route configuration for a specific bind
#[derive(Clone)]
struct Route {
    /// Pattern to match (regex)
    pattern: Regex,
    /// Backend to forward to
    backend: String,
    /// Ship name (for logging)
    ship_name: String,
    /// Circuit breaker for this backend
    circuit_breaker: Arc<Mutex<CircuitBreaker>>,
}

/// Default circuit breaker settings
fn create_circuit_breaker(name: &str) -> CircuitBreaker {
    CircuitBreaker::builder(name)
        .failure_threshold(5)           // Open after 5 failures
        .failure_window_secs(60.0)      // Within 60 seconds
        .half_open_timeout_secs(30.0)   // Try again after 30 seconds
        .success_threshold(2)           // Close after 2 successes
        .on_open(|name| warn!(backend = %name, "Circuit breaker opened"))
        .on_close(|name| info!(backend = %name, "Circuit breaker closed"))
        .build()
}

/// Routes for a specific bind
#[derive(Clone, Default)]
struct BindRoutes {
    routes: Vec<Route>,
}

/// HTTP exposure layer - manages multiple listeners
pub struct HttpExposure {
    /// Named binds (bind_name -> address)
    binds: HashMap<String, Bind>,
    /// Routes per bind (bind_name -> routes)
    routes: HashMap<String, BindRoutes>,
    /// WASM module runtime
    module_runtime: Arc<RwLock<Option<ModuleRuntime>>>,
    /// Module configs for lazy loading
    module_configs: Vec<crate::config::Module>,
}

impl HttpExposure {
    /// Create HTTP exposure from manifest
    ///
    /// Returns None if no binds are configured.
    pub fn from_manifest(manifest: &Manifest) -> Option<Self> {
        if manifest.mothership.bind.is_empty() {
            return None;
        }

        // Clone binds from mothership config
        let binds = manifest.mothership.bind.clone();

        // Build routes per bind from ships
        let mut routes: HashMap<String, BindRoutes> = HashMap::new();

        // Initialize empty routes for each bind
        for bind_name in binds.keys() {
            routes.insert(bind_name.clone(), BindRoutes::default());
        }

        // Process ships and their routes
        for ships in manifest.fleet.values() {
            for ship in ships {
                if ship.routes.is_empty() {
                    continue;
                }

                // Get backend URL from ship's internal bind address
                let backend = match &ship.bind {
                    Some(b) => b.healthcheck_base(),
                    None => {
                        warn!(
                            ship = %ship.name,
                            "Ship has routes but no bind address, skipping"
                        );
                        continue;
                    }
                };

                for route_config in &ship.routes {
                    // Check if the bind exists
                    if !binds.contains_key(&route_config.bind) {
                        warn!(
                            ship = %ship.name,
                            bind = %route_config.bind,
                            "Route references unknown bind, skipping"
                        );
                        continue;
                    }

                    match Regex::new(&route_config.pattern) {
                        Ok(regex) => {
                            let bind_routes = routes
                                .entry(route_config.bind.clone())
                                .or_default();
                            let breaker_name = format!("{}:{}", ship.name, route_config.bind);
                            bind_routes.routes.push(Route {
                                pattern: regex,
                                backend: backend.clone(),
                                ship_name: ship.name.clone(),
                                circuit_breaker: Arc::new(Mutex::new(
                                    create_circuit_breaker(&breaker_name),
                                )),
                            });
                            info!(
                                ship = %ship.name,
                                bind = %route_config.bind,
                                pattern = %route_config.pattern,
                                backend = %backend,
                                "Registered route with circuit breaker"
                            );
                        }
                        Err(e) => {
                            error!(
                                ship = %ship.name,
                                bind = %route_config.bind,
                                pattern = %route_config.pattern,
                                error = %e,
                                "Invalid route pattern"
                            );
                        }
                    }
                }
            }
        }

        Some(Self {
            binds,
            routes,
            module_runtime: Arc::new(RwLock::new(None)),
            module_configs: manifest.modules.clone(),
        })
    }

    /// Run all HTTP servers
    pub async fn run(self, shutdown: tokio::sync::watch::Receiver<bool>) -> anyhow::Result<()> {
        info!(binds = self.binds.len(), "Starting HTTP exposure layer");

        // Load WASM modules if configured
        if !self.module_configs.is_empty() {
            match ModuleRuntime::new() {
                Ok(runtime) => {
                    if let Err(e) = runtime.load_modules(&self.module_configs).await {
                        warn!(error = %e, "Failed to load some WASM modules");
                    }
                    let names = runtime.module_names().await;
                    if !names.is_empty() {
                        info!(modules = ?names, "Loaded WASM modules");
                    }
                    *self.module_runtime.write().await = Some(runtime);
                }
                Err(e) => {
                    warn!(error = %e, "Failed to create WASM runtime, modules disabled");
                }
            }
        }

        let graceful = rama::graceful::Shutdown::default();
        let module_runtime = self.module_runtime.clone();

        // Start a listener for each bind
        for (bind_name, bind_addr) in &self.binds {
            let addr = match bind_addr {
                Bind::Tcp { host, port } => format!("{}:{}", host, port),
                Bind::Unix { path } => {
                    warn!(bind = %bind_name, path = %path, "Unix sockets not yet supported");
                    continue;
                }
            };

            let routes = self.routes.get(bind_name).cloned().unwrap_or_default();
            let bind_name = bind_name.clone();
            let module_runtime = module_runtime.clone();

            info!(bind = %bind_name, addr = %addr, routes = routes.routes.len(), "Starting listener");

            graceful.spawn_task_fn(async move |guard| {
                let tcp_listener = match TcpListener::build().bind(&addr).await {
                    Ok(l) => l,
                    Err(e) => {
                        error!(bind = %bind_name, error = %e, "Failed to bind");
                        return;
                    }
                };

                let exec = Executor::graceful(guard.clone());
                let routes = Arc::new(routes);
                let module_runtime = module_runtime.clone();

                let http_service = HttpServer::auto(exec).service(
                    (TraceLayer::new_for_http(),).into_layer(service_fn(move |req: Request| {
                        let routes = routes.clone();
                        let modules = module_runtime.clone();
                        async move { reverse_proxy(req, &routes, modules).await }
                    })),
                );

                tcp_listener
                    .serve_graceful(
                        guard,
                        BodyLimitLayer::symmetric(10 * 1024 * 1024).into_layer(http_service),
                    )
                    .await;
            });
        }

        // Wait for shutdown signal
        let mut shutdown = shutdown;
        loop {
            tokio::select! {
                _ = shutdown.changed() => {
                    if *shutdown.borrow() {
                        break;
                    }
                }
                _ = tokio::time::sleep(Duration::from_secs(1)) => {}
            }
        }

        graceful
            .shutdown_with_limit(Duration::from_secs(30))
            .await
            .map_err(|e| anyhow::anyhow!("Graceful shutdown failed: {}", e))?;

        info!("HTTP exposure layer stopped");
        Ok(())
    }
}

/// Reverse proxy handler
async fn reverse_proxy(
    req: Request,
    routes: &BindRoutes,
    module_runtime: Arc<RwLock<Option<ModuleRuntime>>>,
) -> Result<Response, Infallible> {
    let path = req.uri().path().to_string();
    let method = req.method().to_string();
    let query = req.uri().query().map(|s| s.to_string());

    // Extract headers for module processing
    let headers: std::collections::HashMap<String, String> = req
        .headers()
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
        .collect();

    // Process request through WASM modules
    let request_info = RequestInfo {
        method: method.clone(),
        path: path.clone(),
        query: query.clone(),
        headers: headers.clone(),
    };

    let modules = module_runtime.read().await;
    if let Some(ref runtime) = *modules {
        match runtime.process_request(&path, &request_info).await {
            ModuleAction::Block { status, body } => {
                return Ok(Response::builder()
                    .status(StatusCode::from_u16(status).unwrap_or(StatusCode::FORBIDDEN))
                    .body(Body::from(body))
                    .unwrap());
            }
            ModuleAction::Modify {
                path: new_path,
                headers: new_headers,
            } => {
                if new_path.is_some() || new_headers.is_some() {
                    info!(new_path = ?new_path, "Module modified request");
                }
            }
            ModuleAction::Continue => {}
        }
    }
    drop(modules);

    // Find matching route
    let route = routes.routes.iter().find(|r| r.pattern.is_match(&path));

    let route = match route {
        Some(r) => r,
        None => {
            return Ok(Response::builder()
                .status(StatusCode::NOT_FOUND)
                .body(Body::from("No route matched"))
                .unwrap());
        }
    };

    let backend = &route.backend;
    let ship_name = &route.ship_name;
    let circuit_breaker = &route.circuit_breaker;

    // Check circuit breaker state
    {
        let breaker = circuit_breaker.lock().await;
        if breaker.is_open() {
            warn!(ship = %ship_name, backend = %backend, "Circuit breaker open, rejecting request");
            return Ok(Response::builder()
                .status(StatusCode::SERVICE_UNAVAILABLE)
                .body(Body::from("Service temporarily unavailable"))
                .unwrap());
        }
    }

    tracing::debug!(ship = %ship_name, path = %path, backend = %backend, "Routing request");

    // Rewrite URI to backend
    let uri = req.uri();
    let new_uri = format!(
        "{}{}{}",
        backend,
        uri.path(),
        uri.query().map(|q| format!("?{}", q)).unwrap_or_default()
    );

    let new_uri: Uri = match new_uri.parse() {
        Ok(u) => u,
        Err(e) => {
            error!(error = %e, uri = %new_uri, "Failed to parse backend URI");
            return Ok(Response::builder()
                .status(StatusCode::INTERNAL_SERVER_ERROR)
                .body(Body::from("Invalid backend URI"))
                .unwrap());
        }
    };

    // Build new request with rewritten URI
    let (mut parts, body) = req.into_parts();
    parts.uri = new_uri;
    let req = Request::from_parts(parts, body);

    // Forward to backend with timeout
    let client = EasyHttpWebClient::default();
    let timeout = Duration::from_secs(30);
    let start = std::time::Instant::now();
    let ctx: Context<()> = Context::default();
    let resp = match tokio::time::timeout(timeout, client.serve(ctx, req)).await {
        Ok(Ok(resp)) => {
            // Record success with duration
            let duration_ms = start.elapsed().as_secs_f64() * 1000.0;
            let breaker = circuit_breaker.lock().await;
            breaker.record_success(duration_ms);
            resp
        }
        Ok(Err(e)) => {
            // Record failure with duration
            let duration_ms = start.elapsed().as_secs_f64() * 1000.0;
            {
                let breaker = circuit_breaker.lock().await;
                breaker.record_failure(duration_ms);
            }
            error!(error = %e, backend = %backend, "Backend request failed");
            return Ok(Response::builder()
                .status(StatusCode::BAD_GATEWAY)
                .body(Body::from("Backend unavailable"))
                .unwrap());
        }
        Err(_) => {
            // Timeout - record as failure (30s)
            {
                let breaker = circuit_breaker.lock().await;
                breaker.record_failure(30_000.0);
            }
            error!(backend = %backend, timeout_secs = 30, "Backend request timed out");
            return Ok(Response::builder()
                .status(StatusCode::GATEWAY_TIMEOUT)
                .body(Body::from("Backend timeout"))
                .unwrap());
        }
    };

    // Process response through WASM modules
    let response_headers: std::collections::HashMap<String, String> = resp
        .headers()
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
        .collect();

    let response_info = ResponseInfo {
        status: resp.status().as_u16(),
        headers: response_headers,
    };

    let modules = module_runtime.read().await;
    if let Some(ref runtime) = *modules {
        match runtime.process_response(&path, &response_info).await {
            ModuleAction::Block { status, body } => {
                return Ok(Response::builder()
                    .status(StatusCode::from_u16(status).unwrap_or(StatusCode::FORBIDDEN))
                    .body(Body::from(body))
                    .unwrap());
            }
            ModuleAction::Modify {
                headers: new_headers,
                ..
            } => {
                if new_headers.is_some() {
                    info!("Module modified response headers");
                }
            }
            ModuleAction::Continue => {}
        }
    }

    Ok(resp)
}
