//! OTLP traces: a real `tracing_subscriber::Layer` → OTLP `Span` bridge
//! (D1-traces, bead `oraclemcp-040-epic-wp-d-1il.3`).
//!
//! This is a genuine component, not a shim. It:
//! - captures span open/close, field values, and start/end timing;
//! - maps `tracing` fields to OTLP span attributes (and OTel `db.*` semantic
//!   conventions where non-leaking);
//! - **redacts secrets** — SQL bind values, passwords, tokens, wallet secrets
//!   never reach a span attribute (every field/attr passes [`Redactor`]);
//! - threads **W3C trace/span IDs**: a 16-byte trace id is generated per root
//!   span (or adopted from an inbound `traceparent` field for context
//!   propagation from the MCP client) and inherited by children; each span gets
//!   a fresh 8-byte span id with the parent linked via `parent_span_id`;
//! - **batches** completed spans into bounded [`SpanBatch`]es and hands them to
//!   the export pump; configurable head sampling drops whole traces up front.
//!
//! The encoded `ExportTraceServiceRequest` protobuf is hand-rolled (asupersync's
//! own trace encoder is `cfg(fuzz)`-gated). Export egress is the pump's job; the
//! layer only fills the batch buffer — it never blocks the request path.

use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

use tracing::field::{Field, Visit};
use tracing::span::{Attributes, Id, Record};
use tracing::{Event, Subscriber};
use tracing_subscriber::Layer;
use tracing_subscriber::layer::Context;
use tracing_subscriber::registry::LookupSpan;

use asupersync::observability::ExportError;

use super::config::OtlpConfig;
use super::proto::{
    ExportTraceServiceRequest, InstrumentationScope, Resource, ResourceSpans, SPAN_KIND_INTERNAL,
    SPAN_KIND_SERVER, STATUS_CODE_ERROR, STATUS_CODE_OK, STATUS_CODE_UNSET, ScopeSpans, Span,
    Status, key_value,
};
use super::redact::Redactor;

/// OTel schema URL the exported scope is anchored to.
pub const OTEL_SCHEMA_URL: &str = "https://opentelemetry.io/schemas/1.37.0";
/// Instrumentation scope name.
pub const SCOPE_NAME: &str = "oraclemcp.telemetry";
/// Instrumentation scope version.
pub const SCOPE_VERSION: &str = env!("CARGO_PKG_VERSION");

/// A completed span ready for OTLP encoding.
#[derive(Clone, Debug, PartialEq)]
pub struct FinishedSpan {
    /// 16-byte W3C trace id.
    pub trace_id: [u8; 16],
    /// 8-byte W3C span id.
    pub span_id: [u8; 8],
    /// 8-byte parent span id (zeroed for a root span).
    pub parent_span_id: [u8; 8],
    /// Span name.
    pub name: String,
    /// OTLP span kind.
    pub kind: i32,
    /// Start time (unix nanos).
    pub start_time_unix_nano: u64,
    /// End time (unix nanos).
    pub end_time_unix_nano: u64,
    /// Redacted attributes.
    pub attributes: Vec<(String, String)>,
    /// Status code (`Unset`/`Ok`/`Error`).
    pub status_code: i32,
    /// W3C `tracestate`, threaded from the inbound context if any.
    pub trace_state: String,
}

impl FinishedSpan {
    fn to_proto(&self) -> Span {
        let parent = if self.parent_span_id == [0u8; 8] {
            Vec::new()
        } else {
            self.parent_span_id.to_vec()
        };
        Span {
            trace_id: self.trace_id.to_vec(),
            span_id: self.span_id.to_vec(),
            trace_state: self.trace_state.clone(),
            parent_span_id: parent,
            name: self.name.clone(),
            kind: self.kind,
            start_time_unix_nano: self.start_time_unix_nano,
            end_time_unix_nano: self.end_time_unix_nano,
            attributes: self
                .attributes
                .iter()
                .map(|(k, v)| key_value(k.clone(), v.clone()))
                .collect(),
            dropped_attributes_count: 0,
            status: Some(Status {
                message: String::new(),
                code: self.status_code,
            }),
        }
    }
}

/// Build an `ExportTraceServiceRequest` from a batch of finished spans.
#[must_use]
pub fn build_request(config: &OtlpConfig, spans: &[FinishedSpan]) -> ExportTraceServiceRequest {
    ExportTraceServiceRequest {
        resource_spans: vec![ResourceSpans {
            resource: Some(Resource {
                attributes: vec![
                    key_value("service.name", config.service_name.clone()),
                    key_value("telemetry.sdk.name", "oraclemcp"),
                    key_value("telemetry.sdk.version", SCOPE_VERSION),
                ],
            }),
            scope_spans: vec![ScopeSpans {
                scope: Some(InstrumentationScope {
                    name: SCOPE_NAME.to_owned(),
                    version: SCOPE_VERSION.to_owned(),
                }),
                spans: spans.iter().map(FinishedSpan::to_proto).collect(),
                schema_url: OTEL_SCHEMA_URL.to_owned(),
            }],
            schema_url: OTEL_SCHEMA_URL.to_owned(),
        }],
    }
}

/// Encode + send a trace request through asupersync's Tokio-free exporter.
///
/// # Errors
/// Returns the asupersync `ExportError` if the OTLP request fails after retries.
pub async fn export_request(
    cx: &asupersync::Cx,
    config: &OtlpConfig,
    request: &ExportTraceServiceRequest,
) -> Result<(), ExportError> {
    let exporter = super::build_http_exporter(&config.traces_endpoint, config);
    exporter.send_otlp_protobuf(cx, request.to_bytes()).await
}

// ===========================================================================
// W3C id generation
// ===========================================================================

/// A deterministic-ish, lock-free id generator (splitmix64-seeded counter). Not
/// cryptographic — span/trace ids only need uniqueness within a trace, and a
/// process-unique counter mixed with the start time gives that without pulling a
/// CSPRNG dependency. Each `next_*` advances the counter.
#[derive(Debug)]
struct IdGen {
    counter: AtomicU64,
}

impl IdGen {
    fn new(seed: u64) -> Self {
        Self {
            counter: AtomicU64::new(seed | 1),
        }
    }

    fn next_u64(&self) -> u64 {
        let n = self
            .counter
            .fetch_add(0x9e37_79b9_7f4a_7c15, Ordering::Relaxed);
        splitmix64(n)
    }

    fn next_trace_id(&self) -> [u8; 16] {
        let hi = self.next_u64();
        let lo = self.next_u64();
        let mut out = [0u8; 16];
        out[..8].copy_from_slice(&hi.to_be_bytes());
        out[8..].copy_from_slice(&lo.to_be_bytes());
        // OTLP forbids an all-zero trace id.
        if out == [0u8; 16] {
            out[15] = 1;
        }
        out
    }

    fn next_span_id(&self) -> [u8; 8] {
        let v = self.next_u64();
        let mut out = v.to_be_bytes();
        if out == [0u8; 8] {
            out[7] = 1;
        }
        out
    }
}

fn splitmix64(mut z: u64) -> u64 {
    z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
    z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
    z ^ (z >> 31)
}

/// Parse a W3C `traceparent` header value
/// (`00-<32 hex trace id>-<16 hex span id>-<2 hex flags>`).
///
/// Returns `(trace_id, parent_span_id, sampled)` on a well-formed value.
#[must_use]
pub fn parse_traceparent(value: &str) -> Option<([u8; 16], [u8; 8], bool)> {
    let parts: Vec<&str> = value.trim().split('-').collect();
    if parts.len() != 4 || parts[0] != "00" {
        return None;
    }
    let trace_id = hex_to_array::<16>(parts[1])?;
    let span_id = hex_to_array::<8>(parts[2])?;
    if trace_id == [0u8; 16] || span_id == [0u8; 8] {
        return None;
    }
    let flags = u8::from_str_radix(parts[3], 16).ok()?;
    Some((trace_id, span_id, flags & 0x01 != 0))
}

fn hex_to_array<const N: usize>(hex: &str) -> Option<[u8; N]> {
    // Operate on bytes, not the &str: a non-ASCII (multi-byte UTF-8) char in an
    // attacker-controlled `traceparent` header would otherwise make a byte-index
    // slice land off a char boundary and PANIC. Validate ASCII-hex up front and
    // index the byte slice in pairs so every input is rejected with `None`.
    let bytes = hex.as_bytes();
    if bytes.len() != N * 2 || !bytes.iter().all(u8::is_ascii_hexdigit) {
        return None;
    }
    let mut out = [0u8; N];
    for (i, byte) in out.iter_mut().enumerate() {
        let hi = (bytes[i * 2] as char).to_digit(16)?;
        let lo = (bytes[i * 2 + 1] as char).to_digit(16)?;
        *byte = (hi * 16 + lo) as u8;
    }
    Some(out)
}

// ===========================================================================
// Field visitor
// ===========================================================================

/// Captures span/event fields into redacted attributes + recognises the special
/// `traceparent` field for inbound W3C context propagation and `otel.status`.
#[derive(Default)]
struct FieldVisitor {
    attributes: Vec<(String, String)>,
    traceparent: Option<String>,
    status_error: bool,
}

impl FieldVisitor {
    fn record(&mut self, redactor: &Redactor, field_name: &str, value: String) {
        match field_name {
            // Inbound W3C context from the MCP client — captured, never emitted
            // as an attribute (it would be redundant with the threaded ids).
            "traceparent" => self.traceparent = Some(value),
            "otel.status_code" => {
                if value.eq_ignore_ascii_case("error") {
                    self.status_error = true;
                }
            }
            _ => {
                if let Some((safe_key, safe_value)) = redactor.filter(field_name, &value) {
                    self.attributes.push((safe_key.to_owned(), safe_value));
                }
            }
        }
    }
}

impl Visit for FieldVisitor {
    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
        // Defer redaction to `record`; we can't hold the Redactor in the Visit
        // impl, so stash raw then redact in `consume`. Store debug-formatted.
        self.attributes
            .push((field.name().to_owned(), format!("{value:?}")));
    }

    fn record_str(&mut self, field: &Field, value: &str) {
        self.attributes
            .push((field.name().to_owned(), value.to_owned()));
    }

    fn record_i64(&mut self, field: &Field, value: i64) {
        self.attributes
            .push((field.name().to_owned(), value.to_string()));
    }

    fn record_u64(&mut self, field: &Field, value: u64) {
        self.attributes
            .push((field.name().to_owned(), value.to_string()));
    }

    fn record_bool(&mut self, field: &Field, value: bool) {
        self.attributes
            .push((field.name().to_owned(), value.to_string()));
    }
}

impl FieldVisitor {
    /// Drain raw captures, applying the redaction policy + special-field routing.
    fn consume(self, redactor: &Redactor) -> ConsumedFields {
        let mut out = ConsumedFields::default();
        let raw = self.attributes;
        for (name, value) in raw {
            let mut routed = FieldVisitor::default();
            routed.record(redactor, &name, value);
            out.attributes.extend(routed.attributes);
            if routed.traceparent.is_some() {
                out.traceparent = routed.traceparent;
            }
            out.status_error |= routed.status_error;
        }
        out
    }
}

#[derive(Default)]
struct ConsumedFields {
    attributes: Vec<(String, String)>,
    traceparent: Option<String>,
    status_error: bool,
}

// ===========================================================================
// Per-span state (stored in registry extensions)
// ===========================================================================

#[derive(Debug)]
struct SpanState {
    trace_id: [u8; 16],
    span_id: [u8; 8],
    parent_span_id: [u8; 8],
    name: String,
    kind: i32,
    start_time_unix_nano: u64,
    attributes: Mutex<Vec<(String, String)>>,
    status_error: std::sync::atomic::AtomicBool,
    sampled: bool,
    trace_state: String,
}

// ===========================================================================
// The Layer
// ===========================================================================

/// Sink the layer hands finished, sampled span batches to. Implemented by the
/// export pump; in tests, a collecting buffer.
pub trait SpanSink: Send + Sync + 'static {
    /// Submit a completed span. Non-blocking; the sink drops on overflow.
    fn submit(&self, span: FinishedSpan);
}

/// A `tracing` layer that bridges spans to OTLP. Generic over the sink so tests
/// can collect spans without a live exporter.
pub struct OtlpTraceLayer<K: SpanSink> {
    sink: Arc<K>,
    redactor: Redactor,
    sample_ratio: f64,
    idgen: Arc<IdGen>,
}

impl<K: SpanSink> OtlpTraceLayer<K> {
    /// Build a layer feeding `sink`, head-sampling at `sample_ratio` ∈ [0,1].
    #[must_use]
    pub fn new(sink: Arc<K>, redactor: Redactor, sample_ratio: f64) -> Self {
        let seed = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos() as u64)
            .unwrap_or(0x1234_5678);
        Self {
            sink,
            redactor,
            sample_ratio: sample_ratio.clamp(0.0, 1.0),
            idgen: Arc::new(IdGen::new(seed)),
        }
    }

    fn sampled(&self, trace_id: &[u8; 16]) -> bool {
        if self.sample_ratio >= 1.0 {
            return true;
        }
        if self.sample_ratio <= 0.0 {
            return false;
        }
        // Deterministic per-trace decision: hash the trace id into [0,1).
        let mut key = 0u64;
        for chunk in trace_id.chunks(8) {
            let mut b = [0u8; 8];
            b[..chunk.len()].copy_from_slice(chunk);
            key ^= u64::from_be_bytes(b);
        }
        let unit = (splitmix64(key) >> 11) as f64 / 9_007_199_254_740_992.0;
        unit < self.sample_ratio
    }
}

fn now_unix_nano() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0)
}

fn span_kind_for(name: &str) -> i32 {
    // request→dispatch→classify→DB call→serialize: a server span at the request
    // root, internal otherwise. (Client kind is reserved for the DB-call span
    // once oracledb's `tracing` feature emits it.)
    if name.eq_ignore_ascii_case("request") || name.eq_ignore_ascii_case("mcp.request") {
        SPAN_KIND_SERVER
    } else {
        SPAN_KIND_INTERNAL
    }
}

impl<S, K> Layer<S> for OtlpTraceLayer<K>
where
    S: Subscriber + for<'a> LookupSpan<'a>,
    K: SpanSink,
{
    fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
        let Some(span) = ctx.span(id) else { return };

        // Capture fields + any inbound traceparent.
        let mut visitor = FieldVisitor::default();
        attrs.record(&mut visitor);
        let consumed = visitor.consume(&self.redactor);

        // Resolve trace + parent ids: inherit from parent span if present, else
        // adopt an inbound traceparent, else mint a fresh root trace.
        let (trace_id, parent_span_id, trace_state, inherited_sampled) =
            if let Some(parent) = span.parent() {
                if let Some(state) = parent.extensions().get::<SpanState>() {
                    (
                        state.trace_id,
                        state.span_id,
                        state.trace_state.clone(),
                        Some(state.sampled),
                    )
                } else {
                    (self.idgen.next_trace_id(), [0u8; 8], String::new(), None)
                }
            } else if let Some((tid, pid, sampled)) =
                consumed.traceparent.as_deref().and_then(parse_traceparent)
            {
                // W3C context propagation from the MCP client.
                (tid, pid, String::new(), Some(sampled))
            } else {
                (self.idgen.next_trace_id(), [0u8; 8], String::new(), None)
            };

        let sampled = inherited_sampled.unwrap_or_else(|| self.sampled(&trace_id));

        let state = SpanState {
            trace_id,
            span_id: self.idgen.next_span_id(),
            parent_span_id,
            name: span.name().to_owned(),
            kind: span_kind_for(span.name()),
            start_time_unix_nano: now_unix_nano(),
            attributes: Mutex::new(consumed.attributes),
            status_error: std::sync::atomic::AtomicBool::new(consumed.status_error),
            sampled,
            trace_state,
        };
        span.extensions_mut().insert(state);
    }

    fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) {
        let Some(span) = ctx.span(id) else { return };
        let mut visitor = FieldVisitor::default();
        values.record(&mut visitor);
        let consumed = visitor.consume(&self.redactor);
        let ext = span.extensions();
        if let Some(state) = ext.get::<SpanState>() {
            if let Ok(mut attrs) = state.attributes.lock() {
                attrs.extend(consumed.attributes);
            }
            if consumed.status_error {
                state.status_error.store(true, Ordering::Relaxed);
            }
        }
    }

    fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
        // An event at ERROR level marks the enclosing span as errored.
        if *event.metadata().level() != tracing::Level::ERROR {
            return;
        }
        if let Some(span) = ctx.event_span(event)
            && let Some(state) = span.extensions().get::<SpanState>()
        {
            state.status_error.store(true, Ordering::Relaxed);
        }
    }

    fn on_close(&self, id: Id, ctx: Context<'_, S>) {
        let Some(span) = ctx.span(&id) else { return };
        let ext = span.extensions();
        let Some(state) = ext.get::<SpanState>() else {
            return;
        };
        if !state.sampled {
            return; // head-dropped: never leaves the process.
        }
        let attributes = state
            .attributes
            .lock()
            .map(|a| a.clone())
            .unwrap_or_default();
        let status_code = if state.status_error.load(Ordering::Relaxed) {
            STATUS_CODE_ERROR
        } else if attributes.iter().any(|(k, _)| k == "status") {
            STATUS_CODE_OK
        } else {
            STATUS_CODE_UNSET
        };
        let finished = FinishedSpan {
            trace_id: state.trace_id,
            span_id: state.span_id,
            parent_span_id: state.parent_span_id,
            name: state.name.clone(),
            kind: state.kind,
            start_time_unix_nano: state.start_time_unix_nano,
            end_time_unix_nano: now_unix_nano(),
            attributes,
            status_code,
            trace_state: state.trace_state.clone(),
        };
        self.sink.submit(finished);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use prost::Message;
    use tracing::subscriber::with_default;
    use tracing_subscriber::Registry;
    use tracing_subscriber::layer::SubscriberExt;

    #[derive(Default)]
    struct CollectingSink {
        spans: Mutex<Vec<FinishedSpan>>,
    }
    impl SpanSink for CollectingSink {
        fn submit(&self, span: FinishedSpan) {
            self.spans.lock().unwrap().push(span);
        }
    }

    fn cfg() -> OtlpConfig {
        OtlpConfig::from_lookup(|k| {
            (k == "OTEL_EXPORTER_OTLP_ENDPOINT").then(|| "http://c:4318".to_owned())
        })
        .expect("on")
    }

    #[test]
    fn parses_w3c_traceparent() {
        let (tid, pid, sampled) =
            parse_traceparent("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
                .expect("valid traceparent");
        assert_eq!(tid[0], 0x4b);
        assert_eq!(pid[0], 0x00);
        assert_eq!(pid[7], 0xb7);
        assert!(sampled);
        assert!(parse_traceparent("garbage").is_none());
        assert!(parse_traceparent("01-aa-bb-01").is_none());
    }

    #[test]
    fn non_ascii_traceparent_is_rejected_not_panicking() {
        // A multi-byte UTF-8 char makes the byte length equal the expected hex
        // length while landing a byte-index slice off a char boundary. The parse
        // must return None, never panic (a traceparent parse failure must not
        // unwind out of telemetry).
        // "é" is 2 bytes (0xC3 0xA9): a 16-char trace-id field built from 8 of
        // them is 16 bytes long — the old length-only guard would slice into it.
        let trace_id = "é".repeat(8); // 16 bytes, 8 chars
        let span_id = "é".repeat(4); // 8 bytes, 4 chars
        let tp = format!("00-{trace_id}-{span_id}-01");
        assert!(parse_traceparent(&tp).is_none());
        // Direct: a non-hex but correct-byte-length input is rejected.
        assert!(hex_to_array::<16>(&"é".repeat(8)).is_none());
        assert!(hex_to_array::<8>("zzzzzzzzzzzzzzzz").is_none());
        // A lone multi-byte char with the right byte count still rejects.
        assert!(hex_to_array::<1>("é").is_none());
    }

    #[test]
    fn span_tree_is_captured_with_threaded_ids_and_redaction() {
        let sink = Arc::new(CollectingSink::default());
        let layer = OtlpTraceLayer::new(sink.clone(), Redactor::new(), 1.0);
        let subscriber = Registry::default().with(layer);

        with_default(subscriber, || {
            let root = tracing::info_span!("request", tool = "oracle_query");
            let _g = root.enter();
            {
                let child = tracing::info_span!(
                    "db.call",
                    row_count = 5_u64,
                    password = "hunter2",
                    bind_0 = "secret-value"
                );
                let _gc = child.enter();
            }
        });

        let spans = sink.spans.lock().unwrap().clone();
        assert_eq!(spans.len(), 2, "root + child closed");
        // Children share the root's trace id; child links the root as parent.
        let root = spans.iter().find(|s| s.name == "request").expect("root");
        let child = spans.iter().find(|s| s.name == "db.call").expect("child");
        assert_eq!(child.trace_id, root.trace_id, "child inherits trace id");
        assert_eq!(child.parent_span_id, root.span_id, "parent linked");
        assert_eq!(root.parent_span_id, [0u8; 8], "root has no parent");
        assert_eq!(root.kind, SPAN_KIND_SERVER, "request root is a server span");

        // Redaction: password + bind dropped, row_count kept.
        let keys: Vec<&str> = child.attributes.iter().map(|(k, _)| k.as_str()).collect();
        assert!(!keys.contains(&"password"), "password dropped from span");
        assert!(!keys.contains(&"bind_0"), "bind value dropped from span");
        assert!(keys.contains(&"row_count"), "safe attribute kept");
    }

    #[test]
    fn head_sampling_drops_unsampled_traces() {
        let sink = Arc::new(CollectingSink::default());
        let layer = OtlpTraceLayer::new(sink.clone(), Redactor::new(), 0.0);
        let subscriber = Registry::default().with(layer);
        with_default(subscriber, || {
            let s = tracing::info_span!("request");
            let _g = s.enter();
        });
        assert!(
            sink.spans.lock().unwrap().is_empty(),
            "ratio 0.0 drops every span before export"
        );
    }

    #[test]
    fn finished_spans_encode_to_valid_otlp() {
        let sink = Arc::new(CollectingSink::default());
        let layer = OtlpTraceLayer::new(sink.clone(), Redactor::new(), 1.0);
        let subscriber = Registry::default().with(layer);
        with_default(subscriber, || {
            let s = tracing::info_span!("request", tool = "oracle_query", status = "ok");
            let _g = s.enter();
        });
        let spans = sink.spans.lock().unwrap().clone();
        let req = build_request(&cfg(), &spans);
        let bytes = req.to_bytes();
        let decoded = ExportTraceServiceRequest::decode(bytes.as_slice()).expect("decodes");
        assert_eq!(decoded, req, "trace request roundtrips");
        let span = &decoded.resource_spans[0].scope_spans[0].spans[0];
        assert_eq!(span.trace_id.len(), 16);
        assert_eq!(span.span_id.len(), 8);
        assert_eq!(span.status.as_ref().unwrap().code, STATUS_CODE_OK);
    }
}
