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

use axum::extract::MatchedPath;
use axum::http::{HeaderMap, Request, Response};
use opentelemetry::Context;
use opentelemetry::global;
use opentelemetry::propagation::TextMapCompositePropagator;
use opentelemetry::trace::TracerProvider as _;
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)
}

/// HTTP server metrics for axum-style routers. Paths without a matched route
/// (static assets, the SPA fallback) are skipped so raw URLs never become
/// metric labels.
pub fn metrics_layer() -> axum_otel_metrics::HttpMetricsLayer {
    axum_otel_metrics::HttpMetricsLayerBuilder::new()
        .with_skipper(axum_otel_metrics::PathSkipper::new(skip_metrics_path))
        .build()
}

fn skip_metrics_path(path: &str) -> bool {
    matches!(path, "/healthz" | "/readyz" | "/favicon.ico")
        || path.starts_with("/assets/")
        || path.starts_with("/.well-known/")
}

#[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()
}

pub fn http_span<B>(request: &Request<B>) -> Span {
    let method = request.method().as_str();
    let route = request
        .extensions()
        .get::<MatchedPath>()
        .map_or_else(|| request.uri().path(), |path| path.as_str());
    let name = format!("{method} {route}");
    let span = tracing::info_span!(
        "http.server",
        "otel.name" = %name,
        "otel.kind" = "server",
        "http.request.method" = %method,
        "url.path" = %request.uri().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.extensions().get::<crate::middleware::RequestId>() {
        span.record("request_id", request_id.0.as_str());
    }
    set_remote_parent(&span, request.headers());
    span
}

pub fn on_response<B>(response: &Response<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));
    }
}

/// `tower_http::trace::MakeSpan` adapter shared by every axum/tonic host.
#[derive(Clone, Copy, Default)]
pub struct HttpSpan;

impl<B> tower_http::trace::MakeSpan<B> for HttpSpan {
    fn make_span(&mut self, request: &Request<B>) -> Span {
        http_span(request)
    }
}

/// `tower_http::trace::OnResponse` adapter keeping the request log line and
/// recording the HTTP semantic-convention attributes on the span.
#[derive(Clone, Copy, Default)]
pub struct HttpOnResponse;

impl<B> tower_http::trace::OnResponse<B> for HttpOnResponse {
    fn on_response(self, response: &Response<B>, latency: Duration, span: &Span) {
        on_response(response, latency, span);
        tracing::info!(
            parent: span,
            status = response.status().as_u16(),
            latency_ms = latency.as_millis() as u64,
            "finished processing request",
        );
    }
}

pub type HttpTraceLayer = tower_http::trace::TraceLayer<
    tower_http::classify::SharedClassifier<tower_http::classify::ServerErrorsAsFailures>,
    HttpSpan,
    tower_http::trace::DefaultOnRequest,
    HttpOnResponse,
>;

/// Ready-made server trace layer for hosts that build plain axum routers.
pub fn http_trace_layer() -> HttpTraceLayer {
    tower_http::trace::TraceLayer::new_for_http()
        .make_span_with(HttpSpan)
        .on_response(HttpOnResponse)
}

/// 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(crate::middleware::REQUEST_ID_HEADER, 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");
    }
}