//! How long it takes to get onto the blocking pool, sampled through the run.
//!
//! The description layer's whole runtime cost is that quasi's router is sync, so
//! quasi-axum dispatches every described request on `spawn_blocking` and the
//! handler holds that thread across its database or upstream call. This server
//! shares that pool with argon2 hashing, content exports and the file scanner,
//! so the question a load run has to answer is whether described requests starve
//! them.
//!
//! Throughput does not answer it and neither does per-endpoint latency: a pool
//! at its limit shows up as everything getting slower together, which is
//! indistinguishable from the box being busy. What separates them is DISPATCH
//! DELAY — how long a fresh `spawn_blocking` waits before its closure starts
//! running. On an unsaturated pool that is microseconds whatever the load;
//! it only grows when every thread is occupied and tokio has to wait for one or
//! spawn another.
//!
//! `RuntimeMetrics::num_blocking_threads` would say it directly, but it is
//! behind `tokio_unstable` and this tree does not build with it. This measures
//! the same saturation from the outside, with no cfg and no build flag: the
//! probe is what a request arriving at that instant would have experienced.
//!
//! Read it alongside the argon2 signup mean (`POST /join/step/account`), which
//! is the same fact seen from the other end: signup is the heaviest genuine
//! blocking-pool user in the mix, so if described requests are crowding the pool
//! both numbers move together.

use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// Samples dispatch delay until it is stopped.
pub(super) struct BlockingProbe {
    samples: Arc<Mutex<Vec<Duration>>>,
    sampler: tokio::task::JoinHandle<()>,
}

impl BlockingProbe {
    /// Start sampling every `interval`.
    ///
    /// One in-flight sample at a time, on purpose: the probe measures the pool,
    /// and a probe that queued its own work would be measuring itself.
    pub(super) fn start(interval: Duration) -> Self {
        let samples = Arc::new(Mutex::new(Vec::new()));
        let into = Arc::clone(&samples);

        let sampler = tokio::spawn(async move {
            loop {
                let queued = Instant::now();
                // The closure records how long it waited to BEGIN, not how long
                // it ran. It does nothing else, so the two are not confusable.
                let waited = tokio::task::spawn_blocking(move || queued.elapsed()).await;
                if let Ok(waited) = waited {
                    into.lock().unwrap().push(waited);
                }
                tokio::time::sleep(interval).await;
            }
        });

        BlockingProbe { samples, sampler }
    }

    /// Stop sampling and summarise.
    pub(super) fn finish(self) -> BlockingReport {
        self.sampler.abort();
        let mut samples = std::mem::take(&mut *self.samples.lock().unwrap());
        samples.sort();
        BlockingReport::of(&samples)
    }
}

/// What the probe saw over one run.
pub(super) struct BlockingReport {
    pub count: usize,
    pub p50: Duration,
    pub p95: Duration,
    pub p99: Duration,
    pub max: Duration,
    /// Samples that waited longer than [`STALL`]. The number that matters: on a
    /// pool with headroom it is zero, and it stops being zero before mean
    /// latency moves at all.
    pub stalls: usize,
}

/// The line above which a dispatch delay is a queue rather than scheduling
/// noise. Dispatch onto an idle pool is single-digit microseconds; a millisecond
/// means the sample waited for a thread.
pub(super) const STALL: Duration = Duration::from_millis(1);

impl BlockingReport {
    /// Print the probe's summary under the endpoint table.
    pub(super) fn print(&self) {
        println!("  Blocking-pool dispatch delay ({} samples):", self.count);
        if self.count == 0 {
            println!("    no samples");
            println!();
            return;
        }
        println!(
            "    p50 {}  p95 {}  p99 {}  max {}",
            format_dur(self.p50),
            format_dur(self.p95),
            format_dur(self.p99),
            format_dur(self.max),
        );
        println!(
            "    stalls over {}: {} ({:.1}%)",
            format_dur(STALL),
            self.stalls,
            self.stalls as f64 / self.count as f64 * 100.0,
        );
        if self.stalls == 0 {
            println!("    the pool had headroom throughout");
        }
        println!();
    }

    fn of(sorted: &[Duration]) -> Self {
        if sorted.is_empty() {
            return BlockingReport {
                count: 0,
                p50: Duration::ZERO,
                p95: Duration::ZERO,
                p99: Duration::ZERO,
                max: Duration::ZERO,
                stalls: 0,
            };
        }
        let at = |pct: f64| {
            let idx = ((sorted.len() as f64 * pct) as usize).min(sorted.len() - 1);
            sorted[idx]
        };
        BlockingReport {
            count: sorted.len(),
            p50: at(0.50),
            p95: at(0.95),
            p99: at(0.99),
            max: sorted[sorted.len() - 1],
            stalls: sorted.iter().filter(|d| **d > STALL).count(),
        }
    }
}

/// Same rendering the endpoint table uses, so the two read against each other.
fn format_dur(d: Duration) -> String {
    let us = d.as_micros();
    if us >= 1000 {
        format!("{:.1}ms", us as f64 / 1000.0)
    } else {
        format!("{us}us")
    }
}