//! Prometheus-format metrics server
//!
//! Exposes metrics on 127.0.0.1:{port}/metrics

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use rama::{
    http::{Body, Request, Response, StatusCode, server::HttpServer},
    rt::Executor,
    service::service_fn,
    tcp::server::TcpListener,
};
use tokio::sync::watch;
use tracing::{error, info};

use crate::fleet::Fleet;

/// Metrics registry - thread-safe counters and gauges
pub struct MetricsRegistry {
    /// Request counters per route (bind:pattern -> count)
    request_counts: dashmap::DashMap<String, AtomicU64>,
    /// Request latency sum per route (for avg calculation)
    request_latency_sum: dashmap::DashMap<String, AtomicU64>,
}

impl MetricsRegistry {
    pub fn new() -> Self {
        Self {
            request_counts: dashmap::DashMap::new(),
            request_latency_sum: dashmap::DashMap::new(),
        }
    }

    /// Record a request
    pub fn record_request(&self, route: &str, latency_ms: u64) {
        self.request_counts
            .entry(route.to_string())
            .or_insert_with(|| AtomicU64::new(0))
            .fetch_add(1, Ordering::Relaxed);

        self.request_latency_sum
            .entry(route.to_string())
            .or_insert_with(|| AtomicU64::new(0))
            .fetch_add(latency_ms, Ordering::Relaxed);
    }

    /// Get all request counts
    pub fn get_request_counts(&self) -> Vec<(String, u64)> {
        self.request_counts
            .iter()
            .map(|e| (e.key().clone(), e.value().load(Ordering::Relaxed)))
            .collect()
    }

    /// Get all latency sums
    pub fn get_latency_sums(&self) -> Vec<(String, u64)> {
        self.request_latency_sum
            .iter()
            .map(|e| (e.key().clone(), e.value().load(Ordering::Relaxed)))
            .collect()
    }
}

impl Default for MetricsRegistry {
    fn default() -> Self {
        Self::new()
    }
}

/// Metrics server
pub struct MetricsServer {
    port: u16,
    fleet: Arc<Fleet>,
    registry: Arc<MetricsRegistry>,
}

impl MetricsServer {
    pub fn new(port: u16, fleet: Arc<Fleet>, registry: Arc<MetricsRegistry>) -> Self {
        Self { port, fleet, registry }
    }

    /// Run the metrics server
    pub async fn run(self, mut shutdown: watch::Receiver<bool>) {
        let addr = format!("127.0.0.1:{}", self.port);

        let fleet = self.fleet;
        let registry = self.registry;

        let tcp_listener = match TcpListener::build().bind(&addr).await {
            Ok(l) => l,
            Err(e) => {
                error!(error = %e, addr = %addr, "Failed to bind metrics server");
                return;
            }
        };

        info!(addr = %addr, "Metrics server listening");

        let exec = Executor::default();
        let http_service = HttpServer::auto(exec).service(service_fn(move |req: Request| {
            let fleet = fleet.clone();
            let registry = registry.clone();
            async move {
                let path = req.uri().path();

                match path {
                    "/metrics" => {
                        let body = render_metrics(&fleet, &registry).await;
                        Ok::<_, std::convert::Infallible>(Response::builder()
                            .status(StatusCode::OK)
                            .header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
                            .body(Body::from(body))
                            .unwrap())
                    }
                    "/health" => {
                        Ok(Response::builder()
                            .status(StatusCode::OK)
                            .body(Body::from("ok"))
                            .unwrap())
                    }
                    _ => {
                        Ok(Response::builder()
                            .status(StatusCode::NOT_FOUND)
                            .body(Body::from("Not found. Try /metrics"))
                            .unwrap())
                    }
                }
            }
        }));

        tokio::select! {
            _ = tcp_listener.serve(http_service) => {
                info!("Metrics server stopped");
            }
            _ = shutdown.changed() => {
                info!("Metrics server shutting down");
            }
        }
    }
}

/// Render metrics in Prometheus format
async fn render_metrics(fleet: &Fleet, registry: &MetricsRegistry) -> String {
    let mut output = String::with_capacity(4096);

    // Ship status gauge (1 = running, 0 = not running)
    output.push_str("# HELP mothership_ship_status Ship status (1=running, 0=stopped)\n");
    output.push_str("# TYPE mothership_ship_status gauge\n");

    for (name, ship) in &fleet.ships {
        let snapshot = ship.snapshot().await;
        let status_value = match snapshot.status() {
            crate::fleet::ShipStatus::Running => 1,
            _ => 0,
        };
        output.push_str(&format!(
            "mothership_ship_status{{ship=\"{}\",group=\"{}\"}} {}\n",
            name, ship.group, status_value
        ));
    }

    // Ship health gauge (1 = healthy, 0 = unhealthy)
    output.push_str("\n# HELP mothership_ship_healthy Ship health (1=healthy, 0=unhealthy)\n");
    output.push_str("# TYPE mothership_ship_healthy gauge\n");

    for (name, ship) in &fleet.ships {
        let snapshot = ship.snapshot().await;
        let healthy = match snapshot.status() {
            crate::fleet::ShipStatus::Running => 1,
            crate::fleet::ShipStatus::Unhealthy => 0,
            _ => 0,
        };
        output.push_str(&format!(
            "mothership_ship_healthy{{ship=\"{}\",group=\"{}\"}} {}\n",
            name, ship.group, healthy
        ));
    }

    // Ship restart count
    output.push_str("\n# HELP mothership_ship_restarts_total Total restart count per ship\n");
    output.push_str("# TYPE mothership_ship_restarts_total counter\n");

    for (name, ship) in &fleet.ships {
        let restarts = ship.restart_count().await;
        output.push_str(&format!(
            "mothership_ship_restarts_total{{ship=\"{}\",group=\"{}\"}} {}\n",
            name, ship.group, restarts
        ));
    }

    // Request counts per route
    let counts = registry.get_request_counts();
    if !counts.is_empty() {
        output.push_str("\n# HELP mothership_requests_total Total requests per route\n");
        output.push_str("# TYPE mothership_requests_total counter\n");

        for (route, count) in counts {
            output.push_str(&format!(
                "mothership_requests_total{{route=\"{}\"}} {}\n",
                route.replace('"', "\\\""), count
            ));
        }
    }

    // Request latency sum (for calculating averages)
    let latencies = registry.get_latency_sums();
    if !latencies.is_empty() {
        output.push_str("\n# HELP mothership_request_latency_ms_sum Sum of request latencies in ms\n");
        output.push_str("# TYPE mothership_request_latency_ms_sum counter\n");

        for (route, sum) in latencies {
            output.push_str(&format!(
                "mothership_request_latency_ms_sum{{route=\"{}\"}} {}\n",
                route.replace('"', "\\\""), sum
            ));
        }
    }

    // Fleet info
    output.push_str("\n# HELP mothership_fleet_ships_total Total number of ships\n");
    output.push_str("# TYPE mothership_fleet_ships_total gauge\n");
    output.push_str(&format!("mothership_fleet_ships_total {}\n", fleet.ships.len()));

    // Groups info
    output.push_str("\n# HELP mothership_fleet_groups_total Total number of groups\n");
    output.push_str("# TYPE mothership_fleet_groups_total gauge\n");
    output.push_str(&format!("mothership_fleet_groups_total {}\n", fleet.groups.len()));

    output
}
