//! Whether this node's clock agrees with its peers' (`Transport::clock_offset`, from
//! their PONGs). Each peer's clock is known within half a round trip; this node's own
//! is off when it differs from most of its peers' by more than `--clock-skew-warn` for
//! sure, then the log says so (every `WARN_EVERY` while it lasts). A clock set anew
//! between two checks (by NTP, a resumed machine, by hand) by more than the limit shows
//! as a jump, this node's when most peers' clocks moved alike, else those peers'; the
//! log says so too. Clocks are the operating system's to set (NTP); the propagation
//! figures only take this node's own offset off (`AppState::clock_ahead_ms`).
//!
//! Two nodes alone can't tell whose clock is off, or jumped: each says it is its own.

use std::collections::HashMap;

use std::time::Duration;

use serde::Serialize;
use tokio::time::Instant;
use tracing::{info, warn};

use crate::locks::MutexExt;
use crate::state::AppState;
use crate::transport::paths::ClockOffset;

/// While this node's clock is off, the log says so this often.
const WARN_EVERY: Duration = Duration::from_secs(600);

/// How this node's clock compares with its peers'.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Skew {
    /// Peers whose clocks were compared (active ones with a PONG lately).
    pub(crate) peers: usize,
    /// How far this node's clock is ahead of theirs (behind if negative): the median.
    pub(crate) offset_ms: Option<i64>,
    /// Peers whose clocks are behind ours by more than the limit for sure, and ahead.
    pub(crate) behind: usize,
    pub(crate) ahead: usize,
    /// Whether ours is off: more than half of them differ from it beyond the limit,
    /// the same way. Never with a limit of 0.
    pub(crate) off: bool,
    /// The limit (`--clock-skew-warn`).
    pub(crate) limit_ms: u64,
}

/// Ours against the peers' clocks `offsets` (each ahead of ours), with `limit`.
pub(crate) fn of(offsets: &[ClockOffset], limit: Duration) -> Skew {
    let limit_ms = u64::try_from(limit.as_millis()).unwrap_or(u64::MAX);
    let beyond = |offset_ms: i64, within_ms: u64| {
        offset_ms > 0 && offset_ms.unsigned_abs() > within_ms.saturating_add(limit_ms)
    };
    let ahead = offsets
        .iter()
        .filter(|o| beyond(o.offset_ms, o.within_ms))
        .count();
    let behind = offsets
        .iter()
        .filter(|o| beyond(-o.offset_ms, o.within_ms))
        .count();
    let mut ours: Vec<i64> = offsets.iter().map(|o| -o.offset_ms).collect();
    ours.sort_unstable();
    let peers = offsets.len();
    Skew {
        peers,
        offset_ms: ours.get(peers / 2).copied(),
        behind,
        ahead,
        off: limit_ms > 0 && (ahead * 2 > peers || behind * 2 > peers),
        limit_ms,
    }
}

/// The clocks of the peers that are active now, by peer.
async fn offsets(state: &AppState) -> Vec<(String, ClockOffset)> {
    let mut offsets = Vec::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        if peer_state.lock().await.liveness.is_dormant() {
            continue;
        }
        if let Some(offset) = state.transport.clock_offset(&peer) {
            offsets.push((peer, offset));
        }
    }
    offsets
}

/// Ours against the clocks of the peers that are active now.
pub(crate) async fn measure(state: &AppState) -> Skew {
    let offsets: Vec<ClockOffset> = offsets(state).await.into_iter().map(|(_, o)| o).collect();
    of(&offsets, state.clock_skew_warn)
}

/// What the checks remember between them (`AppState::clock_skew`).
#[derive(Debug, Default)]
pub(crate) struct Watch {
    /// When the log last said this node's clock is off, while it is.
    warned: Option<Instant>,
    /// Each peer's clock at the last check.
    last: HashMap<String, ClockOffset>,
}

/// Clocks set anew between two checks, by more than the limit for sure.
#[derive(Debug, Default, PartialEq, Eq)]
struct Jumps {
    /// Ours, by how much (forward if positive): most peers' clocks moved alike.
    ours: Option<i64>,
    /// Otherwise the peers' whose clocks moved, each by how much.
    peers: Vec<(String, i64)>,
}

/// Which clocks jumped from `last` to `now` (each peer's ahead of ours), by more than
/// `limit_ms` beyond both bounds. None with a limit of 0.
fn jumps(
    last: &HashMap<String, ClockOffset>,
    now: &[(String, ClockOffset)],
    limit_ms: u64,
) -> Jumps {
    if limit_ms == 0 {
        return Jumps::default();
    }
    let moves: Vec<(&String, i64, bool)> = now
        .iter()
        .filter_map(|(peer, offset)| {
            let before = last.get(peer)?;
            let moved = offset.offset_ms - before.offset_ms;
            let bound = offset
                .within_ms
                .saturating_add(before.within_ms)
                .saturating_add(limit_ms);
            Some((peer, moved, moved.unsigned_abs() > bound))
        })
        .collect();
    let forward = moves
        .iter()
        .filter(|(_, moved, sure)| *sure && *moved > 0)
        .count();
    let back = moves
        .iter()
        .filter(|(_, moved, sure)| *sure && *moved < 0)
        .count();
    if forward * 2 > moves.len() || back * 2 > moves.len() {
        // Ours moved the other way, by the median.
        let mut ours: Vec<i64> = moves.iter().map(|(_, moved, _)| -moved).collect();
        ours.sort_unstable();
        return Jumps {
            ours: ours.get(ours.len() / 2).copied(),
            peers: Vec::new(),
        };
    }
    Jumps {
        ours: None,
        peers: moves
            .into_iter()
            .filter(|(_, _, sure)| *sure)
            .map(|(peer, moved, _)| (peer.clone(), moved))
            .collect(),
    }
}

/// Notes how far this node's clock is ahead of its peers' (`AppState::clock_ahead_ms`),
/// and says in the log when it is off, and when it agrees again. From the peer worker,
/// every few seconds.
pub(crate) async fn check(state: &AppState) {
    let offsets = offsets(state).await;
    let skew = of(
        &offsets.iter().map(|(_, o)| *o).collect::<Vec<_>>(),
        state.clock_skew_warn,
    );
    if let Some(offset_ms) = skew.offset_ms {
        state
            .clock_ahead_ms
            .store(offset_ms, std::sync::atomic::Ordering::Relaxed);
    }
    let now = Instant::now();
    let mut watch = state.clock_skew.lock_or_recover();
    let jumped = jumps(&watch.last, &offsets, skew.limit_ms);
    if let Some(ms) = jumped.ours {
        warn!(
            jump_ms = ms,
            peers = offsets.len(),
            "This node's clock jumped (set anew)"
        );
    }
    for (peer, ms) in &jumped.peers {
        info!(peer = %peer, jump_ms = ms, "A peer's clock jumped (set anew)");
    }
    watch.last = offsets.into_iter().collect();
    let warned = &mut watch.warned;
    if skew.off {
        if warned.is_none_or(|at| now.saturating_duration_since(at) >= WARN_EVERY) {
            warn!(
                offset_ms = skew.offset_ms,
                peers = skew.peers,
                behind = skew.behind,
                ahead = skew.ahead,
                limit_ms = state.clock_skew_warn.as_millis() as u64,
                "This node's clock differs from most of its peers'; check its time synchronisation (NTP)"
            );
            *warned = Some(now);
        }
    } else if warned.take().is_some() {
        info!(
            offset_ms = skew.offset_ms,
            peers = skew.peers,
            "This node's clock agrees with its peers' again"
        );
    }
}

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

    fn offset(offset_ms: i64, within_ms: u64) -> ClockOffset {
        ClockOffset {
            offset_ms,
            within_ms,
        }
    }

    /// Off only beyond the limit and the bound, and only against most peers.
    #[test]
    fn off_against_most_peers_for_sure() {
        let limit = Duration::from_secs(2);
        // Every peer 5 s ahead: ours is 5 s behind.
        let behind = of(
            &[offset(5_000, 100), offset(5_100, 50), offset(4_900, 300)],
            limit,
        );
        assert_eq!(behind.offset_ms, Some(-5_000));
        assert_eq!((behind.ahead, behind.behind, behind.off), (3, 0, true));
        // One peer off: that peer's clock, not ours.
        let one = of(
            &[offset(5_000, 100), offset(10, 50), offset(-20, 30)],
            limit,
        );
        assert_eq!((one.ahead, one.off), (1, false));
        // Within the bound: not for sure.
        let unsure = of(&[offset(2_500, 600), offset(2_500, 600)], limit);
        assert!(!unsure.off);
        // A limit of 0: never.
        assert!(!of(&[offset(60_000, 10)], Duration::ZERO).off);
        // No peers: nothing known.
        let none = of(&[], limit);
        assert_eq!((none.peers, none.offset_ms, none.off), (0, None, false));
    }

    /// Most peers' clocks moved alike: ours jumped; a few moved: theirs.
    #[test]
    fn jumps_are_ours_when_most_moved_alike() {
        let peers = |offsets: &[i64]| -> Vec<(String, ClockOffset)> {
            offsets
                .iter()
                .enumerate()
                .map(|(i, &ms)| (format!("p{i}"), offset(ms, 50)))
                .collect()
        };
        let last: HashMap<_, _> = peers(&[0, 10, -20]).into_iter().collect();
        // Ours went 30 s forward: theirs are 30 s behind it now.
        let ours = jumps(&last, &peers(&[-30_000, -29_990, -30_020]), 2_000);
        assert_eq!(ours.ours, Some(30_000));
        assert!(ours.peers.is_empty());
        // One peer's went 10 s back.
        let one = jumps(&last, &peers(&[-10_000, 10, -20]), 2_000);
        assert_eq!(one.ours, None);
        assert_eq!(one.peers, vec![("p0".to_owned(), -10_000)]);
        // Within the bounds and the limit: nothing.
        assert_eq!(
            jumps(&last, &peers(&[1_000, 10, -20]), 2_000),
            Jumps::default()
        );
        // A limit of 0: never.
        assert_eq!(
            jumps(&last, &peers(&[-30_000, -30_000, -30_000]), 0),
            Jumps::default()
        );
    }
}