use core::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use arc_swap::ArcSwap;

use crate::ResolveReason;

mod pb {
    pub use crate::proto::confidence::flags::resolver::v1::telemetry_data::{
        BucketSpan, ResolveLatency, ResolveRate, StateAge,
    };
    pub use crate::proto::confidence::flags::resolver::v1::TelemetryData;
}

include!(concat!(env!("OUT_DIR"), "/telemetry_config.rs"));

// REASON_COUNT is generated by build.rs from the ResolveReason proto enum.

/// VictoriaMetrics-compatible bucket ratio: 10^(1/18), i.e. 18 buckets per
/// decimal order of magnitude (~13.6% width per bucket).
const LN_RATIO: f64 = core::f64::consts::LN_10 / 18.0;

/// Number of buckets needed to cover the full u32 range: ceil(ln(2^32) / LN_RATIO) + 1.
/// With LN_RATIO ≈ 0.128 this is 174 buckets (~1.4KB), so we always allocate for
/// the entire u32 space — no clamping needed on the client side.
const BUCKET_COUNT: usize = 174;

/// Lock-free exponential histogram for positive integer observations.
///
/// Bucket boundaries are at `exp(exponent * LN_RATIO)`. The bucket array
/// covers the full u32 range so every observation maps to a real bucket.
pub struct Histogram {
    sum: AtomicU64,
    count: AtomicU64,
    buckets: [AtomicU64; BUCKET_COUNT],
}

impl Histogram {
    fn new() -> Self {
        // SAFETY: AtomicU64 is valid when zero-initialized.
        // std's array::from_fn would work but generates poor code for 174 elements.
        let buckets = unsafe { core::mem::zeroed() };
        Histogram {
            sum: AtomicU64::new(0),
            count: AtomicU64::new(0),
            buckets,
        }
    }

    /// Record an observation.
    fn observe(&self, value: u32) {
        self.sum.fetch_add(value as u64, Ordering::Relaxed);
        self.count.fetch_add(1, Ordering::Relaxed);

        let idx = if value == 0 {
            0
        } else {
            let k = ((value as f64).ln() / LN_RATIO).floor() as usize;
            k.min(self.buckets.len().saturating_sub(1))
        };
        if let Some(bucket) = self.buckets.get(idx) {
            bucket.fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Snapshot the histogram counters into plain u64 values.
    fn snapshot(&self) -> HistogramSnapshot {
        HistogramSnapshot {
            sum: self.sum.load(Ordering::Relaxed),
            count: self.count.load(Ordering::Relaxed),
            buckets: self
                .buckets
                .iter()
                .map(|b| b.load(Ordering::Relaxed))
                .collect(),
        }
    }

    /// Build BucketSpans from a slice of bucket counts (used for proto encoding).
    fn bucket_spans(buckets: &[u32]) -> Vec<pb::BucketSpan> {
        let mut spans: Vec<pb::BucketSpan> = Vec::new();
        let mut current_span: Option<(i32, Vec<u32>)> = None;

        for (i, &v) in buckets.iter().enumerate() {
            let exponent = i as i32;
            if v > 0 {
                match &mut current_span {
                    Some((_, counts)) => counts.push(v),
                    None => current_span = Some((exponent, vec![v])),
                }
            } else if let Some((offset, counts)) = current_span.take() {
                spans.push(pb::BucketSpan { offset, counts });
            }
        }
        if let Some((offset, counts)) = current_span {
            spans.push(pb::BucketSpan { offset, counts });
        }

        spans
    }
}

/// Plain snapshot of cumulative telemetry counters.
///
/// Used for delta computation between flushes and as the future intermediate
/// representation for Prometheus text format serialization.
#[derive(Clone, Default)]
#[cfg_attr(feature = "json", derive(serde::Serialize, serde::Deserialize))]
pub struct TelemetrySnapshot {
    pub latency: HistogramSnapshot,
    pub resolve_rates: Vec<u64>,
    pub memory_bytes: u64,
}

#[derive(Clone, Default)]
#[cfg_attr(feature = "json", derive(serde::Serialize, serde::Deserialize))]
pub struct HistogramSnapshot {
    pub sum: u64,
    pub count: u64,
    pub buckets: Vec<u64>,
}

/// Configuration for Prometheus text format rendering.
pub struct PrometheusConfig {
    /// Number of histogram buckets per decimal order of magnitude.
    /// The internal histogram always stores 174 buckets (18 per decade).
    /// This controls how many are emitted in the text output by striding.
    /// Valid range: 1..=18. Values outside this range are clamped.
    /// A value of 0 is treated as the default (18 — no striding).
    pub buckets_per_decade: u32,
    /// When true, emit OpenMetrics text format instead of Prometheus exposition format.
    /// Differences: integer values rendered as floats, `# EOF` appended.
    pub openmetrics: bool,
}

impl Default for PrometheusConfig {
    fn default() -> Self {
        PrometheusConfig {
            buckets_per_decade: 18,
            openmetrics: false,
        }
    }
}

impl TelemetrySnapshot {
    /// Accumulate a `TelemetryData` delta into this cumulative snapshot.
    ///
    /// Expands compressed `BucketSpan`s back into the flat bucket array and
    /// adds all counters. Gauge fields (memory_bytes) are replaced with the
    /// latest value.
    #[allow(clippy::indexing_slicing, clippy::arithmetic_side_effects)]
    pub fn accumulate_delta(&mut self, td: &pb::TelemetryData) {
        if let Some(latency) = &td.resolve_latency {
            self.latency.sum = self.latency.sum.wrapping_add(latency.sum as u64);
            self.latency.count = self.latency.count.wrapping_add(latency.count as u64);

            // Expand BucketSpans into flat bucket array
            for span in &latency.buckets {
                let base = match usize::try_from(span.offset) {
                    Ok(b) if b < BUCKET_COUNT => b,
                    _ => continue, // skip malformed span
                };
                for (i, &count) in span.counts.iter().enumerate() {
                    let idx = base.saturating_add(i);
                    if idx >= BUCKET_COUNT {
                        break;
                    }
                    if idx >= self.latency.buckets.len() {
                        self.latency.buckets.resize(idx.saturating_add(1), 0);
                    }
                    // Safety: idx < BUCKET_COUNT and we just resized to at least idx+1
                    self.latency.buckets[idx] =
                        self.latency.buckets[idx].wrapping_add(count as u64);
                }
            }
        }

        for rate in &td.resolve_rate {
            let idx = rate.reason as usize;
            if idx >= self.resolve_rates.len() {
                self.resolve_rates.resize(idx.saturating_add(1), 0);
            }
            // Safety: we just resized to at least idx+1
            self.resolve_rates[idx] = self.resolve_rates[idx].wrapping_add(rate.count as u64);
        }

        if td.memory_bytes > 0 {
            self.memory_bytes = td.memory_bytes;
        }
    }

    /// Format the snapshot as Prometheus exposition text.
    ///
    /// All values are cumulative counters, matching what Prometheus expects.
    /// The histogram uses cumulative `le` buckets derived from the exponential
    /// bucket boundaries.
    ///
    /// `resolver_id` is included as a `resolver_id="..."` label on every metric,
    /// allowing outputs from multiple WASM instances to be concatenated into
    /// a single scrape endpoint.
    pub fn to_prometheus(&self, resolver_id: &str, config: &PrometheusConfig) -> String {
        let mut out = String::new();
        // fmt::Write for String is infallible; ignore the Ok.
        let _ = self.write_prometheus(&mut out, resolver_id, config);
        out
    }

    fn write_prometheus(
        &self,
        w: &mut dyn fmt::Write,
        resolver_id: &str,
        config: &PrometheusConfig,
    ) -> fmt::Result {
        self.write_histogram(w, resolver_id, config)?;
        self.write_resolve_rates(w, resolver_id, config)?;
        self.write_memory(w, resolver_id, config)?;
        if config.openmetrics {
            writeln!(w, "# EOF")?;
        }
        Ok(())
    }

    fn write_histogram(
        &self,
        w: &mut dyn fmt::Write,
        resolver_id: &str,
        config: &PrometheusConfig,
    ) -> fmt::Result {
        if self.latency.count == 0 {
            return Ok(());
        }

        let bpd = if config.buckets_per_decade == 0 {
            18usize
        } else {
            config.buckets_per_decade.clamp(1, 18) as usize
        };
        // Ceiling division: emit at most bpd buckets per decade.
        // bpd is in 1..=18, so checked_div cannot return None.
        let stride = 18usize
            .saturating_add(bpd)
            .saturating_sub(1)
            .checked_div(bpd)
            .unwrap_or(1);

        writeln!(
            w,
            "# HELP confidence_resolve_latency_microseconds Latency of flag resolve operations in microseconds."
        )?;
        writeln!(
            w,
            "# TYPE confidence_resolve_latency_microseconds histogram"
        )?;

        let mut cumulative: u64 = 0;
        // Track the next bucket index to emit (stride-1, stride*2-1, ...).
        // Advance next_emit at every stride boundary regardless of data;
        // only skip the actual write when cumulative is still zero.
        let mut next_emit = stride.wrapping_sub(1);
        // OpenMetrics requires integer values to be rendered as floats.
        let suffix = if config.openmetrics { ".0" } else { "" };
        for (i, &count) in self.latency.buckets.iter().enumerate() {
            cumulative = cumulative.wrapping_add(count);
            if i != next_emit {
                continue;
            }
            next_emit = next_emit.saturating_add(stride);
            if cumulative == 0 {
                continue;
            }
            let upper = ((i as f64 + 1.0) * LN_RATIO).exp();
            writeln!(
                w,
                "confidence_resolve_latency_microseconds_bucket{{resolver_id=\"{resolver_id}\",le=\"{upper:.6e}\"}} {cumulative}{suffix}"
            )?;
        }
        writeln!(
            w,
            "confidence_resolve_latency_microseconds_bucket{{resolver_id=\"{resolver_id}\",le=\"+Inf\"}} {}{suffix}",
            self.latency.count
        )?;
        writeln!(
            w,
            "confidence_resolve_latency_microseconds_sum{{resolver_id=\"{resolver_id}\"}} {}{suffix}",
            self.latency.sum
        )?;
        writeln!(
            w,
            "confidence_resolve_latency_microseconds_count{{resolver_id=\"{resolver_id}\"}} {}{suffix}",
            self.latency.count
        )?;

        Ok(())
    }

    fn write_resolve_rates(
        &self,
        w: &mut dyn fmt::Write,
        resolver_id: &str,
        config: &PrometheusConfig,
    ) -> fmt::Result {
        let has_any = self.resolve_rates.iter().any(|&c| c > 0);
        if !has_any {
            return Ok(());
        }

        let suffix = if config.openmetrics { ".0" } else { "" };
        // OpenMetrics: TYPE/HELP use the base name; sample lines keep _total.
        let type_name = if config.openmetrics {
            "confidence_resolves"
        } else {
            "confidence_resolves_total"
        };
        writeln!(
            w,
            "# HELP {type_name} Total number of flag resolve operations."
        )?;
        writeln!(w, "# TYPE {type_name} counter")?;
        for (i, &count) in self.resolve_rates.iter().enumerate() {
            if count == 0 {
                continue;
            }
            let label = ResolveReason::try_from(i as i32)
                .map(|r| r.as_str_name())
                .unwrap_or("UNKNOWN");
            writeln!(
                w,
                "confidence_resolves_total{{resolver_id=\"{resolver_id}\",reason=\"{label}\"}} {count}{suffix}"
            )?;
        }

        Ok(())
    }

    fn write_memory(
        &self,
        w: &mut dyn fmt::Write,
        resolver_id: &str,
        config: &PrometheusConfig,
    ) -> fmt::Result {
        if self.memory_bytes == 0 {
            return Ok(());
        }
        let suffix = if config.openmetrics { ".0" } else { "" };
        writeln!(
            w,
            "# HELP confidence_memory_bytes Current memory usage of the resolver in bytes."
        )?;
        writeln!(w, "# TYPE confidence_memory_bytes gauge")?;
        writeln!(
            w,
            "confidence_memory_bytes{{resolver_id=\"{resolver_id}\"}} {}{suffix}",
            self.memory_bytes
        )
    }
}

/// Concurrent telemetry collector.
///
/// All methods are safe to call from multiple threads without locking.
/// Call [`Telemetry::snapshot`] to read the current cumulative state, or
/// [`Telemetry::delta_snapshot`] to atomically produce a delta since the last flush.
pub struct Telemetry {
    resolve_latency: Histogram,

    /// Resolve rate counters, one per reason variant.
    resolve_rates: Box<[AtomicU64]>,

    /// Millisecond epoch of the last state update.
    last_state_update: AtomicU64,

    /// Provider called at snapshot time to read current memory usage in bytes.
    memory_provider: Box<dyn Fn() -> u64 + Send + Sync>,
}

impl Telemetry {
    pub fn new() -> Self {
        Self::with_memory_provider(|| 0)
    }

    pub fn with_memory_provider(memory_provider: impl Fn() -> u64 + Send + Sync + 'static) -> Self {
        let resolve_rates: Vec<AtomicU64> = (0..REASON_COUNT).map(|_| AtomicU64::new(0)).collect();
        Telemetry {
            resolve_latency: Histogram::new(),
            resolve_rates: resolve_rates.into_boxed_slice(),
            last_state_update: AtomicU64::new(0),
            memory_provider: Box::new(memory_provider),
        }
    }

    /// Record a resolve latency observation in microseconds.
    pub fn record_latency_us(&self, micros: u32) {
        self.resolve_latency.observe(micros);
    }

    /// Increment the resolve rate counter for the given reason.
    pub fn mark_resolve(&self, reason: ResolveReason) {
        if let Some(counter) = self.resolve_rates.get(reason as usize) {
            counter.fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Update the last known state timestamp (millisecond epoch).
    pub fn set_last_state_update(&self, epoch_ms: u64) {
        self.last_state_update.store(epoch_ms, Ordering::Relaxed);
    }

    /// Snapshot the current cumulative telemetry state.
    ///
    /// All values are cumulative (monotonically increasing). Returns a
    /// [`TelemetrySnapshot`] with raw u64 counters. Not perfectly atomic across
    /// all fields, but each individual counter load is atomic — acceptable for metrics.
    pub fn snapshot(&self) -> TelemetrySnapshot {
        TelemetrySnapshot {
            latency: self.resolve_latency.snapshot(),
            resolve_rates: self
                .resolve_rates
                .iter()
                .map(|c| c.load(Ordering::Relaxed))
                .collect(),
            memory_bytes: (self.memory_provider)(),
        }
    }

    /// Produce a [`TelemetryData`] proto containing deltas since the last flush.
    ///
    /// Atomically swaps `last` with the current snapshot and computes deltas
    /// using wrapping subtraction. Counter fields are cast to u32 for the proto.
    /// Gauge fields (`memory_bytes`, `state_age`) are passed through as latest values.
    ///
    /// Lock-free — safe for both single-threaded WASM and multi-threaded native use.
    pub fn delta_snapshot(&self, last: &ArcSwap<TelemetrySnapshot>) -> pb::TelemetryData {
        let current = self.snapshot();
        let previous = last.swap(Arc::new(current.clone()));

        // Delta histogram
        let delta_sum = current.latency.sum.wrapping_sub(previous.latency.sum) as u32;
        let delta_count = current.latency.count.wrapping_sub(previous.latency.count) as u32;

        let resolve_latency = if delta_count > 0 {
            let delta_buckets: Vec<u32> = current
                .latency
                .buckets
                .iter()
                .enumerate()
                .map(|(i, c)| {
                    let p = previous.latency.buckets.get(i).copied().unwrap_or(0);
                    c.wrapping_sub(p) as u32
                })
                .collect();

            let spans = Histogram::bucket_spans(&delta_buckets);

            Some(pb::ResolveLatency {
                sum: delta_sum,
                count: delta_count,
                buckets: spans,
                ln_ratio: LN_RATIO,
            })
        } else {
            None
        };

        // Delta resolve rates
        let resolve_rate: Vec<pb::ResolveRate> = current
            .resolve_rates
            .iter()
            .enumerate()
            .filter_map(|(i, c)| {
                let p = previous.resolve_rates.get(i).copied().unwrap_or(0);
                let delta = c.wrapping_sub(p) as u32;
                if delta > 0 {
                    Some(pb::ResolveRate {
                        count: delta,
                        reason: i as i32,
                    })
                } else {
                    None
                }
            })
            .collect();

        // Gauge fields: latest values as-is
        let last_update = self.last_state_update.load(Ordering::Relaxed);
        let state_age = if last_update > 0 {
            Some(pb::StateAge {
                last_state_update: last_update,
            })
        } else {
            None
        };

        pb::TelemetryData {
            sdk: None,
            resolve_latency,
            resolve_rate,
            state_age,
            memory_bytes: (self.memory_provider)(),
            resolver_version: crate::version::VERSION.to_string(),
        }
    }
}

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

pub fn build_request_telemetry(
    latency_us: Option<u32>,
    reasons: &[ResolveReason],
) -> pb::TelemetryData {
    let resolve_latency = latency_us.map(|us| {
        let idx = if us == 0 {
            0
        } else {
            let k = ((us as f64).ln() / LN_RATIO).floor() as usize;
            k.min(BUCKET_COUNT.saturating_sub(1))
        };
        pb::ResolveLatency {
            sum: us,
            count: 1,
            buckets: vec![pb::BucketSpan {
                offset: idx as i32,
                counts: vec![1],
            }],
            ln_ratio: LN_RATIO,
        }
    });

    let mut reason_counts: Vec<pb::ResolveRate> = Vec::new();
    for reason in reasons {
        let r = *reason as i32;
        if let Some(entry) = reason_counts.iter_mut().find(|e| e.reason == r) {
            entry.count = entry.count.saturating_add(1);
        } else {
            reason_counts.push(pb::ResolveRate {
                count: 1,
                reason: r,
            });
        }
    }

    pb::TelemetryData {
        resolve_latency,
        resolve_rate: reason_counts,
        resolver_version: crate::version::VERSION.to_string(),
        ..Default::default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn histogram_empty() {
        let hist = Histogram::new();
        let snap = hist.snapshot();
        assert_eq!(snap.count, 0);
        assert_eq!(snap.sum, 0);
    }

    #[test]
    fn histogram_basic_observation() {
        let hist = Histogram::new();
        hist.observe(100);
        hist.observe(200);
        hist.observe(300);

        let snap = hist.snapshot();
        assert_eq!(snap.count, 3);
        assert_eq!(snap.sum, 600);
    }

    #[test]
    fn histogram_snapshot_does_not_reset() {
        let hist = Histogram::new();
        hist.observe(100);
        hist.observe(200);

        let snap1 = hist.snapshot();
        assert_eq!(snap1.count, 2);
        assert_eq!(snap1.sum, 300);

        // Second snapshot should show same values
        let snap2 = hist.snapshot();
        assert_eq!(snap2.count, 2);
        assert_eq!(snap2.sum, 300);

        // Add more observations
        hist.observe(50);
        let snap3 = hist.snapshot();
        assert_eq!(snap3.count, 3);
        assert_eq!(snap3.sum, 350);
    }

    #[test]
    fn histogram_zero_lands_in_first_bucket() {
        let hist = Histogram::new();
        hist.observe(0);
        hist.observe(0);

        let snap = hist.snapshot();
        assert_eq!(snap.count, 2);
        assert_eq!(snap.buckets[0], 2);
    }

    #[test]
    fn histogram_covers_full_u32_range() {
        let hist = Histogram::new();
        hist.observe(1);
        hist.observe(1_000_000);
        hist.observe(u32::MAX);

        let snap = hist.snapshot();
        assert_eq!(snap.count, 3);
        let total: u64 = snap.buckets.iter().sum();
        assert_eq!(total, 3);
        // All three should land in different buckets
        let nonzero: Vec<_> = snap.buckets.iter().filter(|&&b| b > 0).collect();
        assert_eq!(nonzero.len(), 3);
    }

    #[test]
    fn histogram_bucket_spans() {
        let hist = Histogram::new();

        // Create a pattern with gaps
        hist.observe(10);
        hist.observe(11);
        hist.observe(12);
        // gap
        hist.observe(100);
        hist.observe(101);

        let snap = hist.snapshot();
        let total: u64 = snap.buckets.iter().sum();
        assert_eq!(total, 5);

        // Build spans and verify gaps create multiple spans
        let delta_buckets: Vec<u32> = snap.buckets.iter().map(|&b| b as u32).collect();
        let spans = Histogram::bucket_spans(&delta_buckets);
        assert!(spans.len() > 1, "should have gaps between spans");
        let span_total: u32 = spans.iter().flat_map(|s| &s.counts).sum();
        assert_eq!(span_total, 5);
    }

    #[test]
    fn telemetry_new_is_empty() {
        let tel = Telemetry::new();
        let snap = tel.snapshot();

        assert_eq!(snap.latency.count, 0);
        assert_eq!(snap.latency.sum, 0);
        assert!(snap.resolve_rates.iter().all(|&r| r == 0));
    }

    #[test]
    fn telemetry_record_latency() {
        let tel = Telemetry::new();
        tel.record_latency_us(1000);
        tel.record_latency_us(2000);
        tel.record_latency_us(3000);

        let snap = tel.snapshot();
        assert_eq!(snap.latency.count, 3);
        assert_eq!(snap.latency.sum, 6000);
    }

    #[test]
    fn telemetry_mark_resolve() {
        let tel = Telemetry::new();
        tel.mark_resolve(ResolveReason::Match);
        tel.mark_resolve(ResolveReason::Match);
        tel.mark_resolve(ResolveReason::NoTreatmentMatch);

        let snap = tel.snapshot();
        assert_eq!(snap.resolve_rates[ResolveReason::Match as usize], 2);
        assert_eq!(
            snap.resolve_rates[ResolveReason::NoTreatmentMatch as usize],
            1
        );
    }

    #[test]
    fn telemetry_snapshot_does_not_reset() {
        let tel = Telemetry::new();

        tel.record_latency_us(100);
        tel.mark_resolve(ResolveReason::Match);

        let snap1 = tel.snapshot();
        assert_eq!(snap1.latency.count, 1);
        assert_eq!(snap1.resolve_rates[ResolveReason::Match as usize], 1);

        // Second snapshot should show same values
        let snap2 = tel.snapshot();
        assert_eq!(snap2.latency.count, 1);
        assert_eq!(snap2.resolve_rates[ResolveReason::Match as usize], 1);

        // Add more data
        tel.record_latency_us(200);
        tel.mark_resolve(ResolveReason::NoTreatmentMatch);

        let snap3 = tel.snapshot();
        assert_eq!(snap3.latency.count, 2);
        assert_eq!(snap3.resolve_rates[ResolveReason::Match as usize], 1);
        assert_eq!(
            snap3.resolve_rates[ResolveReason::NoTreatmentMatch as usize],
            1
        );
    }

    #[test]
    fn telemetry_concurrent_updates() {
        use std::thread;

        let tel = Arc::new(Telemetry::new());
        let mut handles = vec![];

        for i in 0..10 {
            let tel_clone = Arc::clone(&tel);
            let handle = thread::spawn(move || {
                for _ in 0..100 {
                    tel_clone.record_latency_us(i * 10);
                    tel_clone.mark_resolve(ResolveReason::Match);
                }
            });
            handles.push(handle);
        }

        for handle in handles {
            handle.join().unwrap();
        }

        let snap = tel.snapshot();
        assert_eq!(snap.latency.count, 1000);
        assert_eq!(snap.resolve_rates[ResolveReason::Match as usize], 1000);
    }

    #[test]
    fn delta_snapshot_basic() {
        let tel = Telemetry::new();
        let last = ArcSwap::from_pointee(TelemetrySnapshot::default());

        // First flush: delta should equal full snapshot
        tel.record_latency_us(100);
        tel.mark_resolve(ResolveReason::Match);

        let delta1 = tel.delta_snapshot(&last);
        assert_eq!(delta1.resolve_latency.as_ref().unwrap().count, 1);
        assert_eq!(delta1.resolve_latency.as_ref().unwrap().sum, 100);
        assert_eq!(delta1.resolve_rate.len(), 1);
        assert_eq!(delta1.resolve_rate[0].count, 1);
        assert_eq!(delta1.resolve_rate[0].reason, ResolveReason::Match as i32);

        // Second flush: delta should only show new observations
        tel.record_latency_us(200);
        tel.record_latency_us(300);
        tel.mark_resolve(ResolveReason::Match);
        tel.mark_resolve(ResolveReason::NoTreatmentMatch);

        let delta2 = tel.delta_snapshot(&last);
        assert_eq!(delta2.resolve_latency.as_ref().unwrap().count, 2);
        assert_eq!(delta2.resolve_latency.as_ref().unwrap().sum, 500);

        let match_rate = delta2
            .resolve_rate
            .iter()
            .find(|r| r.reason == ResolveReason::Match as i32)
            .unwrap();
        assert_eq!(match_rate.count, 1); // delta, not cumulative 2

        let no_match_rate = delta2
            .resolve_rate
            .iter()
            .find(|r| r.reason == ResolveReason::NoTreatmentMatch as i32)
            .unwrap();
        assert_eq!(no_match_rate.count, 1);
    }

    #[test]
    fn delta_snapshot_no_activity() {
        let tel = Telemetry::new();
        let last = ArcSwap::from_pointee(TelemetrySnapshot::default());

        tel.record_latency_us(100);
        let _ = tel.delta_snapshot(&last);

        // No new activity
        let delta = tel.delta_snapshot(&last);
        assert!(delta.resolve_latency.is_none());
        assert!(delta.resolve_rate.is_empty());
    }

    #[test]
    fn delta_snapshot_gauges_always_present() {
        let tel = Telemetry::new();
        let last = ArcSwap::from_pointee(TelemetrySnapshot::default());

        tel.set_last_state_update(1000);
        let _ = tel.delta_snapshot(&last);

        // Gauge values should appear even if no counter changes
        let delta = tel.delta_snapshot(&last);
        assert!(delta.resolve_latency.is_none()); // no counter activity
        assert_eq!(delta.state_age.unwrap().last_state_update, 1000);
    }

    #[test]
    fn prometheus_empty_snapshot() {
        let snap = TelemetrySnapshot::default();
        assert_eq!(snap.to_prometheus("w0", &PrometheusConfig::default()), "");
    }

    #[test]
    fn prometheus_histogram_and_counters() {
        let tel = Telemetry::new();
        tel.record_latency_us(100);
        tel.record_latency_us(500);
        tel.mark_resolve(ResolveReason::Match);
        tel.mark_resolve(ResolveReason::Match);
        tel.mark_resolve(ResolveReason::NoSegmentMatch);

        let config = PrometheusConfig::default();
        let prom = tel.snapshot().to_prometheus("w0", &config);

        // HELP and TYPE headers
        assert!(prom.contains("# HELP confidence_resolve_latency_microseconds"));
        assert!(prom.contains("# TYPE confidence_resolve_latency_microseconds histogram"));
        // Must have +Inf bucket with resolver_id label
        assert!(prom.contains(
            r#"confidence_resolve_latency_microseconds_bucket{resolver_id="w0",le="+Inf"} 2"#
        ));
        // Sum and count with resolver_id label
        assert!(
            prom.contains(r#"confidence_resolve_latency_microseconds_sum{resolver_id="w0"} 600"#)
        );
        assert!(
            prom.contains(r#"confidence_resolve_latency_microseconds_count{resolver_id="w0"} 2"#)
        );

        // Counters with resolver_id label
        assert!(prom.contains("# HELP confidence_resolves_total"));
        assert!(prom.contains("# TYPE confidence_resolves_total counter"));
        assert!(prom.contains(
            r#"confidence_resolves_total{resolver_id="w0",reason="RESOLVE_REASON_MATCH"} 2"#
        ));
        assert!(prom.contains(
            r#"confidence_resolves_total{resolver_id="w0",reason="RESOLVE_REASON_NO_SEGMENT_MATCH"} 1"#
        ));
        // Zero-count reasons should be omitted
        assert!(!prom.contains("FLAG_ARCHIVED"));
    }

    #[test]
    fn prometheus_cumulative_buckets() {
        let tel = Telemetry::new();
        // Two observations in different buckets
        tel.record_latency_us(10);
        tel.record_latency_us(10_000);

        let config = PrometheusConfig::default();
        let prom = tel.snapshot().to_prometheus("w0", &config);

        // Parse all le buckets and verify they are monotonically non-decreasing
        let bucket_counts: Vec<u64> = prom
            .lines()
            .filter(|l| l.contains("le=\"") && !l.contains("+Inf"))
            .map(|l| l.rsplit_once(' ').unwrap().1.parse::<u64>().unwrap())
            .collect();

        for pair in bucket_counts.windows(2) {
            assert!(
                pair[1] >= pair[0],
                "Prometheus buckets must be cumulative, got {} then {}",
                pair[0],
                pair[1]
            );
        }
    }

    #[test]
    fn prometheus_distinct_instances_concatenate() {
        let tel = Telemetry::new();
        tel.record_latency_us(100);
        tel.mark_resolve(ResolveReason::Match);

        let snap = tel.snapshot();
        let config = PrometheusConfig::default();
        let mut combined = snap.to_prometheus("w0", &config);
        combined.push_str(&snap.to_prometheus("w1", &config));

        // Both resolver_ids present
        assert!(combined.contains(r#"resolver_id="w0""#));
        assert!(combined.contains(r#"resolver_id="w1""#));
    }

    #[test]
    fn prometheus_bucket_striding() {
        let tel = Telemetry::new();
        tel.record_latency_us(100);
        tel.record_latency_us(10_000);
        tel.record_latency_us(1_000_000);

        let snap = tel.snapshot();

        // Default (18 per decade) — count all le= lines
        let full = snap.to_prometheus(
            "w0",
            &PrometheusConfig {
                buckets_per_decade: 18,
                ..PrometheusConfig::default()
            },
        );
        let full_count = full.lines().filter(|l| l.contains("le=\"")).count();

        // 9 per decade — should produce roughly half the bucket lines
        let half = snap.to_prometheus(
            "w0",
            &PrometheusConfig {
                buckets_per_decade: 9,
                ..PrometheusConfig::default()
            },
        );
        let half_count = half.lines().filter(|l| l.contains("le=\"")).count();

        // 6 per decade — should produce roughly a third
        let third = snap.to_prometheus(
            "w0",
            &PrometheusConfig {
                buckets_per_decade: 6,
                ..PrometheusConfig::default()
            },
        );
        let third_count = third.lines().filter(|l| l.contains("le=\"")).count();

        assert!(
            half_count < full_count,
            "9 bpd ({half_count}) should produce fewer buckets than 18 bpd ({full_count})"
        );
        assert!(
            third_count < half_count,
            "6 bpd ({third_count}) should produce fewer buckets than 9 bpd ({half_count})"
        );

        // +Inf must always be present
        assert!(half.contains(r#"le="+Inf""#));
        assert!(third.contains(r#"le="+Inf""#));

        // Cumulative correctness: last non-Inf bucket count should equal +Inf count
        // (since all observations are within u32 range)
        let inf_line = half.lines().find(|l| l.contains("+Inf")).unwrap();
        let inf_count: u64 = inf_line.rsplit_once(' ').unwrap().1.parse().unwrap();
        assert_eq!(inf_count, 3);
    }

    #[test]
    fn openmetrics_parses_successfully() {
        let tel = Telemetry::with_memory_provider(|| 1_048_576);
        tel.record_latency_us(100);
        tel.record_latency_us(500);
        tel.mark_resolve(ResolveReason::Match);
        tel.mark_resolve(ResolveReason::NoSegmentMatch);

        let config = PrometheusConfig {
            openmetrics: true,
            ..PrometheusConfig::default()
        };
        let output = tel.snapshot().to_prometheus("w0", &config);

        // Must end with # EOF
        assert!(output.trim_end().ends_with("# EOF"));

        // Must contain float-formatted values
        assert!(output.contains(".0\n"));

        // Must parse with a strict OpenMetrics parser
        let result = openmetrics_parser::openmetrics::parse_openmetrics(&output);
        assert!(
            result.is_ok(),
            "OpenMetrics parser rejected output: {:?}\n\nRaw:\n{output}",
            result.err()
        );
    }

    #[test]
    fn openmetrics_with_bucket_striding() {
        let tel = Telemetry::with_memory_provider(|| 1_048_576);
        tel.record_latency_us(100);
        tel.record_latency_us(10_000);
        tel.mark_resolve(ResolveReason::Match);

        let config = PrometheusConfig {
            buckets_per_decade: 9,
            openmetrics: true,
        };
        let output = tel.snapshot().to_prometheus("w0", &config);

        let result = openmetrics_parser::openmetrics::parse_openmetrics(&output);
        assert!(
            result.is_ok(),
            "OpenMetrics parser rejected strided output: {:?}\n\nRaw:\n{output}",
            result.err()
        );
    }

    #[test]
    fn prometheus_bucket_stride_zero_means_default() {
        let tel = Telemetry::new();
        tel.record_latency_us(100);

        let snap = tel.snapshot();
        let default_out = snap.to_prometheus(
            "w0",
            &PrometheusConfig {
                buckets_per_decade: 18,
                ..PrometheusConfig::default()
            },
        );
        let zero_out = snap.to_prometheus(
            "w0",
            &PrometheusConfig {
                buckets_per_decade: 0,
                ..PrometheusConfig::default()
            },
        );
        assert_eq!(default_out, zero_out);
    }

    #[test]
    fn accumulate_delta_basic() {
        let mut snap = TelemetrySnapshot::default();
        let td = pb::TelemetryData {
            resolve_latency: Some(pb::ResolveLatency {
                sum: 500,
                count: 2,
                buckets: vec![pb::BucketSpan {
                    offset: 5,
                    counts: vec![1, 1],
                }],
                ln_ratio: LN_RATIO,
            }),
            resolve_rate: vec![pb::ResolveRate {
                reason: ResolveReason::Match as i32,
                count: 3,
            }],
            memory_bytes: 4096,
            ..Default::default()
        };

        snap.accumulate_delta(&td);
        assert_eq!(snap.latency.sum, 500);
        assert_eq!(snap.latency.count, 2);
        assert_eq!(snap.latency.buckets[5], 1);
        assert_eq!(snap.latency.buckets[6], 1);
        assert_eq!(snap.resolve_rates[ResolveReason::Match as usize], 3);
        assert_eq!(snap.memory_bytes, 4096);

        // Second delta accumulates counters, replaces gauge
        snap.accumulate_delta(&td);
        assert_eq!(snap.latency.sum, 1000);
        assert_eq!(snap.latency.count, 4);
        assert_eq!(snap.latency.buckets[5], 2);
        assert_eq!(snap.resolve_rates[ResolveReason::Match as usize], 6);
        assert_eq!(snap.memory_bytes, 4096);
    }

    #[test]
    fn accumulate_delta_negative_offset_skipped() {
        let mut snap = TelemetrySnapshot::default();
        let td = pb::TelemetryData {
            resolve_latency: Some(pb::ResolveLatency {
                sum: 100,
                count: 1,
                buckets: vec![
                    pb::BucketSpan {
                        offset: -1,
                        counts: vec![99],
                    },
                    pb::BucketSpan {
                        offset: 3,
                        counts: vec![1],
                    },
                ],
                ln_ratio: LN_RATIO,
            }),
            ..Default::default()
        };

        snap.accumulate_delta(&td);
        // Negative offset span skipped, valid span applied
        assert_eq!(snap.latency.count, 1);
        assert_eq!(snap.latency.buckets.get(3).copied().unwrap_or(0), 1);
        // Bucket from negative offset should not exist
        let total: u64 = snap.latency.buckets.iter().sum();
        assert_eq!(total, 1);
    }

    #[test]
    fn accumulate_delta_oversized_offset_skipped() {
        let mut snap = TelemetrySnapshot::default();
        let td = pb::TelemetryData {
            resolve_latency: Some(pb::ResolveLatency {
                sum: 100,
                count: 1,
                buckets: vec![pb::BucketSpan {
                    offset: BUCKET_COUNT as i32 + 10,
                    counts: vec![1],
                }],
                ln_ratio: LN_RATIO,
            }),
            ..Default::default()
        };

        snap.accumulate_delta(&td);
        // Oversized offset span skipped, sum/count still accumulated
        assert_eq!(snap.latency.count, 1);
        assert_eq!(snap.latency.sum, 100);
        let total: u64 = snap.latency.buckets.iter().sum();
        assert_eq!(total, 0);
    }
}
