//! What passing changes on costs (`swarm::relay`): whole nodes on the simulated network,
//! a series of writes, and the messages, bytes and delays it took, per scenario. A
//! benchmark, not a test: `cargo benchmarks` (release, with the other `bench_*`) prints
//! a table per scenario to compare before and after a change; nothing is asserted but
//! that the nodes converge.
//!
//! Each scenario starts its swarm and waits until it has settled (every member known,
//! the same heads, then sync messages no fewer in one half of the quiet phase than in
//! the one before), those two halves being the quiet phase (no writes), then each
//! workload: 100 writes at 5 a second, and until every node has
//! them. What a workload sent is shown as it is, beside what the quiet phase sent in as
//! long: not taken off, since writes make keepalives unneeded (a pair that exchanges
//! anything needs none), so the quiet phase's count isn't part of a workload's.

use std::fmt::Write;
use std::time::Duration;

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

use super::{SYNC_REPLY_TIMEOUT, Sim, write};
use crate::state::{AppState, Timing};
use crate::swarm::bootstrap::Start;
use crate::swarm::members;
use crate::swarm::stats::{Propagation, Relay};
use crate::transport::sim::{DELIVERY_LIMIT, LinkParams};
use crate::transport::traffic::{ByKind, Counter};

/// Writes per workload, and how many a second.
const WRITES: usize = 100;
const WRITE_EVERY: Duration = Duration::from_millis(200);
/// Longer than the tests' (`simtest::KEEPALIVE`), so that keepalives don't drown what
/// the writes cost; as long as the relay tests'.
const KEEPALIVE: Duration = Duration::from_secs(5);
/// The quiet phase: long enough to see the keepalives' rate (a STATUS every third, 15 s).
const QUIET: Duration = Duration::from_secs(20);
/// Only the scenarios whose names contain this, if set: `BENCH_SCENARIO=NAT`.
const ONLY: &str = "BENCH_SCENARIO";
/// If set, all scenarios at once (`BENCH_PARALLEL=1`): several times faster on many
/// cores, but they share them, so delays (and what depends on timing) may come out
/// higher than alone; message counts barely.
const PARALLEL: &str = "BENCH_PARALLEL";
const SETTLE: Duration = Duration::from_secs(180);

/// How the swarm is made.
struct Scenario {
    name: &'static str,
    nodes: usize,
    /// The transports of node `i` (`sim`, `sim2`).
    on: fn(usize) -> &'static [&'static str],
    /// Before the nodes start: links cut, delays.
    net: fn(&Sim, &[String]),
    /// The node that writes in the one-writer workload.
    writer: usize,
}

fn sim_only(_: usize) -> &'static [&'static str] {
    &["sim"]
}

fn clean(_: &Sim, _: &[String]) {}

/// Node 1 reaches node 2 alone, as one behind NAT with a single link.
fn writer_behind_nat(sim: &Sim, ids: &[String]) {
    let cut = LinkParams {
        loss: 1.0,
        ..LinkParams::default()
    };
    for (i, id) in ids.iter().enumerate() {
        if i != 1 && i != 2 {
            sim.net.set_link(&ids[1], id, cut.clone());
            sim.net.set_link(id, &ids[1], cut.clone());
        }
    }
}

/// Nodes 0 to 2 on both nets (bridges), the first half of the rest on `sim`, the
/// other half on `sim2`.
fn two_transports(i: usize) -> &'static [&'static str] {
    match i {
        0..3 => &["sim", "sim2"],
        3..16 => &["sim"],
        _ => &["sim2"],
    }
}

/// As the soak test's: 2% lost, 1% twice, 50 to 500 ms.
fn lossy(sim: &Sim, _: &[String]) {
    sim.net.set_default_link(LinkParams {
        loss: 0.02,
        duplicate: 0.01,
        delay: (Duration::from_millis(50), Duration::from_millis(500)),
        ..LinkParams::default()
    });
}

const SCENARIOS: &[Scenario] = &[
    Scenario {
        name: "full mesh, 13 nodes",
        nodes: 13,
        on: sim_only,
        net: clean,
        writer: 1,
    },
    Scenario {
        name: "writer behind NAT (one link), 13 nodes",
        nodes: 13,
        on: sim_only,
        net: writer_behind_nat,
        writer: 1,
    },
    Scenario {
        name: "two transports, 3 bridges, 30 nodes",
        nodes: 30,
        on: two_transports,
        net: clean,
        writer: 5,
    },
    Scenario {
        name: "partial mesh, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: clean,
        writer: 1,
    },
    Scenario {
        name: "partial mesh, lossy and slow links, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: lossy,
        writer: 1,
    },
    Scenario {
        name: "partial mesh, 100 nodes",
        nodes: 100,
        on: sim_only,
        net: clean,
        writer: 1,
    },
];

#[tokio::test(flavor = "multi_thread", worker_threads = 16)]
#[ignore = "a benchmark: cargo benchmarks"]
async fn bench_propagation() {
    // Past the test harness's `test … ...`, so each report is on a line of its own.
    println!();
    let only = std::env::var(ONLY).unwrap_or_default();
    let chosen: Vec<(u64, &'static Scenario)> = (70..)
        .zip(SCENARIOS)
        .filter(|(_, scenario)| scenario.name.contains(&only))
        .collect();
    if std::env::var_os(PARALLEL).is_some() {
        // All at once, each reported when all are done, in order.
        let running: Vec<_> = chosen
            .into_iter()
            .map(|(seed, scenario)| tokio::spawn(run(seed, scenario)))
            .collect();
        for report in running {
            print!("{}", report.await.expect("scenario panicked"));
        }
    } else {
        for (seed, scenario) in chosen {
            print!("{}", run(seed, scenario).await);
        }
    }
}

/// What all nodes have sent, counted, and their relay counters and propagation delays
/// summed: at one moment.
#[derive(Clone, Copy)]
struct Totals {
    at: Instant,
    sent: ByKind,
    relay: Relay,
    propagation: Propagation,
}

impl Totals {
    fn of(nodes: &[AppState]) -> Self {
        let mut totals = Self {
            at: Instant::now(),
            sent: ByKind::default(),
            relay: Relay::default(),
            propagation: Propagation::default(),
        };
        for node in nodes {
            let sent = node.traffic.total().sent;
            for (total, theirs) in counters_mut(&mut totals.sent)
                .into_iter()
                .zip(counters(&sent))
            {
                total.messages += theirs.messages;
                total.bytes += theirs.bytes;
            }
            let stats = node.stats.snapshot();
            let relay = &mut totals.relay;
            relay.announce_rounds += stats.relay.announce_rounds;
            relay.treeless_rounds += stats.relay.treeless_rounds;
            relay.slow_rounds += stats.relay.slow_rounds;
            relay.pulls += stats.relay.pulls;
            relay.pulls_queued += stats.relay.pulls_queued;
            relay.tree_forwards += stats.relay.tree_forwards;
            let propagation = &mut totals.propagation;
            for (sum, count) in propagation
                .buckets
                .iter_mut()
                .zip(stats.propagation.buckets)
            {
                *sum += count;
            }
            propagation.count += stats.propagation.count;
            propagation.sum_ms += stats.propagation.sum_ms;
        }
        totals
    }
}

/// The kinds, in the order of the report.
const KINDS: [&str; 9] = [
    "SYNC",
    "STATUS",
    "STATUS_REPLY",
    "BLOB",
    "HELLO",
    "HELLO_REPLY",
    "PING/PONG",
    "WHO/ME",
    "LINKS",
];

fn counters(by: &ByKind) -> [Counter; 9] {
    [
        by.sync,
        by.status,
        by.status_reply,
        by.blob,
        by.hello,
        by.hello_reply,
        by.paths,
        by.identity,
        by.links,
    ]
}

fn counters_mut(by: &mut ByKind) -> [&mut Counter; 9] {
    [
        &mut by.sync,
        &mut by.status,
        &mut by.status_reply,
        &mut by.blob,
        &mut by.hello,
        &mut by.hello_reply,
        &mut by.paths,
        &mut by.identity,
        &mut by.links,
    ]
}

/// Runs `scenario`; returns its report.
async fn run(seed: u64, scenario: &'static Scenario) -> String {
    let mut out = String::new();
    let sim = Sim::with_identity(seed);
    let ids: Vec<String> = (0..scenario.nodes).map(|i| format!("n{i:03}")).collect();
    (scenario.net)(&sim, &ids);
    let timing = |config: &mut crate::node::NodeConfig| {
        config.timing = Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
        // As in production (`--neighbours`).
        config.neighbours_k = 12;
    };
    let mut nodes = vec![sim.start_on_with(&ids[0], (scenario.on)(0), &[], Start::Genesis, timing)];
    for (i, id) in ids.iter().enumerate().skip(1) {
        // The writer behind NAT reaches node 2 alone.
        let seed = if i == 1 {
            ids[2].as_str()
        } else {
            ids[0].as_str()
        };
        nodes.push(sim.start_on_with(id, (scenario.on)(i), &[seed], Start::Join, timing));
    }
    let all: Vec<&AppState> = nodes.iter().collect();
    let all = all.as_slice();
    sim.wait_ready(all, SETTLE).await;
    let count = scenario.nodes;
    sim.wait_until(
        "every node knows every member",
        SETTLE,
        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, SETTLE).await;
    // Neighbours chosen, handshakes done, what joining set off over: sync messages no
    // fewer in a half of the quiet phase than in the one before (some go on for good:
    // alive stamps, registrations). Those two halves are the quiet phase.
    let settling = Instant::now();
    let mut halves = vec![Totals::of(&nodes)];
    loop {
        tokio::time::sleep(QUIET / 2).await;
        halves.push(Totals::of(&nodes));
        let count = |i: usize| halves[i + 1].sent.sync.messages - halves[i].sent.sync.messages;
        let n = halves.len();
        if (n >= 3 && count(n - 2) >= count(n - 3)) || settling.elapsed() > SETTLE {
            break;
        }
    }
    let n = halves.len();
    let quiet = (halves[n.saturating_sub(3)], halves[n - 1]);

    let _ = writeln!(
        out,
        "\n{} ({} writes at {} a second each)",
        scenario.name,
        WRITES,
        1000 / WRITE_EVERY.as_millis()
    );
    for (w, (workload, writer)) in [
        ("one writer", Some(scenario.writer)),
        ("writers in turn", None),
    ]
    .into_iter()
    .enumerate()
    {
        let start = Totals::of(&nodes);
        for i in 0..WRITES {
            let node = &nodes[writer.unwrap_or(i % nodes.len())];
            // A path of each workload's own: writing what is there already changes
            // nothing, and nothing would pass.
            write(node, &format!("bench/{w}/{i}"), json!(i)).await;
            tokio::time::sleep(WRITE_EVERY).await;
        }
        let written = Instant::now();
        sim.wait_converged(all, SETTLE).await;
        let converged = written.elapsed();
        report(
            &mut out,
            workload,
            quiet,
            (start, Totals::of(&nodes)),
            converged,
        );
    }
    out
}

/// The workload's counts (`during`), and the quiet phase's for as long.
fn report(
    out: &mut String,
    workload: &str,
    quiet: (Totals, Totals),
    during: (Totals, Totals),
    converged: Duration,
) {
    let took = during.1.at - during.0.at;
    let scale = took.as_secs_f64() / (quiet.1.at - quiet.0.at).as_secs_f64();
    let writes = WRITES as f64;
    let _ = writeln!(
        out,
        "  {workload}: {:.1} s, converged {} ms after the last write",
        took.as_secs_f64(),
        converged.as_millis()
    );
    let _ = writeln!(
        out,
        "    {:<13} {:>9} {:>9} {:>11}   (quiet phase, as long: msgs)",
        "kind", "msgs", "/write", "bytes/write"
    );
    let rows = KINDS
        .iter()
        .zip(
            counters(&quiet.0.sent)
                .into_iter()
                .zip(counters(&quiet.1.sent)),
        )
        .zip(
            counters(&during.0.sent)
                .into_iter()
                .zip(counters(&during.1.sent)),
        );
    let (mut all_messages, mut all_bytes) = (0.0, 0.0);
    for ((kind, (q0, q1)), (d0, d1)) in rows {
        let quiet_messages = (q1.messages - q0.messages) as f64 * scale;
        let messages = (d1.messages - d0.messages) as f64;
        let bytes = (d1.bytes - d0.bytes) as f64;
        all_messages += messages;
        all_bytes += bytes;
        let _ = writeln!(
            out,
            "    {kind:<13} {messages:>9.0} {:>9.1} {:>11.0}   ({quiet_messages:.0})",
            messages / writes,
            bytes / writes
        );
    }
    let _ = writeln!(
        out,
        "    {:<13} {all_messages:>9.0} {:>9.1} {:>11.0}",
        "all",
        all_messages / writes,
        all_bytes / writes
    );
    let (r0, r1) = (during.0.relay, during.1.relay);
    let _ = writeln!(
        out,
        "    relay per write: {:.1} announcement rounds ({:.1} without a tree, {:.1} slow), {:.1} pulls ({:.1} queued), {:.1} along writers' trees",
        (r1.announce_rounds - r0.announce_rounds) as f64 / writes,
        (r1.treeless_rounds - r0.treeless_rounds) as f64 / writes,
        (r1.slow_rounds - r0.slow_rounds) as f64 / writes,
        (r1.pulls - r0.pulls) as f64 / writes,
        (r1.pulls_queued - r0.pulls_queued) as f64 / writes,
        (r1.tree_forwards - r0.tree_forwards) as f64 / writes,
    );
    let mut propagation = Propagation::default();
    let (p0, p1) = (during.0.propagation, during.1.propagation);
    for ((sum, a), b) in propagation
        .buckets
        .iter_mut()
        .zip(p0.buckets)
        .zip(p1.buckets)
    {
        *sum = b - a;
    }
    propagation.count = p1.count - p0.count;
    propagation.sum_ms = p1.sum_ms - p0.sum_ms;
    let within = |quantile| {
        super::soak::percentile_bound(&propagation, quantile)
            .map_or("more than an hour".to_owned(), |ms| format!("{ms} ms"))
    };
    let _ = writeln!(
        out,
        "    propagation: {} arrivals, mean {} ms, half within {}, 99% within {}",
        propagation.count,
        propagation.sum_ms / propagation.count.max(1),
        within(0.5),
        within(0.99),
    );
}