//! Round-trip times to peers. Mostly the paths layer's (`transport::paths`): the
//! smoothed round-trip time of the path a peer's messages go over, measured by PINGs
//! (keepalives, and the paths layer's own), taken whenever a PONG from the peer arrives
//! (`measured`). Before a peer's path is measured (while joining: no keepalives yet),
//! HELLO → HELLO_REPLY gives one, smoothed (exponential moving average, weight 1/4);
//! a HELLO_REPLY only counts when exactly one HELLO was outstanding: with two, we can't
//! tell which one the answer belongs to (an answer to the older one would look too
//! fast).
//!
//! Without a recent value (nodes move; an old value says little) it is unknown. A lost
//! PING or HELLO is no sample: a peer that stops answering goes dormant, and is given up
//! as a catch-up source, by itself. Silence alone doesn't count: we don't ask
//! non-neighbours anything, so their quiet is ours.

use std::collections::HashMap;
use std::time::Duration;

use tokio::time::Instant;

#[derive(Debug, Default)]
pub(crate) struct RttTable {
    hellos: HashMap<String, Outstanding>,
    /// From the paths layer, as last measured.
    from_paths: HashMap<String, Sample>,
    /// From HELLOs, smoothed.
    from_hellos: HashMap<String, Sample>,
}

#[derive(Debug, Clone, Copy)]
struct Sample {
    rtt: Duration,
    at: Instant,
}

#[derive(Debug, Clone, Copy)]
struct Outstanding {
    sent: Instant,
    count: u32,
}

impl RttTable {
    /// The paths layer's round-trip time to `peer` (`Transport::rtt`), as a PONG from
    /// it arrived.
    pub(crate) fn measured(&mut self, peer: &str, rtt: Duration, now: Instant) {
        self.from_paths
            .insert(peer.to_owned(), Sample { rtt, at: now });
    }

    pub(crate) fn hello_sent(&mut self, peer: &str, now: Instant) {
        self.hellos
            .entry(peer.to_owned())
            .and_modify(|outstanding| outstanding.count += 1)
            .or_insert(Outstanding {
                sent: now,
                count: 1,
            });
    }

    pub(crate) fn hello_reply_received(&mut self, peer: &str, now: Instant) {
        let Some(outstanding) = self.hellos.remove(peer) else {
            return;
        };
        if outstanding.count != 1 {
            return;
        }
        let sample = now.saturating_duration_since(outstanding.sent);
        let rtt = match self.from_hellos.get(peer) {
            Some(old) => (old.rtt * 3 + sample) / 4,
            None => sample,
        };
        self.from_hellos
            .insert(peer.to_owned(), Sample { rtt, at: now });
    }

    /// The round-trip time: the paths layer's if measured within `fresh_for`, else
    /// HELLOs' if sampled within it; None without either.
    pub(crate) fn effective(
        &self,
        peer: &str,
        now: Instant,
        fresh_for: Duration,
    ) -> Option<Duration> {
        let fresh = |sample: &&Sample| now.saturating_duration_since(sample.at) < fresh_for;
        self.from_paths
            .get(peer)
            .filter(fresh)
            .or_else(|| self.from_hellos.get(peer).filter(fresh))
            .map(|sample| sample.rtt)
    }

    /// The round-trip time of every peer that has one.
    pub(crate) fn all_effective(
        &self,
        now: Instant,
        fresh_for: Duration,
    ) -> impl Iterator<Item = (&str, Duration)> {
        let mut peers: Vec<&String> = self
            .from_paths
            .keys()
            .chain(self.from_hellos.keys())
            .collect();
        peers.sort_unstable();
        peers.dedup();
        peers.into_iter().filter_map(move |peer| {
            self.effective(peer, now, fresh_for)
                .map(|rtt| (peer.as_str(), rtt))
        })
    }
}

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

    const FRESH: Duration = Duration::from_secs(300);

    #[test]
    fn hellos_sample_only_unambiguous_answers_and_smooth() {
        let mut table = RttTable::default();
        let start = Instant::now();
        let at = |ms| start + Duration::from_millis(ms);
        table.hello_sent("a", start);
        table.hello_reply_received("a", at(100));
        assert_eq!(
            table.effective("a", at(100), FRESH),
            Some(Duration::from_millis(100))
        );
        table.hello_sent("a", at(100));
        table.hello_reply_received("a", at(300));
        assert_eq!(
            table.effective("a", at(300), FRESH),
            Some(Duration::from_millis(125))
        );
        // Two HELLOs out: the HELLO_REPLY can't be attributed.
        table.hello_sent("a", at(300));
        table.hello_sent("a", at(400));
        table.hello_reply_received("a", at(410));
        assert_eq!(
            table.effective("a", at(410), FRESH),
            Some(Duration::from_millis(125))
        );
        // An unsolicited HELLO_REPLY changes nothing.
        table.hello_reply_received("b", start);
        assert_eq!(table.effective("b", start, FRESH), None);
    }

    #[test]
    fn the_path_wins_nothing_waits_and_old_values_expire() {
        let mut table = RttTable::default();
        let start = Instant::now();
        table.hello_sent("a", start);
        table.hello_reply_received("a", start + Duration::from_millis(80));
        table.measured("a", Duration::from_millis(20), start);
        assert_eq!(
            table.effective("a", start, FRESH),
            Some(Duration::from_millis(20))
        );
        // A HELLO unanswered for long raises nothing.
        table.hello_sent("a", start);
        assert_eq!(
            table.effective("a", start + Duration::from_secs(40), FRESH),
            Some(Duration::from_millis(20))
        );
        assert_eq!(
            table.all_effective(start, FRESH).collect::<Vec<_>>(),
            [("a", Duration::from_millis(20))]
        );
        // No value within the freshness window: unknown.
        let stale = start + FRESH + Duration::from_secs(1);
        assert_eq!(table.effective("a", stale, FRESH), None);
    }
}