use std::sync::{LazyLock, OnceLock};
use std::time::Instant;

use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter, UpDownCounter};
use opentelemetry::{KeyValue, global};
use opentelemetry_otlp::{MetricExporter, WithExportConfig, WithHttpConfig};
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};

use crate::config::TelemetryConfig;

pub const RPC_DURATION: &str = "rpc.server.duration";
pub const RPC_REQUESTS: &str = "rpc.server.requests";
pub const RPC_ACTIVE: &str = "rpc.server.active_requests";

struct RpcMetrics {
    requests: Counter<u64>,
    duration: Histogram<f64>,
    active: UpDownCounter<i64>,
}

fn rpc() -> &'static RpcMetrics {
    static METRICS: OnceLock<RpcMetrics> = OnceLock::new();
    METRICS.get_or_init(|| {
        let meter = global::meter("identity.grpc");
        RpcMetrics {
            requests: meter.u64_counter(RPC_REQUESTS).build(),
            duration: meter.f64_histogram(RPC_DURATION).with_unit("ms").build(),
            active: meter.i64_up_down_counter(RPC_ACTIVE).build(),
        }
    })
}

/// 在途 RPC 守卫：构造时 +1、Drop 时 -1；`finish` 记录计数、时延与状态。
pub struct ActiveRpcGuard {
    service: &'static str,
    started: Instant,
}

impl ActiveRpcGuard {
    pub fn new(service: &'static str) -> Self {
        let metrics = rpc();
        metrics
            .active
            .add(1, &[KeyValue::new("rpc.service", service)]);
        Self {
            service,
            started: Instant::now(),
        }
    }

    /// 记录一次 RPC 结束。`status` 为 `OK` 之外的取值时同时把 span 标成错误。
    pub fn finish<T>(&self, method: &'static str, result: &Result<T, tonic::Status>) {
        let status_code = status_label(result);
        if let Some(error) = result.as_ref().err() {
            use tracing_opentelemetry::OpenTelemetrySpanExt;
            tracing::Span::current().set_status(opentelemetry::trace::Status::error(
                error.message().to_owned(),
            ));
        }

        let metrics = rpc();
        let attributes = [
            KeyValue::new("rpc.system", "grpc"),
            KeyValue::new("rpc.service", self.service),
            KeyValue::new("rpc.method", method),
            KeyValue::new("rpc.grpc.status_code", status_code),
            KeyValue::new("error", status_code != "OK"),
        ];
        metrics.requests.add(1, &attributes);
        metrics
            .duration
            .record(self.started.elapsed().as_secs_f64() * 1000.0, &attributes);
    }
}

impl Drop for ActiveRpcGuard {
    fn drop(&mut self) {
        rpc()
            .active
            .add(-1, &[KeyValue::new("rpc.service", self.service)]);
    }
}

pub fn status_label<T>(result: &Result<T, tonic::Status>) -> &'static str {
    match result {
        Ok(_) => "OK",
        Err(status) => match status.code() {
            tonic::Code::Ok => "OK",
            tonic::Code::Cancelled => "CANCELLED",
            tonic::Code::Unknown => "UNKNOWN",
            tonic::Code::InvalidArgument => "INVALID_ARGUMENT",
            tonic::Code::DeadlineExceeded => "DEADLINE_EXCEEDED",
            tonic::Code::NotFound => "NOT_FOUND",
            tonic::Code::AlreadyExists => "ALREADY_EXISTS",
            tonic::Code::PermissionDenied => "PERMISSION_DENIED",
            tonic::Code::ResourceExhausted => "RESOURCE_EXHAUSTED",
            tonic::Code::FailedPrecondition => "FAILED_PRECONDITION",
            tonic::Code::Aborted => "ABORTED",
            tonic::Code::OutOfRange => "OUT_OF_RANGE",
            tonic::Code::Unimplemented => "UNIMPLEMENTED",
            tonic::Code::Internal => "INTERNAL",
            tonic::Code::Unavailable => "UNAVAILABLE",
            tonic::Code::DataLoss => "DATA_LOSS",
            tonic::Code::Unauthenticated => "UNAUTHENTICATED",
        },
    }
}

pub fn build(
    config: &TelemetryConfig,
    resource: Resource,
    warnings: &mut Vec<String>,
) -> SdkMeterProvider {
    let mut builder = SdkMeterProvider::builder().with_resource(resource);

    if let Some(endpoint) = config.metrics_url() {
        match exporter(config, &endpoint) {
            Ok(exporter) => {
                let reader = PeriodicReader::builder(exporter)
                    .with_interval(config.metrics.interval)
                    .build();
                builder = builder.with_reader(reader);
            }
            Err(error) => warnings.push(format!(
                "otlp metric exporter disabled: {error} (endpoint {endpoint})"
            )),
        }
    }

    builder.build()
}

fn exporter(
    config: &TelemetryConfig,
    endpoint: &str,
) -> Result<MetricExporter, opentelemetry_otlp::ExporterBuildError> {
    let mut builder = MetricExporter::builder()
        .with_http()
        .with_endpoint(endpoint)
        .with_temporality(Temporality::default())
        .with_timeout(config.timeout);
    if !config.metrics.signal.headers.is_empty() {
        builder = builder.with_headers(config.metrics.signal.headers.clone());
    }
    builder.build()
}

fn app_meter() -> Meter {
    global::meter("gitdata")
}

static AUTH_LOGINS: LazyLock<Counter<u64>> = LazyLock::new(|| {
    app_meter()
        .u64_counter("auth.logins")
        .with_description("Login attempts by outcome")
        .build()
});

static PUSH_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {
    app_meter()
        .u64_counter("push.deliveries")
        .with_description("Web push deliveries by outcome")
        .build()
});

static IM_CONNECTIONS: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {
    app_meter()
        .i64_up_down_counter("im.connections")
        .with_description("Open chat WebSocket connections")
        .build()
});

static CHAT_WEBHOOK_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {
    app_meter()
        .u64_counter("chat.webhook.deliveries")
        .with_description("Outbound chat webhook deliveries by outcome")
        .build()
});

static CHAT_FANOUT: LazyLock<Counter<u64>> = LazyLock::new(|| {
    app_meter()
        .u64_counter("chat.fanout")
        .with_description("Chat fan-out jobs by outcome")
        .build()
});

static QUEUE_DEPTH: LazyLock<Gauge<i64>> = LazyLock::new(|| {
    app_meter()
        .i64_gauge("queue.depth")
        .with_description("Pending queue entries by topic")
        .build()
});

static DEPENDENCY_UP: LazyLock<Gauge<i64>> = LazyLock::new(|| {
    app_meter()
        .i64_gauge("health.dependency.up")
        .with_description("Sampled dependency health (1 = up, 0 = down)")
        .build()
});

pub fn auth_login(outcome: &'static str) {
    AUTH_LOGINS.add(1, &[KeyValue::new("outcome", outcome)]);
}

pub fn push_delivery(outcome: &'static str) {
    PUSH_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);
}

pub fn im_connection(delta: i64) {
    IM_CONNECTIONS.add(delta, &[]);
}

/// `Drop`-based guard so every socket exit path (incl. early returns) balances
/// the connection gauge.
pub struct ImConnectionGuard;

impl ImConnectionGuard {
    #[must_use]
    pub fn new() -> Self {
        im_connection(1);
        Self
    }
}

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

impl Drop for ImConnectionGuard {
    fn drop(&mut self) {
        im_connection(-1);
    }
}

pub fn chat_webhook_delivery(outcome: &'static str) {
    CHAT_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);
}

pub fn chat_fanout(lane: &'static str, outcome: &'static str) {
    CHAT_FANOUT.add(
        1,
        &[
            KeyValue::new("lane", lane),
            KeyValue::new("outcome", outcome),
        ],
    );
}

pub fn queue_depth(topic: &'static str, depth: i64) {
    QUEUE_DEPTH.record(depth, &[KeyValue::new("topic", topic)]);
}

pub fn health_dependency(dependency: &'static str, up: bool) {
    DEPENDENCY_UP.record(
        if up { 1 } else { 0 },
        &[KeyValue::new("dependency", dependency)],
    );
}