//! Which side of a pair of nodes leads, for duties only one side needs to do (opening
//! the sync after a reset, sending keepalive PINGs).
//!
//! Both sides decide on their own, from the two keys alone, and agree. The leader is
//! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a
//! global order would make the node with the smallest key lead every one of its pairs
//! and the one with the largest lead none, so every such duty would pile up on the same
//! few nodes. With the hash, each node leads about half of its pairs.

use sha2::{Digest, Sha256};

const LABEL: &[u8] = b"continuum/pair-leader/1";

/// Whether `local` leads its pair with `peer`. Exactly one of `leads(a, b)` and
/// `leads(b, a)` is true for any two different keys.
pub(crate) fn leads(local: &str, peer: &str) -> bool {
    let (low, high) = if local <= peer {
        (local, peer)
    } else {
        (peer, local)
    };
    let mut hash = Sha256::new();
    // Length-prefixed fields, so no two different pairs hash the same input.
    for field in [LABEL, low.as_bytes(), high.as_bytes()] {
        hash.update((field.len() as u32).to_be_bytes());
        hash.update(field);
    }
    let leader = if hash.finalize()[0] & 1 == 0 { low } else { high };
    leader == local
}

#[cfg(test)]
mod tests {
    use super::*;

    fn key(n: u32) -> String {
        format!("{:064x}", u64::from(n).wrapping_mul(0x9e37_79b9_7f4a_7c15))
    }

    #[test]
    fn exactly_one_side_leads() {
        for a in 0..50 {
            for b in 0..50 {
                if a != b {
                    assert_ne!(leads(&key(a), &key(b)), leads(&key(b), &key(a)));
                }
            }
        }
    }

    #[test]
    fn leading_is_spread_across_nodes() {
        let nodes: Vec<String> = (0..64).map(key).collect();
        let mut smallest_key_leads = 0;
        for (i, node) in nodes.iter().enumerate() {
            let led = nodes
                .iter()
                .filter(|peer| *peer != node && leads(node, peer))
                .count();
            // 63 pairs per node: about half each; far from all or none.
            assert!((10..=53).contains(&led), "node {i} leads {led} of 63 pairs");
            for peer in &nodes {
                if node < peer && leads(node, peer) {
                    smallest_key_leads += 1;
                }
            }
        }
        // Not simply the smaller key of each pair (64 * 63 / 2 = 2016 pairs).
        assert!((800..=1216).contains(&smallest_key_leads), "{smallest_key_leads}");
    }
}