use std::sync::Arc;

use apollo_opentelemetry::default_instrumentation_scope;
use apollo_opentelemetry::metrics::{HistogramExt, RecordDurationGuard};
use opentelemetry::global::BoxedTracer;
use opentelemetry::metrics::Histogram;
use opentelemetry::trace::{TraceContextExt as _, Tracer as _, TracerProvider as _};
use opentelemetry::{Context as OtelContext, KeyValue};

// The span and the duration histogram share the `apollo.blocking.task` prefix so trace and
// metric queries can correlate on a single identifier. See OTel naming guidance:
// https://opentelemetry.io/docs/specs/semconv/general/naming/
const TASK_NAME: &str = "apollo.blocking.task";
const DURATION_NAME: &str = "apollo.blocking.task.duration";
const SPAN_ATTR_NAME: &str = "apollo.blocking.name";
const SPAN_ATTR_RESULT: &str = "apollo.blocking.result";
const METRIC_ATTR_NAME: &str = "name";
const METRIC_ATTR_RESULT: &str = "result";

pub(crate) const RESULT_SUCCESS: &str = "success";
pub(crate) const RESULT_CANCELLED: &str = "cancelled";
pub(crate) const RESULT_PANICKED: &str = "panicked";

pub(crate) const EVENT_CANCELLED: &str = "apollo.blocking.task.cancelled";
pub(crate) const EVENT_PANICKED: &str = "apollo.blocking.task.panicked";

/// Cached tracer + meter instruments shared across all clones of an adapter.
pub(crate) struct Instruments {
    tracer: BoxedTracer,
    task_duration: Histogram<f64>,
}

impl Instruments {
    pub(crate) fn new() -> Arc<Self> {
        let scope = default_instrumentation_scope!().clone();
        let tracer = opentelemetry::global::tracer_provider().tracer_with_scope(scope.clone());
        let meter = opentelemetry::global::meter_with_scope(scope);
        let task_duration = meter
            .f64_histogram(DURATION_NAME)
            .with_description("Wall-clock duration of blocking tasks dispatched")
            .with_unit("s")
            .build();
        Arc::new(Self {
            tracer,
            task_duration,
        })
    }

    pub(crate) fn start(&self, name: &'static str) -> InstrumentationState {
        let parent_cx = OtelContext::current();
        let span = self.tracer.build_with_context(
            opentelemetry::trace::SpanBuilder::from_name(TASK_NAME)
                .with_attributes([KeyValue::new(SPAN_ATTR_NAME, name)]),
            &parent_cx,
        );
        let span_cx = parent_cx.with_span(span);

        let duration = self
            .task_duration
            .record_duration_on_drop(vec![KeyValue::new(METRIC_ATTR_NAME, name)]);

        InstrumentationState {
            span_cx,
            duration,
            finalized: false,
        }
    }
}

/// Per-call instrumentation state. Owned by `run_blocking` on the worker thread, so the span
/// and histogram guard drop only when the blocking work actually exits.
///
/// On normal completion `run_blocking` calls [`finalize`] with `success`; on cooperative
/// cancellation it calls [`record_event_and_finalize`] with `cancelled`. The [`Drop`] impl
/// covers the remaining cases: it records `panicked` when an unwind is in progress (e.g. the
/// user closure panicked) and `cancelled` otherwise (e.g. the tokio runtime aborted a queued
/// `spawn_blocking` task during shutdown).
pub(crate) struct InstrumentationState {
    span_cx: OtelContext,
    duration: RecordDurationGuard,
    finalized: bool,
}

impl InstrumentationState {
    pub(crate) fn record_event_and_finalize(
        &mut self,
        event_name: &'static str,
        result: &'static str,
    ) {
        if self.finalized {
            return;
        }
        self.span_cx.span().add_event(event_name, vec![]);
        self.finalize(result);
    }

    pub(crate) fn finalize(&mut self, result: &'static str) {
        if self.finalized {
            return;
        }
        self.finalized = true;
        self.span_cx
            .span()
            .set_attribute(KeyValue::new(SPAN_ATTR_RESULT, result));
        self.duration.set(KeyValue::new(METRIC_ATTR_RESULT, result));
    }
}

impl Drop for InstrumentationState {
    fn drop(&mut self) {
        // `panicking()` distinguishes a panic-driven unwind through `run_blocking` from any
        // other drop path (normal early-return, queued task aborted at runtime shutdown, etc.).
        if std::thread::panicking() {
            self.record_event_and_finalize(EVENT_PANICKED, RESULT_PANICKED);
        } else {
            self.record_event_and_finalize(EVENT_CANCELLED, RESULT_CANCELLED);
        }
    }
}
