use std::sync::Arc;

use anyhow::Result;
use async_trait::async_trait;
use bytes::Bytes;
use folk_api::rpc::BoxFuture;
use folk_api::{PluginContext, RpcMethodDef, ServerPlugin};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use serde::Deserialize;
use tokio::net::TcpListener;
use tracing::info;

use crate::config::MetricsConfig;
use crate::custom::CustomMetrics;
use crate::handler::MetricsHandler;

pub struct MetricsPlugin {
    config: MetricsConfig,
}

impl MetricsPlugin {
    pub fn new(config: MetricsConfig) -> Self {
        Self { config }
    }
}

#[derive(Deserialize)]
struct IncrementRequest {
    name: String,
    #[serde(default)]
    labels: Vec<String>,
    #[serde(default = "default_increment_value")]
    value: u64,
}

fn default_increment_value() -> u64 {
    1
}

#[derive(Deserialize)]
struct ObserveRequest {
    name: String,
    #[serde(default)]
    labels: Vec<String>,
    value: f64,
}

#[derive(Deserialize)]
struct SetRequest {
    name: String,
    #[serde(default)]
    labels: Vec<String>,
    value: i64,
}

#[async_trait]
impl ServerPlugin for MetricsPlugin {
    fn name(&self) -> &'static str {
        "metrics"
    }

    /// Phase 79: the metrics scrape server lives in the master process. Workers
    /// record into the shared-memory registry; the master renders `/metrics`.
    fn placement(&self) -> folk_api::Placement {
        folk_api::Placement::MasterOnly
    }

    async fn run(&self, ctx: PluginContext) -> Result<()> {
        let listener = TcpListener::bind(self.config.listen).await?;
        info!(
            listen = %self.config.listen,
            metrics_path = %self.config.metrics_path,
            health_path = %self.config.health_path,
            ready_path = %self.config.ready_path,
            collectors = self.config.collectors.len(),
            "metrics server listening",
        );

        // Create custom metrics from config
        let custom = ctx.metrics_registry.as_ref().map(|reg| {
            Arc::new(CustomMetrics::from_config(
                &self.config.collectors,
                &self.config.prefix,
                reg.as_ref(),
            ))
        });

        // Register RPC handlers
        if let Some(rpc) = &ctx.rpc_registrar {
            // metrics.increment — increment a custom counter
            if let Some(cm) = &custom {
                let cm = cm.clone();
                rpc.register_raw(
                    "metrics.increment".into(),
                    Arc::new(move |payload: Bytes| -> BoxFuture<'static, Result<Bytes>> {
                        let cm = cm.clone();
                        Box::pin(async move {
                            let req: IncrementRequest = serde_json::from_slice(&payload)?;
                            let labels: Vec<&str> = req.labels.iter().map(String::as_str).collect();
                            cm.increment(&req.name, &labels, req.value)?;
                            Ok(Bytes::from(serde_json::to_vec(&"ok")?))
                        })
                    }),
                )
                .await;
            }

            // metrics.observe — observe a histogram value
            if let Some(cm) = &custom {
                let cm = cm.clone();
                rpc.register_raw(
                    "metrics.observe".into(),
                    Arc::new(move |payload: Bytes| -> BoxFuture<'static, Result<Bytes>> {
                        let cm = cm.clone();
                        Box::pin(async move {
                            let req: ObserveRequest = serde_json::from_slice(&payload)?;
                            let labels: Vec<&str> = req.labels.iter().map(String::as_str).collect();
                            cm.observe(&req.name, &labels, req.value)?;
                            Ok(Bytes::from(serde_json::to_vec(&"ok")?))
                        })
                    }),
                )
                .await;
            }

            // metrics.set — set a gauge value
            if let Some(cm) = &custom {
                let cm = cm.clone();
                rpc.register_raw(
                    "metrics.set".into(),
                    Arc::new(move |payload: Bytes| -> BoxFuture<'static, Result<Bytes>> {
                        let cm = cm.clone();
                        Box::pin(async move {
                            let req: SetRequest = serde_json::from_slice(&payload)?;
                            let labels: Vec<&str> = req.labels.iter().map(String::as_str).collect();
                            cm.set(&req.name, &labels, req.value)?;
                            Ok(Bytes::from(serde_json::to_vec(&"ok")?))
                        })
                    }),
                )
                .await;
            }

            // metrics.render — return current Prometheus snapshot
            let registry = ctx.metrics_registry.clone();
            let disabled = self.config.disabled_prefixes();
            rpc.register_raw(
                "metrics.render".into(),
                Arc::new(move |_: Bytes| -> BoxFuture<'static, Result<Bytes>> {
                    let registry = registry.clone();
                    let disabled = disabled.clone();
                    Box::pin(async move {
                        let body = match &registry {
                            Some(reg) => {
                                let rendered = reg.render();
                                if disabled.is_empty() {
                                    rendered
                                } else {
                                    crate::handler::filter_metrics(&rendered, &disabled)
                                }
                            }
                            None => "# metrics registry not available\n".into(),
                        };
                        Ok(Bytes::from(serde_json::to_vec(&body)?))
                    })
                }),
            )
            .await;
        }

        let handler = Arc::new(MetricsHandler {
            metrics: ctx.metrics_registry.clone(),
            health: ctx.health_registry.clone(),
            metrics_path: self.config.metrics_path.clone(),
            health_path: self.config.health_path.clone(),
            ready_path: self.config.ready_path.clone(),
            disabled_prefixes: self.config.disabled_prefixes(),
        });

        let mut shutdown = ctx.shutdown.clone();
        loop {
            tokio::select! {
                accept = listener.accept() => {
                    let (stream, _) = accept?;
                    let h = handler.clone();
                    tokio::spawn(async move {
                        let io = TokioIo::new(stream);
                        let _ = http1::Builder::new()
                            .serve_connection(io, service_fn(move |req| {
                                let h = h.clone();
                                async move { h.handle(req).await }
                            }))
                            .await;
                    });
                }
                _ = shutdown.changed() => {
                    if *shutdown.borrow() { break; }
                }
            }
        }
        Ok(())
    }

    fn rpc_methods(&self) -> Vec<RpcMethodDef> {
        vec![
            RpcMethodDef::new("metrics.render", "current Prometheus metrics snapshot"),
            RpcMethodDef::new("metrics.increment", "increment a custom counter"),
            RpcMethodDef::new("metrics.observe", "observe a custom histogram value"),
            RpcMethodDef::new("metrics.set", "set a custom gauge value"),
        ]
    }
}
