quorum-set 0.2.0

use std::collections::BTreeSet;

use super::QuorumTreeSpec;
use crate::Node;
use crate::QuorumTree;
use crate::QuorumTreeError;

fn id(i: u64) -> Node<u64> {
    Node::Id(i)
}

fn qset(quorum_size: u64, nodes: &[u64]) -> QuorumTreeSpec<u64> {
    QuorumTreeSpec::new(quorum_size, nodes.iter().copied().map(id)).unwrap()
}

fn set(quorum_size: u64, nodes: &[u64]) -> Node<u64> {
    Node::Subtree(QuorumTree::new(quorum_size, nodes.iter().copied().map(id)).unwrap())
}

fn raft_config(nodes: &[u64]) -> Node<u64> {
    let majority = nodes.len() as u64 / 2 + 1;
    set(majority, nodes)
}

fn btreeset(nodes: &[u64]) -> BTreeSet<u64> {
    nodes.iter().copied().collect()
}

fn all_subsets(ids: &[u64]) -> Vec<Vec<u64>> {
    (0..(1 << ids.len()))
        .map(|mask| {
            ids.iter()
                .enumerate()
                .filter(|(i, _)| mask & (1 << i) != 0)
                .map(|(_, id)| *id)
                .collect::<Vec<_>>()
        })
        .collect()
}

fn is_subset(lhs: &[u64], rhs: &[u64]) -> bool {
    lhs.iter().all(|id| rhs.contains(id))
}

fn assert_is_quorum_monotonic(qset: &QuorumTreeSpec<u64>, ids: &[u64]) {
    let subsets = all_subsets(ids);

    for lhs in &subsets {
        if !qset.is_quorum(lhs) {
            continue;
        }

        for rhs in &subsets {
            if is_subset(lhs, rhs) {
                assert!(qset.is_quorum(rhs), "lhs: {lhs:?}, rhs: {rhs:?}");
            }
        }
    }
}

fn is_majority_quorum(config: &BTreeSet<u64>, ids: &[u64]) -> bool {
    let majority = config.len() / 2 + 1;
    config.iter().filter(|id| ids.contains(id)).count() >= majority
}

fn assert_quorum_matches_raft_joint(
    qset: &QuorumTreeSpec<u64>,
    configs: Vec<BTreeSet<u64>>,
    ids: &[u64],
) {
    for selected in all_subsets(ids) {
        assert_eq!(
            configs.iter().all(|config| is_majority_quorum(config, &selected)),
            qset.is_quorum(&selected),
            "selected ids: {selected:?}"
        );
    }
}

#[test]
fn test_0_nodes_quorum_0() {
    let qset = qset(0, &[]);

    assert_eq!(0, qset.quorum_size());

    assert!(qset.is_quorum(&[]));
    assert!(qset.is_quorum(&[1]));
}

#[test]
fn test_0_nodes_quorum_1() {
    let err = QuorumTreeSpec::<u64>::new(1, []).unwrap_err();

    assert_eq!(
        QuorumTreeError::UnsatisfiableQuorum {
            quorum_size: 1,
            num_children: 0
        },
        err
    );
}

#[test]
fn test_3_nodes_quorum_0() {
    let qset = qset(0, &[1, 2, 3]);

    assert_eq!(0, qset.quorum_size());

    assert!(qset.is_quorum(&[]));
    assert!(qset.is_quorum(&[1]));
    assert!(qset.is_quorum(&[1, 2, 3]));
}

#[test]
fn test_3_nodes_quorum_1() {
    let qset = qset(1, &[1, 2, 3]);

    assert_eq!(1, qset.quorum_size());

    assert!(!qset.is_quorum(&[]));
    assert!(qset.is_quorum(&[1]));
    assert!(qset.is_quorum(&[2]));
    assert!(qset.is_quorum(&[1, 2, 3]));
}

#[test]
fn test_3_nodes() {
    let qset = QuorumTreeSpec::new(2, vec![id(1), id(2), id(3)]).unwrap();

    assert_eq!(2, qset.quorum_size());

    assert!(!qset.is_quorum(&[]));
    assert!(!qset.is_quorum(&[1]));
    assert!(!qset.is_quorum(&[2]));
    assert!(!qset.is_quorum(&[3]));

    assert!(qset.is_quorum(&[1, 2]));
    assert!(qset.is_quorum(&[1, 3]));
    assert!(qset.is_quorum(&[2, 3]));
    assert!(qset.is_quorum(&[1, 2, 3]));
}

#[test]
fn test_4_nodes() {
    let qset = QuorumTreeSpec::new(2, vec![id(1), id(2), id(3), id(4)]).unwrap();

    assert_eq!(2, qset.quorum_size());

    assert!(!qset.is_quorum(&[]));
    assert!(!qset.is_quorum(&[1]));
    assert!(!qset.is_quorum(&[2]));
    assert!(!qset.is_quorum(&[3]));

    assert!(qset.is_quorum(&[1, 2]));
    assert!(qset.is_quorum(&[1, 3]));
    assert!(qset.is_quorum(&[1, 4]));
    assert!(qset.is_quorum(&[2, 3]));
    assert!(qset.is_quorum(&[2, 4]));
    assert!(qset.is_quorum(&[3, 4]));
    assert!(qset.is_quorum(&[1, 2, 3]));
    assert!(qset.is_quorum(&[1, 2, 4]));
    assert!(qset.is_quorum(&[1, 3, 4]));
    assert!(qset.is_quorum(&[2, 3, 4]));
    assert!(qset.is_quorum(&[1, 2, 3, 4]));
}

#[test]
fn test_4_nodes_quorum_4() {
    let qset = qset(4, &[1, 2, 3, 4]);

    assert_eq!(4, qset.quorum_size());

    assert!(!qset.is_quorum(&[]));
    assert!(!qset.is_quorum(&[1]));
    assert!(!qset.is_quorum(&[1, 2, 3]));
    assert!(qset.is_quorum(&[1, 2, 3, 4]));
}

#[test]
fn test_5_nodes_quorum_4() {
    let qset = qset(4, &[1, 2, 3, 4, 5]);

    assert_eq!(4, qset.quorum_size());

    assert!(!qset.is_quorum(&[]));
    assert!(!qset.is_quorum(&[1]));
    assert!(!qset.is_quorum(&[1, 2, 3]));
    assert!(qset.is_quorum(&[1, 2, 3, 4]));
    assert!(qset.is_quorum(&[1, 2, 3, 4, 5]));
}

#[test]
fn test_joint() {
    let read_tree =
        QuorumTreeSpec::new(1, [raft_config(&[1, 2, 3]), raft_config(&[4, 5, 6])]).unwrap();
    let write_tree =
        QuorumTreeSpec::new(2, [raft_config(&[1, 2, 3]), raft_config(&[4, 5, 6])]).unwrap();

    assert_eq!(1, read_tree.quorum_size());
    assert_eq!(2, write_tree.quorum_size());

    assert!(!read_tree.is_quorum(&[]));
    assert!(!read_tree.is_quorum(&[1]));
    assert!(!read_tree.is_quorum(&[1, 4]));

    assert!(read_tree.is_quorum(&[1, 2]));
    assert!(read_tree.is_quorum(&[4, 5]));
    assert!(read_tree.is_quorum(&[1, 2, 4, 5]));

    assert!(!write_tree.is_quorum(&[]));
    assert!(!write_tree.is_quorum(&[1]));
    assert!(!write_tree.is_quorum(&[1, 2]));
    assert!(!write_tree.is_quorum(&[4, 5]));
    assert!(!write_tree.is_quorum(&[1, 4]));

    assert!(write_tree.is_quorum(&[1, 2, 4, 5]));
    assert!(write_tree.is_quorum(&[1, 3, 4, 6]));
    assert!(write_tree.is_quorum(&[1, 2, 3, 4, 5, 6]));

    assert_quorum_matches_raft_joint(
        &write_tree,
        vec![btreeset(&[1, 2, 3]), btreeset(&[4, 5, 6])],
        &[1, 2, 3, 4, 5, 6],
    );

    let write_tree =
        QuorumTreeSpec::new(2, [raft_config(&[1, 2, 3, 4]), raft_config(&[3, 4, 5, 6])]).unwrap();

    assert_quorum_matches_raft_joint(
        &write_tree,
        vec![btreeset(&[1, 2, 3, 4]), btreeset(&[3, 4, 5, 6])],
        &[1, 2, 3, 4, 5, 6],
    );
}

#[test]
fn test_is_quorum_is_monotonic() {
    assert_is_quorum_monotonic(&qset(0, &[1, 2, 3]), &[1, 2, 3]);
    assert_is_quorum_monotonic(&qset(2, &[1, 2, 3]), &[1, 2, 3]);
    assert_is_quorum_monotonic(&qset(4, &[1, 2, 3, 4]), &[1, 2, 3, 4]);

    let qset = QuorumTreeSpec::new(2, [raft_config(&[1, 2, 3]), raft_config(&[4, 5, 6])]).unwrap();
    assert_is_quorum_monotonic(&qset, &[1, 2, 3, 4, 5, 6]);

    let qset = QuorumTreeSpec::new(2, [
        set(2, &[1, 2, 3]),
        set(2, &[4, 5, 6]),
        set(2, &[7, 8, 9]),
    ])
    .unwrap();
    assert_is_quorum_monotonic(&qset, &[1, 2, 3, 4, 5, 6, 7, 8, 9]);
}

#[test]
fn test_group() {
    let qset = QuorumTreeSpec::new(2, [
        set(2, &[1, 2, 3]),
        set(2, &[4, 5, 6]),
        set(2, &[7, 8, 9]),
    ])
    .unwrap();

    assert_eq!(2, qset.quorum_size());

    assert!(!qset.is_quorum(&[]));
    assert!(!qset.is_quorum(&[1, 2]));
    assert!(!qset.is_quorum(&[1, 4, 7]));
    assert!(!qset.is_quorum(&[1, 2, 4]));

    assert!(qset.is_quorum(&[1, 2, 4, 5]));
    assert!(qset.is_quorum(&[1, 3, 7, 9]));
    assert!(qset.is_quorum(&[4, 6, 7, 8]));
    assert!(qset.is_quorum(&[1, 2, 3, 4, 5, 6, 7, 8, 9]));
}

#[test]
fn test_duplicate_subtrees_are_rejected() {
    // Canonically equal subtrees are duplicates even when built in a different node order.
    let err = QuorumTreeSpec::new(2, [set(2, &[1, 2, 3]), set(2, &[3, 2, 1])]).unwrap_err();

    assert_eq!(
        QuorumTreeError::DuplicateChild {
            canonical_id: "Subtree=2/(Id=1,Id=2,Id=3)".to_string(),
        },
        err
    );
}