//! Whom each neighbour reaches: a node's 2-hop neighbourhood, as measured, not as the
//! member list says.
//!
//! A node knows its own links, and from them whom it reaches: the members it has an
//! active session with (handshake done, neither dormant nor unreachable) that answer
//! (`PeerState::answer_overdue`: none left a keepalive or sync message unanswered for
//! `Timing::unreached_after`, asked again halfway, with nothing else heard from it; a
//! member gone shows within seconds, not after `dormant_after`). It tells its
//! neighbours so in its STATUS (`Status::reaches`, each member by its pair key,
//! `topology::pair_key`, 8 bytes instead of a node ID, with its round trip to each as a
//! bucket: `Status::round_trips`, `rtt::bucket`): in every keepalive STATUS and
//! STATUS_REPLY, and soon to every neighbour when it changed (the peer worker looks
//! every 5 s, and at once when a member we reach stops answering), but not in
//! announcements of heads, which go often. So every node knows whom its neighbours
//! reach, as of their latest STATUS, for as long as the session lasts (a
//! neighbour that goes dormant, or starts anew, says it again). Not for a set time:
//! while a pair exchanges changes it sends no keepalives, and so no STATUS, for as
//! long as the writes go on. Nothing of it goes into the document (but where a link
//! stays down for a while: `swarm::working_links`), and nobody knows more than two hops
//! out.
//!
//! What it is for: passing changes on around links that don't work, and to exactly the
//! members the sender doesn't reach; a member a neighbour reaches is one worth trying,
//! and a way to it. Not for choosing multipoint relays (as OLSR does): over the member
//! list's graph, which is close to random, neighbourhoods hardly overlap, and nearly
//! every neighbour would be one (13.7 of 13.7 at 250 members and more), saving next to
//! nothing but in a full mesh.

use std::collections::{BTreeSet, HashMap};

use tokio::time::Instant;
use tracing::debug;

use crate::locks::MutexExt;
use crate::state::{AppState, HelloState, Standing};
use crate::swarm::peer::announce_heads;
use crate::swarm::{members, rtt, topology};

/// The most members a node may say it reaches (`said`): more than it could hold
/// sessions with.
pub(crate) const MAX_REACHES: usize = 4096;

/// A reach set as a neighbour said it (`Status::reaches`), with its round trips if
/// they match it: sorted by pair key (as we send ours: `measure`), so it can be looked
/// up by halving (`swarm::tree`); None if it is larger than `MAX_REACHES`.
pub(crate) fn said(
    reaches: &[u64],
    round_trips: Option<&[u8]>,
) -> Option<(Vec<u64>, Option<Vec<u8>>)> {
    if reaches.len() > MAX_REACHES {
        return None;
    }
    let round_trips = round_trips.filter(|round_trips| round_trips.len() == reaches.len());
    if reaches.windows(2).all(|pair| pair[0] < pair[1]) {
        return Some((reaches.to_vec(), round_trips.map(<[u8]>::to_vec)));
    }
    let mut pairs: Vec<(u64, Option<u8>)> = reaches
        .iter()
        .enumerate()
        .map(|(i, key)| (*key, round_trips.map(|round_trips| round_trips[i])))
        .collect();
    pairs.sort_unstable_by_key(|(key, _)| *key);
    pairs.dedup_by_key(|(key, _)| *key);
    let round_trips = round_trips.map(|_| pairs.iter().filter_map(|(_, rtt)| *rtt).collect());
    Some((pairs.into_iter().map(|(key, _)| key).collect(), round_trips))
}

/// Whom we reach, as last told our neighbours, and our round trip to each.
#[derive(Default)]
pub(crate) struct Reach {
    published: Option<Vec<u64>>,
    /// One per member in `published`, in its order (`rtt::bucket`).
    round_trips: Vec<u8>,
}

/// The pair keys of the members we reach now, sorted: those with an active session
/// that answer; and our round trip to each, as a bucket, held near a bound to the one
/// told before (`before`, by pair key: `rtt::bucket`).
async fn measure(state: &AppState, before: &HashMap<u64, u8>) -> (Vec<u64>, Vec<u8>) {
    let now = Instant::now();
    let mut reached: Vec<(u64, String)> = reached_members(state, now)
        .await
        .into_iter()
        .map(|peer| (topology::pair_key(&peer), peer))
        .collect();
    reached.sort_unstable();
    let rtts = state.lock_rtt();
    let round_trips = reached
        .iter()
        .map(|(key, peer)| {
            let rtt = rtts.effective(peer, now, state.timing.rtt_fresh_for());
            rtt::bucket(rtt, before.get(key).copied())
        })
        .collect();
    (
        reached.into_iter().map(|(key, _)| key).collect(),
        round_trips,
    )
}

/// The members we reach now: those with an active session (handshake done, neither
/// dormant nor unreachable) that answer.
pub(crate) async fn reached_members(state: &AppState, now: Instant) -> BTreeSet<String> {
    let mut reached = BTreeSet::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        let peer_state = peer_state.lock().await;
        if peer_state.hello_state == HelloState::Established
            && !peer_state.liveness.is_dormant()
            && !peer_state.is_unreachable()
            && !peer_state.answer_overdue(now)
            && matches!(peer_state.standing, Standing::Member { .. })
        {
            reached.insert(peer);
        }
    }
    reached
}

/// Whom we reach, as we tell it (`Status::reaches`); empty until first measured.
pub(crate) fn published(state: &AppState) -> Vec<u64> {
    published_with_round_trips(state).0
}

/// Whom we reach and our round trip to each, as we tell it (`Status::reaches`,
/// `Status::round_trips`); empty until first measured.
pub(crate) fn published_with_round_trips(state: &AppState) -> (Vec<u64>, Vec<u8>) {
    let reach = state.reach.lock_or_recover();
    match &reach.published {
        Some(keys) => (keys.clone(), reach.round_trips.clone()),
        None => (Vec::new(), Vec::new()),
    }
}

/// With the peer worker's periodic pass (every 5 s), and at once when a member we said
/// we reach stops answering (`reached_stops_answering`): whom we reach now; if it changed,
/// our neighbours are told at once (a STATUS each, unanswered unless they lack some of
/// our heads). A change that is undone within a pass isn't told.
pub(crate) async fn check(state: &AppState) {
    let before: HashMap<u64, u8> = {
        let reach = state.reach.lock_or_recover();
        reach
            .published
            .iter()
            .flatten()
            .copied()
            .zip(reach.round_trips.iter().copied())
            .collect()
    };
    let (now, round_trips) = measure(state, &before).await;
    {
        let mut reach = state.reach.lock_or_recover();
        if reach.published.as_ref() == Some(&now) && reach.round_trips == round_trips {
            return;
        }
        reach.published = Some(now.clone());
        reach.round_trips = round_trips;
    }
    debug!(
        reached = now.len(),
        "Whom we reach changed; telling our neighbours"
    );
    for (peer, peer_state) in state.peers_snapshot().await {
        let established = peer_state.lock().await.hello_state == HelloState::Established;
        if !established || !topology::is_neighbour(state, &peer) {
            continue;
        }
        if let Err(error) = tell(state, &peer).await {
            debug!(peer = %peer, error = %format!("{error:#}"), "Telling whom we reach failed");
        }
    }
}

/// Whether `peer`, whose answer is overdue (`PeerState::answer_overdue`), is one we
/// said we reach: then the peer worker checks at once (`check`).
pub(crate) fn reached_stops_answering(state: &AppState, peer: &str) -> bool {
    state
        .reach
        .lock_or_recover()
        .published
        .as_ref()
        .is_some_and(|keys| keys.contains(&topology::pair_key(peer)))
}

/// A STATUS to `peer` with whom we reach: as an announcement (unanswered unless it lacks
/// some of our heads), but with `reaches`.
async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> {
    announce_heads(state, peer, true).await
}

/// The members a peer reaches, as it last said in this session (`said`, kept with its
/// state, cleared when the session ends: `PeerState::reset_sync`); by pair key.
pub(crate) fn of(said: Option<&(Instant, Vec<u64>)>) -> Option<Vec<u64>> {
    said.map(|(_, reaches)| reaches.clone())
}

/// The members of our member list by pair key, to name what neighbours say they reach.
pub(crate) async fn names(state: &AppState) -> HashMap<u64, String> {
    let replica = state.doc.read().await;
    members::members(&replica.doc)
        .into_iter()
        .map(|member| (topology::pair_key(&member), member))
        .collect()
}

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

    /// A reach set comes sorted, its round trips along; too large, it is ignored.
    #[test]
    fn reach_sets_are_sorted_and_bounded() {
        assert_eq!(
            said(&[1, 3, 5], Some(&[9, 8, 7])),
            Some((vec![1, 3, 5], Some(vec![9, 8, 7])))
        );
        assert_eq!(
            said(&[5, 1, 3, 1], Some(&[7, 9, 8, 9])),
            Some((vec![1, 3, 5], Some(vec![9, 8, 7])))
        );
        assert_eq!(said(&[2, 1], Some(&[1])), Some((vec![1, 2], None)));
        assert_eq!(said(&vec![0; MAX_REACHES + 1], None), None);
    }
}