//! Measurement: a dependency-free latency histogram, backpressure-aware status
//! classification, and per-stage aggregation/reporting.
//!
//! WHY THESE BUCKETS: the server distinguishes 429 (write-pool / WAL
//! backpressure — slow the write rate), 408 (request-timeout from the 100-deep
//! concurrency queue — lower concurrency), and 503 (leader/region down or
//! draining — re-target). A load report that lumps them as "errors" is useless
//! for capacity work, so [`StatusClass`] keeps them apart and the verdict reads
//! the mix to name the ceiling we hit.

use std::time::Duration;

use crate::workload::OpKind;

// ── Latency histogram ────────────────────────────────────────────────────────
//
// Log-spaced buckets with linear interpolation inside the crossing bucket. Min /
// max / mean are tracked exactly; only the percentiles are bucket-estimated, to
// ~3-4% (≈30 buckets per decade). HdrHistogram would be marginally tighter but a
// new lock entry; this is accurate enough to find a capacity knee and matches the
// repo's dependency-light posture (the engine's own latency metric uses fixed
// 1µs–10ms buckets, docs/ops/monitoring.md).

const MIN_NS: f64 = 1_000.0; // 1µs — finer than that is noise over a network hop
const GROWTH: f64 = 1.0772; // ~30 buckets/decade
const BUCKETS: usize = 300; // 1µs * 1.0772^300 ≈ 9.4e12ns ≈ 2.6h — far past any real tail

/// A single operation's latency distribution. Exact count/min/max/sum, bucketed
/// percentiles.
#[derive(Clone)]
pub struct LatencyHistogram {
    counts: Vec<u64>,
    total: u64,
    min_ns: u64,
    max_ns: u64,
}

impl Default for LatencyHistogram {
    fn default() -> Self {
        Self {
            counts: vec![0; BUCKETS],
            total: 0,
            min_ns: u64::MAX,
            max_ns: 0,
        }
    }
}

impl LatencyHistogram {
    fn index(value_ns: u64) -> usize {
        let v = value_ns as f64;
        if v <= MIN_NS {
            return 0;
        }
        let idx = (v / MIN_NS).log(GROWTH).floor() as usize;
        idx.min(BUCKETS - 1)
    }

    /// Lower bound (ns) of bucket `i`.
    fn bound(i: usize) -> f64 {
        MIN_NS * GROWTH.powi(i as i32)
    }

    pub fn record(&mut self, latency: Duration) {
        let ns = u64::try_from(latency.as_nanos()).unwrap_or(u64::MAX);
        self.counts[Self::index(ns)] += 1;
        self.total += 1;
        self.min_ns = self.min_ns.min(ns);
        self.max_ns = self.max_ns.max(ns);
    }

    /// Estimated p-th percentile (`p` in 0.0..=1.0), linearly interpolated within
    /// the crossing bucket. Returns 0 for an empty histogram.
    pub fn percentile(&self, p: f64) -> Duration {
        if self.total == 0 {
            return Duration::ZERO;
        }
        let target = (p * self.total as f64).ceil().max(1.0) as u64;
        let mut cumulative = 0u64;
        for (i, &c) in self.counts.iter().enumerate() {
            if c == 0 {
                continue;
            }
            if cumulative + c >= target {
                // Interpolate within [bound(i), bound(i+1)) by how far into this
                // bucket's mass `target` falls. Clamp to the exact min/max so a
                // p100 never reads below the bucket floor or above the real max.
                let into = (target - cumulative) as f64 / c as f64;
                let lo = Self::bound(i);
                let hi = Self::bound(i + 1);
                let est = (hi - lo).mul_add(into, lo);
                let clamped = est.clamp(self.min_ns as f64, self.max_ns as f64);
                return Duration::from_nanos(clamped as u64);
            }
            cumulative += c;
        }
        Duration::from_nanos(self.max_ns)
    }

    /// Exact worst-case latency (not bucket-estimated) — the tail a load report
    /// must surface alongside p999.
    pub const fn max(&self) -> Duration {
        Duration::from_nanos(if self.total == 0 { 0 } else { self.max_ns })
    }
}

// ── Status classification ────────────────────────────────────────────────────

/// What a single request's result *means* for capacity, not just its code.
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum StatusClass {
    /// 2xx — the operation succeeded.
    Ok,
    /// 429 — write-pool / WAL backpressure (honor retry, slow the write rate).
    Backpressure,
    /// 408 — request-timeout: queued past 30s behind the 100-concurrency limit.
    Timeout,
    /// 503 — leader/region unreachable or draining (re-target / wait).
    Unavailable,
    /// Other 4xx (400/401/413/404…) — a client/usage bug, not load.
    ClientError,
    /// 5xx other than 503 — a server fault under load.
    ServerError,
    /// No HTTP response at all: connect refused, TLS, client-side timeout, reset.
    Transport,
}

impl StatusClass {
    pub const fn from_status(code: u16) -> Self {
        match code {
            200..=299 => Self::Ok,
            429 => Self::Backpressure,
            408 => Self::Timeout,
            503 => Self::Unavailable,
            400..=499 => Self::ClientError,
            _ => Self::ServerError,
        }
    }

    const fn idx(self) -> usize {
        match self {
            Self::Ok => 0,
            Self::Backpressure => 1,
            Self::Timeout => 2,
            Self::Unavailable => 3,
            Self::ClientError => 4,
            Self::ServerError => 5,
            Self::Transport => 6,
        }
    }
}

const STATUS_KINDS: usize = 7;

/// The result of one issued request, sent from a worker to the collector.
pub struct Outcome {
    pub op: OpKind,
    pub class: StatusClass,
    /// Wall-clock from the request's *intended* send time to its completion —
    /// the coordinated-omission-corrected latency (includes any client backlog).
    pub latency: Duration,
}

// ── Per-op + per-stage aggregation ───────────────────────────────────────────

#[derive(Clone, Default)]
pub struct OpStats {
    pub hist: LatencyHistogram,
    pub status: [u64; STATUS_KINDS],
}

impl OpStats {
    fn record(&mut self, class: StatusClass, latency: Duration) {
        self.status[class.idx()] += 1;
        // Only successful requests' latencies describe service time; a 429/503
        // rejected fast would otherwise flatter the percentiles. Errors are
        // counted in `status`, excluded from the latency picture.
        if class == StatusClass::Ok {
            self.hist.record(latency);
        }
    }

    pub const fn total(&self) -> u64 {
        let mut sum = 0u64;
        let mut i = 0;
        while i < STATUS_KINDS {
            sum += self.status[i];
            i += 1;
        }
        sum
    }

    pub const fn ok(&self) -> u64 {
        self.status[0]
    }
    pub const fn backpressure(&self) -> u64 {
        self.status[1]
    }
    pub const fn timeout(&self) -> u64 {
        self.status[2]
    }
    pub const fn unavailable(&self) -> u64 {
        self.status[3]
    }
    pub const fn client_error(&self) -> u64 {
        self.status[4]
    }
    pub const fn server_error(&self) -> u64 {
        self.status[5]
    }
    pub const fn transport(&self) -> u64 {
        self.status[6]
    }
    pub const fn errors(&self) -> u64 {
        self.total() - self.ok()
    }
}

/// Everything one ramp stage produced. Per-op stats plus a roll-up.
pub struct StageStats {
    pub ops: Vec<OpStats>, // indexed by OpKind::idx()
    pub elapsed: Duration,
    /// Requests the client could not even dispatch (in-flight cap hit) — a signal
    /// the LOAD GENERATOR is saturated, not the server. Non-zero ⇒ the reported
    /// achieved RPS is a client-limited floor, not the server's ceiling.
    pub client_shed: u64,
    /// Mean delay between a request's intended send time and its actual dispatch.
    /// Growing scheduling delay ⇒ the generator is falling behind the target rate.
    pub mean_schedule_lag: Duration,
}

impl StageStats {
    pub fn new() -> Self {
        Self {
            ops: vec![OpStats::default(); OpKind::COUNT],
            elapsed: Duration::ZERO,
            client_shed: 0,
            mean_schedule_lag: Duration::ZERO,
        }
    }

    pub fn record(&mut self, outcome: &Outcome) {
        self.ops[outcome.op.idx()].record(outcome.class, outcome.latency);
    }

    pub fn total(&self) -> u64 {
        self.ops.iter().map(OpStats::total).sum()
    }
    pub fn total_ok(&self) -> u64 {
        self.ops.iter().map(OpStats::ok).sum()
    }
    pub fn total_errors(&self) -> u64 {
        self.ops.iter().map(OpStats::errors).sum()
    }
    pub fn error_rate(&self) -> f64 {
        let t = self.total();
        if t == 0 {
            0.0
        } else {
            self.total_errors() as f64 / t as f64
        }
    }
    /// Achieved throughput: completed requests per second over the stage.
    pub fn achieved_rps(&self) -> f64 {
        let s = self.elapsed.as_secs_f64();
        if s <= 0.0 {
            0.0
        } else {
            self.total() as f64 / s
        }
    }

    /// Sum of a status class across every op (for the verdict).
    pub fn class_total(&self, pick: fn(&OpStats) -> u64) -> u64 {
        self.ops.iter().map(pick).sum()
    }
}

fn fmt_dur(d: Duration) -> String {
    let us = d.as_nanos() as f64 / 1000.0;
    if us < 1000.0 {
        format!("{us:.0}µs")
    } else if us < 1_000_000.0 {
        format!("{:.2}ms", us / 1000.0)
    } else {
        format!("{:.2}s", us / 1_000_000.0)
    }
}

/// Render a per-stage report block as a human-readable table.
pub fn render_stage(label: &str, target_rps: f64, stats: &StageStats) -> String {
    let mut out = String::new();
    out.push_str(&format!(
        "\n── stage {label}  (target {target_rps:.0} rps, achieved {:.0} rps, {:.1}s) ──\n",
        stats.achieved_rps(),
        stats.elapsed.as_secs_f64(),
    ));
    out.push_str(&format!(
        "{:<10} {:>8} {:>8} {:>9} {:>9} {:>9} {:>9} {:>9}  {:>5} {:>4} {:>4} {:>4} {:>5} {:>4}\n",
        "op",
        "count",
        "ok/s",
        "p50",
        "p90",
        "p99",
        "p999",
        "max",
        "429",
        "408",
        "503",
        "4xx",
        "5xx",
        "tx",
    ));
    for kind in OpKind::ALL {
        let s = &stats.ops[kind.idx()];
        if s.total() == 0 {
            continue;
        }
        let ok_rps = s.ok() as f64 / stats.elapsed.as_secs_f64().max(1e-9);
        out.push_str(&format!(
            "{:<10} {:>8} {:>8.0} {:>9} {:>9} {:>9} {:>9} {:>9}  {:>5} {:>4} {:>4} {:>4} {:>5} {:>4}\n",
            kind.label(),
            s.total(),
            ok_rps,
            fmt_dur(s.hist.percentile(0.50)),
            fmt_dur(s.hist.percentile(0.90)),
            fmt_dur(s.hist.percentile(0.99)),
            fmt_dur(s.hist.percentile(0.999)),
            fmt_dur(s.hist.max()),
            s.backpressure(),
            s.timeout(),
            s.unavailable(),
            s.client_error(),
            s.server_error(),
            s.transport(),
        ));
    }
    out.push_str(&format!(
        "{:<10} {:>8} {:>8.0} {:>52}  {:>5} {:>4} {:>4} {:>4} {:>5} {:>4}\n",
        "ALL",
        stats.total(),
        stats.total_ok() as f64 / stats.elapsed.as_secs_f64().max(1e-9),
        "",
        stats.class_total(OpStats::backpressure),
        stats.class_total(OpStats::timeout),
        stats.class_total(OpStats::unavailable),
        stats.class_total(OpStats::client_error),
        stats.class_total(OpStats::server_error),
        stats.class_total(OpStats::transport),
    ));
    out.push_str(&format!(
        "error rate {:.2}%  |  client-shed {}  |  schedule-lag {}\n",
        stats.error_rate() * 100.0,
        stats.client_shed,
        fmt_dur(stats.mean_schedule_lag),
    ));
    out
}

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

    #[test]
    fn histogram_percentiles_are_ordered_and_bounded() {
        let mut h = LatencyHistogram::default();
        for ns in 1..=10_000u64 {
            h.record(Duration::from_nanos(ns * 1000)); // 1µs..10ms
        }
        let p50 = h.percentile(0.50);
        let p99 = h.percentile(0.99);
        assert!(p50 < p99, "p50 {p50:?} should be < p99 {p99:?}");
        assert!(p99 <= h.max(), "p99 {p99:?} above exact max {:?}", h.max());
        assert!(
            h.max() >= Duration::from_millis(9),
            "max should be ~10ms, got {:?}",
            h.max()
        );
        // 50th percentile of a uniform 1µs..10ms ≈ 5ms; allow the bucket error.
        let p50_ms = p50.as_secs_f64() * 1000.0;
        assert!((p50_ms - 5.0).abs() < 0.5, "p50 {p50_ms}ms not ~5ms");
    }

    #[test]
    fn status_class_maps_backpressure_codes() {
        assert!(matches!(
            StatusClass::from_status(429),
            StatusClass::Backpressure
        ));
        assert!(matches!(
            StatusClass::from_status(408),
            StatusClass::Timeout
        ));
        assert!(matches!(
            StatusClass::from_status(503),
            StatusClass::Unavailable
        ));
        assert!(matches!(StatusClass::from_status(201), StatusClass::Ok));
        assert!(matches!(
            StatusClass::from_status(400),
            StatusClass::ClientError
        ));
        assert!(matches!(
            StatusClass::from_status(500),
            StatusClass::ServerError
        ));
    }
}