//! What the router exports for its dashboard, through the swarm collector.
//!
//! - `hyperhive.notify_router.events`: forge events taken off the stream, by
//!   `event_type` and `outcome` (`routed`, `unmatched`, `ignored`, `unusable`,
//!   `retried`).
//! - `hyperhive.notify_router.event.duration`: seconds to handle one event, by
//!   `event_type`.
//! - `hyperhive.notify_router.todos`: todo ops queued for an agent, by `op`
//!   (`upsert`, `clear`) and `outcome` (`delivered`, `failed`).
//! - `hyperhive.notify_router.review_requests`: review requests made to the
//!   controller, by `outcome` (`requested`, `skipped`, `failed`).
//! - `hyperhive.notify_router.pull.failures`: pulls from the stream that failed.
//! - `hyperhive.notify_router.consumer.pending`, `.ack_pending`, `.redelivered`:
//!   the router's own consumer on the events stream, read at the start of each
//!   pull: events not yet delivered to it, delivered and not yet acked, and
//!   delivered more than once. A growing `pending` is the router falling behind.
//!
//! The provider is built from the standard `OTEL_*` environment, so without a
//! collector the instruments do nothing.

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use opentelemetry::KeyValue;
use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider as _};
use opentelemetry_sdk::metrics::SdkMeterProvider;

const METER: &str = "hyperhive.notify_router";

/// Seconds. Handling one event is a few queue and store round trips, in
/// milliseconds; a review request waits on the controller's answer, up to 30
/// seconds. The SDK's own bounds start at 5 and are made for milliseconds.
const DURATION_BOUNDARIES: [f64; 12] = [
    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0,
];

/// What came of handling one event.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Handled {
    /// Names no issue or pull request: nothing to route.
    Ignored,
    /// No flow matched.
    Unmatched,
    /// At least one flow matched and its ops are queued.
    Routed,
    /// Can never be handled; dropped for good.
    Unusable,
    /// The queue or the store failed; it comes back.
    Retried,
}

impl Handled {
    const fn name(self) -> &'static str {
        match self {
            Self::Ignored => "ignored",
            Self::Unmatched => "unmatched",
            Self::Routed => "routed",
            Self::Unusable => "unusable",
            Self::Retried => "retried",
        }
    }
}

/// The router's instruments, and the provider that exports them.
pub struct Metrics {
    /// Kept alive: it exports only while it lives.
    _provider: Option<SdkMeterProvider>,
    events: Counter<u64>,
    duration: Histogram<f64>,
    todos: Counter<u64>,
    reviews: Counter<u64>,
    pull_failures: Counter<u64>,
    consumer: Arc<Consumer>,
}

/// The consumer's counts as of the last read, for the gauges to report.
#[derive(Default)]
struct Consumer {
    pending: AtomicU64,
    ack_pending: AtomicU64,
    redelivered: AtomicU64,
}

impl Metrics {
    /// The instruments on the provider the environment names, or no-op ones.
    #[must_use]
    pub fn from_env() -> Self {
        let provider = swarm_secret_client::meter_provider::from_env(METER, "notify-router");
        let meter = provider
            .as_ref()
            .map_or_else(|| opentelemetry::global::meter(METER), |p| p.meter(METER));
        Self::on(&meter, provider)
    }

    fn on(meter: &Meter, provider: Option<SdkMeterProvider>) -> Self {
        let consumer = Arc::new(Consumer::default());
        for (name, read) in [
            (
                "pending",
                (|c: &Consumer| &c.pending) as fn(&Consumer) -> &AtomicU64,
            ),
            ("ack_pending", |c| &c.ack_pending),
            ("redelivered", |c| &c.redelivered),
        ] {
            let consumer = Arc::clone(&consumer);
            meter
                .u64_observable_gauge(format!("{METER}.consumer.{name}"))
                .with_callback(move |obs| obs.observe(read(&consumer).load(Ordering::Relaxed), &[]))
                .build();
        }
        Self {
            _provider: provider,
            events: meter.u64_counter(format!("{METER}.events")).build(),
            duration: meter
                .f64_histogram(format!("{METER}.event.duration"))
                .with_unit("s")
                .with_boundaries(DURATION_BOUNDARIES.to_vec())
                .build(),
            todos: meter.u64_counter(format!("{METER}.todos")).build(),
            reviews: meter
                .u64_counter(format!("{METER}.review_requests"))
                .build(),
            pull_failures: meter.u64_counter(format!("{METER}.pull.failures")).build(),
            consumer,
        }
    }

    /// One event handled, after `took`.
    pub fn event(&self, event_type: &str, handled: Handled, took: Duration) {
        let event_type = KeyValue::new("event_type", event_type.to_owned());
        self.events.add(
            1,
            &[event_type.clone(), KeyValue::new("outcome", handled.name())],
        );
        self.duration.record(took.as_secs_f64(), &[event_type]);
    }

    /// One todo op queued, or failed to be.
    pub fn todo(&self, op: &'static str, delivered: bool) {
        let outcome = if delivered { "delivered" } else { "failed" };
        self.todos.add(
            1,
            &[KeyValue::new("op", op), KeyValue::new("outcome", outcome)],
        );
    }

    /// One review request answered: `requested`, `skipped` or `failed`.
    pub fn review_request(&self, outcome: &'static str) {
        self.reviews.add(1, &[KeyValue::new("outcome", outcome)]);
    }

    /// One pull from the stream failed.
    pub fn pull_failed(&self) {
        self.pull_failures.add(1, &[]);
    }

    /// The consumer's counts, as the stream reports them.
    pub fn consumer(&self, pending: u64, ack_pending: usize, redelivered: usize) {
        let as_u64 = |n: usize| u64::try_from(n).unwrap_or(u64::MAX);
        self.consumer.pending.store(pending, Ordering::Relaxed);
        self.consumer
            .ack_pending
            .store(as_u64(ack_pending), Ordering::Relaxed);
        self.consumer
            .redelivered
            .store(as_u64(redelivered), Ordering::Relaxed);
    }
}

#[cfg(test)]
mod tests {
    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
    use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader};

    use super::*;

    /// Metrics on an in-memory exporter, and the exporter.
    fn recorded() -> (Metrics, SdkMeterProvider, InMemoryMetricExporter) {
        let exporter = InMemoryMetricExporter::default();
        let reader = PeriodicReader::builder(exporter.clone()).build();
        let provider = SdkMeterProvider::builder().with_reader(reader).build();
        let metrics = Metrics::on(&provider.meter("test"), None);
        (metrics, provider, exporter)
    }

    /// Every data point of the sum or gauge `name`, as its labels and value.
    fn points(exporter: &InMemoryMetricExporter, name: &str) -> Vec<(Vec<(String, String)>, u64)> {
        let mut out = Vec::new();
        for resource in exporter.get_finished_metrics().unwrap() {
            for scope in resource.scope_metrics() {
                for metric in scope.metrics().filter(|m| m.name() == name) {
                    // The SDK keeps attributes in no fixed order, so sort them.
                    let labels = |attrs: &mut dyn Iterator<Item = &KeyValue>| {
                        let mut all = attrs
                            .map(|kv| (kv.key.to_string(), kv.value.to_string()))
                            .collect::<Vec<_>>();
                        all.sort();
                        all
                    };
                    match metric.data() {
                        AggregatedMetrics::U64(MetricData::Sum(sum)) => {
                            for p in sum.data_points() {
                                out.push((labels(&mut p.attributes()), p.value()));
                            }
                        }
                        AggregatedMetrics::U64(MetricData::Gauge(gauge)) => {
                            for p in gauge.data_points() {
                                out.push((labels(&mut p.attributes()), p.value()));
                            }
                        }
                        _ => {}
                    }
                }
            }
        }
        out
    }

    fn pair(k: &str, v: &str) -> (String, String) {
        (k.to_owned(), v.to_owned())
    }

    #[test]
    fn an_event_counts_under_its_type_and_outcome() {
        let (metrics, provider, exporter) = recorded();
        metrics.event("issues", Handled::Routed, Duration::from_millis(5));
        metrics.event("issues", Handled::Routed, Duration::from_millis(7));
        metrics.event("issues", Handled::Retried, Duration::from_millis(1));
        provider.force_flush().unwrap();
        let mut got = points(&exporter, "hyperhive.notify_router.events");
        got.sort();
        assert_eq!(
            got,
            vec![
                (
                    vec![pair("event_type", "issues"), pair("outcome", "retried")],
                    1
                ),
                (
                    vec![pair("event_type", "issues"), pair("outcome", "routed")],
                    2
                ),
            ]
        );
    }

    #[test]
    fn todos_and_review_requests_count_by_outcome() {
        let (metrics, provider, exporter) = recorded();
        metrics.todo("upsert", true);
        metrics.todo("clear", false);
        metrics.review_request("requested");
        metrics.pull_failed();
        provider.force_flush().unwrap();
        let mut todos = points(&exporter, "hyperhive.notify_router.todos");
        todos.sort();
        assert_eq!(
            todos,
            vec![
                (vec![pair("op", "clear"), pair("outcome", "failed")], 1),
                (vec![pair("op", "upsert"), pair("outcome", "delivered")], 1),
            ]
        );
        assert_eq!(
            points(&exporter, "hyperhive.notify_router.review_requests"),
            vec![(vec![pair("outcome", "requested")], 1)]
        );
        assert_eq!(
            points(&exporter, "hyperhive.notify_router.pull.failures"),
            vec![(vec![], 1)]
        );
    }

    #[test]
    fn the_consumer_gauges_report_the_last_counts() {
        let (metrics, provider, exporter) = recorded();
        metrics.consumer(12, 3, 1);
        metrics.consumer(4, 0, 1);
        provider.force_flush().unwrap();
        for (name, want) in [("pending", 4), ("ack_pending", 0), ("redelivered", 1)] {
            assert_eq!(
                points(
                    &exporter,
                    &format!("hyperhive.notify_router.consumer.{name}")
                ),
                vec![(vec![], want)],
                "{name}"
            );
        }
    }
}