//! Which links work, as every member records its own in its registry entry (`links`,
//! `members::RecordedLinks`): where they differ from the member list's graph
//! (`topology::member_list_graph`), which takes every member as reaching every other.
//!
//! Whom a member reaches it tells its neighbours (`swarm::reach`), but only them, and
//! only for as long as their sessions last; the writers' trees are computed by every
//! node alike from what every node has, and that is the document. So each member
//! records its neighbours by the graph that it doesn't reach (`unreached`), and its
//! neighbours it reaches that the graph doesn't link it to (`reached`: a ring that goes
//! round a member it doesn't reach, a member behind NAT that reaches out to one the
//! graph doesn't give it), each only once it has stayed so for a while: `DOWN_FOR`
//! keepalive intervals for a link down, or one the graph doesn't give (a ring goes
//! round a member only while it is away, so a short absence writes nothing), `UP_FOR`
//! for a link up again, so a link that drops for a keepalive isn't written down, and
//! one back is soon again. A link either end records
//! as unreached is one that doesn't work; one either records as reached, one that does;
//! so a link the other end records as unreached already isn't recorded by this one.
//! Only neighbours (`topology::our_neighbours`, those we sync with) count as reached:
//! not every member we have a session with (one that took us as its neighbour, or that
//! was one before the member list was known), as changes pass on along a tree only
//! where a sync opens. A member that went away is recorded as unreached by its
//! neighbours (they can't tell it from a link that broke), and taken out once they
//! reach it again. So a node first asks up to `PROBE_VIA` of
//! the member's own neighbours that it reaches whether they reach it (a PROBE, as SWIM
//! asks others before it suspects; answered from what they know, contacting nobody),
//! and records the link unless some say they don't and none says it does: a member
//! away, which its neighbours hold dormant, costs no writes, where each of them would
//! write twice (going and coming back). Asked ones that can't tell, or don't answer,
//! leave the link recorded. A node that reaches nobody at all is the
//! one away, and records no link as unreached.
//!
//! The record changes rarely: on a stable swarm, never; it is written only when a link
//! has been down or up for its hold.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::time::Duration;

use tokio::time::Instant;
use tracing::{debug, info, warn};

use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::members::{self, RecordedLinks};
use crate::swarm::{reach, topology};
use crate::transport::{ParsedMessage, ProbeReply, ProbeRequest};

/// Keepalive intervals a link stays down before it is recorded so (`unreached`, or out
/// of `reached`), and a link the graph doesn't give stays up before it is recorded
/// (into `reached`): as long as a silent peer takes to be held dormant
/// (`dormant_after`).
const DOWN_FOR: u32 = 4;
/// Keepalive intervals a link stays up before it is recorded so (out of `unreached`).
const UP_FOR: u32 = 1;

/// Members asked about a member at once at most: of its
/// neighbours by the graph that we reach, the ones with the lowest pair score with it,
/// so askers spread out alike.
const PROBE_VIA: usize = 3;
/// How long the members asked have to answer (a PROBE_REPLY comes at once: a round
/// trip, and losses).
const PROBE_WAIT: Duration = Duration::from_secs(10);

/// What we asked others about a member: when, with which nonce,
/// whom, and their answers so far (`ProbeReply::reached`).
struct Probe {
    asked: Instant,
    nonce: u64,
    via: BTreeSet<String>,
    answers: HashMap<String, Option<bool>>,
}

/// What the answers to a `Probe` say.
#[derive(Debug, PartialEq, Eq)]
enum Verdict {
    /// Not all in yet, and not overdue.
    Waiting,
    /// One of those asked reaches the member: our link to it is what doesn't work.
    Confirmed,
    /// Some of those asked don't reach it either, and none does: it is away.
    Away,
}

impl Probe {
    /// One answer that it reaches the member settles it; otherwise once all answered,
    /// or the rest are overdue: away if one said it doesn't reach it, else confirmed
    /// (nobody could tell, or nobody answered: the link is recorded).
    fn verdict(&self, now: Instant) -> Verdict {
        if self.answers.values().any(|reached| *reached == Some(true)) {
            return Verdict::Confirmed;
        }
        let all_in = self.via.iter().all(|peer| self.answers.contains_key(peer));
        if !all_in && now.saturating_duration_since(self.asked) < PROBE_WAIT {
            return Verdict::Waiting;
        }
        if self.answers.values().any(|reached| *reached == Some(false)) {
            Verdict::Away
        } else {
            Verdict::Confirmed
        }
    }
}

/// Since when each member's link has differed from our record.
#[derive(Default)]
pub(crate) struct Pending {
    since: HashMap<String, Instant>,
    /// What we asked others about each member, for links held long
    /// enough that await confirmation.
    probes: HashMap<String, Probe>,
    /// Checks done (`check`), for the benchmark to wait for one (`settled`).
    #[cfg(test)]
    checks: u64,
    /// Records written, for the soak (`written`); and of those, how many changed
    /// `unreached`, how many `reached`.
    #[cfg(test)]
    written: u64,
    #[cfg(test)]
    written_unreached: u64,
    #[cfg(test)]
    written_reached: u64,
}

/// How many records we wrote that changed `unreached`, and `reached`.
#[cfg(test)]
pub(crate) fn written_by_set(state: &AppState) -> (u64, u64) {
    let pending = state.working_links.lock_or_recover();
    (pending.written_unreached, pending.written_reached)
}

/// How many times we wrote our record.
#[cfg(test)]
pub(crate) fn written(state: &AppState) -> u64 {
    state.working_links.lock_or_recover().written
}

/// How many checks were done, and whether none of our links differs from our record
/// (nothing awaits its hold).
#[cfg(test)]
pub(crate) fn settled(state: &AppState) -> (u64, bool) {
    let pending = state.working_links.lock_or_recover();
    (pending.checks, pending.since.is_empty())
}

/// With the peer worker's periodic pass (every 5 s): whom we reach now, against the
/// member list's graph and our record; a difference that held for long enough goes into
/// the record.
pub(crate) async fn check(state: &AppState) {
    if !state.is_ready().await {
        return;
    }
    let me = state.local_id().to_owned();
    let Some(graph) = topology::member_list_graph(state) else {
        // No shared graph (`Topology::graph`): nothing to differ from; counted as a
        // check all the same (`settled`).
        #[cfg(test)]
        {
            state.working_links.lock_or_recover().checks += 1;
        }
        return;
    };
    let Some(ours) = graph.get(&me) else {
        // Not listed yet.
        return;
    };
    let now = Instant::now();
    let reached = reach::reached_members(state, now).await;
    let neighbours: BTreeSet<String> = topology::our_neighbours(state).into_iter().collect();
    let links = RecordedLinks {
        unreached: ours.difference(&reached).cloned().collect(),
        reached: reached
            .iter()
            .cloned()
            .filter(|member| {
                neighbours.contains(member) && graph.contains_key(member) && !ours.contains(member)
            })
            .collect(),
    };
    // Our record, and links the other end already records as unreached: either end's
    // record takes a link out, so ours needs no copy (half the writes for a link that
    // broke, and a link we can't confirm but the other end could is settled). One we
    // record already stays: dropping it as the other end drops its own, both at once,
    // would leave none.
    let (recorded, links) = {
        let replica = state.doc.read().await;
        let recorded = members::links(&replica.doc, &me);
        let mut links = links;
        links.unreached.retain(|member| {
            recorded.unreached.contains(member)
                || !members::links(&replica.doc, member).unreached.contains(&me)
        });
        (recorded, links)
    };
    // Reaching nobody, we are the one away: no link of ours is
    // recorded as unreached, which would go out, and be taken out, once we're back.
    let mut links = links;
    if reached.is_empty() {
        links.unreached.clear();
    }
    // Links newly unreached that others confirm: asked, they
    // reach the member, or can't tell; or there is nobody to ask.
    let confirmed = {
        let pending = state.working_links.lock_or_recover();
        links
            .unreached
            .difference(&recorded.unreached)
            .filter(|member| {
                via(&graph, &reached, member).is_empty()
                    || pending
                        .probes
                        .get(*member)
                        .is_some_and(|probe| probe.verdict(now) == Verdict::Confirmed)
            })
            .cloned()
            .collect::<BTreeSet<String>>()
    };
    let (next, to_probe) = {
        let mut pending = state.working_links.lock_or_recover();
        #[cfg(test)]
        {
            pending.checks += 1;
        }
        let (next, awaiting) = decide(
            &mut pending,
            &recorded,
            &links,
            now,
            state.timing.keepalive,
            Some(&confirmed),
        );
        // Asked again once away for another hold (it may be back by then, or the link
        // be all that's down).
        let hold = state.timing.keepalive * DOWN_FOR;
        pending.probes.retain(|member, _| awaiting.contains(member));
        let mut to_probe = Vec::new();
        for member in awaiting {
            let due = pending.probes.get(&member).is_none_or(|probe| {
                probe.verdict(now) == Verdict::Away
                    && now.saturating_duration_since(probe.asked) >= PROBE_WAIT + hold
            });
            if due {
                let probe = Probe {
                    asked: now,
                    nonce: crate::transport::paths::random_nonce(),
                    via: via(&graph, &reached, &member).into_iter().collect(),
                    answers: HashMap::new(),
                };
                to_probe.push((member.clone(), probe.nonce, probe.via.clone()));
                pending.probes.insert(member, probe);
            }
        }
        (next, to_probe)
    };
    for (member, nonce, asked) in to_probe {
        debug!(member = %member, via = ?asked, "Link down for its hold; asking whether others reach the member");
        let (epoch, epoch_seq) = state.epoch_stamp();
        let request = ParsedMessage::Probe(ProbeRequest {
            epoch,
            epoch_seq,
            nonce,
            member: member.clone(),
        });
        for peer in asked {
            if let Err(error) = state.transport.send_message(&peer, &request).await {
                debug!(peer = %peer, error = %format!("{error:#}"), "PROBE failed");
            }
        }
    }
    if next == recorded {
        return;
    }
    let (unreached, reached) = (next.unreached.len(), next.reached.len());
    #[cfg(test)]
    let changed = (
        next.unreached != recorded.unreached,
        next.reached != recorded.reached,
    );
    match state
        .mutate_doc(move |doc| members::set_links(doc, &me, &next))
        .await
    {
        Ok(()) => {
            #[cfg(test)]
            {
                let mut pending = state.working_links.lock_or_recover();
                pending.written += 1;
                pending.written_unreached += u64::from(changed.0);
                pending.written_reached += u64::from(changed.1);
            }
            info!(
                unreached,
                reached, "Links that differ from the member list's graph recorded"
            )
        }
        Err(error) => warn!(error = %error, "Failed to record our links"),
    }
}

/// A PROBE from `peer`: whether we reach the member it names, from what we know (no
/// contact made): an active session that answers (`reach::reached_members`); a session
/// that doesn't answer now (not yet held dormant: asks come after as long as that
/// takes), held dormant, or HELLOs unanswered; else we can't tell. Only members are
/// answered.
pub(crate) async fn answer(state: &AppState, peer: &str, request: ProbeRequest) {
    if topology::listed(state, peer) != Some(true) {
        return;
    }
    let now = Instant::now();
    let reached = if reach::reached_members(state, now)
        .await
        .contains(&request.member)
    {
        Some(true)
    } else {
        match state.get_peer_state(&request.member).await {
            Some(member) => {
                let member = member.lock().await;
                (member.hello_state == crate::state::HelloState::Established
                    || member.liveness.is_dormant()
                    || member.is_unreachable())
                .then_some(false)
            }
            None => None,
        }
    };
    let (epoch, epoch_seq) = state.epoch_stamp();
    let reply = ParsedMessage::ProbeReply(ProbeReply {
        epoch,
        epoch_seq,
        nonce: request.nonce,
        member: request.member,
        reached,
    });
    if let Err(error) = state.transport.send_message(peer, &reply).await {
        debug!(peer = %peer, error = %format!("{error:#}"), "PROBE_REPLY failed");
    }
}

/// A PROBE_REPLY from `peer`: its answer kept, if it answers what we asked it.
pub(crate) fn answered(state: &AppState, peer: &str, reply: ProbeReply) {
    let mut pending = state.working_links.lock_or_recover();
    if let Some(probe) = pending.probes.get_mut(&reply.member)
        && probe.nonce == reply.nonce
        && probe.via.contains(peer)
    {
        probe.answers.insert(peer.to_owned(), reply.reached);
    }
}

/// The members to ask whether they reach `member`: its
/// neighbours by the graph that we reach (the likeliest to have been in touch with
/// it), up to `PROBE_VIA`, by the lowest pair score with it.
fn via(
    graph: &BTreeMap<String, BTreeSet<String>>,
    reached: &BTreeSet<String>,
    member: &str,
) -> Vec<String> {
    let theirs = graph.get(member);
    let mut via: Vec<&String> = reached
        .iter()
        .filter(|peer| *peer != member && theirs.is_some_and(|theirs| theirs.contains(*peer)))
        .collect();
    via.sort_by_key(|peer| topology::pair_score(peer, member));
    via.into_iter().take(PROBE_VIA).cloned().collect()
}

/// The record to keep: `recorded`, but where a member's link has differed from it as
/// `now_links` says since long enough before `now` (`pending`): `DOWN_FOR` keepalive
/// intervals (newly unreached, no longer reached, or newly reached: a substitute
/// neighbour while one is away), `UP_FOR` otherwise (reached again, or no longer our
/// neighbour by the graph). A link newly
/// unreached goes in only if `confirmed` has it (all do without): those held long
/// enough that it hasn't are returned, to ask others about.
fn decide(
    pending: &mut Pending,
    recorded: &RecordedLinks,
    now_links: &RecordedLinks,
    now: Instant,
    keepalive: Duration,
    confirmed: Option<&BTreeSet<String>>,
) -> (RecordedLinks, BTreeSet<String>) {
    let mut awaiting = BTreeSet::new();
    let members: BTreeSet<&String> = recorded
        .unreached
        .iter()
        .chain(&recorded.reached)
        .chain(&now_links.unreached)
        .chain(&now_links.reached)
        .collect();
    pending.since.retain(|member, _| members.contains(member));
    let mut next = recorded.clone();
    for member in members {
        let unreached = now_links.unreached.contains(member);
        let reached = now_links.reached.contains(member);
        let was_unreached = recorded.unreached.contains(member);
        let was_reached = recorded.reached.contains(member);
        if unreached == was_unreached && reached == was_reached {
            pending.since.remove(member);
            continue;
        }
        let long = (unreached && !was_unreached) || (was_reached != reached);
        let hold = keepalive * if long { DOWN_FOR } else { UP_FOR };
        let since = *pending.since.entry(member.clone()).or_insert(now);
        if now.saturating_duration_since(since) < hold {
            continue;
        }
        if unreached
            && !was_unreached
            && confirmed.is_some_and(|confirmed| !confirmed.contains(member))
        {
            awaiting.insert(member.clone());
            continue;
        }
        pending.since.remove(member);
        for (set, is) in [
            (&mut next.unreached, unreached),
            (&mut next.reached, reached),
        ] {
            if is {
                set.insert(member.clone());
            } else {
                set.remove(member);
            }
        }
    }
    (next, awaiting)
}

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

    fn set(members: &[&str]) -> BTreeSet<String> {
        members.iter().map(|member| (*member).to_owned()).collect()
    }

    fn links(unreached: &[&str], reached: &[&str]) -> RecordedLinks {
        RecordedLinks {
            unreached: set(unreached),
            reached: set(reached),
        }
    }

    /// A link goes into the record once down for `DOWN_FOR` keepalives, not for one; out
    /// once up for `UP_FOR`; one reached the graph doesn't give, in and out after
    /// `DOWN_FOR`.
    #[test]
    fn links_are_recorded_after_their_hold() {
        let keepalive = Duration::from_secs(5);
        let start = Instant::now();
        let at = |secs: u64| start + Duration::from_secs(secs);
        let mut pending = Pending::default();
        let none = links(&[], &[]);

        // Down for a keepalive, then up again: not recorded.
        assert_eq!(
            decide(
                &mut pending,
                &none,
                &links(&["b"], &[]),
                at(0),
                keepalive,
                None
            )
            .0,
            none
        );
        assert_eq!(
            decide(
                &mut pending,
                &none,
                &links(&["b"], &[]),
                at(5),
                keepalive,
                None
            )
            .0,
            none
        );
        assert_eq!(
            decide(&mut pending, &none, &none, at(10), keepalive, None).0,
            none
        );
        // Down again: from then on, `DOWN_FOR` keepalives.
        assert_eq!(
            decide(
                &mut pending,
                &none,
                &links(&["b"], &[]),
                at(15),
                keepalive,
                None
            )
            .0,
            none
        );
        assert_eq!(
            decide(
                &mut pending,
                &none,
                &links(&["b"], &[]),
                at(30),
                keepalive,
                None
            )
            .0,
            none
        );
        let down = links(&["b"], &[]);
        assert_eq!(
            decide(&mut pending, &none, &down, at(35), keepalive, None).0,
            down
        );

        // Up again: out after `UP_FOR`.
        assert_eq!(
            decide(&mut pending, &down, &none, at(40), keepalive, None).0,
            down
        );
        assert_eq!(
            decide(&mut pending, &down, &none, at(45), keepalive, None).0,
            none
        );

        // Reached beyond the graph: in after `DOWN_FOR`, not `UP_FOR`; gone: out after
        // `DOWN_FOR`.
        let beyond = links(&[], &["c"]);
        assert_eq!(
            decide(&mut pending, &none, &beyond, at(50), keepalive, None).0,
            none
        );
        assert_eq!(
            decide(&mut pending, &none, &beyond, at(55), keepalive, None).0,
            none
        );
        assert_eq!(
            decide(&mut pending, &none, &beyond, at(70), keepalive, None).0,
            beyond
        );
        assert_eq!(
            decide(&mut pending, &beyond, &none, at(75), keepalive, None).0,
            beyond
        );
        assert_eq!(
            decide(&mut pending, &beyond, &none, at(95), keepalive, None).0,
            none
        );
    }

    /// With asks, a link down for its hold goes in only once
    /// confirmed (one of those asked reaches the member); until then it is returned to
    /// ask about, and stays pending. Other changes aren't asked about.
    #[test]
    fn unreached_links_wait_for_confirmation() {
        let keepalive = Duration::from_secs(5);
        let start = Instant::now();
        let at = |secs: u64| start + Duration::from_secs(secs);
        let mut pending = Pending::default();
        let none = links(&[], &[]);
        let down = links(&["b"], &[]);
        let nobody = BTreeSet::new();

        assert_eq!(
            decide(&mut pending, &none, &down, at(0), keepalive, Some(&nobody)),
            (none.clone(), BTreeSet::new())
        );
        // Held long enough, not confirmed: asked about, not recorded.
        assert_eq!(
            decide(&mut pending, &none, &down, at(20), keepalive, Some(&nobody)),
            (none.clone(), set(&["b"]))
        );
        assert_eq!(
            decide(&mut pending, &none, &down, at(25), keepalive, Some(&nobody)),
            (none.clone(), set(&["b"]))
        );
        // Confirmed: recorded.
        assert_eq!(
            decide(
                &mut pending,
                &none,
                &down,
                at(30),
                keepalive,
                Some(&set(&["b"]))
            ),
            (down.clone(), BTreeSet::new())
        );
        // Coming out needs no confirmation.
        assert_eq!(
            decide(&mut pending, &down, &none, at(31), keepalive, Some(&nobody)).0,
            down
        );
        assert_eq!(
            decide(&mut pending, &down, &none, at(36), keepalive, Some(&nobody)).0,
            none
        );
    }
}