//! Mapping from nm metrics to OpenTelemetry instruments.

use std::hash::BuildHasher;
use std::sync::Arc;

use foldhash::fast::RandomState;
use hashbrown::HashTable;
use hashbrown::hash_table::Entry;
use nm::{EventName, Magnitude, Report};
use opentelemetry::KeyValue;
use opentelemetry::metrics::{Counter, Gauge, Meter};

use crate::state::CollectionState;

/// Suffix for the sum metric name.
const SUM_SUFFIX: &str = "_sum";

/// Suffix for histogram bucket metric name.
const BUCKET_SUFFIX: &str = "_bucket";

/// Attribute key for histogram bucket upper bound (Prometheus convention).
const LE_ATTRIBUTE: &str = "le";

/// Cached OpenTelemetry instruments for a single nm event.
///
/// All instruments are created together when an event is first seen, avoiding repeated
/// string formatting on every metric lookup.
#[derive(Debug)]
struct EventInstruments {
    /// Counter for event count (named after the event itself, e.g. `http_requests`).
    count_counter: Counter<u64>,

    /// Gauge for event sum (named `{event}_sum`).
    sum_gauge: Gauge<i64>,

    /// Counter for histogram buckets (named `{event}_bucket`), if the event has a histogram.
    /// Different bucket bounds are distinguished by the `le` attribute, not by instrument name.
    bucket_counter: Option<Counter<u64>>,

    /// Precomputed `[KeyValue; 1]` attribute slice for each histogram bucket.
    ///
    /// Indexed by bucket index, matching the order from `histogram.magnitudes()`.
    /// Building the full `KeyValue` once per bucket at registration time eliminates the
    /// per-export `Arc<str>` clone and `KeyValue` construction that the export hot path
    /// would otherwise perform on every call to `add_bucket_delta`.
    ///
    /// `[KeyValue; 1]` (not `KeyValue`) so the hot path can pass the attribute slice to
    /// `Counter::add` as `&self.bucket_attrs[i]` without constructing a `&[KeyValue]`
    /// reference each call. The fixed-size array's `&[KeyValue; 1]` deref-coerces to
    /// `&[KeyValue]` directly.
    bucket_attrs: Vec<[KeyValue; 1]>,
}

/// Manages OpenTelemetry instruments for nm metrics.
///
/// Creates instruments on-demand as events appear in reports and caches them for reuse.
/// All instruments for an event are created together, and string formatting only happens
/// once per event (not on every metric export).
#[derive(Debug)]
pub(crate) struct InstrumentRegistry {
    meter: Meter,

    // See the matching comment on `CollectionState::hasher` in `state.rs` for the design
    // rationale. In short: we use `hashbrown::HashTable` instead of `HashMap` so the
    // `entry(hash, eq, hasher)` API can clone `EventName` only on cache miss, which stable
    // `HashMap` cannot do without double-hashing on hits. Hashing is still `foldhash`.
    hasher: RandomState,

    /// Cached instruments per event name.
    events: HashTable<(EventName, EventInstruments)>,
}

impl InstrumentRegistry {
    /// Creates a new instrument registry with the given meter.
    pub(crate) fn new(meter: Meter) -> Self {
        Self {
            meter,
            hasher: RandomState::default(),
            events: HashTable::new(),
        }
    }

    /// Gets or creates instruments for an event.
    ///
    /// If histogram magnitudes are provided, the bucket counter and cached bucket attribute
    /// arrays are also created. This avoids creating them for events without histogram data.
    fn instruments(
        &mut self,
        event_name: &EventName,
        magnitudes: Option<impl Iterator<Item = Magnitude>>,
    ) -> &EventInstruments {
        let meter = &self.meter;

        let hash = self.hasher.hash_one(event_name);
        let hasher = &self.hasher;
        // See `CollectionState::event_state` for the per-closure breakdown; the same three-
        // closure contract (lookup hash, equality, growth-time rehash) applies here.
        let instruments = match self.events.entry(
            hash,
            |(existing, _)| existing == event_name,
            |(existing, _)| hasher.hash_one(existing),
        ) {
            Entry::Occupied(occupied) => &mut occupied.into_mut().1,
            Entry::Vacant(vacant) => {
                let sum_name = format!("{event_name}{SUM_SUFFIX}");
                let new_instruments = EventInstruments {
                    count_counter: meter.u64_counter(event_name.to_string()).build(),
                    sum_gauge: meter.i64_gauge(sum_name).build(),
                    bucket_counter: None,
                    bucket_attrs: Vec::new(),
                };
                // Cache miss — clone the key here so the hit path stays clone-free.
                &mut vacant
                    .insert((event_name.clone(), new_instruments))
                    .into_mut()
                    .1
            }
        };

        // Lazily create the bucket counter and cache bucket attributes if histogram data
        // provided. Building the `KeyValue` once per bucket here eliminates per-export
        // `Arc<str>` clone and `KeyValue::new` work in `add_bucket_delta`.
        if let Some(mags) = magnitudes
            && instruments.bucket_counter.is_none()
        {
            let bucket_name = format!("{event_name}{BUCKET_SUFFIX}");
            instruments.bucket_counter = Some(meter.u64_counter(bucket_name).build());
            instruments.bucket_attrs = mags
                .map(|m| [KeyValue::new(LE_ATTRIBUTE, format_bucket_bound(m))])
                .collect();
        }

        instruments
    }
}

/// Exports an nm report to OpenTelemetry instruments.
///
/// This function processes each event in the report, computes deltas where needed,
/// and records values to the appropriate OpenTelemetry instruments.
pub(crate) fn export_report(
    report: &Report,
    state: &mut CollectionState,
    instruments: &mut InstrumentRegistry,
) {
    for event in report.events() {
        let event_name = event.name();
        let event_state = state.event_state(event_name);
        let histogram = event.histogram();

        // Get cached instruments for this event (creates them if first time seeing this event).
        // Pass histogram magnitudes if present so bucket counter and bounds can be created.
        let event_instruments =
            instruments.instruments(event_name, histogram.map(nm::Histogram::magnitudes));

        // Export count as counter (delta).
        let count_delta = event_state.count_delta(event.count());
        add_count_delta(&event_instruments.count_counter, count_delta);

        // Export sum as gauge (absolute value).
        event_instruments.sum_gauge.record(event.sum(), &[]);

        // Export histogram buckets if present.
        if let Some(histogram) = histogram {
            let bucket_counter = event_instruments
                .bucket_counter
                .as_ref()
                .expect("bucket counter was created during instruments() call");

            for (i, (_magnitude, _cumulative, delta)) in event_state
                .histogram_deltas(histogram.magnitudes(), histogram.counts())
                .enumerate()
            {
                let attrs = event_instruments
                    .bucket_attrs
                    .get(i)
                    .expect("bucket_attrs length matches histogram bucket count");
                add_bucket_delta(bucket_counter, delta, attrs);
            }
        }
    }
}

/// Adds a count delta to a counter if positive.
///
/// This is a separate function to allow skipping equivalent mutations - adding 0 to a counter
/// produces the same observable result as not calling add at all.
#[cfg_attr(test, mutants::skip)]
fn add_count_delta(counter: &Counter<u64>, delta: u64) {
    if delta > 0 {
        counter.add(delta, &[]);
    }
}

/// Adds a bucket delta to a counter if positive.
///
/// `attrs` is a fixed-size attribute array stored in the registry. `&[KeyValue; 1]`
/// auto-coerces to `&[KeyValue]` at the `Counter::add` call site, so no per-call
/// slice or `KeyValue` construction is required on the hot path.
///
/// This is a separate function to allow skipping equivalent mutations - adding 0 to a counter
/// produces the same observable result as not calling add at all.
#[cfg_attr(test, mutants::skip)]
fn add_bucket_delta(counter: &Counter<u64>, delta: u64, attrs: &[KeyValue; 1]) {
    if delta > 0 {
        counter.add(delta, attrs);
    }
}

/// Formats a bucket bound for the `le` attribute.
///
/// `Magnitude::MAX` is formatted as `+Inf` following Prometheus conventions.
fn format_bucket_bound(magnitude: Magnitude) -> Arc<str> {
    if magnitude == Magnitude::MAX {
        Arc::from("+Inf")
    } else {
        Arc::from(magnitude.to_string())
    }
}

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
    use nm::{EventMetrics, Histogram};
    use opentelemetry::metrics::MeterProvider;
    use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider};

    use super::*;

    fn create_test_provider() -> (SdkMeterProvider, InMemoryMetricExporter) {
        let exporter = InMemoryMetricExporter::default();
        let reader = PeriodicReader::builder(exporter.clone()).build();
        let provider = SdkMeterProvider::builder().with_reader(reader).build();
        (provider, exporter)
    }

    #[test]
    fn format_bucket_bound_regular_values() {
        assert_eq!(format_bucket_bound(10).as_ref(), "10");
        assert_eq!(format_bucket_bound(0).as_ref(), "0");
        assert_eq!(format_bucket_bound(-5).as_ref(), "-5");
        assert_eq!(format_bucket_bound(1000).as_ref(), "1000");
    }

    #[test]
    fn format_bucket_bound_max_is_plus_inf() {
        assert_eq!(format_bucket_bound(Magnitude::MAX).as_ref(), "+Inf");
    }

    // OpenTelemetry SDK uses system time calls not available under Miri isolation.
    #[cfg_attr(miri, ignore)]
    #[test]
    fn export_report_simple_event() {
        let (provider, exporter) = create_test_provider();
        let meter = provider.meter("test");

        let mut state = CollectionState::new();
        let mut instruments = InstrumentRegistry::new(meter);

        let event = EventMetrics::fake("test_event", 100, 5000, None);
        let report = Report::fake(vec![event]);

        export_report(&report, &mut state, &mut instruments);

        // Force flush to ensure metrics are exported.
        provider.force_flush().unwrap();

        let metrics = exporter.get_finished_metrics().unwrap();
        assert!(!metrics.is_empty());
    }

    // OpenTelemetry SDK uses system time calls not available under Miri isolation.
    #[cfg_attr(miri, ignore)]
    #[test]
    fn export_report_event_with_histogram() {
        static BUCKETS: &[Magnitude] = &[10, 50, 100, 500];

        let (provider, exporter) = create_test_provider();
        let meter = provider.meter("test");

        let mut state = CollectionState::new();
        let mut instruments = InstrumentRegistry::new(meter);

        let histogram = Histogram::fake(BUCKETS, vec![5, 12, 8, 3], 2);
        let event = EventMetrics::fake("latency_ms", 30, 4567, Some(histogram));
        let report = Report::fake(vec![event]);

        export_report(&report, &mut state, &mut instruments);

        provider.force_flush().unwrap();

        let metrics = exporter.get_finished_metrics().unwrap();
        assert!(!metrics.is_empty());
    }

    // OpenTelemetry SDK uses system time calls not available under Miri isolation.
    #[cfg_attr(miri, ignore)]
    #[test]
    fn export_report_delta_computation() {
        static BUCKETS: &[Magnitude] = &[10, 50];

        let (provider, _exporter) = create_test_provider();
        let meter = provider.meter("test");

        let mut state = CollectionState::new();
        let mut instruments = InstrumentRegistry::new(meter);

        // First collection.
        let histogram1 = Histogram::fake(BUCKETS, vec![5, 10], 2);
        let event1 = EventMetrics::fake("test_event", 17, 100, Some(histogram1));
        let report1 = Report::fake(vec![event1]);

        export_report(&report1, &mut state, &mut instruments);

        // Verify state was updated.
        let event_state = state.event_state(&"test_event".into());
        assert_eq!(event_state.count, 17);
        assert_eq!(event_state.histogram_buckets, vec![5, 15, 17]); // Cumulative: 5, 15, 17.

        // Second collection with more data.
        let histogram2 = Histogram::fake(BUCKETS, vec![8, 15], 4);
        let event2 = EventMetrics::fake("test_event", 27, 200, Some(histogram2));
        let report2 = Report::fake(vec![event2]);

        export_report(&report2, &mut state, &mut instruments);

        // Verify state was updated again.
        let event_state = state.event_state(&"test_event".into());
        assert_eq!(event_state.count, 27);
        assert_eq!(event_state.histogram_buckets, vec![8, 23, 27]); // Cumulative: 8, 23, 27.
    }

    // OpenTelemetry SDK uses system time calls not available under Miri isolation.
    #[cfg_attr(miri, ignore)]
    #[test]
    fn export_report_multiple_events() {
        let (provider, exporter) = create_test_provider();
        let meter = provider.meter("test");

        let mut state = CollectionState::new();
        let mut instruments = InstrumentRegistry::new(meter);

        let event1 = EventMetrics::fake("event_a", 10, 100, None);
        let event2 = EventMetrics::fake("event_b", 20, 200, None);
        let report = Report::fake(vec![event1, event2]);

        export_report(&report, &mut state, &mut instruments);

        provider.force_flush().unwrap();

        let metrics = exporter.get_finished_metrics().unwrap();
        assert!(!metrics.is_empty());
    }

    // OpenTelemetry SDK uses system time calls not available under Miri isolation.
    #[cfg_attr(miri, ignore)]
    #[test]
    fn instrument_registry_preserves_entries_across_table_growth() {
        // The underlying `HashTable` grows when capacity is exceeded, which calls the
        // rehash closure passed to `entry()` on every existing entry. This test inserts
        // enough events to force multiple grows and then re-exports the same events to
        // verify that lookups still find the existing entries — otherwise entries would
        // land in the wrong buckets after a grow and the second export would create
        // duplicate `EventInstruments`, leaving the table with more than `NUM_EVENTS`
        // entries.
        const NUM_EVENTS: u64 = 64;

        let (provider, _exporter) = create_test_provider();
        let meter = provider.meter("test");

        let mut state = CollectionState::new();
        let mut instruments = InstrumentRegistry::new(meter);

        // First pass: register every event, forcing the `HashTable` to grow.
        let events: Vec<_> = (0..NUM_EVENTS)
            .map(|i| EventMetrics::fake(format!("growth_event_{i}"), i.saturating_add(1), 0, None))
            .collect();
        let report = Report::fake(events);
        export_report(&report, &mut state, &mut instruments);

        let expected_len = usize::try_from(NUM_EVENTS).unwrap();
        assert_eq!(instruments.events.len(), expected_len);

        // Second pass with the same events: every lookup must hit the existing entry.
        let events2: Vec<_> = (0..NUM_EVENTS)
            .map(|i| {
                EventMetrics::fake(
                    format!("growth_event_{i}"),
                    i.saturating_add(1).saturating_mul(2),
                    0,
                    None,
                )
            })
            .collect();
        let report2 = Report::fake(events2);
        export_report(&report2, &mut state, &mut instruments);

        // If the rehash closure produced inconsistent hashes for any existing entry,
        // the second export would have inserted a duplicate instead of reusing it.
        assert_eq!(instruments.events.len(), expected_len);
    }

    // OpenTelemetry SDK uses system time calls not available under Miri isolation.
    #[cfg_attr(miri, ignore)]
    #[test]
    fn export_report_zero_count_delta_does_not_add_to_counter() {
        static BUCKETS: &[Magnitude] = &[10, 50];

        let (provider, _exporter) = create_test_provider();
        let meter = provider.meter("test");

        let mut state = CollectionState::new();
        let mut instruments = InstrumentRegistry::new(meter);

        // First collection establishes baseline.
        let histogram1 = Histogram::fake(BUCKETS, vec![5, 10], 2);
        let event1 = EventMetrics::fake("test_event", 100, 500, Some(histogram1));
        let report1 = Report::fake(vec![event1]);
        export_report(&report1, &mut state, &mut instruments);

        // Second collection with same count - zero delta.
        let histogram2 = Histogram::fake(BUCKETS, vec![5, 10], 2);
        let event2 = EventMetrics::fake("test_event", 100, 500, Some(histogram2));
        let report2 = Report::fake(vec![event2]);

        // This should complete without error even when delta is zero.
        export_report(&report2, &mut state, &mut instruments);

        // Verify the state shows zero delta was computed.
        let event_state = state.event_state(&"test_event".into());
        assert_eq!(event_state.count, 100);
    }

    // OpenTelemetry SDK uses system time calls not available under Miri isolation.
    #[cfg_attr(miri, ignore)]
    #[test]
    fn export_report_zero_bucket_delta_does_not_add_to_counter() {
        static BUCKETS: &[Magnitude] = &[10, 50];

        let (provider, _exporter) = create_test_provider();
        let meter = provider.meter("test");

        let mut state = CollectionState::new();
        let mut instruments = InstrumentRegistry::new(meter);

        // First collection establishes baseline.
        let histogram1 = Histogram::fake(BUCKETS, vec![5, 10], 2);
        let event1 = EventMetrics::fake("test_event", 10, 100, Some(histogram1));
        let report1 = Report::fake(vec![event1]);
        export_report(&report1, &mut state, &mut instruments);

        // Second collection with same histogram bucket counts - zero bucket deltas.
        let histogram2 = Histogram::fake(BUCKETS, vec![5, 10], 2);
        let event2 = EventMetrics::fake("test_event", 10, 100, Some(histogram2));
        let report2 = Report::fake(vec![event2]);

        // This should complete without error even when bucket deltas are zero.
        export_report(&report2, &mut state, &mut instruments);

        // Verify histogram state is unchanged (same cumulative values).
        let event_state = state.event_state(&"test_event".into());
        assert_eq!(event_state.histogram_buckets, vec![5, 15, 17]);
    }
}
