//! This node's links, as it measures them, for the dashboard's map (`own_links`): its
//! own in the state, and to any page that subscribes to them
//! (`GET /admin/dashboard/links/own/events`, also from another member's page).

use crate::state::AppState;
use crate::swarm::topology;
use crate::transport::paths::{self, ClockOffset};
use crate::transport::{Connection, Destination};

/// Links listed at most (a node has about `--neighbours`, more if others prefer it).
const MAX_LINKS: usize = 256;
/// Other transports in a link at most.
const MAX_OTHERS: usize = 8;
/// One of this node's links.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Link {
    /// The peer's node ID.
    pub(crate) peer: String,
    /// The transport its messages go over now (the scheme of the path in use), if any.
    pub(crate) transport: Option<String>,
    /// Only through a relay (libp2p).
    pub(crate) relayed: bool,
    /// The path in use is down: none to the peer answers (`paths`), whatever its
    /// connection was.
    pub(crate) down: bool,
    /// The round trip on that path, in microseconds, if measured.
    pub(crate) rtt_us: Option<u32>,
    /// Silent for long (`peers.md`, dormancy).
    pub(crate) dormant: bool,
    /// The peer's other transports (`MAX_OTHERS` at most): where else it proved
    /// itself, those paths up.
    pub(crate) others: Vec<OtherPath>,
    /// How far the peer's clock is ahead of this node's (`paths::ClockOffset`), if
    /// measured: with other members', the page shows whose clock is off.
    pub(crate) clock: Option<ClockOffset>,
}

/// A transport of a link besides the one in use.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct OtherPath {
    /// Its scheme (`mycelium`, `libp2p`, `tor`).
    pub(crate) transport: String,
    /// The round trip of its fastest path, in microseconds, if any is measured.
    pub(crate) rtt_us: Option<u32>,
    /// Only through a relay (libp2p): no direct connection at any of its paths.
    pub(crate) relayed: bool,
}

/// This node's links: its neighbours, and the peers it holds dormant, as it measures
/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its
/// link stays listed, dormant, until it is heard from again: left out, the link would
/// show only as that peer last reported it, from before.
pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
    let mut links = Vec::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        let dormant = peer_state.lock().await.liveness.is_dormant();
        if !dormant && !topology::is_neighbour(state, &peer) {
            continue;
        }
        let destination = state.transport.destination(&peer);
        let transport = destination
            .as_ref()
            .and_then(|destination| scheme_of(&destination.address));
        links.push(Link {
            others: others(destination.as_ref(), transport.as_deref()),
            transport,
            relayed: destination
                .as_ref()
                .is_some_and(|destination| destination.connection == Some(Connection::Relayed)),
            // Ranked up before down: the first is down only if all are.
            down: destination
                .as_ref()
                .and_then(|destination| destination.paths.first())
                .is_some_and(|path| path.down),
            // The path in use's (`Transport::rtt`, the first of `destination.paths`).
            rtt_us: state
                .transport
                .rtt(&peer)
                .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
            dormant,
            clock: state.transport.clock_offset(&peer),
            peer,
        });
    }
    links.sort_by(|a, b| a.peer.cmp(&b.peer));
    links.truncate(MAX_LINKS);
    links
}

/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it
/// has one.
fn scheme_of(address: &str) -> Option<String> {
    Some(paths::scheme_of(address))
        .filter(|scheme| !scheme.is_empty())
        .map(str::to_owned)
}

/// The transports besides `used` the peer proved itself at, with a path up: each with
/// the round trip of its fastest path, if any is measured, and whether all of them go
/// through a relay.
fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> {
    let mut others: Vec<OtherPath> = Vec::new();
    for path in destination.map_or(&[][..], |destination| &destination.paths[..]) {
        let Some(transport) = scheme_of(&path.address) else {
            continue;
        };
        if path.down || used == Some(transport.as_str()) {
            continue;
        }
        let rtt_us = path
            .rtt_ms
            .filter(|ms| ms.is_finite() && *ms >= 0.0)
            .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32);
        let relayed = path.connection == Some(Connection::Relayed);
        match others.iter_mut().find(|other| other.transport == transport) {
            Some(other) => {
                other.rtt_us = match (other.rtt_us, rtt_us) {
                    (Some(a), Some(b)) => Some(a.min(b)),
                    (a, b) => a.or(b),
                };
                other.relayed &= relayed;
            }
            None => others.push(OtherPath {
                transport,
                rtt_us,
                relayed,
            }),
        }
    }
    others.sort_by(|a, b| a.transport.cmp(&b.transport));
    others.truncate(MAX_OTHERS);
    others
}

/// Whether `new` says something `old` didn't: another peer, transport, relay or
/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock
/// measured or moved beyond both bounds. Round trips and clock samples wobble; small
/// moves aren't news.
pub(crate) fn changed(old: &[Link], new: &[Link]) -> bool {
    let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) {
        (Some(a), Some(b)) => {
            a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms)
        }
        (a, b) => a.is_some() != b.is_some(),
    };
    let moved = |a: Option<u32>, b: Option<u32>| match (a, b) {
        (Some(a), Some(b)) => {
            let by = a.abs_diff(b);
            by > 1000 && by > a.max(b) / 5
        }
        (a, b) => a.is_some() != b.is_some(),
    };
    old.len() != new.len()
        || old.iter().zip(new).any(|(a, b)| {
            a.peer != b.peer
                || a.transport != b.transport
                || a.relayed != b.relayed
                || a.down != b.down
                || a.dormant != b.dormant
                || moved(a.rtt_us, b.rtt_us)
                || clock_moved(a.clock, b.clock)
                || a.others.len() != b.others.len()
                || a.others.iter().zip(&b.others).any(|(x, y)| {
                    x.transport != y.transport
                        || x.relayed != y.relayed
                        || moved(x.rtt_us, y.rtt_us)
                })
        })
}

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

    fn link(peer: &str, rtt_us: Option<u32>) -> Link {
        Link {
            peer: peer.into(),
            transport: Some("libp2p".into()),
            relayed: false,
            down: false,
            rtt_us,
            dormant: false,
            others: Vec::new(),
            clock: None,
        }
    }

    #[test]
    fn only_news_is_a_change() {
        let old = [link("a", Some(10_000))];
        assert!(!changed(&old, &[link("a", Some(11_000))]));
        assert!(changed(&old, &[link("a", Some(13_000))]));
        assert!(changed(&old, &[link("a", None)]));
        assert!(changed(&old, &[link("b", Some(10_000))]));
        assert!(changed(&old, &[]));
        let mut dormant = link("a", Some(10_000));
        dormant.dormant = true;
        assert!(changed(&old, &[dormant]));
        let mut down = link("a", Some(10_000));
        down.down = true;
        assert!(changed(&old, &[down]));
        // Small round trips: a millisecond's wobble isn't news.
        assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))]));
        // A clock: news when measured, and when it moved beyond both bounds.
        let clocked = |offset_ms| Link {
            clock: Some(ClockOffset {
                offset_ms,
                within_ms: 50,
            }),
            ..link("a", Some(10_000))
        };
        assert!(changed(&old, &[clocked(0)]));
        assert!(!changed(&[clocked(0)], &[clocked(80)]));
        assert!(changed(&[clocked(0)], &[clocked(5_000)]));
    }
}