//! Many nodes under realistic conditions for a while. Ignored by default (a few
//! minutes in a debug build): `cargo test soak -- --ignored --nocapture`.

use std::time::Duration;

use serde_json::json;

use super::{Sim, write};
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::swarm::members;
use crate::swarm::stats::Propagation;
use crate::transport::sim::LinkParams;

const NODES: usize = 25;
/// One per second, round robin over the nodes.
const WRITES: usize = 60;
/// 99% of changes reach a node within this, from the writer's commit. Measured: p99
/// within the 2.5 s bucket, mean about 0.7 s (October 2026); the bound is the next
/// bucket up.
const P99_BOUND_MS: u64 = 5_000;

/// 25 nodes (a partial mesh: 8 neighbours each) over links with 2% loss, 1%
/// duplication and 50–500 ms delay; one write per second for a minute. All converge,
/// and 99% of changes reach each node within `P99_BOUND_MS`.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore]
async fn soak_25_nodes() {
    let sim = Sim::new(25);
    sim.net.set_default_link(LinkParams {
        loss: 0.02,
        duplicate: 0.01,
        delay: (Duration::from_millis(50), Duration::from_millis(500)),
        ..LinkParams::default()
    });
    let ids: Vec<String> = (0..NODES).map(|i| format!("n{i:02}")).collect();
    let mut nodes = vec![sim.start(&ids[0], &[], Start::Genesis)];
    for id in &ids[1..] {
        nodes.push(sim.start(id, &[ids[0].as_str()], Start::Join));
    }
    let all: Vec<&AppState> = nodes.iter().collect();
    let all = all.as_slice();
    sim.wait_ready(all, Duration::from_secs(120)).await;
    sim.wait_until(
        "every node knows every member",
        Duration::from_secs(60),
        move || async move {
            for node in all {
                if members::members(&node.doc.read().await.doc).len() < NODES {
                    return false;
                }
            }
            true
        },
    )
    .await;

    for i in 0..WRITES {
        write(&nodes[i % NODES], &format!("soak/{i}"), json!(i)).await;
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
    sim.wait_converged(all, Duration::from_secs(60)).await;

    let mut total = Propagation::default();
    for node in all {
        let propagation = node.stats.snapshot().propagation;
        for (sum, count) in total.buckets.iter_mut().zip(propagation.buckets) {
            *sum += count;
        }
        total.count += propagation.count;
        total.sum_ms += propagation.sum_ms;
    }
    let p99 = percentile_bound(&total, 0.99);
    println!(
        "propagation: {} changes, mean {} ms, p99 within {}, buckets {:?}; net {:?}",
        total.count,
        total.sum_ms / total.count.max(1),
        p99.map_or("more than an hour".to_owned(), |ms| format!("{ms} ms")),
        total.buckets().collect::<Vec<_>>(),
        sim.net.counts(),
    );
    assert!(total.count > 0);
    assert!(
        p99.is_some_and(|ms| ms <= P99_BOUND_MS),
        "p99 propagation above {P99_BOUND_MS} ms"
    );
}

/// The upper bound of the bucket in which the `quantile` of changes is reached; None if
/// only in the last (unbounded) one.
fn percentile_bound(propagation: &Propagation, quantile: f64) -> Option<u64> {
    let wanted = (propagation.count as f64 * quantile).ceil() as u64;
    let mut seen = 0;
    for (bound, count) in propagation.buckets() {
        seen += count;
        if seen >= wanted {
            return bound;
        }
    }
    None
}