//! Every member's clock against the swarm's, for the map: from the clocks each member
//! reports of its peers (`Link::clock`), chained from this node along the narrowest
//! bounds, then taken against their median. A member is off when it differs from the
//! median beyond its chained bound and the limit (`--clock-skew-warn`); with fewer than
//! three members compared, none is (two can't tell whose clock is off). An estimate for
//! the eye: each node judges its own clock by its peers' (`swarm::skew`).

use std::cmp::Reverse;
use std::collections::{BTreeMap, BinaryHeap, HashMap};

use serde::Serialize;

use crate::transport::Link;

/// A member's clock against the swarm's.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(super) struct MemberClock {
    /// How far it is ahead of the median (behind if negative).
    pub(super) offset_ms: i64,
    /// Give or take: the bounds chained to it from this node.
    pub(super) within_ms: u64,
    pub(super) off: bool,
}

/// From `local` and `reports` (each member's links), with `limit_ms`; by member.
pub(super) fn swarm<'a>(
    local: &str,
    reports: impl IntoIterator<Item = (&'a str, &'a [Link])>,
    limit_ms: u64,
) -> BTreeMap<String, MemberClock> {
    // Each way between two members: how far the far end's clock is ahead, give or take.
    let mut edges: HashMap<&str, Vec<(&str, i64, u64)>> = HashMap::new();
    for (reporter, links) in reports {
        for link in links {
            if let Some(clock) = link.clock {
                edges.entry(reporter).or_default().push((
                    link.peer.as_str(),
                    clock.offset_ms,
                    clock.within_ms,
                ));
                edges.entry(link.peer.as_str()).or_default().push((
                    reporter,
                    -clock.offset_ms,
                    clock.within_ms,
                ));
            }
        }
    }
    // Ahead of this node's, along the narrowest chained bounds (Dijkstra).
    let mut reached: HashMap<&str, (i64, u64)> = HashMap::new();
    let mut queue = BinaryHeap::from([Reverse((0_u64, local, 0_i64))]);
    while let Some(Reverse((within, member, offset))) = queue.pop() {
        if reached.contains_key(member) {
            continue;
        }
        reached.insert(member, (offset, within));
        for &(next, step, step_within) in edges.get(member).into_iter().flatten() {
            if !reached.contains_key(next) {
                queue.push(Reverse((
                    within.saturating_add(step_within),
                    next,
                    offset.saturating_add(step),
                )));
            }
        }
    }
    let mut offsets: Vec<i64> = reached.values().map(|(offset, _)| *offset).collect();
    offsets.sort_unstable();
    let median = offsets.get(offsets.len() / 2).copied().unwrap_or(0);
    let judged = reached.len() >= 3 && limit_ms > 0;
    reached
        .into_iter()
        .map(|(member, (offset, within))| {
            let offset_ms = offset - median;
            let off = judged && offset_ms.unsigned_abs() > within.saturating_add(limit_ms);
            (
                member.to_owned(),
                MemberClock {
                    offset_ms,
                    within_ms: within,
                    off,
                },
            )
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::transport::paths::ClockOffset;

    fn link(peer: &str, offset_ms: i64) -> Link {
        Link {
            peer: peer.into(),
            transport: None,
            relayed: false,
            down: false,
            rtt_us: None,
            dormant: false,
            others: Vec::new(),
            clock: Some(ClockOffset {
                offset_ms,
                within_ms: 20,
            }),
        }
    }

    /// Chained through the reports: the one clock 10 s ahead is off, wherever it is.
    #[test]
    fn the_odd_clock_out() {
        // a – b – c – d, d's clock 10 s ahead; a only knows b.
        let a = [link("b", 0)];
        let b = [link("a", 0), link("c", 5)];
        let c = [link("d", 10_000)];
        let clocks = swarm("a", [("a", &a[..]), ("b", &b[..]), ("c", &c[..])], 2_000);
        assert_eq!(clocks.len(), 4);
        assert!(clocks["d"].off);
        assert!(clocks["d"].offset_ms > 9_900);
        assert!(!clocks["a"].off && !clocks["b"].off && !clocks["c"].off);
        assert_eq!(clocks["d"].within_ms, 60);
        // Two alone: none.
        let two = swarm("a", [("a", &[link("b", 10_000)][..])], 2_000);
        assert!(two.values().all(|clock| !clock.off));
    }
}