use super::*;
use std::collections::VecDeque;

fn members(n: usize) -> Vec<String> {
    (0..n).map(|i| format!("{:064x}", i * 7919 + 13)).collect()
}

fn all_reachable(_: &str) -> bool {
    false
}

fn none_stable(_: &str) -> bool {
    false
}

fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> {
    super::graph(members, k, &all_reachable, &none_stable)
}

fn connected(g: &BTreeMap<String, BTreeSet<String>>) -> bool {
    let Some(start) = g.keys().next() else {
        return true;
    };
    let mut seen = BTreeSet::from([start.clone()]);
    let mut queue = VecDeque::from([start.clone()]);
    while let Some(m) = queue.pop_front() {
        for next in &g[&m] {
            if seen.insert(next.clone()) {
                queue.push_back(next.clone());
            }
        }
    }
    seen.len() == g.len()
}

fn edges(graph: &BTreeMap<String, BTreeSet<String>>) -> BTreeSet<(String, String)> {
    graph
        .iter()
        .flat_map(|(a, ns)| {
            ns.iter().map(move |b| {
                if a < b {
                    (a.clone(), b.clone())
                } else {
                    (b.clone(), a.clone())
                }
            })
        })
        .collect()
}

fn hex(bytes: [u8; 32]) -> String {
    crate::transport::Hex(&bytes).to_string()
}

/// Over random swarms (size, `k`, who is reachable, who is stable): links only
/// between reachable members, never to oneself, symmetric; a full mesh while small,
/// otherwise at least `k` links each and connected; independent of the order members
/// are listed in; and each member's own view (`neighbours`) agrees with the graph.
#[test]
fn random_swarms() {
    crate::testrand::for_seeds(300, |rng| {
        let n = rng.between(1, 60);
        let k = rng.between(2, 14);
        let mut all: Vec<String> = (0..n).map(|_| hex(rng.bytes32())).collect();
        let unreachable: BTreeSet<String> =
            all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
        let stable: BTreeSet<String> = all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
        let is_unreachable = |m: &str| unreachable.contains(m);
        let is_stable = |m: &str| stable.contains(m);

        let g = super::graph(&all, k, &is_unreachable, &is_stable);
        // Owned: `all` is shuffled below.
        let live: BTreeSet<String> = all
            .iter()
            .filter(|m| !unreachable.contains(*m))
            .cloned()
            .collect();
        assert_eq!(g.keys().cloned().collect::<BTreeSet<_>>(), live);
        for (a, ns) in &g {
            assert!(!ns.contains(a), "{a} links to itself");
            for b in ns {
                assert!(live.contains(b), "{a} links to unreachable {b}");
                assert!(g[b].contains(a), "{a} -> {b} but not back");
            }
        }
        if live.len() <= k + 1 {
            assert!(
                g.values().all(|ns| ns.len() + 1 == live.len()),
                "not a full mesh"
            );
        } else {
            assert!(
                g.values().all(|ns| ns.len() >= k),
                "a member with fewer than k links"
            );
            assert!(connected(&g), "not connected");
        }

        rng.shuffle(&mut all);
        assert_eq!(super::graph(&all, k, &is_unreachable, &is_stable), g);

        if let Some(local) = live.iter().nth(rng.below(live.len())) {
            let others: Vec<String> = all.iter().filter(|m| *m != local).cloned().collect();
            assert_eq!(
                neighbours(&others, local, k, &is_unreachable, &is_stable),
                g[local]
            );
        }
    });
}

#[test]
fn small_swarms_are_a_full_mesh() {
    let ms = members(13);
    for m in &ms {
        assert_eq!(
            neighbours(&ms, m, 12, &all_reachable, &none_stable).len(),
            12
        );
    }
}

#[test]
fn links_are_symmetric_and_degree_at_least_k() {
    let ms = members(250);
    let g = graph(&ms, 8);
    for (a, ns) in &g {
        assert!(ns.len() >= 8, "{a} has {} neighbours", ns.len());
        assert!(!ns.contains(a));
        for b in ns {
            assert!(g[b].contains(a), "{a} -> {b} not symmetric");
        }
    }
}

#[test]
fn graph_is_connected() {
    for (n, k) in [(250, 8), (250, 2), (60, 3)] {
        assert!(connected(&graph(&members(n), k)), "n={n} k={k}");
    }
}

#[test]
fn one_member_joining_changes_few_links() {
    let ms = members(200);
    let before = edges(&graph(&ms, 8));
    let mut more = ms.clone();
    more.push(format!("{:064x}", 999_999));
    let after = edges(&graph(&more, 8));
    let changed = before.symmetric_difference(&after).count();
    assert!(changed <= 4 * 8, "{changed} links changed");
}

#[test]
fn reachable_members_stay_connected_when_most_are_offline() {
    let ms = members(100);
    // Every tenth member is online.
    let online: BTreeSet<String> = ms.iter().step_by(10).cloned().collect();
    let offline = |m: &str| !online.contains(m);
    let g = super::graph(&ms, 12, &offline, &none_stable);
    assert_eq!(g.len(), 10);
    // Ten reachable members and k = 12: a full mesh among them.
    assert!(g.values().all(|ns| ns.len() == 9));

    let online: BTreeSet<String> = ms.iter().step_by(4).cloned().collect();
    let offline = |m: &str| !online.contains(m);
    let g = super::graph(&ms, 8, &offline, &none_stable);
    assert_eq!(g.len(), 25);
    assert!(connected(&g));
    assert!(
        g.values()
            .all(|ns| ns.len() >= 8 && ns.iter().all(|n| online.contains(n)))
    );
}

#[test]
fn stable_members_become_hubs_with_bounded_degree() {
    let ms = members(250);
    let stable: BTreeSet<String> = ms.iter().step_by(12).take(20).cloned().collect();
    let is_stable = |m: &str| stable.contains(m);
    let g = super::graph(&ms, 12, &all_reachable, &is_stable);
    assert!(connected(&g));

    let degree = |pred: &dyn Fn(&String) -> bool| -> (usize, f64) {
        let degrees: Vec<usize> = g
            .iter()
            .filter(|(m, _)| pred(m))
            .map(|(_, ns)| ns.len())
            .collect();
        let max = *degrees.iter().max().unwrap();
        (
            max,
            degrees.iter().sum::<usize>() as f64 / degrees.len() as f64,
        )
    };
    let (stable_max, stable_mean) = degree(&|m| stable.contains(m));
    let (_, transient_mean) = degree(&|m| !stable.contains(m));
    // 230 transient members put about 5 of their 10 scored slots on 20 stable ones.
    assert!(
        stable_mean > 3.0 * transient_mean,
        "{stable_mean} vs {transient_mean}"
    );
    assert!(
        stable_max <= 100,
        "a stable member has {stable_max} neighbours"
    );
}

/// A hub's degree in a swarm of `n` with `stable_count` stable members stays within
/// the bound.
fn hub_degree_is_bounded(n: usize, stable_count: usize) {
    // Few stable members among many: without the fan-in budget each would be chosen
    // by most of the swarm (about 130 neighbours with 10 of 250). With it, a hub has
    // at most the budget plus the links others choose by pair score and the ring.
    let k = 12;
    let ms = members(n);
    let stable: BTreeSet<String> = ms.iter().step_by(n / stable_count).cloned().collect();
    let is_stable = |m: &str| stable.contains(m);
    let g = super::graph(&ms, k, &all_reachable, &is_stable);
    assert!(connected(&g), "n={n} stable={stable_count}");
    let hub_max = g
        .iter()
        .filter(|(m, _)| stable.contains(*m))
        .map(|(_, ns)| ns.len())
        .max()
        .unwrap();
    assert!(
        hub_max <= 8 * k,
        "n={n} stable={stable_count}: a stable member has {hub_max} neighbours"
    );
}

/// Two transports with members on one or the other, and two members on both: every
/// link is over a shared transport, and the two members on both connect the rest.
#[test]
fn members_on_two_transports_bridge_the_others() {
    let k = 8;
    let ms = members(201);
    let on = |names: &[&str]| -> BTreeSet<String> {
        names.iter().map(|name| (*name).to_owned()).collect()
    };
    let transports: BTreeMap<String, BTreeSet<String>> = ms
        .iter()
        .enumerate()
        .map(|(i, m)| {
            let transports = match i {
                0 | 1 => on(&["a", "b"]),
                i if i % 2 == 0 => on(&["a"]),
                _ => on(&["b"]),
            };
            (m.clone(), transports)
        })
        .collect();
    let no_stable = |_: &str| false;
    let mut graph: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
    for (m, own) in &transports {
        for n in neighbours_across(&transports, m, own, k, &all_reachable, &no_stable) {
            assert!(
                !transports[&n].is_disjoint(own),
                "{m} and {n} share no transport"
            );
            graph.entry(m.clone()).or_default().insert(n.clone());
            graph.entry(n).or_default().insert(m.clone());
        }
    }
    assert_eq!(graph.len(), ms.len());
    assert!(connected(&graph));
    let bridge = &ms[0];
    assert!(graph[bridge].iter().any(|n| !transports[n].contains("b")));
    assert!(graph[bridge].iter().any(|n| !transports[n].contains("a")));

    // One transport: the one graph, as without transports.
    let one: BTreeMap<String, BTreeSet<String>> =
        ms.iter().map(|m| (m.clone(), on(&[""]))).collect();
    for m in &ms[..5] {
        assert_eq!(
            neighbours_across(&one, m, &on(&[""]), k, &all_reachable, &no_stable),
            neighbours(&ms, m, k, &all_reachable, &no_stable)
        );
    }
}

#[test]
fn schemes_of_addresses() {
    let addresses = |all: &[&str]| -> Vec<String> { all.iter().map(|a| (*a).to_owned()).collect() };
    assert_eq!(
        schemes(&addresses(&[
            "mycelium:ab",
            "libp2p:/ip4/1.2.3.4/tcp/1",
            "libp2p:/ip6/::1/tcp/1"
        ])),
        BTreeSet::from(["libp2p".to_owned(), "mycelium".to_owned()])
    );
    assert_eq!(schemes(&addresses(&["a"])), BTreeSet::from([String::new()]));
    assert!(schemes(&[]).is_empty());
}

#[test]
fn hub_degree_is_bounded_however_few_stable_members() {
    for (n, stable_count) in [(250, 10), (400, 5)] {
        hub_degree_is_bounded(n, stable_count);
    }
}

/// The same at the sizes the bound was designed for; slow in debug builds, so run
/// on request (`cargo test --release -- --ignored`).
#[test]
#[ignore]
fn hub_degree_is_bounded_in_large_swarms() {
    for (n, stable_count) in [(1000, 10), (1000, 20), (2000, 20)] {
        hub_degree_is_bounded(n, stable_count);
    }
}

/// What passing one change on costs over the graph the member list gives, by the
/// number of messages, while some links don't work (`down`: a share of them, chosen by
/// their members' keys): as now (every node that gets it tells all its neighbours but
/// the one it came from, over the links that work), and along the writer's tree.
///
/// The tree (`swarm::tree`): every member's distance from the writer over the member
/// list's graph (as every node computes it alike), and each member's parent the
/// neighbour it reaches (as it tells its neighbours, `swarm::reach`) that comes first by
/// distance, then whether it reaches a member a hop closer, then pair score, of those
/// before it (by the same, then pair key: no cycles). Whoever has a
/// change of the writer's passes it on to its children in the writer's tree, however
/// it got it. A member with no parent it reaches waits for an announcement from a
/// neighbour that has it (a pull), then passes it on to its own children.
///
/// Reported per change: messages along the tree, pulls, the share that got it along the
/// tree alone, and how many announcement rounds the last one waited. Arrival in
/// breadth-first order: no delays, no losses. A benchmark: `cargo benchmarks`.
#[test]
#[ignore = "a benchmark: cargo benchmarks"]
fn bench_writer_trees() {
    use std::collections::{HashMap, HashSet};
    // Along the tree from `from`, to all below it: the messages counted in `total`.
    fn along<'a>(
        from: &'a str,
        children: &HashMap<&'a str, Vec<&'a str>>,
        has: &mut HashSet<&'a str>,
        total: &mut usize,
    ) {
        let mut queue = VecDeque::from([from]);
        has.insert(from);
        while let Some(m) = queue.pop_front() {
            for &child in children.get(m).map_or(&[][..], Vec::as_slice) {
                *total += 1;
                if has.insert(child) {
                    queue.push_back(child);
                }
            }
        }
    }
    println!();
    let k = 12;
    for n in [13, 50, 100, 250, 1000, 2000] {
        let ms = members(n);
        let g = graph(&ms, k);
        let key: HashMap<&str, u64> = ms.iter().map(|m| (m.as_str(), pair_key(m))).collect();
        let degree: f64 = g.values().map(BTreeSet::len).sum::<usize>() as f64 / n as f64;
        let sources: Vec<&str> = ms
            .iter()
            .step_by((n / 10).max(1))
            .map(String::as_str)
            .collect();
        for down_percent in [0, 5, 20] {
            let works =
                |a: &str, b: &str| g[a].contains(b) && mix(key[a] ^ key[b]) % 100 >= down_percent;
            let (mut now_total, mut tree_total, mut pulls_total) = (0usize, 0usize, 0usize);
            let (mut eager_total, mut reached_total) = (0usize, 0usize);
            let (mut depth_max, mut children_max, mut rounds_max) = (0usize, 0usize, 0usize);
            for &source in &sources {
                // As now, over the links that work.
                let mut seen: HashSet<&str> = HashSet::from([source]);
                let mut queue: VecDeque<(&str, Option<&str>)> = VecDeque::from([(source, None)]);
                while let Some((m, from)) = queue.pop_front() {
                    for next in &g[m] {
                        if Some(next.as_str()) == from || !works(m, next) {
                            continue;
                        }
                        now_total += 1;
                        if seen.insert(next.as_str()) {
                            queue.push_back((next.as_str(), Some(m)));
                        }
                    }
                }
                // Distances over the member list's graph.
                let mut depth: HashMap<&str, usize> = HashMap::from([(source, 0)]);
                let mut order = vec![source];
                let mut at = 0;
                while at < order.len() {
                    let m = order[at];
                    at += 1;
                    for next in &g[m] {
                        if !depth.contains_key(next.as_str()) {
                            depth.insert(next.as_str(), depth[m] + 1);
                            order.push(next.as_str());
                        }
                    }
                }
                depth_max = depth_max.max(depth.values().copied().max().unwrap_or(0));
                // Each member's parent: one it reaches, before it.
                // Whether a member reaches one a hop closer (the writer: yes).
                let closer = |m: &str| {
                    depth[m] == 0
                        || g[m]
                            .iter()
                            .any(|o| depth[o.as_str()] + 1 == depth[m] && works(m, o))
                };
                let rank = |m: &str| (depth[m], !closer(m), key[m]);
                let mut children: HashMap<&str, Vec<&str>> = HashMap::new();
                for &m in &order[1..] {
                    let parent = g[m]
                        .iter()
                        .map(String::as_str)
                        .filter(|p| rank(p) < rank(m) && works(m, p))
                        .min_by_key(|p| (depth[p], !closer(p), pair_score(p, m)));
                    if let Some(parent) = parent {
                        children.entry(parent).or_default().push(m);
                    }
                }
                children_max = children_max.max(children.values().map(Vec::len).max().unwrap_or(0));
                // Along the tree from whoever has it; then pulls where announcements
                // reach, round by round.
                let mut has: HashSet<&str> = HashSet::new();
                along(source, &children, &mut has, &mut tree_total);
                eager_total += has.len();
                let mut rounds = 0;
                loop {
                    let pulling: Vec<&str> = ms
                        .iter()
                        .map(String::as_str)
                        .filter(|m| {
                            !has.contains(m)
                                && g[*m]
                                    .iter()
                                    .any(|x| has.contains(x.as_str()) && works(m, x))
                        })
                        .collect();
                    if pulling.is_empty() {
                        break;
                    }
                    rounds += 1;
                    for m in pulling {
                        if !has.contains(m) {
                            pulls_total += 1;
                            along(m, &children, &mut has, &mut tree_total);
                        }
                    }
                }
                rounds_max = rounds_max.max(rounds);
                reached_total += has.len();
            }
            let changes = sources.len() as f64;
            let share = |count: usize| 100.0 * count as f64 / changes / n as f64;
            println!(
                "{n:>5} members, {degree:.1} neighbours, {down_percent:>2}% of links down: per change \
                 {:>6.0} messages now; along the writer's tree {:>5.0}, and {:>4.0} pulls \
                 ({:.1}% along the tree alone, {:.1}% in all, the last after {rounds_max} \
                 announcement rounds; {depth_max} hops, {children_max} children at most)",
                now_total as f64 / changes,
                tree_total as f64 / changes,
                pulls_total as f64 / changes,
                share(eager_total),
                share(reached_total),
            );
        }
    }
}

/// The neighbours a swarm records count; without one, or with one out of range, the
/// node's own.
#[test]
fn the_swarms_neighbours_count() {
    use crate::document::testing::change;
    use crate::document::{Document, store};
    let mut doc = Document::new();
    assert_eq!(swarm_neighbours(&doc, 12), 12);
    for (recorded, expected) in [
        (serde_json::json!(8), 8),
        (serde_json::json!(1), 12),
        (serde_json::json!(1000), 12),
        (serde_json::json!("eight"), 12),
    ] {
        change(&mut doc, |doc| {
            store::write_path(doc, &neighbours_path(), &recorded).unwrap()
        });
        assert_eq!(swarm_neighbours(&doc, 12), expected, "{recorded}");
    }
    assert!(affects_neighbours(&neighbours_path()));
}