quorum-set 0.6.2

use std::collections::BTreeSet;
use std::num::NonZeroU64;
use std::sync::Arc;

use maplit::btreeset;

use crate::quorum::QuorumIntersection;
use crate::quorum::QuorumSet;
use crate::quorum::ThresholdQuorumSet;
use crate::quorum::verify_intersection;

/// Returns a rule that accepts a set holding at least `quorum_size` of `ids`.
pub(super) fn threshold(quorum_size: u64, ids: BTreeSet<u64>) -> ThresholdQuorumSet<u64> {
    let quorum_size = NonZeroU64::new(quorum_size).expect("quorum_size is at least 1");
    ThresholdQuorumSet::new(quorum_size, ids)
}

/// Returns every subset of `ids`, ordered by the bit mask that selects it.
pub(super) fn all_subsets(ids: &BTreeSet<u64>) -> Vec<BTreeSet<u64>> {
    let ids = ids.iter().copied().collect::<Vec<_>>();
    let mut subsets = vec![];
    for mask in 0..1u64 << ids.len() {
        let selected = ids.iter().enumerate().filter(|(i, _)| mask & (1u64 << i) != 0);
        let subset = selected.map(|(_, id)| *id).collect();
        subsets.push(subset);
    }
    subsets
}

/// Returns every rule over a subset of `ids`, with each `quorum_size` from 1 to one above the
/// subset size. The largest size gives a rule that accepts no set.
pub(super) fn all_thresholds(ids: &BTreeSet<u64>) -> Vec<ThresholdQuorumSet<u64>> {
    let mut thresholds = vec![];
    for subset in all_subsets(ids) {
        let max_quorum_size = subset.len() as u64 + 1;
        for quorum_size in 1..=max_quorum_size {
            thresholds.push(threshold(quorum_size, subset.clone()));
        }
    }
    thresholds
}

#[test]
fn test_is_quorum_accepts_every_set_holding_quorum_size_ids() {
    let ids = btreeset! {1, 2, 3, 4, 5};
    let three_of_five = threshold(3, ids.clone());

    let subsets = all_subsets(&ids);
    let mut accepted = vec![];
    for subset in &subsets {
        let is_quorum = three_of_five.is_quorum(subset.iter());
        if is_quorum {
            accepted.push(subset.clone());
        }
    }

    // 10 subsets hold 3 IDs, 5 hold 4, and 1 holds all 5.
    let want = subsets.into_iter().filter(|subset| subset.len() >= 3).collect::<Vec<_>>();
    assert_eq!(16, want.len());
    assert_eq!(want, accepted);
}

#[test]
fn test_is_quorum_counts_each_rule_id_once() {
    let two_of_three = threshold(2, btreeset! {1, 2, 3});

    // A repeated ID counts once, and an ID outside the rule never counts.
    let repeated = two_of_three.is_quorum([1, 1, 1].iter());
    assert!(!repeated);
    let outside = two_of_three.is_quorum([1, 4, 5].iter());
    assert!(!outside);
    let two_ids = two_of_three.is_quorum([1, 1, 2].iter());
    assert!(two_ids);
}

#[test]
fn test_quorum_size_above_id_count_accepts_no_set() {
    let four_of_three = threshold(4, btreeset! {1, 2, 3});
    let all_ids = four_of_three.is_quorum([1, 2, 3, 4].iter());
    assert!(!all_ids);

    let one_of_none = threshold(1, btreeset! {});
    let empty_input = one_of_none.is_quorum(std::iter::empty());
    assert!(!empty_input);
    let some_ids = one_of_none.is_quorum([1].iter());
    assert!(!some_ids);
}

#[test]
fn test_rules_share_one_id_set() {
    let voters = Arc::new(btreeset! {1, 2, 3});
    let three = NonZeroU64::new(3).unwrap();
    let read = ThresholdQuorumSet::new(NonZeroU64::MIN, voters.clone());
    let write = ThresholdQuorumSet::new(three, voters.clone());

    let read_shares = Arc::ptr_eq(&voters, read.id_set());
    assert!(read_shares);
    let write_shares = Arc::ptr_eq(&voters, write.id_set());
    assert!(write_shares);

    assert_eq!(1, read.quorum_size());
    assert_eq!(3, write.quorum_size());

    let ids = read.ids().collect::<Vec<_>>();
    assert_eq!(vec![1, 2, 3], ids);
}

#[test]
fn test_intersects_with() {
    let voters = btreeset! {1, 2, 3, 4, 5};
    let two_of_five = threshold(2, voters.clone());
    let three_of_five = threshold(3, voters.clone());
    let four_of_five = threshold(4, voters);

    // 2 + 4 > 5: a quorum of 2 and a quorum of 4 always share an ID.
    let sizes_exceed_ids = two_of_five.intersects_with(&four_of_five);
    assert_eq!(Some(true), sizes_exceed_ids);

    // 2 + 3 = 5: {1,2} and {3,4,5} are disjoint quorums.
    let sizes_equal_ids = two_of_five.intersects_with(&three_of_five);
    assert_eq!(Some(false), sizes_equal_ids);

    // A quorum of 3 of {1,2,3,4} holds at least 2 of {1,2,3}, so it meets every quorum of 2 of
    // {1,2,3}.
    let old = threshold(2, btreeset! {1, 2, 3});
    let new = threshold(3, btreeset! {1, 2, 3, 4});
    let different_ids = old.intersects_with(&new);
    assert_eq!(Some(true), different_ids);
}

#[test]
fn test_intersects_with_agrees_with_verify_intersection() {
    let thresholds = all_thresholds(&btreeset! {1, 2, 3, 4});

    // The check always answers, and every answer matches the exact relation.
    for a in &thresholds {
        for b in &thresholds {
            let intersects = a.intersects_with(b);
            let exact = verify_intersection(a, b);
            assert_eq!(Some(exact), intersects, "a: {a:?}, b: {b:?}");
        }
    }
}

#[test]
fn test_dual() {
    // A rule that needs 2 of 5 IDs has a dual that needs 4 of them, over the same ID set.
    let rule = threshold(2, btreeset! {1, 2, 3, 4, 5});
    let dual = rule.dual().unwrap();
    assert_eq!(4, dual.quorum_size());
    let shared = Arc::ptr_eq(rule.id_set(), dual.id_set());
    assert!(shared);

    // The dual of the dual is the rule again.
    let dual_of_dual = dual.dual().unwrap();
    assert_eq!(rule, dual_of_dual);

    // A rule with no quorum has no dual.
    let no_quorum = threshold(6, btreeset! {1, 2, 3, 4, 5});
    assert_eq!(None, no_quorum.dual());
    let no_id = threshold(1, btreeset! {});
    assert_eq!(None, no_id.dual());
}

#[test]
fn test_dual_accepts_exactly_the_sets_that_meet_every_quorum() {
    let ids = btreeset! {1, 2, 3, 4};
    let subsets = all_subsets(&ids);

    for rule in all_thresholds(&ids) {
        let Some(dual) = rule.dual() else {
            // Only a rule with no quorum has no dual.
            let no_quorum = rule.quorum_size() > rule.id_set().len() as u64;
            assert!(no_quorum, "{rule:?}");
            continue;
        };

        let quorums: Vec<&BTreeSet<u64>> =
            subsets.iter().filter(|subset| rule.is_quorum(subset.iter())).collect();
        for subset in &subsets {
            let meets_every_quorum = quorums.iter().all(|quorum| !quorum.is_disjoint(subset));
            let accepts = dual.is_quorum(subset.iter());
            assert_eq!(
                meets_every_quorum, accepts,
                "rule={rule:?}, subset={subset:?}"
            );
        }

        let intersects = rule.intersects_with(&dual);
        assert_eq!(Some(true), intersects, "{rule:?}");
    }
}