//! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a
//! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its
//! neighbours and the peers it holds dormant (`own_links`), each with the transport its
//! messages go over, whether only through a relay, the round trip and whether it is
//! dormant: as the member measures them, which no other node can.
//!
//! Every node answers, whatever its build: one with a dashboard asks any member, which
//! may have none. Only members (in the member list) are answered, as the links name
//! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing
//! is written to the document: asked for while someone looks, forgotten after.
//!
//! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again
//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a
//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
//! subscribes (round trips change without telling), until the subscription runs out
//! (the asker renews it with its next LINKS, so one that is gone stops it by not
//! renewing) or a LINKS with 0 ends it.
//!
//! A member the asker can't reach (on none of its transports: one on Tor alone, asked
//! from one on libp2p alone) is asked through a peer that can (`relay_for`): a LINKS
//! naming the members (`of`) asks the receiver to ask them on the sender's behalf, and
//! to pass their links on (`relayed`). The receiver asks each itself, or through a peer
//! of its own while hops are left (`hops`), for as long as the sender asked, and holds
//! what arrives a while to send it together: the window (`batch_ms`) is the asker's for
//! the whole way (at most `MAX_BATCH`), shared out among the hops that may be left, so
//! the links reach the asker within it however long the chain (every hop holds
//! `MIN_BATCH` at least). Such a
//! subscription isn't ended by a message: it runs out, at every hop, unless the asker
//! renews it, so one that is gone (or no longer looks) stops it within a lease.

use std::collections::{HashMap, HashSet};
use std::sync::Mutex;
use std::time::Duration;

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

use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::{members, topology};
use crate::transport::paths::{self, ClockOffset};
use crate::transport::{
    Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, RelayedLinks,
};

/// Links one reply lists at most (a member has about `--neighbours`, more if others
/// prefer it).
pub(crate) const MAX_LINKS: usize = 256;
/// Other transports in a link at most.
pub(crate) const MAX_OTHERS: usize = 8;
/// A member answers each other one at most this often.
const ANSWER_EVERY: Duration = Duration::from_secs(5);
/// Replies kept from members at most (one each), and requests out.
const MAX_KEPT: usize = 4096;
/// Subscriptions a member keeps at most, and how long one lasts at most.
const MAX_SUBSCRIBERS: usize = 64;
const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300);
/// How often a member with subscribers looks whether its links changed.
const WATCH_EVERY: Duration = Duration::from_secs(2);
/// A reply to a LINKS that doesn't subscribe is taken for this long.
const REPLY_WITHIN: Duration = Duration::from_secs(30);
/// Members a relayed LINKS names at most, and that one LINKS_REPLY passes on at most.
const MAX_RELAYED: usize = 64;
/// Bytes of relayed links one LINKS_REPLY carries at most, as encoded (one member's
/// alone may be more): what a member reports is usually small (about 12 links of 150
/// bytes), but may be up to `MAX_LINKS` links, and 64 such would make a message of
/// megabytes.
const MAX_RELAYED_BYTES: usize = 256 * 1024;
/// How long relayed links are held to be sent together, along the whole way: the
/// asker's window (at most `MAX_BATCH`), each hop holding at least `MIN_BATCH`.
pub(crate) const DEFAULT_BATCH: Duration = Duration::from_millis(250);
const MIN_BATCH: Duration = Duration::from_millis(10);
const MAX_BATCH: Duration = Duration::from_secs(5);
/// Members a peer may have us ask for it, at most: as many at once (`RELAY_BURST`), and
/// on average this many per second (`RELAY_RATE`): a dashboard asking every member
/// (`MAX_ASKED` there, 512) every 30 s needs less. The rest of a request is dropped.
const RELAY_BURST: f64 = 512.0;
const RELAY_RATE: f64 = 32.0;
/// Nodes a relayed LINKS goes through at most, the first included.
const MAX_HOPS: u8 = 3;

/// The requests out and the replies in, and when each asker was last answered.
#[derive(Default)]
pub(crate) struct Links {
    inner: Mutex<Inner>,
    /// Someone subscribed (`answer`): `watch` looks again.
    subscribed: tokio::sync::Notify,
}

#[derive(Default)]
struct Inner {
    /// Our LINKS out, by member: its nonce, and until when replies to it are taken (its
    /// subscription's end, or `REPLY_WITHIN`).
    asked: HashMap<String, (u64, Instant)>,
    /// The latest LINKS_REPLY of each member: when, and its links.
    replies: HashMap<String, (Instant, Vec<Link>)>,
    /// When we last answered each member.
    answered: HashMap<String, Instant>,
    /// Who subscribed to our links: the nonce to answer with, and until when.
    subscribers: HashMap<String, (u64, Instant)>,
    /// The links subscribers were last sent (`watch`).
    published: Option<Vec<Link>>,
    /// Our relayed LINKS out, by the peer asked to relay: its nonce, until when replies
    /// to it are taken, and the members asked for through it.
    via: HashMap<String, (u64, Instant, HashSet<String>)>,
    /// What we ask on others' behalf, by member asked for: who asked (`Relaying`).
    relaying: HashMap<String, HashMap<String, Relaying>>,
    /// Relayed links due to an asker, by asker: the members whose links changed.
    due: HashMap<String, HashSet<String>>,
    /// How many members each peer may still have us ask for it (`RELAY_BURST`,
    /// `RELAY_RATE`), as of when.
    relay_budget: HashMap<String, (f64, Instant)>,
}

/// One asker's relayed subscription to a member.
#[derive(Clone, Copy)]
struct Relaying {
    /// The nonce to answer with.
    nonce: u64,
    /// Until when, if still wanted.
    until: Instant,
    /// How long its links are held to be sent with others.
    window: Duration,
    /// Asked for once (no subscription): ends with the first links passed on.
    once: bool,
}

impl Links {
    /// The latest links each member replied with, and how long ago.
    #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
    pub(crate) fn replies(&self) -> Vec<(String, Duration, Vec<Link>)> {
        let now = Instant::now();
        self.inner
            .lock_or_recover()
            .replies
            .iter()
            .map(|(member, (at, links))| {
                (
                    member.clone(),
                    now.saturating_duration_since(*at),
                    links.clone(),
                )
            })
            .collect()
    }

    /// Who is subscribed to our links (not run out).
    #[cfg(test)]
    pub(crate) fn subscribers(&self) -> Vec<String> {
        let now = Instant::now();
        self.inner
            .lock_or_recover()
            .subscribers
            .iter()
            .filter(|(_, (_, until))| now < *until)
            .map(|(peer, _)| peer.clone())
            .collect()
    }

    /// The members we ask on others' behalf (relayed, not run out), and who asked.
    #[cfg(test)]
    pub(crate) fn relaying(&self) -> Vec<(String, Vec<String>)> {
        let now = Instant::now();
        self.inner
            .lock_or_recover()
            .relaying
            .iter()
            .map(|(member, askers)| {
                let askers: Vec<String> = askers
                    .iter()
                    .filter(|(_, relaying)| now < relaying.until)
                    .map(|(asker, _)| asker.clone())
                    .collect();
                (member.clone(), askers)
            })
            .filter(|(_, askers)| !askers.is_empty())
            .collect()
    }

    /// The members our subscriptions are with, and the peers relaying for us (not run
    /// out).
    #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
    pub(crate) fn subscribed(&self) -> Vec<String> {
        let now = Instant::now();
        let inner = self.inner.lock_or_recover();
        inner
            .asked
            .iter()
            .filter(|(_, (_, until))| now < *until)
            .map(|(member, _)| member.clone())
            .chain(
                inner
                    .via
                    .iter()
                    .filter(|(_, (_, until, _))| now < *until)
                    .map(|(relay, _)| relay.clone()),
            )
            .collect()
    }
}

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

/// Asks `member` for its links (a LINKS), and with `subscribe` for its changes for
/// that long; its replies land in `AppState::links`.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) {
    let nonce = {
        let now = Instant::now();
        let mut inner = state.links.inner.lock_or_recover();
        if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) {
            inner.asked.clear();
        }
        let until = now + subscribe.max(REPLY_WITHIN);
        // The same nonce while its replies are taken: a renewal doesn't drop a reply on
        // its way under the last one (in a chain of relays, every hop renews at once,
        // `relay`), as `ask_via` keeps one per relay.
        match inner.asked.get_mut(member) {
            Some((nonce, taken)) if now < *taken => {
                *taken = (*taken).max(until);
                *nonce
            }
            _ => {
                let nonce = crate::transport::paths::random_nonce();
                inner.asked.insert(member.to_owned(), (nonce, until));
                nonce
            }
        }
    };
    let (epoch, epoch_seq) = state.epoch_stamp();
    let request = ParsedMessage::Links(LinksRequest {
        epoch,
        epoch_seq,
        nonce,
        subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX),
        of: Vec::new(),
        batch_ms: 0,
        hops: 0,
    });
    if let Err(error) = state.transport.send_message(member, &request).await {
        debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
    }
}

/// Asks `members` for their links (`ask`): those we can reach ourselves, the others
/// through peers that can (`relay_for`, `ask_via`), with our window for what they pass
/// on (`DEFAULT_BATCH`).
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: Duration) {
    let (direct, through) = split(state, members, None).await;
    for member in direct {
        let state = state.clone();
        tokio::spawn(async move { ask(&state, &member, subscribe).await });
    }
    for (relay, members) in through {
        ask_via(
            state,
            &relay,
            members,
            subscribe,
            DEFAULT_BATCH,
            MAX_HOPS - 1,
        )
        .await;
    }
}

/// `members` we can reach ourselves, and the others by the peer to ask them through
/// (none for those no peer leads to); not `exclude` (who asked us).
async fn split(
    state: &AppState,
    members: Vec<String>,
    exclude: Option<&str>,
) -> (Vec<String>, HashMap<String, Vec<String>>) {
    let mut direct = Vec::new();
    let mut through: HashMap<String, Vec<String>> = HashMap::new();
    for member in members {
        if member == state.local_id() || Some(member.as_str()) == exclude {
            continue;
        }
        if state.transport.can_reach(&member) {
            direct.push(member);
        } else if let Some(relay) = relay_for(state, &member, exclude).await {
            through.entry(relay).or_default().push(member);
        }
    }
    (direct, through)
}

/// The peer to ask `member` through, which we can't reach: an active member we reach
/// that, by the member list, is on a transport `member` is on; else one on a transport
/// we aren't on (it may know one that is). Of several, the one with the lowest pair score
/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`.
async fn relay_for(state: &AppState, member: &str, exclude: Option<&str>) -> Option<String> {
    let theirs = topology::transports_of(state, member);
    let ours = topology::transports_of(state, state.local_id());
    let mut best: Option<(bool, u64, String)> = None;
    for (peer, peer_state) in state.peers_snapshot().await {
        if peer == member || Some(peer.as_str()) == exclude || !state.transport.can_reach(&peer) {
            continue;
        }
        {
            let peer_state = peer_state.lock().await;
            if peer_state.liveness.is_dormant()
                || !matches!(peer_state.standing, crate::state::Standing::Member { .. })
            {
                continue;
            }
        }
        let on = topology::transports_of(state, &peer);
        let shares = !on.is_disjoint(&theirs);
        if !shares && on.is_subset(&ours) {
            continue;
        }
        let score = topology::pair_score(&peer, member);
        // Sharing a transport with the member first, then the lowest score.
        let better = best.as_ref().is_none_or(|(best_shares, best_score, _)| {
            (shares, std::cmp::Reverse(score)) > (*best_shares, std::cmp::Reverse(*best_score))
        });
        if better {
            best = Some((shares, score, peer));
        }
    }
    best.map(|(_, _, peer)| peer)
}

/// Asks `relay` to ask `members` on our behalf (a LINKS naming them), with `subscribe`,
/// passing their links on within `window`, through `hops` more nodes at most: in LINKS
/// of `MAX_RELAYED` members at most. One nonce per relay while its replies are taken, so
/// its replies for every member we asked it for are taken.
async fn ask_via(
    state: &AppState,
    relay: &str,
    members: Vec<String>,
    subscribe: Duration,
    window: Duration,
    hops: u8,
) {
    let nonce = {
        let now = Instant::now();
        let mut inner = state.links.inner.lock_or_recover();
        if inner.via.len() >= MAX_KEPT && !inner.via.contains_key(relay) {
            inner.via.retain(|_, (_, until, _)| now < *until);
        }
        let until = now + subscribe.max(REPLY_WITHIN);
        let entry = inner.via.entry(relay.to_owned()).or_insert_with(|| {
            (
                crate::transport::paths::random_nonce(),
                until,
                HashSet::new(),
            )
        });
        if now >= entry.1 {
            entry.0 = crate::transport::paths::random_nonce();
            entry.2.clear();
        }
        entry.1 = entry.1.max(until);
        entry.2.extend(members.iter().cloned());
        entry.0
    };
    let (epoch, epoch_seq) = state.epoch_stamp();
    for members in members.chunks(MAX_RELAYED) {
        let request = ParsedMessage::Links(LinksRequest {
            epoch,
            epoch_seq,
            nonce,
            subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX),
            of: members.to_vec(),
            batch_ms: u32::try_from(window.as_millis()).unwrap_or(u32::MAX),
            hops,
        });
        if let Err(error) = state.transport.send_message(relay, &request).await {
            debug!(relay = %relay, error = %format!("{error:#}"), "Relayed LINKS failed");
        }
    }
}

/// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any
/// more. A peer relaying for us isn't told: what it asks for us runs out by itself.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) async fn unsubscribe(state: &AppState, member: &str) {
    let nonce = {
        let mut inner = state.links.inner.lock_or_recover();
        if inner.via.remove(member).is_some() && !inner.asked.contains_key(member) {
            return;
        }
        match inner.asked.remove(member) {
            Some((nonce, _)) => nonce,
            None => return,
        }
    };
    let (epoch, epoch_seq) = state.epoch_stamp();
    let request = ParsedMessage::Links(LinksRequest {
        epoch,
        epoch_seq,
        nonce,
        subscribe_secs: 0,
        of: Vec::new(),
        batch_ms: 0,
        hops: 0,
    });
    if let Err(error) = state.transport.send_message(member, &request).await {
        debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
    }
}

/// A LINKS from `peer`: answered if it's a member, and not answered lately.
pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) {
    let member = {
        let replica = state.doc.read().await;
        members::members(&replica.doc)
            .iter()
            .any(|member| member == peer)
    };
    if !member {
        debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member");
        return;
    }
    if !request.of.is_empty() {
        relay(state, peer, request).await;
        return;
    }
    {
        let now = Instant::now();
        let mut inner = state.links.inner.lock_or_recover();
        // A subscription is taken, renewed or ended even when the reply is due later.
        if request.subscribe_secs == 0 {
            inner.subscribers.remove(peer);
        } else if inner.subscribers.len() < MAX_SUBSCRIBERS || inner.subscribers.contains_key(peer)
        {
            let lasts =
                Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION);
            inner
                .subscribers
                .insert(peer.to_owned(), (request.nonce, now + lasts));
            state.links.subscribed.notify_one();
        }
        if inner
            .answered
            .get(peer)
            .is_some_and(|at| now.saturating_duration_since(*at) < ANSWER_EVERY)
        {
            return;
        }
        if inner.answered.len() >= MAX_KEPT {
            inner
                .answered
                .retain(|_, at| now.saturating_duration_since(*at) < ANSWER_EVERY);
        }
        inner.answered.insert(peer.to_owned(), now);
    }
    let (epoch, epoch_seq) = state.epoch_stamp();
    let reply = ParsedMessage::LinksReply(LinksReply {
        epoch,
        epoch_seq,
        nonce: request.nonce,
        links: own_links(state).await,
        relayed: Vec::new(),
    });
    if let Err(error) = state.transport.send_message(peer, &reply).await {
        debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed");
    }
}

/// A relayed LINKS from `asker` (`answer`): the members it names are asked on its
/// behalf, ourselves or through a peer of ours while hops are left, and what we know of
/// them already is passed on at once (within its window); the rest as it arrives.
async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
    let now = Instant::now();
    let lease = Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION);
    let once = lease.is_zero();
    // What is left of the asker's window, shared out among this hop and those that may
    // follow: ours is a share, the rest goes on.
    let left = Duration::from_millis(u64::from(request.batch_ms)).min(MAX_BATCH);
    let window = (left / (u32::from(request.hops) + 1)).max(MIN_BATCH);
    let rest = left.saturating_sub(window);
    let mut members: Vec<String> = request
        .of
        .into_iter()
        .filter(|member| member != asker && member != state.local_id())
        .take(MAX_RELAYED)
        .collect();
    let known: Vec<String> = {
        let mut inner = state.links.inner.lock_or_recover();
        // Within the asker's budget: each member costs one.
        if inner.relay_budget.len() >= MAX_KEPT && !inner.relay_budget.contains_key(asker) {
            inner.relay_budget.retain(|_, (budget, at)| {
                *budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE
                    < RELAY_BURST
            });
        }
        let (budget, at) = inner
            .relay_budget
            .entry(asker.to_owned())
            .or_insert((RELAY_BURST, now));
        *budget = (*budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE)
            .min(RELAY_BURST);
        *at = now;
        let allowed = (*budget as usize).min(members.len());
        *budget -= allowed as f64;
        if allowed < members.len() {
            debug!(asker = %asker, asked = members.len(), allowed, "Relayed LINKS over its budget; asking for part of it");
            members.truncate(allowed);
        }
        for member in &members {
            if inner.relaying.len() >= MAX_KEPT && !inner.relaying.contains_key(member) {
                inner
                    .relaying
                    .retain(|_, askers| askers.values().any(|relaying| now < relaying.until));
            }
            let askers = inner.relaying.entry(member.clone()).or_default();
            askers.retain(|_, relaying| now < relaying.until);
            if askers.len() >= MAX_SUBSCRIBERS && !askers.contains_key(asker) {
                continue;
            }
            askers.insert(
                asker.to_owned(),
                Relaying {
                    nonce: request.nonce,
                    // Asked once: for as long as its reply is awaited.
                    until: now + if once { REPLY_WITHIN } else { lease },
                    window,
                    once,
                },
            );
        }
        members
            .iter()
            .filter(|member| {
                inner
                    .replies
                    .get(*member)
                    .is_some_and(|(at, _)| now.saturating_duration_since(*at) < REPLY_WITHIN)
            })
            .cloned()
            .collect()
    };
    for member in known {
        forward(state, &member);
    }
    let (direct, through) = split(state, members, Some(asker)).await;
    for member in direct {
        // Asked once while we subscribe anyway (our own dashboard, or another asker's):
        // what comes is passed on, and a LINKS with 0 would end that subscription.
        let subscribed = once
            && state
                .links
                .inner
                .lock_or_recover()
                .asked
                .get(&member)
                .is_some_and(|(_, until)| now + REPLY_WITHIN < *until);
        if !subscribed {
            ask(state, &member, lease).await;
        }
    }
    if request.hops > 0 {
        for (next, members) in through {
            ask_via(state, &next, members, lease, rest, request.hops - 1).await;
        }
    }
}

/// `member`'s links changed (or arrived): due to those who asked us for them
/// (`relaying`), each sent with what else is due to it at the end of its window.
fn forward(state: &AppState, member: &str) {
    let now = Instant::now();
    let mut starting = Vec::new();
    {
        let mut inner = state.links.inner.lock_or_recover();
        let askers: Vec<(String, Duration)> = match inner.relaying.get_mut(member) {
            Some(askers) => {
                askers.retain(|_, relaying| now < relaying.until);
                askers
                    .iter()
                    .map(|(asker, relaying)| (asker.clone(), relaying.window))
                    .collect()
            }
            None => return,
        };
        for (asker, window) in askers {
            let due = inner.due.entry(asker.clone()).or_default();
            if due.is_empty() {
                starting.push((asker, window));
            }
            due.insert(member.to_owned());
        }
    }
    for (asker, window) in starting {
        let state = state.clone();
        tokio::spawn(async move {
            tokio::time::sleep(window).await;
            pass_on(&state, &asker).await;
        });
    }
}

/// What is due to `asker`: the links of the members it asked for, as we have them, in
/// as few LINKS_REPLYs as `MAX_RELAYED` allows (one per nonce it asked with).
async fn pass_on(state: &AppState, asker: &str) {
    let now = Instant::now();
    let mut by_nonce: HashMap<u64, Vec<RelayedLinks>> = HashMap::new();
    {
        let mut inner = state.links.inner.lock_or_recover();
        let due = inner.due.remove(asker).unwrap_or_default();
        for member in due {
            let Some(relaying) = inner
                .relaying
                .get(&member)
                .and_then(|askers| askers.get(asker))
                .copied()
                .filter(|relaying| now < relaying.until)
            else {
                continue;
            };
            let Some((at, links)) = inner.replies.get(&member) else {
                continue;
            };
            let age_ms =
                u32::try_from(now.saturating_duration_since(*at).as_millis()).unwrap_or(u32::MAX);
            by_nonce
                .entry(relaying.nonce)
                .or_default()
                .push(RelayedLinks {
                    member: member.clone(),
                    age_ms,
                    links: links.clone(),
                });
            if relaying.once
                && let Some(askers) = inner.relaying.get_mut(&member)
            {
                askers.remove(asker);
            }
        }
    }
    let (epoch, epoch_seq) = state.epoch_stamp();
    for (nonce, relayed) in by_nonce {
        for part in by_size(relayed) {
            let reply = ParsedMessage::LinksReply(LinksReply {
                epoch,
                epoch_seq,
                nonce,
                links: Vec::new(),
                relayed: part,
            });
            if let Err(error) = state.transport.send_message(asker, &reply).await {
                debug!(asker = %asker, error = %format!("{error:#}"), "Relayed LINKS_REPLY failed");
            }
        }
    }
}

/// `relayed` in parts for a LINKS_REPLY each: `MAX_RELAYED` members' links at most, and
/// `MAX_RELAYED_BYTES` as encoded (but for one member's alone).
fn by_size(relayed: Vec<RelayedLinks>) -> Vec<Vec<RelayedLinks>> {
    let mut parts: Vec<Vec<RelayedLinks>> = Vec::new();
    let mut bytes = 0;
    for links in relayed {
        let size = rmp_serde::encode::to_vec(&links).map_or(0, |encoded| encoded.len());
        let full = parts.last().is_none_or(|part| {
            part.len() >= MAX_RELAYED || (!part.is_empty() && bytes + size > MAX_RELAYED_BYTES)
        });
        if full {
            parts.push(Vec::new());
            bytes = 0;
        }
        bytes += size;
        parts.last_mut().expect("just made").push(links);
    }
    parts
}

/// A LINKS_REPLY from `peer`: its own links kept if it answers our latest LINKS to it,
/// and those it relays if it answers our latest relayed one, while their replies are
/// taken (a subscription's go on); then passed on to whoever asked us for them.
pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) {
    let now = Instant::now();
    let mut arrived = Vec::new();
    {
        let mut inner = state.links.inner.lock_or_recover();
        let own = inner
            .asked
            .get(peer)
            .is_some_and(|(nonce, until)| *nonce == reply.nonce && now < *until);
        let via: Option<HashSet<String>> = inner
            .via
            .get(peer)
            .filter(|(nonce, until, _)| *nonce == reply.nonce && now < *until)
            .map(|(_, _, members)| members.clone());
        let mut keep = |member: String, age: Duration, mut links: Vec<Link>| {
            links.truncate(MAX_LINKS);
            for link in &mut links {
                link.others.truncate(MAX_OTHERS);
            }
            if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(&member) {
                inner.replies.clear();
            }
            let at = now.checked_sub(age).unwrap_or(now);
            // Not older news over newer.
            if inner
                .replies
                .get(&member)
                .is_some_and(|(known, _)| *known > at)
            {
                return;
            }
            inner.replies.insert(member.clone(), (at, links));
            arrived.push(member);
        };
        if own {
            keep(peer.to_owned(), Duration::ZERO, reply.links);
        }
        if let Some(members) = via {
            for relayed in reply.relayed.into_iter().take(MAX_RELAYED) {
                if members.contains(&relayed.member) {
                    keep(
                        relayed.member,
                        Duration::from_millis(u64::from(relayed.age_ms)),
                        relayed.links,
                    );
                }
            }
        }
    }
    for member in arrived {
        forward(state, &member);
    }
}

/// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's
/// links; sent to each of them when they changed since they were last sent. Without
/// any, it waits for one. Runs for the node's life.
pub(crate) async fn watch(state: AppState) {
    let mut every = tokio::time::interval(WATCH_EVERY);
    every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        every.tick().await;
        let subscribers: Vec<(String, u64)> = {
            let now = Instant::now();
            let mut inner = state.links.inner.lock_or_recover();
            inner.subscribers.retain(|_, (_, until)| now < *until);
            if inner.subscribers.is_empty() {
                inner.published = None;
            }
            inner
                .subscribers
                .iter()
                .map(|(peer, (nonce, _))| (peer.clone(), *nonce))
                .collect()
        };
        if subscribers.is_empty() {
            // A new subscriber was just answered (`answer`, or within `ANSWER_EVERY`):
            // the next look is due an interval later.
            state.links.subscribed.notified().await;
            every.reset();
            continue;
        }
        let links = own_links(&state).await;
        {
            let mut inner = state.links.inner.lock_or_recover();
            if inner
                .published
                .as_ref()
                .is_some_and(|published| !changed(published, &links))
            {
                continue;
            }
            inner.published = Some(links.clone());
        }
        let (epoch, epoch_seq) = state.epoch_stamp();
        for (peer, nonce) in subscribers {
            let reply = ParsedMessage::LinksReply(LinksReply {
                epoch,
                epoch_seq,
                nonce,
                links: links.clone(),
                relayed: Vec::new(),
            });
            if let Err(error) = state.transport.send_message(&peer, &reply).await {
                debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed");
            }
        }
    }
}

/// 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.
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::*;

    #[test]
    fn relayed_links_are_sent_in_parts_by_count_and_size() {
        let relayed = |links: usize| RelayedLinks {
            member: "m".into(),
            age_ms: 0,
            links: (0..links)
                .map(|i| link(&format!("peer-{i:0>70}"), Some(1500)))
                .collect(),
        };
        // Small: by count.
        let parts = by_size((0..130).map(|_| relayed(2)).collect());
        assert_eq!(parts.iter().map(Vec::len).collect::<Vec<_>>(), [64, 64, 2]);
        // Large: by size, each part within the bytes but for one alone.
        let big = rmp_serde::encode::to_vec(&relayed(MAX_LINKS))
            .unwrap()
            .len();
        assert!(20 * big > MAX_RELAYED_BYTES, "{big}");
        let parts = by_size((0..20).map(|_| relayed(MAX_LINKS)).collect());
        assert_eq!(parts.iter().map(Vec::len).sum::<usize>(), 20);
        for part in &parts {
            let bytes: usize = part
                .iter()
                .map(|links| rmp_serde::encode::to_vec(links).unwrap().len())
                .sum();
            assert!(part.len() == 1 || bytes <= MAX_RELAYED_BYTES, "{bytes}");
        }
        assert!(parts.len() > 1);
    }

    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)]));
    }
}