//! 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). 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: each warns.

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,
    }
}

/// Ours against the clocks of the peers that are active now.
pub(crate) async fn measure(state: &AppState) -> Skew {
    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(offset);
        }
    }
    of(&offsets, state.clock_skew_warn)
}

/// 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 skew = measure(state).await;
    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 warned = state.clock_skew_warned.lock_or_recover();
    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));
    }
}