quorum-set 0.6.2

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

use crate::quorum::QuorumIntersection;
use crate::quorum::QuorumSet;

/// A quorum rule that accepts an ID set holding at least `quorum_size` of its IDs.
///
/// A read rule and a write rule over the same voters can share one ID set: build both from one
/// [`Arc`], or build the read rule from the write rule with [`dual()`](Self::dual). Two rules over
/// the same IDs have quorum intersection exactly when the sum of their quorum sizes exceeds the
/// number of IDs.
///
/// # Examples
///
/// ```
/// use std::collections::BTreeSet;
/// use std::num::NonZeroU64;
/// use std::sync::Arc;
///
/// use quorum_set::{QuorumIntersection, QuorumSet, ThresholdQuorumSet};
///
/// let voters = Arc::new(BTreeSet::from([1, 2, 3, 4, 5]));
/// let read = ThresholdQuorumSet::new(NonZeroU64::new(2).unwrap(), voters.clone());
/// let write = ThresholdQuorumSet::new(NonZeroU64::new(4).unwrap(), voters);
///
/// let read_accepts_two = read.is_quorum([1, 2].iter());
/// assert!(read_accepts_two);
/// let write_accepts_three = write.is_quorum([1, 2, 3].iter());
/// assert!(!write_accepts_three);
///
/// // 2 + 4 > 5, so every read quorum intersects every write quorum.
/// let intersects = read.intersects_with(&write);
/// assert_eq!(Some(true), intersects);
///
/// let shared = Arc::ptr_eq(read.id_set(), write.id_set());
/// assert!(shared);
/// ```
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ThresholdQuorumSet<ID> {
    ids: Arc<BTreeSet<ID>>,
    quorum_size: NonZeroU64,
}

impl<ID> ThresholdQuorumSet<ID> {
    /// Builds a rule that accepts an ID set holding at least `quorum_size` of `ids`.
    ///
    /// `ids` is a [`BTreeSet`], or an [`Arc`] of one to share it with another rule. A
    /// `quorum_size` above the number of `ids` gives a rule that accepts no set.
    pub fn new(quorum_size: NonZeroU64, ids: impl Into<Arc<BTreeSet<ID>>>) -> Self {
        let ids = ids.into();
        Self { ids, quorum_size }
    }

    /// Returns the shared ID set. [`QuorumSet::ids()`] yields the same IDs as an iterator.
    pub fn id_set(&self) -> &Arc<BTreeSet<ID>> {
        &self.ids
    }

    /// Returns the number of IDs that an accepted set must hold.
    pub fn quorum_size(&self) -> u64 {
        self.quorum_size.get()
    }

    /// Returns the rule over the same IDs that accepts exactly the sets that meet every quorum of
    /// `self`. It shares the [`Arc`] of IDs with `self`.
    ///
    /// A set meets every quorum of a rule that needs `quorum_size` of `n` IDs exactly when it holds
    /// at least `n + 1 - quorum_size` of them, so the dual needs that many. It is the threshold
    /// rule over these IDs with the smallest quorum size that intersects `self`, so it fits as the
    /// read rule of a write rule `self`. The dual of the dual is `self` again.
    ///
    /// Returns `None` when `quorum_size` exceeds the number of IDs. Such a rule has no quorum, so
    /// every set meets all of its quorums, the empty set included, and no `ThresholdQuorumSet`
    /// accepts the empty set.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::collections::BTreeSet;
    /// use std::num::NonZeroU64;
    /// use std::sync::Arc;
    ///
    /// use quorum_set::{QuorumIntersection, ThresholdQuorumSet};
    ///
    /// // A write quorum needs 2 of 5 IDs, so a read quorum needs 4 of them.
    /// let two = NonZeroU64::new(2).unwrap();
    /// let write = ThresholdQuorumSet::new(two, BTreeSet::from([1, 2, 3, 4, 5]));
    /// let read = write.dual().unwrap();
    /// assert_eq!(4, read.quorum_size());
    ///
    /// let shared = Arc::ptr_eq(write.id_set(), read.id_set());
    /// assert!(shared);
    /// let intersects = read.intersects_with(&write);
    /// assert_eq!(Some(true), intersects);
    /// ```
    pub fn dual(&self) -> Option<Self> {
        let id_count = self.ids.len() as u64;
        let spare = id_count.checked_sub(self.quorum_size())?;
        let quorum_size = NonZeroU64::MIN.saturating_add(spare);
        let ids = Arc::clone(&self.ids);
        Some(Self { ids, quorum_size })
    }
}

impl<ID> QuorumSet for ThresholdQuorumSet<ID>
where ID: Ord + Clone
{
    type Id = ID;
    type Iter = btree_set::IntoIter<ID>;

    fn is_quorum<'a, I>(&self, ids: I) -> bool
    where
        ID: 'a,
        I: Iterator<Item = &'a ID> + Clone,
    {
        let quorum_size = self.quorum_size();
        let mut count = 0;
        for id in self.ids.iter() {
            if ids.clone().any(|candidate| candidate == id) {
                count += 1;
                if count >= quorum_size {
                    return true;
                }
            }
        }
        false
    }

    fn ids(&self) -> Self::Iter {
        let ids = self.ids.as_ref().clone();
        ids.into_iter()
    }
}

/// Exact quorum intersection for two threshold rules.
///
/// A smallest quorum of `self` first takes the IDs that `other` lacks, and then takes shared IDs
/// only as needed. A smallest quorum of `other` does the same. Two disjoint quorums exist exactly
/// when the shared IDs are enough for both, so the check is exact and runs in O(n).
impl<ID> QuorumIntersection<ThresholdQuorumSet<ID>> for ThresholdQuorumSet<ID>
where ID: Ord + Clone
{
    /// Return `Some(true)` when every quorum of `self` intersects every quorum of `other`, and
    /// `Some(false)` otherwise. This check never returns `None`.
    ///
    /// A rule whose `quorum_size` exceeds its number of IDs has no quorum, so it has quorum
    /// intersection with every rule.
    fn intersects_with(&self, other: &ThresholdQuorumSet<ID>) -> Option<bool> {
        let only_self = self.ids.difference(&other.ids).count() as u64;
        let only_other = other.ids.difference(&self.ids).count() as u64;
        let shared = self.ids.intersection(&other.ids).count() as u64;

        let need_self = self.quorum_size().saturating_sub(only_self);
        let need_other = other.quorum_size().saturating_sub(only_other);
        let need_both = need_self.saturating_add(need_other);

        Some(need_both > shared)
    }
}