//! Many nodes under realistic conditions for a while. Ignored by default (a few
//! minutes in a debug build): `cargo test soak -- --ignored --nocapture`; the soak with
//! churn takes over ten minutes (`soak_with_churn`).

use std::time::Duration;

use serde_json::json;
use tokio::time::Instant;

use super::{SYNC_REPLY_TIMEOUT, Sim, write};
use crate::locks::MutexExt;
use crate::node::NodeConfig;
use crate::state::{AppState, Timing};
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.
pub(super) 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
}

/// The soak with churn: its members, their keepalive (a day of keepalive-scaled timers in
/// minutes), and each workload's length (one writer, then writers in turn), two writes
/// a second.
const CHURN_NODES: usize = 30;
const CHURN_KEEPALIVE: Duration = Duration::from_millis(500);
const CHURN_WORKLOAD: Duration = Duration::from_secs(300);
const CHURN_WRITE_EVERY: Duration = Duration::from_millis(500);
/// A member away (all to and from it lost) for `AWAY_FOR` every `AWAY_EVERY`, in turns
/// (`n10` to `n19`); one restarted (its process gone, a new one on the same ID, joining
/// again) every `RESTART_EVERY`, in turns (`n20` to `n24`).
const AWAY_EVERY: Duration = Duration::from_secs(20);
const AWAY_FOR: Duration = Duration::from_secs(15);
const RESTART_EVERY: Duration = Duration::from_secs(120);
/// How many neighbours each node has (`topology::our_neighbours`), `when`.
fn neighbour_counts(out: &mut String, nodes: &[AppState], when: &str) {
    use std::fmt::Write;
    let counts: Vec<usize> = nodes
        .iter()
        .map(|node| crate::swarm::topology::our_neighbours(node).len())
        .collect();
    let _ = writeln!(
        out,
        "    neighbours {when}: {:.1} on average, {} to {}",
        counts.iter().sum::<usize>() as f64 / counts.len().max(1) as f64,
        counts.iter().min().copied().unwrap_or(0),
        counts.iter().max().copied().unwrap_or(0),
    );
}

/// How long the quiet phase at the end is watched.
const CHURN_QUIET: Duration = Duration::from_secs(30);

/// What grows over a long run, and what churn sets off: 30 members on Mycelium's round
/// trips (`bench::mycelium_like`: each link its own, 2% lost, 1% twice), keepalive
/// 500 ms; one writer for 5 minutes, then writers in turn for 5; a member away for 15 s
/// every 20 s, a restart every 2 minutes, a member joining halfway, a remake near the
/// end. Asserted: every node on the remake's epoch, converged; peers and the tables
/// kept per peer no larger than the swarm; tasks alive within twice their number after
/// the warm-up. Reported: propagation, what counts as sent per session, heals, the
/// links recorded at the end (`swarm::working_links`), and what the quiet phase sends.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore]
async fn soak_with_churn() {
    let sim = Sim::new(31);
    let ids: Vec<String> = (0..CHURN_NODES).map(|i| format!("n{i:02}")).collect();
    super::bench::mycelium_like(&sim, &ids);
    let timing = |config: &mut NodeConfig| {
        config.timing = Timing::new(
            CHURN_KEEPALIVE,
            SYNC_REPLY_TIMEOUT,
            crate::transport::sim::DELIVERY_LIMIT,
        );
        // With `SIM_TOPOLOGY=hyparview`, HyParView's 5, as in production
        // (`config::derive_defaults`).
        if config.topology == crate::swarm::topology::Kind::Hyparview {
            config.neighbours_k = 5;
        }
    };
    let genesis = ids[0].as_str();
    let mut nodes = vec![sim.start_with(genesis, &[], Start::Genesis, timing)];
    for id in &ids[1..CHURN_NODES - 1] {
        nodes.push(sim.start_with(id, &[genesis], Start::Join, timing));
    }
    {
        let all: Vec<&AppState> = nodes.iter().collect();
        let all = all.as_slice();
        sim.wait_ready(all, Duration::from_secs(120)).await;
        let count = all.len();
        sim.wait_until(
            "every node knows every member",
            Duration::from_secs(120),
            move || async move {
                for node in all {
                    if members::members(&node.doc.read().await.doc).len() < count {
                        return false;
                    }
                }
                true
            },
        )
        .await;
        sim.wait_converged(all, Duration::from_secs(120)).await;
    }
    let warm_tasks = tokio::runtime::Handle::current()
        .metrics()
        .num_alive_tasks();

    let start = Instant::now();
    let total = CHURN_WORKLOAD * 2;
    let (mut next_away, mut next_restart) = (start + AWAY_EVERY, start + RESTART_EVERY);
    let (mut away_turn, mut restart_turn) = (0, 0);
    let mut away: Option<(String, Instant)> = None;
    let (mut joined, mut remade) = (false, false);
    let mut writes = 0usize;
    // PlumTree's links along the way (nothing for another dissemination): once the one
    // writer is done, and before the remake; with the PRUNEs and GRAFTs sent until then.
    let mut along = String::new();
    let plumtree_sent = |nodes: &[AppState]| -> u64 {
        nodes
            .iter()
            .map(|node| node.traffic.total().sent.plumtree.messages)
            .sum()
    };
    while start.elapsed() < total {
        let now = Instant::now();
        if let Some((id, until)) = &away
            && now >= *until
        {
            sim.net.bring_up(id);
            away = None;
        }
        if away.is_none() && now >= next_away {
            let id = ids[10 + away_turn % 10].clone();
            sim.net.take_down(&id);
            away = Some((id, now + AWAY_FOR));
            away_turn += 1;
            next_away += AWAY_EVERY;
        }
        if now >= next_restart {
            let id = ids[20 + restart_turn % 5].clone();
            let restarted = sim.restart_with(&id, &[genesis], Start::Join, timing);
            for node in &mut nodes {
                if node.alias.as_deref() == Some(id.as_str()) {
                    *node = restarted.clone();
                }
            }
            restart_turn += 1;
            next_restart += RESTART_EVERY;
        }
        if !joined && start.elapsed() >= CHURN_WORKLOAD {
            let mut links = String::new();
            super::bench::eager_links(&mut links, &nodes, &sim.net);
            if !links.is_empty() {
                along.push_str(&format!(
                    "    after the one writer ({} PRUNE and GRAFT sent):\n{links}",
                    plumtree_sent(&nodes)
                ));
            }
            nodes.push(sim.start_with(&ids[CHURN_NODES - 1], &[genesis], Start::Join, timing));
            joined = true;
        }
        if !remade && start.elapsed() >= total - CHURN_WORKLOAD / 5 {
            let mut links = String::new();
            super::bench::eager_links(&mut links, &nodes, &sim.net);
            if !links.is_empty() {
                along.push_str(&format!(
                    "    before the remake, writers in turn ({} PRUNE and GRAFT sent):\n{links}",
                    plumtree_sent(&nodes)
                ));
            }
            let (status, report) =
                crate::remake::testing::start(nodes[0].clone(), json!({ "kind": "compact" }), None)
                    .await;
            assert!(status.is_success(), "remake refused: {status} {report}");
            remade = true;
        }
        let writer = if start.elapsed() < CHURN_WORKLOAD {
            &nodes[1]
        } else {
            &nodes[writes % nodes.len()]
        };
        if writer.is_ready().await {
            write(writer, &format!("churn/{writes}"), json!(writes)).await;
        }
        writes += 1;
        tokio::time::sleep(CHURN_WRITE_EVERY).await;
    }
    if let Some((id, _)) = away {
        sim.net.bring_up(&id);
    }

    let all: Vec<&AppState> = nodes.iter().collect();
    let all = all.as_slice();
    sim.wait_until(
        "every node on the remake's epoch, ready",
        Duration::from_secs(180),
        move || async move {
            for node in all {
                if node.local_epoch().generation != 1 || !node.is_ready().await {
                    return false;
                }
            }
            true
        },
    )
    .await;
    sim.wait_converged(all, Duration::from_secs(120)).await;

    let mut out = String::new();
    let mut propagation = Propagation::default();
    let (mut peers, mut rtts, mut throughputs, mut traffic, mut heals) = (0, 0, 0, 0, 0);
    for node in all {
        let theirs = node.stats.snapshot().propagation;
        for (sum, count) in propagation.buckets.iter_mut().zip(theirs.buckets) {
            *sum += count;
        }
        propagation.count += theirs.count;
        propagation.sum_ms += theirs.sum_ms;
        let known = node.peers_snapshot().await;
        peers = peers.max(known.len());
        for (_, peer_state) in &known {
            heals += peer_state.lock().await.incidents.heals;
        }
        rtts = rtts.max(node.lock_rtt().len());
        throughputs = throughputs.max(node.throughput.lock_or_recover().len());
        traffic = traffic.max(node.traffic.len());
    }
    let records = |node: AppState| async move {
        let replica = node.doc.read().await;
        members::members(&replica.doc)
            .into_iter()
            .map(|member| {
                let links = members::links(&replica.doc, &member);
                (member, links)
            })
            .filter(|(_, links)| *links != members::RecordedLinks::default())
            .collect::<Vec<_>>()
    };
    let written = |nodes: &[AppState]| -> u64 {
        nodes.iter().map(crate::swarm::working_links::written).sum()
    };
    let recorded = records(nodes[0].clone()).await.len();
    let written_during = written(&nodes);
    let (unreached_written, reached_written) = nodes
        .iter()
        .map(crate::swarm::working_links::written_by_set)
        .fold((0, 0), |(u, r), (nu, nr)| (u + nu, r + nr));
    let end_tasks = tokio::runtime::Handle::current()
        .metrics()
        .num_alive_tasks();
    println!(
        "soak with churn ({}, {}): {writes} writes, {away_turn} times away, {restart_turn} restarts; \
         propagation: {} changes, mean {} ms, p99 within {}",
        nodes[0].topology().name(),
        nodes[0].dissemination().name(),
        propagation.count,
        propagation.sum_ms / propagation.count.max(1),
        percentile_bound(&propagation, 0.99)
            .map_or("more than an hour".to_owned(), |ms| format!("{ms} ms")),
    );
    super::bench::sent_kept(&mut out, &nodes).await;
    neighbour_counts(&mut out, &nodes, "at the end");
    out.push_str(&along);
    if !along.is_empty() {
        out.push_str(&format!(
            "    at the end ({} PRUNE and GRAFT sent):\n",
            plumtree_sent(&nodes)
        ));
    }
    // For PlumTree and HyParView: their links and views at the end (nothing otherwise).
    super::bench::eager_links(&mut out, &nodes, &sim.net);
    super::bench::active_views(&mut out, &nodes);
    print!("{out}");
    println!(
        "    per node at most: {peers} peers, round trips of {rtts}, throughput of \
         {throughputs}, traffic of {traffic}; {heals} heals; {recorded} members with links \
         recorded, {written_during} records written (by the nodes running at the end; \
         {unreached_written} changing unreached, {reached_written} reached); tasks alive {warm_tasks} after the warm-up, {end_tasks} at the end"
    );

    // The quiet phase: what the swarm sends with nothing to pass on, once the links
    // recorded while members were away are taken out again (every node checked since,
    // with nothing awaiting its hold: `swarm::working_links`).
    let checks: Vec<u64> = nodes
        .iter()
        .map(|node| crate::swarm::working_links::settled(node).0)
        .collect();
    let (nodes_ref, checks_ref) = (&nodes, &checks);
    sim.wait_until(
        "the links recorded",
        Duration::from_secs(60),
        move || async move {
            nodes_ref.iter().zip(checks_ref).all(|(node, before)| {
                let (checks, settled) = crate::swarm::working_links::settled(node);
                checks > *before && settled
            })
        },
    )
    .await;
    let settled_after = written(&nodes) - written_during;
    let written_during = written(&nodes);
    let sent = |nodes: &[AppState]| -> [u64; 4] {
        let mut sum = [0; 4];
        for node in nodes {
            let total = node.traffic.total().sent;
            for (sum, counter) in
                sum.iter_mut()
                    .zip([total.sync, total.status, total.status_reply, total.paths])
            {
                *sum += counter.messages;
            }
        }
        sum
    };
    let before = sent(&nodes);
    tokio::time::sleep(CHURN_QUIET).await;
    let after = sent(&nodes);
    let per_node_minute = |i: usize| {
        (after[i] - before[i]) as f64 * 60.0 / CHURN_QUIET.as_secs_f64() / nodes.len() as f64
    };
    println!(
        "    quiet, per node and minute: {:.1} SYNC, {:.1} STATUS, {:.1} STATUS_REPLY, \
         {:.1} PING/PONG; {} records written (and {settled_after} before it, once all were \
         back)",
        per_node_minute(0),
        per_node_minute(1),
        per_node_minute(2),
        per_node_minute(3),
        written(&nodes) - written_during,
    );
    for (member, links) in records(nodes[0].clone()).await {
        println!(
            "    recorded after the quiet phase by {member}: unreached {:?}, reached {:?}",
            links.unreached, links.reached
        );
    }
    // Again once the swarm had the quiet phase to settle in after the remake's switch:
    // whether what the end shows is the switch's, or stays.
    let mut out = String::new();
    neighbour_counts(&mut out, &nodes, "after the quiet phase");
    let mut links = String::new();
    super::bench::eager_links(&mut links, &nodes, &sim.net);
    super::bench::active_views(&mut links, &nodes);
    if !links.is_empty() {
        out.push_str("    after the quiet phase:\n");
        out.push_str(&links);
    }
    print!("{out}");

    for (what, most) in [
        ("peers", peers),
        ("round trips", rtts),
        ("throughputs", throughputs),
    ] {
        assert!(
            most <= CHURN_NODES,
            "{most} {what} kept by a node of {CHURN_NODES}"
        );
    }
    assert!(
        end_tasks <= warm_tasks * 2,
        "tasks alive grew from {warm_tasks} to {end_tasks}"
    );
}