//! What a resolve costs in wall time, and what figure it is charged, tier by
//! tier, on this host's in-process counter.
//!
//! ```text
//! cargo run --release -p eoc-cascade --example hit_cost
//! ```
//!
//! It prints; it asserts nothing. The figures follow the core the work ran on
//! and the load on the host.

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

use eoc_cache::{LruCache, Stage as _};
use eoc_cascade::Cascade;
use eoc_core::{JouleCost, JouleSource, Query, Response, Stage};
use eoc_graph::{GraphStage, Triple};
use eoc_kv::{KvBackend, KvStage, MemoryKvBackend};
use eoc_meter::detect_in_process;
use eoc_neural::{NeuralBackend, NeuralStage};

/// A neural stage that does nothing and claims no figure, so a resolve it
/// answers costs what the cascade itself adds around it.
struct Nothing;

#[async_trait::async_trait]
impl NeuralBackend for Nothing {
    async fn infer(&self, q: &Query) -> Response {
        Response::new(q.id, String::new(), Stage::Neural, JouleCost::zero())
    }
}

/// The stages the `eoc` CLI wires, without memoization so a KV or graph hit
/// stays one; metered by this process's own counter where it is live, or
/// given no meter at all.
fn cascade(metered: bool) -> Cascade {
    let kv = Box::new(MemoryKvBackend::new());
    kv.put("ping", b"pong".to_vec());
    let graph = GraphStage::new();
    graph.extend([
        Triple::new("Paris", "capital of", "France"),
        Triple::new("Tokyo", "capital of", "Japan"),
        Triple::new("Mars", "fourth planet from", "the Sun"),
    ]);
    let cascade = Cascade::new(
        Arc::new(LruCache::new(1024)),
        Arc::new(KvStage::new(kv)),
        Arc::new(graph),
        Arc::new(NeuralStage::new(Box::new(Nothing))),
    )
    .without_memoization();
    if metered { cascade.with_meter(Arc::from(detect_in_process())) } else { cascade }
}

/// Resolve `q` `n` times and print the wall time of a resolve and the cost
/// each was charged.
async fn timed(cascade: &Cascade, what: &str, q: &Query, n: usize) {
    let mut micros = Vec::with_capacity(n);
    let (mut measured, mut sum_uj, mut stage) = (0usize, 0u64, None);
    let (mut first, mut last) = (None, None);
    let all = Instant::now();
    for _ in 0..n {
        let started = Instant::now();
        let r = cascade.resolve(q.clone()).await;
        micros.push(started.elapsed().as_nanos() as f64 / 1_000.0);
        measured += usize::from(r.joule_cost.source == JouleSource::Measured);
        sum_uj += r.joule_cost.microjoules;
        first.get_or_insert(r.joule_cost);
        last = Some(r.joule_cost);
        stage = Some(r.stage);
    }
    let all = all.elapsed();
    micros.sort_by(f64::total_cmp);
    println!(
        "{what}: {n} resolves answered by {} in {:.3} ms ({:.3} µs each at the median, {:.3} µs at the 95th \
         percentile, {:.3} µs at most); charged {sum_uj} µJ in all, {measured} of {n} tagged measured; first {}, \
         last {}",
        stage.map(|s| s.to_string()).unwrap_or_default(),
        all.as_secs_f64() * 1_000.0,
        micros[n / 2],
        micros[n * 95 / 100],
        micros[n - 1],
        first.map(|c| c.to_string()).unwrap_or_default(),
        last.map(|c| c.to_string()).unwrap_or_default(),
    );
}

/// The first hit on a tier, which calibrates it, then `n` more; then the
/// tier calibrated five times over, to show how the figure moves.
async fn tier(cascade: &Cascade, q: &Query, n: usize) {
    let started = Instant::now();
    let first = cascade.resolve(q.clone()).await;
    let took = started.elapsed();
    let stage = first.stage;
    println!("first {stage} hit: {:.3} ms, {}; {}", took.as_secs_f64() * 1_000.0, first.joule_cost, held(cascade, stage));
    timed(cascade, &stage.to_string(), q, n).await;
    for _ in 0..5 {
        cascade.calibrate(q).await;
        println!("  {stage} calibrated again: {}", held(cascade, stage));
    }
}

/// The calibration the cascade holds for `stage`, in words.
fn held(cascade: &Cascade, stage: Stage) -> String {
    match cascade.calibration(stage) {
        Ok(c) => format!(
            "{} nJ a hit: {} lookups in one span of {} µJ, {} times the {} µJ an empty span read, in {:.3} ms",
            c.per_hit_nanojoules,
            c.batch,
            c.span_microjoules,
            c.span_microjoules / c.empty_span_microjoules.max(1),
            c.empty_span_microjoules,
            c.took.as_secs_f64() * 1_000.0
        ),
        Err(why) => format!("not calibrated ({why})"),
    }
}

/// What the three deterministic misses before a neural resolve spend: one
/// span of the meter around `n` walks that every tier misses.
async fn misses(cascade: &Cascade, n: u64) {
    let q = Query::new("nothing knows this");
    let started = Instant::now();
    let open = cascade.meter().read_microjoules();
    for _ in 0..n {
        let missed = cascade.cache().try_resolve(&q).await.is_none()
            && cascade.kv().try_resolve(&q).await.is_none()
            && cascade.graph().try_resolve(&q).await.is_none();
        assert!(std::hint::black_box(missed));
    }
    let close = cascade.meter().read_microjoules();
    let took = started.elapsed();
    match (open, close) {
        (Ok(open), Ok(close)) if close > open => println!(
            "the three misses: {} nJ and {:.0} ns a walk ({n} walks in one span of {} µJ)",
            (close - open) * 1_000 / n,
            took.as_nanos() as f64 / n as f64,
            close - open
        ),
        _ => println!("the three misses: not read ({:.0} ns a walk)", took.as_nanos() as f64 / n as f64),
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let c = cascade(true);
    println!("meter: {}", c.meter().name());

    let cached = Query::new("seen before");
    c.cache().insert(&cached, "answer");
    tier(&c, &cached, 10_000).await;
    tier(&c, &Query::new("ping"), 10_000).await;
    tier(&c, &Query::new("Mars is the fourth planet from what?"), 10_000).await;

    // A neural stage that does nothing: what the cascade adds around it,
    // its two reads of the meter.
    timed(&c, "neural (a stage that does nothing)", &Query::new("nothing knows this"), 2_000).await;
    misses(&c, 100_000).await;

    // The same with no meter at all: what a resolve takes when nothing is
    // read or calibrated.
    let c = cascade(false);
    c.cache().insert(&cached, "answer");
    timed(&c, "no meter: cache", &cached, 10_000).await;
    timed(&c, "no meter: neural (a stage that does nothing)", &Query::new("nothing knows this"), 2_000).await;
    std::thread::sleep(Duration::from_millis(1));
}