use std::sync::OnceLock;
use std::time::Duration;

use actix_web::body::{BoxBody, MessageBody};
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::middleware::Next;
use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse};
use http::{HeaderMap, Request, Response};
use opentelemetry::propagation::TextMapCompositePropagator;
use opentelemetry::trace::TracerProvider as _;
use opentelemetry::{Context, global};
use opentelemetry_http::{HeaderExtractor, HeaderInjector};
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};
use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};
use opentelemetry_sdk::trace::SdkTracerProvider;
use tracing::{Instrument as _, Span, info, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

/// Installed tracer provider, kept so shutdown can flush the batch processor.
static TRACER_PROVIDER: OnceLock<SdkTracerProvider> = OnceLock::new();
/// Installed meter provider, kept so shutdown can flush the periodic reader.
static METER_PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();

const METRICS_EXPORT_INTERVAL: Duration = Duration::from_secs(30);

pub fn init_tracing() {
    // W3C trace context is what lets `traceparent` survive HTTP, gRPC and
    // queue payloads across hosts; installing it is a no-op for log-only runs.
    global::set_text_map_propagator(TextMapCompositePropagator::new(vec![
        Box::new(TraceContextPropagator::new()),
        Box::new(BaggagePropagator::new()),
    ]));

    // The filter is installed on the registry rather than on the fmt layer so it
    // governs span *creation*: without it every debug-level span in the process
    // (hyper/tower internals) becomes an exported span.
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
    let subscriber = tracing_subscriber::registry()
        .with(tracing_subscriber::fmt::layer())
        .with(filter);

    match tracer_provider() {
        Some(provider) => {
            let _ = TRACER_PROVIDER.set(provider.clone());
            subscriber
                .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("gitdata")))
                .init();
            info!("opentelemetry trace export enabled");
        }
        None => subscriber.init(),
    }
}

pub fn init_meter_provider() -> Result<SdkMeterProvider, String> {
    if !otlp_configured(OtlpSignal::Metrics) {
        return Err("OTLP endpoint is not configured".to_string());
    }

    let exporter = opentelemetry_otlp::MetricExporter::builder()
        .with_http()
        .with_temporality(Temporality::default())
        .build()
        .map_err(|error| error.to_string())?;

    let reader = PeriodicReader::builder(exporter)
        .with_interval(METRICS_EXPORT_INTERVAL)
        .build();

    let provider = SdkMeterProvider::builder().with_reader(reader).build();
    global::set_meter_provider(provider.clone());
    let _ = METER_PROVIDER.set(provider.clone());
    Ok(provider)
}

#[derive(Clone, Copy)]
enum OtlpSignal {
    Traces,
    Metrics,
}

/// Telemetry is opt-in through the standard OTLP environment variables:
/// setting `OTEL_EXPORTER_OTLP_ENDPOINT` turns export on, anything else
/// (including a bad endpoint) only downgrades to log-only. Telemetry must never
/// stop a host from booting, so every failure path here logs and returns `None`.
fn otlp_configured(signal: OtlpSignal) -> bool {
    let specific = match signal {
        OtlpSignal::Traces => "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
        OtlpSignal::Metrics => "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
    };
    ["OTEL_EXPORTER_OTLP_ENDPOINT", specific].iter().any(|key| {
        std::env::var(key)
            .map(|value| !value.trim().is_empty())
            .unwrap_or(false)
    })
}

fn tracer_provider() -> Option<SdkTracerProvider> {
    if !otlp_configured(OtlpSignal::Traces) {
        return None;
    }

    let exporter = match opentelemetry_otlp::SpanExporter::builder()
        .with_http()
        .build()
    {
        Ok(exporter) => exporter,
        Err(err) => {
            eprintln!("failed to build otlp span exporter, traces disabled: {err}");
            return None;
        }
    };

    Some(
        SdkTracerProvider::builder()
            .with_batch_exporter(exporter)
            .build(),
    )
}

/// Adopts the W3C trace context carried by inbound headers, when present.
pub fn set_remote_parent(span: &Span, headers: &HeaderMap) {
    let parent = global::get_text_map_propagator(|propagator| {
        propagator.extract_with_context(&Context::new(), &HeaderExtractor(headers))
    });
    let _ = span.set_parent(parent);
}

/// Continues a trace across a queue boundary; `None` keeps the ambient parent.
pub fn set_parent_from_traceparent(span: &Span, traceparent: Option<&str>) {
    let Some(traceparent) = traceparent else {
        return;
    };
    let mut headers = HeaderMap::new();
    if let Ok(value) = traceparent.parse() {
        headers.insert("traceparent", value);
    }
    set_remote_parent(span, &headers);
}

/// Serialises the current trace context for a queue payload.
pub fn traceparent() -> Option<String> {
    let mut headers = HeaderMap::new();
    global::get_text_map_propagator(|propagator| {
        propagator.inject_context(
            &Span::current().context(),
            &mut HeaderInjector(&mut headers),
        );
    });
    headers
        .get("traceparent")
        .and_then(|value| value.to_str().ok())
        .map(str::to_owned)
}

/// Serialises the request id for a queue payload.
pub fn request_id() -> Option<String> {
    crate::middleware::request_id()
}

fn span_with(
    method: &str,
    path: &str,
    route: &str,
    request_id: Option<&str>,
    headers: &HeaderMap,
) -> Span {
    let name = format!("{method} {route}");
    let span = tracing::info_span!(
        "http.server",
        "otel.name" = %name,
        "otel.kind" = "server",
        "http.request.method" = %method,
        "url.path" = %path,
        "http.route" = %route,
        "http.response.status_code" = tracing::field::Empty,
        "latency_ms" = tracing::field::Empty,
        "request_id" = tracing::field::Empty,
    );
    if let Some(request_id) = request_id {
        span.record("request_id", request_id);
    }
    set_remote_parent(&span, headers);
    span
}

pub fn http_span_actix(request: &HttpRequest) -> Span {
    let route = request
        .match_pattern()
        .unwrap_or_else(|| request.path().to_string());
    let request_id = request
        .extensions()
        .get::<crate::middleware::RequestId>()
        .map(|id| id.0.clone());
    let mut headers = HeaderMap::new();
    for (name, value) in request.headers() {
        if let (Ok(name), Ok(value)) = (
            http::header::HeaderName::from_bytes(name.as_str().as_bytes()),
            http::HeaderValue::from_bytes(value.as_bytes()),
        ) {
            headers.insert(name, value);
        }
    }
    span_with(
        request.method().as_str(),
        request.path(),
        &route,
        request_id.as_deref(),
        &headers,
    )
}

pub fn http_span<B>(request: &Request<B>) -> Span {
    let path = request.uri().path();
    let request_id = request
        .extensions()
        .get::<crate::middleware::RequestId>()
        .map(|id| id.0.clone());
    span_with(
        request.method().as_str(),
        path,
        path,
        request_id.as_deref(),
        request.headers(),
    )
}

pub fn on_response<B>(response: &HttpResponse<B>, latency: Duration, span: &Span) {
    let status = response.status().as_u16();
    span.record("http.response.status_code", i64::from(status));
    span.record("latency_ms", latency.as_millis() as i64);
    if status >= 500 {
        span.record("otel.status_code", "error");
        span.record("error.type", i64::from(status));
    }
}

pub async fn http_telemetry<B>(
    request: ServiceRequest,
    next: Next<B>,
) -> Result<ServiceResponse<BoxBody>, Error>
where
    B: MessageBody + 'static,
{
    let span = http_span_actix(request.request());
    let started = std::time::Instant::now();
    let response = next.call(request).instrument(span.clone()).await?;
    let latency = started.elapsed();
    on_response(response.response(), latency, &span);
    tracing::info!(
        parent: &span,
        status = response.status().as_u16(),
        latency_ms = latency.as_millis() as u64,
        "finished processing request",
    );
    Ok(response.map_into_boxed_body())
}

/// Body-agnostic tower layer for servers whose body type must stay untouched
/// (tonic gRPC): it adopts the remote parent, creates the server span and runs
/// the inner future inside it.
#[derive(Clone, Copy, Default)]
pub struct TraceContextLayer;

impl<S> tower::Layer<S> for TraceContextLayer {
    type Service = TraceContextService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        TraceContextService { inner }
    }
}

#[derive(Clone)]
pub struct TraceContextService<S> {
    inner: S,
}

impl<S, B, ResBody> tower::Service<Request<B>> for TraceContextService<S>
where
    S: tower::Service<Request<B>, Response = Response<ResBody>>,
    S::Future: Send + 'static,
    B: 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = tracing::instrument::Instrumented<S::Future>;

    fn poll_ready(
        &mut self,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, request: Request<B>) -> Self::Future {
        let span = http_span(&request);
        self.inner.call(request).instrument(span)
    }
}

/// Client span for an outbound HTTP call; the trace context of this span is
/// what gets injected into the request, so the remote hop parents to it.
pub fn http_client_span(method: &str, url: &str) -> Span {
    let name = format!("{method} {url}");
    tracing::info_span!(
        "http.client",
        "otel.name" = %name,
        "otel.kind" = "client",
        "http.request.method" = %method,
        "url.full" = %url,
        "http.response.status_code" = tracing::field::Empty,
    )
}

/// Sends a request inside `span` with `traceparent` injected into its headers.
pub async fn send(
    span: Span,
    request: reqwest::RequestBuilder,
) -> Result<reqwest::Response, reqwest::Error> {
    let mut headers = HeaderMap::new();
    global::get_text_map_propagator(|propagator| {
        propagator.inject_context(&span.context(), &mut HeaderInjector(&mut headers));
    });
    if let Some(request_id) = crate::middleware::request_id()
        && let Ok(value) = request_id.parse()
    {
        headers.insert(http::header::HeaderName::from_static("x-request-id"), value);
    }

    let result = request
        .headers(headers)
        .send()
        .instrument(span.clone())
        .await;
    match &result {
        Ok(response) => {
            span.record(
                "http.response.status_code",
                i64::from(response.status().as_u16()),
            );
        }
        Err(_) => {
            span.record("otel.status_code", "error");
        }
    }
    result
}

pub async fn shutdown_signal() {
    let ctrl_c = async {
        if let Err(err) = tokio::signal::ctrl_c().await {
            warn!(error = %err, "failed to install ctrl_c handler");
        }
    };

    #[cfg(unix)]
    let terminate = async {
        use tokio::signal::unix::{SignalKind, signal};
        match signal(SignalKind::terminate()) {
            Ok(mut s) => {
                s.recv().await;
            }
            Err(err) => {
                warn!(error = %err, "failed to install SIGTERM handler");
            }
        }
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        _ = ctrl_c => info!("received SIGINT"),
        _ = terminate => info!("received SIGTERM"),
    }

    shutdown_telemetry();
}

/// Drains spans queued in the batch processor and metrics queued in the
/// periodic reader. Without this the tail of every deployment's telemetry is
/// lost on rollout, which is exactly when you look at it.
pub fn shutdown_telemetry() {
    flush_traces();
    if let Some(provider) = METER_PROVIDER.get()
        && let Err(err) = provider.shutdown()
    {
        warn!(error = %err, "failed to shut down meter provider");
    }
}

pub fn flush_traces() {
    if let Some(provider) = TRACER_PROVIDER.get()
        && let Err(err) = provider.shutdown()
    {
        warn!(error = %err, "failed to flush tracer provider");
    }
}