//! HyParView's active views (`swarm::hyparview`, `--topology hyparview`): links both
//! ends agreed to, kept as members fail.

use std::collections::BTreeSet;
use std::time::Duration;

use serde_json::json;

use super::{Sim, write};
use crate::state::AppState;
use crate::swarm::bootstrap::Start;

const SOON: Duration = Duration::from_secs(60);
const K: usize = 3;

/// Every node by HyParView, changes passed on by PlumTree, each keeping `K` neighbours.
fn hyparview(config: &mut crate::node::NodeConfig) {
    config.topology = crate::swarm::topology::Kind::Hyparview;
    config.dissemination = crate::swarm::dissemination::Kind::Plumtree;
    config.neighbours_k = K;
}

/// Each node's active view, by node ID.
fn views(nodes: &[AppState]) -> Vec<(String, BTreeSet<String>)> {
    nodes
        .iter()
        .map(|node| {
            (
                node.local_id().to_owned(),
                node.topology().active().expect("an active view"),
            )
        })
        .collect()
}

/// Every node has a neighbour, none more than `K` and the ones that made room for a
/// node that had none, and every link is in both ends' views.
fn settled(nodes: &[AppState], up: &BTreeSet<String>) -> bool {
    unsettled(nodes, up).is_empty()
}

/// What keeps the views from being `settled`, one line each: empty or too large views,
/// links in one end's view only.
fn unsettled(nodes: &[AppState], up: &BTreeSet<String>) -> Vec<String> {
    let views = views(nodes);
    let mut wrong = Vec::new();
    for (node, active) in views.iter().filter(|(node, _)| up.contains(node)) {
        if active.is_empty() || active.len() > K + 1 {
            wrong.push(format!("{node}: {} neighbours", active.len()));
        }
        for other in active {
            if !views
                .iter()
                .any(|(id, theirs)| id == other && theirs.contains(node))
            {
                wrong.push(format!("{node} has {other}, which hasn't it"));
            }
        }
    }
    wrong
}

/// Waits until the views are `settled` among `up`, and `also` holds; else fails with
/// each node's active view (by its name in `names`, as `nodes` are in order) and what
/// keeps them unsettled: `Sim::describe` shows the neighbours being asked as well.
async fn wait_settled<F: Fn(&[AppState]) -> bool>(
    nodes: &[AppState],
    names: &[String],
    up: &BTreeSet<String>,
    what: &str,
    timeout: Duration,
    also: F,
) {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        if settled(nodes, up) && also(nodes) {
            return;
        }
        if tokio::time::Instant::now() >= deadline {
            let name = |id: &str| {
                nodes
                    .iter()
                    .position(|node| node.local_id() == id)
                    .map_or(id.to_owned(), |at| names[at].clone())
            };
            let views = views(nodes)
                .into_iter()
                .map(|(node, active)| {
                    format!(
                        "  {}: {}",
                        name(&node),
                        active
                            .iter()
                            .map(|id| name(id))
                            .collect::<Vec<_>>()
                            .join(", ")
                    )
                })
                .collect::<Vec<_>>()
                .join("\n");
            let wrong = unsettled(nodes, up)
                .into_iter()
                .map(|line| {
                    let mut line = line;
                    for node in nodes {
                        line = line.replace(node.local_id(), &name(node.local_id()));
                    }
                    format!("  {line}")
                })
                .collect::<Vec<_>>()
                .join("\n");
            let ways = nodes
                .iter()
                .map(|node| {
                    format!(
                        "{} {}/{}",
                        name(node.local_id()),
                        node.topology().name(),
                        node.dissemination().name()
                    )
                })
                .collect::<Vec<_>>()
                .join(", ");
            panic!(
                "timed out after {timeout:?} waiting until {what}\nactive views:\n{views}\nunsettled:\n{wrong}\nways: {ways}"
            );
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}

/// Twelve members, each keeping three neighbours: the active views settle, symmetric,
/// and a write reaches every member. A member that goes away leaves the views of those
/// that had it, which take others, and the next write reaches every member still up.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn active_views_settle_and_heal() {
    let sim = Sim::with_identity(88);
    let ids: Vec<String> = (0..12).map(|i| format!("h{i:02}")).collect();
    let mut nodes = vec![sim.start_with(&ids[0], &[], Start::Genesis, hyparview)];
    for id in &ids[1..] {
        nodes.push(sim.start_with(id, &[ids[0].as_str()], Start::Join, hyparview));
    }
    let all: Vec<&AppState> = nodes.iter().collect();
    sim.wait_ready(&all, SOON).await;
    // By node ID, as the views hold them (`ids` are the names on the simulated net).
    let up: BTreeSet<String> = nodes.iter().map(|n| n.local_id().to_owned()).collect();
    wait_settled(&nodes, &ids, &up, "the active views settle", SOON, |_| true).await;
    write(&nodes[0], "x", json!(1)).await;
    sim.wait_converged(&all, SOON).await;

    // The member most others have in their views goes away (by its index: its name on
    // the net, its ID in the views).
    let gone_at = {
        let views = views(&nodes);
        (0..nodes.len())
            .max_by_key(|i| {
                views
                    .iter()
                    .filter(|(_, active)| active.contains(nodes[*i].local_id()))
                    .count()
            })
            .expect("a member")
    };
    sim.net.take_down(&ids[gone_at]);
    let gone = nodes[gone_at].local_id().to_owned();
    let up: BTreeSet<String> = nodes
        .iter()
        .map(|n| n.local_id().to_owned())
        .filter(|id| *id != gone)
        .collect();
    wait_settled(
        &nodes,
        &ids,
        &up,
        "the views heal without the member gone",
        SOON * 2,
        |nodes| {
            views(nodes)
                .iter()
                .all(|(node, active)| *node == gone || !active.contains(&gone))
        },
    )
    .await;
    let remaining: Vec<&AppState> = nodes.iter().filter(|n| *n.local_id() != gone).collect();
    write(remaining[0], "y", json!(2)).await;
    sim.wait_converged(&remaining, SOON).await;
}

/// A swarm on the member list's graph and writers' trees switches, as a protocol writer
/// records HyParView and PlumTree: every node swaps, starting from the neighbours it had,
/// the active views settle, and a write reaches every member.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn a_swarm_switches_to_hyparview_at_runtime() {
    let sim = Sim::with_identity(89);
    let three = |config: &mut crate::node::NodeConfig| config.neighbours_k = K;
    let ids: Vec<String> = (0..8).map(|i| format!("s{i:02}")).collect();
    let mut nodes = vec![sim.start_with(&ids[0], &[], Start::Genesis, three)];
    for id in &ids[1..] {
        nodes.push(sim.start_with(id, &[ids[0].as_str()], Start::Join, three));
    }
    let all: Vec<&AppState> = nodes.iter().collect();
    sim.wait_ready(&all, SOON).await;
    assert!(nodes[0].topology().active().is_none());

    nodes[0]
        .mutate_doc(|doc| {
            crate::document::store::write_path(
                doc,
                &crate::swarm::topology::topology_path(),
                &json!("hyparview"),
            )?;
            crate::document::store::write_path(
                doc,
                &crate::swarm::topology::dissemination_path(),
                &json!("plumtree"),
            )
        })
        .await
        .expect("recorded");
    let up: BTreeSet<String> = nodes.iter().map(|n| n.local_id().to_owned()).collect();
    // Longer than a start: a node that switched first asks others still on the member
    // list's graph, which ignore NEIGHBOUR; it gives up on them after 20 s, and asks
    // them again only a minute later (or they ask it, once switched).
    wait_settled(
        &nodes,
        &ids,
        &up,
        "every node switched, and the active views settle",
        SOON * 2,
        |nodes| {
            nodes.iter().all(|node| {
                node.topology().name() == "hyparview" && node.dissemination().name() == "plumtree"
            })
        },
    )
    .await;
    write(&nodes[3], "z", json!(3)).await;
    sim.wait_converged(&all, SOON).await;
}