use std::future::{Future, Ready, ready};
use std::pin::Pin;
use std::rc::Rc;
use std::sync::OnceLock;
use std::time::Instant;

use actix_web::HttpMessage;
use actix_web::body::MessageBody;
use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform, forward_ready};
use opentelemetry::metrics::{Counter, Histogram};
use opentelemetry::trace::Status;
use opentelemetry::{KeyValue, global};
use tracing::Instrument;
use tracing_opentelemetry::OpenTelemetrySpanExt;

use crate::{TRACEPARENT, set_parent_from_traceparent};

pub struct HttpTelemetry {
    scope: &'static str,
}

impl HttpTelemetry {
    pub fn new(service: impl AsRef<str>) -> Self {
        Self {
            scope: Box::leak(format!("{}.http", service.as_ref()).into_boxed_str()),
        }
    }
}

impl<S, B> Transform<S, ServiceRequest> for HttpTelemetry
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = actix_web::Error> + 'static,
    S::Future: 'static,
    B: MessageBody + 'static,
{
    type Response = ServiceResponse<B>;
    type Error = actix_web::Error;
    type Transform = HttpTelemetryMiddleware<S>;
    type InitError = ();
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ready(Ok(HttpTelemetryMiddleware {
            service: Rc::new(service),
            metrics: metrics(self.scope),
        }))
    }
}

pub struct HttpTelemetryMiddleware<S> {
    service: Rc<S>,
    metrics: &'static HttpMetrics,
}

impl<S, B> Service<ServiceRequest> for HttpTelemetryMiddleware<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = actix_web::Error> + 'static,
    S::Future: 'static,
    B: MessageBody + 'static,
{
    type Response = ServiceResponse<B>;
    type Error = actix_web::Error;
    #[allow(clippy::type_complexity)]
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;

    forward_ready!(service);

    fn call(&self, request: ServiceRequest) -> Self::Future {
        let service = Rc::clone(&self.service);
        let metrics = self.metrics;
        let method = request.method().as_str().to_owned();
        let path = request.path().to_owned();
        let request_id = request
            .extensions()
            .get::<envelope::RequestId>()
            .map(|value| value.0.clone())
            .filter(|value| !value.is_empty())
            .unwrap_or_default();
        let traceparent = request
            .headers()
            .get(TRACEPARENT)
            .and_then(|value| value.to_str().ok())
            .map(str::to_owned);

        let span = tracing::info_span!(
            "http.server.request",
            "otel.kind" = "server",
            "http.request.method" = %method,
            "url.path" = %path,
            "http.response.status_code" = tracing::field::Empty,
            request_id = %request_id,
        );
        set_parent_from_traceparent(&span, traceparent.as_deref());

        let instrumented = span.clone();
        let future = async move {
            let started = Instant::now();
            let result = service.call(request).await;
            let elapsed = started.elapsed().as_secs_f64();
            let status = match &result {
                Ok(response) => response.status().as_u16(),
                Err(error) => error.as_response_error().status_code().as_u16(),
            };

            span.record("http.response.status_code", status);
            if status >= 500 {
                span.set_status(Status::error(format!("http {status}")));
            }

            log(method.as_str(), path.as_str(), status, elapsed, &request_id);

            let attributes = [
                KeyValue::new("http.request.method", method),
                KeyValue::new("http.response.status_code", i64::from(status)),
            ];
            metrics.requests.add(1, &attributes);
            metrics.duration.record(elapsed, &attributes);

            result
        };

        Box::pin(future.instrument(instrumented))
    }
}

struct HttpMetrics {
    requests: Counter<u64>,
    duration: Histogram<f64>,
}

/// 静态资源单独降级：一次页面加载会带出几十个 chunk，按 info 打印会淹没请求日志。
/// 探针路径：编排器高频轮询，降到 DEBUG 以免淹没业务日志。
const QUIET_PREFIX: &str = "/assets/";
const PROBE_PATHS: [&str; 2] = ["/healthz", "/readyz"];

fn log(method: &str, path: &str, status: u16, elapsed: f64, request_id: &str) {
    let elapsed_ms = ((elapsed * 1_000.0) * 1_000.0).round() / 1_000.0;
    let request_id = if request_id.is_empty() {
        "-"
    } else {
        request_id
    };
    if path.starts_with(QUIET_PREFIX) || PROBE_PATHS.contains(&path) {
        tracing::debug!(
            "http.request.method" = method,
            "url.path" = path,
            "http.response.status_code" = status,
            "duration_ms" = elapsed_ms,
            "request_id" = request_id,
            "http request"
        );
    } else {
        tracing::info!(
            "http.request.method" = method,
            "url.path" = path,
            "http.response.status_code" = status,
            "duration_ms" = elapsed_ms,
            "request_id" = request_id,
            "http request"
        );
    }
}

fn metrics(scope: &'static str) -> &'static HttpMetrics {
    static METRICS: OnceLock<HttpMetrics> = OnceLock::new();
    METRICS.get_or_init(|| {
        let meter = global::meter(scope);
        HttpMetrics {
            requests: meter.u64_counter("http.server.request.count").build(),
            duration: meter
                .f64_histogram("http.server.request.duration")
                .with_unit("s")
                .build(),
        }
    })
}