//! Which members a node syncs with: its neighbours.
//!
//! In a full mesh every change reaches every node from every other node (Automerge
//! decides per pair, so nobody knows that a third peer already passed it on), so the
//! copies of one write grow with N². Above `k` members, each node keeps only a few
//! neighbours, and changes travel over several hops.
//!
//! Every node computes the graph from the member list and which members it can reach,
//! with no coordination:
//! * a ring ordered by key hash: each member's nearest reachable member in both
//!   directions, which keeps the reachable members connected;
//! * plus, for each member, the `k - 2` reachable others with the lowest pair score (a
//!   hash of both keys, so both ends agree on it), which keeps paths short.
//!
//! A link exists if either end chose it, so links are symmetric and a node has at least
//! `k` neighbours. With at most `k + 1` reachable members the graph is the full mesh:
//! small swarms behave exactly as before. A member joining, leaving or going offline
//! changes only links near it.
//!
//! Members on several transports (Mycelium, libp2p) get such a graph per transport,
//! among the members on it (as their addresses in the member list say), and a node's
//! neighbours are the union of its neighbours on each transport it is on
//! (`neighbours_across`). Each of those graphs is connected (its ring), and a member on
//! two transports is in both: as long as members on several transports link the
//! transports, the swarm stays connected, and such a member has about `k` neighbours per
//! transport. With one transport, it is the one graph as before.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;

use tracing::info;

use crate::locks::RwLockExt;
use crate::state::{AppState, PeerState};
use crate::swarm::members;
use crate::transport::ParsedMessage;

mod graph;

use graph::*;
pub(crate) use graph::{RingAndPairs, pair_key, pair_score};

/// How members' neighbours are chosen: the graph above (`RingAndPairs`) is one way;
/// others can be tried against it in the benchmark (`docs/future-work.md`, Swappable
/// topology and dissemination). One per swarm: a node's neighbours must count it as
/// theirs.
///
/// A topology either computes a graph every node shares (`graph`: the same inputs give
/// every node the same), or keeps its own state from its own messages (HyParView's
/// views), and has none. What needs the shared graph does nothing without one: writers'
/// trees (`member_list_graph`, `swarm::tree`), the links members record where it is
/// wrong (`working_graph`, `swarm::working_links`), indirect probes through a member's
/// neighbours, the dashboard's map of the intended links (`intended_graph`). A
/// dissemination that needs none goes with such a topology (flooding, PlumTree).
#[async_trait::async_trait]
pub(crate) trait Topology: Send + Sync + 'static {
    /// Its name (in logs, and once chosen by the swarm, in the document).
    fn name(&self) -> &'static str;

    /// Every reachable member's neighbours among `members` (those on one transport),
    /// each keeping about `k`, skipping those `unreachable`, preferring those `stable`
    /// as it does. Pure: the same inputs give every node the same graph. None for a
    /// topology without a shared graph.
    fn graph(
        &self,
        members: &[String],
        k: usize,
        unreachable: &dyn Fn(&str) -> bool,
        stable: &dyn Fn(&str) -> bool,
    ) -> Option<BTreeMap<String, BTreeSet<String>>>;

    /// Our neighbours, from what they are chosen from (`Inputs`), each time those may
    /// have changed (`refresh`). By default our part of the shared graph, on each of our
    /// transports (`neighbours_across`); a topology with state of its own keeps them
    /// there.
    fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
        neighbours_across(
            self,
            inputs.transports,
            local,
            inputs.local_transports,
            inputs.k,
            inputs.unreachable,
            inputs.stable,
        )
    }

    /// With every recomputation (`refresh`, at least every 5 s), before `neighbours`:
    /// what a topology with state of its own does about it (asking members to be
    /// neighbours, giving up on those that don't answer). By default nothing.
    async fn maintain(&self, state: &AppState, inputs: &Inputs<'_>) {
        let _ = (state, inputs);
    }

    /// A message of the topology's own from `peer` (HyParView's NEIGHBOUR,
    /// NEIGHBOUR_REPLY, DISCONNECT). By default ignored.
    async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) {
        let _ = (state, peer, message);
    }

    /// Changes whenever the topology's own state changed what `neighbours` gives: the
    /// neighbours are recomputed then, even though nothing they are chosen from did. By
    /// default always the same.
    fn version(&self) -> u64 {
        0
    }

    /// Taking over from another topology, whose neighbours were `neighbours` (the swarm
    /// switched, `swarm_ways`): a topology with state of its own starts from them. By
    /// default nothing.
    fn seed(&self, neighbours: &BTreeSet<String>) {
        let _ = neighbours;
    }

    /// The neighbours each agreed to, for a topology that keeps such (HyParView's active
    /// view), for the tests and the benchmark to look at. By default none.
    #[cfg(test)]
    fn active(&self) -> Option<BTreeSet<String>> {
        None
    }
}

/// The topology and the dissemination in use (`AppState::ways`), and which they are.
pub(crate) struct Ways {
    pub(crate) kinds: (Kind, crate::swarm::dissemination::Kind),
    pub(crate) topology: Arc<dyn Topology>,
    pub(crate) dissemination: Arc<dyn crate::swarm::dissemination::Dissemination>,
}

impl Ways {
    pub(crate) fn new(topology: Kind, dissemination: crate::swarm::dissemination::Kind) -> Self {
        Self {
            kinds: (topology, dissemination),
            topology: topology.make(),
            dissemination: dissemination.make(),
        }
    }
}

/// Where the swarm records its topology and its dissemination (`swarm_ways`); the
/// genesis node writes its `--topology` and `--dissemination` there
/// (`swarm::bootstrap`).
pub(crate) fn topology_path() -> Vec<String> {
    crate::document::meta::path(&[crate::document::meta::TOPOLOGY])
}

pub(crate) fn dissemination_path() -> Vec<String> {
    crate::document::meta::path(&[crate::document::meta::DISSEMINATION])
}

/// The topology and the dissemination the swarm records: its genesis node writes its
/// own (`topology_path`, `dissemination_path`), a protocol writer may change them (every
/// node then swaps to them, `refresh`). Where the document has none (a swarm from
/// before they were recorded), or one this node doesn't know, this node's own (`own`).
pub(crate) fn swarm_ways(
    doc: &crate::document::Document,
    own: (Kind, crate::swarm::dissemination::Kind),
) -> (Kind, crate::swarm::dissemination::Kind) {
    fn recorded<T: serde::de::DeserializeOwned>(
        doc: &crate::document::Document,
        path: &[String],
        what: &str,
        own: T,
    ) -> T {
        match crate::document::store::read_path(doc, path) {
            Ok(Some(value)) => match serde_json::from_value(value.clone()) {
                Ok(recorded) => recorded,
                Err(_) => {
                    warn_once(&format!(
                        "The swarm's {what} ({value}) isn't one this node knows; using --{what}"
                    ));
                    own
                }
            },
            _ => own,
        }
    }
    (
        recorded(doc, &topology_path(), "topology", own.0),
        recorded(doc, &dissemination_path(), "dissemination", own.1),
    )
}

/// Swaps to the topology and the dissemination `kinds` if they aren't the ones in use:
/// fresh, the new topology starting from our neighbours (`Topology::seed`).
fn switch_ways(state: &AppState, kinds: (Kind, crate::swarm::dissemination::Kind)) {
    let neighbours: BTreeSet<String> = our_neighbours(state).into_iter().collect();
    let mut ways = state.ways.write_or_recover();
    if ways.kinds == kinds {
        return;
    }
    let before = (ways.topology.name(), ways.dissemination.name());
    if ways.kinds.0 != kinds.0 {
        ways.topology = kinds.0.make();
        ways.topology.seed(&neighbours);
    }
    if ways.kinds.1 != kinds.1 {
        ways.dissemination = kinds.1.make();
    }
    ways.kinds = kinds;
    info!(
        from_topology = before.0,
        from_dissemination = before.1,
        topology = ways.topology.name(),
        dissemination = ways.dissemination.name(),
        "The swarm switched how neighbours are chosen, or how changes pass on"
    );
    if ways
        .topology
        .graph(&[], 2, &|_| false, &|_| false)
        .is_none()
        && kinds.1 == crate::swarm::dissemination::Kind::WritersTrees
    {
        warn_once(
            "Writers' trees need a graph every node shares, which this topology has none of: changes pass on by announcements alone (a protocol writer may record dissemination \"flood\")",
        );
    }
}

/// The topologies to choose from (`--topology`); one per swarm.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub(crate) enum Kind {
    /// The member list's graph (`RingAndPairs`).
    #[default]
    RingAndPairs,
    /// HyParView's active view, slimmed down (`swarm::hyparview`).
    Hyparview,
}

impl Kind {
    pub(crate) fn make(self) -> Arc<dyn Topology> {
        match self {
            Kind::RingAndPairs => Arc::new(RingAndPairs),
            Kind::Hyparview => Arc::new(crate::swarm::hyparview::HyParView::default()),
        }
    }
}

/// What a node's neighbours are chosen from (`Topology::neighbours`): the members, by
/// the transports each is on (as the member list gives their addresses), ours, the
/// neighbours each keeps, whom we judge unreachable, who is stable.
pub(crate) struct Inputs<'a> {
    pub(crate) transports: &'a BTreeMap<String, BTreeSet<String>>,
    pub(crate) local_transports: &'a BTreeSet<String>,
    pub(crate) k: usize,
    pub(crate) unreachable: &'a (dyn Fn(&str) -> bool + Sync),
    pub(crate) stable: &'a (dyn Fn(&str) -> bool + Sync),
}

/// This node's current neighbours, and the member list they were computed from.
#[derive(Debug, Default)]
pub(crate) struct NeighbourView {
    members: Vec<String>,
    k: usize,
    unreachable: BTreeSet<String>,
    stable: BTreeSet<String>,
    /// The transports (address schemes) of each member, and ours.
    transports: BTreeMap<String, BTreeSet<String>>,
    local_transports: BTreeSet<String>,
    /// Peers that proved they are of our swarm but aren't in our member list (yet): their
    /// registration hasn't reached us. They are neighbours too (`refresh`).
    unlisted: BTreeSet<String>,
    /// Members the member list gives no address for on our transports, that reach us
    /// anyway: they only connect out (no listen address). They are neighbours too while
    /// they say they sync with us (`refresh`).
    outbound: BTreeSet<String>,
    neighbours: BTreeSet<String>,
    /// The topology they were computed with, and its version (`Topology::version`).
    version: (&'static str, u64),
    /// Every member not judged unreachable is a neighbour (the tests wait for it,
    /// `is_full_mesh`). Not the swarm being a full mesh: members on a transport we are
    /// not on, or that we never reached, don't count against it.
    #[cfg(test)]
    full_mesh: bool,
    /// The graph the member list gives, as if every member reached every other
    /// (`member_list_graph`): computed when first asked for, kept while the member list
    /// stays the same. None for a topology without a shared graph.
    whole: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
    /// Where members recorded their links differ from that graph (`swarm::working_links`),
    /// those with a record.
    recorded: BTreeMap<String, members::RecordedLinks>,
    /// That graph with the links members recorded (`working_graph`): computed when first
    /// asked for, kept while the member list and the records stay the same.
    working: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
}

/// Every member's neighbours as the member list gives them, as if every member reached
/// every other: the same on every node that knows the same member list, whatever each
/// reaches (`swarm::tree` builds writers' trees on it). On each transport among the
/// members on it, all together. None until the view is known, and for a topology
/// without a shared graph (`Topology::graph`).
pub(crate) fn member_list_graph(
    state: &AppState,
) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
    let view = state.neighbours.read_or_recover();
    let view = view.as_ref()?;
    view.whole
        .get_or_init(|| {
            let mut whole: BTreeMap<String, BTreeSet<String>> = view
                .members
                .iter()
                .map(|member| (member.clone(), BTreeSet::new()))
                .collect();
            let transports: BTreeSet<&String> = view.transports.values().flatten().collect();
            for transport in transports {
                let on_it: Vec<String> = view
                    .transports
                    .iter()
                    .filter(|(_, on)| on.contains(transport))
                    .map(|(member, _)| member.clone())
                    .collect();
                let graph = state
                    .topology()
                    .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m))?;
                for (member, neighbours) in graph {
                    whole.entry(member).or_default().extend(neighbours);
                }
            }
            Some(Arc::new(whole))
        })
        .clone()
}

/// The member list's graph (`member_list_graph`) with the links members recorded
/// (`swarm::working_links`): without those either end recorded as unreached, with those
/// one recorded as reached (`with_recorded`). The same on every node that has the same
/// document; what the writers' trees are built on (`swarm::tree`). None until the view
/// is known, and for a topology without a shared graph.
pub(crate) fn working_graph(state: &AppState) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
    let whole = member_list_graph(state)?;
    let view = state.neighbours.read_or_recover();
    let view = view.as_ref()?;
    view.working
        .get_or_init(|| {
            Some(if view.recorded.is_empty() {
                whole
            } else {
                Arc::new(with_recorded(&whole, &view.recorded))
            })
        })
        .clone()
}

/// `graph` with the links members recorded (`recorded`): those one recorded as reached
/// added, between members `graph` has; then those either end recorded as unreached
/// taken out (a link one end reaches and the other doesn't counts as down).
pub(crate) fn with_recorded(
    graph: &BTreeMap<String, BTreeSet<String>>,
    recorded: &BTreeMap<String, members::RecordedLinks>,
) -> BTreeMap<String, BTreeSet<String>> {
    let mut graph = graph.clone();
    for (member, links) in recorded {
        for other in &links.reached {
            if member != other && graph.contains_key(member) && graph.contains_key(other) {
                graph
                    .entry(member.clone())
                    .or_default()
                    .insert(other.clone());
                graph
                    .entry(other.clone())
                    .or_default()
                    .insert(member.clone());
            }
        }
    }
    for (member, links) in recorded {
        for other in &links.unreached {
            if let Some(neighbours) = graph.get_mut(member) {
                neighbours.remove(other);
            }
            if let Some(neighbours) = graph.get_mut(other) {
                neighbours.remove(member);
            }
        }
    }
    graph
}

/// Whether `peer` is in the member list, as the view was last computed from it; None
/// while the view isn't known.
pub(crate) fn listed(state: &AppState, peer: &str) -> Option<bool> {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map(|view| view.members.iter().any(|member| member == peer))
}

/// Whether the neighbours are computed from the member list yet (before, everyone
/// counts as a neighbour).
pub(crate) fn view_known(state: &AppState) -> bool {
    state.neighbours.read_or_recover().is_some()
}

/// Whether every reachable member is our neighbour (also while the view is unknown).
#[cfg(test)]
pub(crate) fn is_full_mesh(state: &AppState) -> bool {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .is_none_or(|view| view.full_mesh)
}

/// The current view in numbers, once known.
#[derive(Debug, Clone)]
pub(crate) struct Summary {
    /// Registered members, us included.
    pub(crate) members: usize,
    /// Members other than us not currently judged unreachable.
    pub(crate) reachable: usize,
    pub(crate) stable: usize,
    pub(crate) neighbours: usize,
    pub(crate) stable_neighbours: usize,
    /// For each of our transports (address scheme), the neighbours the member list
    /// gives an address on it: one on two transports counts on both, one it gives none
    /// for (not listed yet, or only connecting out) on none.
    pub(crate) neighbours_by_transport: BTreeMap<String, usize>,
}

pub(crate) fn summary(state: &AppState) -> Option<Summary> {
    let local = state.local_id();
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map(|view| Summary {
            members: view.members.len(),
            reachable: view
                .members
                .iter()
                .filter(|m| m.as_str() != local && !view.unreachable.contains(*m))
                .count(),
            stable: view.stable.len(),
            neighbours: view.neighbours.len(),
            stable_neighbours: view
                .neighbours
                .iter()
                .filter(|m| view.stable.contains(*m))
                .count(),
            neighbours_by_transport: view
                .local_transports
                .iter()
                .map(|transport| {
                    let on_it = view
                        .neighbours
                        .iter()
                        .filter(|m| {
                            view.transports
                                .get(*m)
                                .is_some_and(|theirs| theirs.contains(transport))
                        })
                        .count();
                    (transport.clone(), on_it)
                })
                .collect(),
        })
}

/// Whether `peer` is in the member list the neighbours were computed from (false until
/// the view is known).
pub(crate) fn is_listed(state: &AppState, peer: &str) -> bool {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .is_some_and(|view| view.members.iter().any(|member| member == peer))
}

/// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync
/// openings. Everyone is a neighbour until the view is known (while joining, or before
/// any member registered), so joining works as in a full mesh. We always *answer*
/// anyone who contacts us.
/// Our neighbours, as the view was last computed; none while it isn't known (when
/// everyone counts: `is_neighbour`).
pub(crate) fn our_neighbours(state: &AppState) -> Vec<String> {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map(|view| view.neighbours.iter().cloned().collect())
        .unwrap_or_default()
}

pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
    match &*state.neighbours.read_or_recover() {
        Some(view) => view.neighbours.contains(peer),
        None => true,
    }
}

/// Every member's neighbours as the member list makes them, as this node sees it (whom
/// it can reach, who is stable): the graph each node computes its own part of. Ours as
/// they are (with members that only connect out, or aren't listed yet). For a map of
/// the swarm (the dashboard); None until the view is known. A link exists if either end
/// chose it: draw one if either lists the other. About `members²` work: compute it when
/// the member list may have changed, not on every look. None too for a topology without
/// a shared graph.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> {
    member_list_graph(state)?;
    let topology = state.topology();
    let view = state.neighbours.read_or_recover();
    let view = view.as_ref()?;
    let graph = view
        .members
        .iter()
        .map(|member| {
            (
                member.clone(),
                view.neighbours_of(&*topology, state.local_id(), member),
            )
        })
        .collect();
    Some(graph)
}

/// Whether we reach `peer` ourselves, as choosing neighbours judges it: on one of our
/// transports, and not unanswered (three HELLOs, or dormant: `PeerState::is_unreachable`).
/// One we have no word of yet counts as reached until it fails to answer.
pub(crate) async fn reaches(state: &AppState, peer: &str) -> bool {
    if !state.transport.can_reach(peer) {
        return false;
    }
    match state.get_peer_state(peer).await {
        Some(peer_state) => !peer_state.lock().await.is_unreachable(),
        None => true,
    }
}

/// The transports `member` is on, by the member list (the schemes of its addresses);
/// ours as we are on them. None known: empty.
pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> {
    let view = state.neighbours.read_or_recover();
    match view.as_ref() {
        Some(view) if member == state.local_id() => view.local_transports.clone(),
        Some(view) => view.transports.get(member).cloned().unwrap_or_default(),
        None if member == state.local_id() => state.transport.transports().into_iter().collect(),
        None => BTreeSet::new(),
    }
}

impl NeighbourView {
    fn neighbours_of(
        &self,
        topology: &dyn Topology,
        local: &str,
        member: &str,
    ) -> BTreeSet<String> {
        if member == local {
            return self.neighbours.clone();
        }
        let own = self.transports.get(member).cloned().unwrap_or_default();
        neighbours_across(
            topology,
            &self.transports,
            member,
            &own,
            self.k,
            &|member| self.unreachable.contains(member),
            &|member| self.stable.contains(member),
        )
    }
}

/// How the neighbours changed.
pub(crate) struct NeighbourChange {
    pub(crate) added: Vec<String>,
    pub(crate) removed: Vec<String>,
}

/// Runs `change` on a peer's state, and has the neighbours recomputed (`run_worker`) if
/// it changed what they are chosen from (`PeerState::neighbour_inputs`). Changes to a
/// peer that may matter to the neighbours go through here.
pub(crate) fn change_peer<T>(
    state: &AppState,
    peer_state: &mut PeerState,
    change: impl FnOnce(&mut PeerState) -> T,
) -> T {
    let before = peer_state.neighbour_inputs();
    let result = change(peer_state);
    if peer_state.neighbour_inputs() != before {
        state.topology_changed.notify_one();
    }
    result
}

/// Recomputes the neighbours whenever something they depend on may have changed
/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`),
/// a peer's state (`change_peer`), and every 5 s as a fallback (`swarm::peer`). A burst
/// of notifications coalesces into one recomputation.
pub(crate) async fn run_worker(state: AppState) {
    info!(
        topology = state.topology().name(),
        dissemination = state.dissemination().name(),
        "Choosing neighbours"
    );
    loop {
        state.topology_changed.notified().await;
        if let Some(change) = refresh(&state).await {
            crate::swarm::peer::apply_neighbour_change(&state, change).await;
        }
    }
}

/// Whether a change at `path` may change the neighbours: one in the member list, or the
/// root (the whole document replaced, as after an epoch switch).
pub(crate) fn affects_neighbours(path: &[String]) -> bool {
    path.is_empty()
        || path.starts_with(&members::members_dir())
        || path.starts_with(&neighbours_path())
        || path.starts_with(&topology_path())
        || path.starts_with(&dissemination_path())
}

/// The most neighbours a swarm may ask each member to keep (`swarm_neighbours`).
const MAX_NEIGHBOURS: u64 = 256;

/// Where the swarm records its neighbours (`swarm_neighbours`); the genesis node writes
/// its `--neighbours` there (`swarm::bootstrap`).
pub(crate) fn neighbours_path() -> Vec<String> {
    crate::document::meta::path(&[crate::document::meta::NEIGHBOURS])
}

/// The neighbours each member keeps, as the swarm records it: its genesis node writes
/// its `--neighbours` (`neighbours_path`), a protocol writer may change it (every
/// node then recomputes its neighbours, and the writers' trees with them). The same on
/// every node, as the member list's graph must be (`member_list_graph`). Where the
/// document has none (a swarm from before it was recorded), or one that isn't a number
/// from 2 to `MAX_NEIGHBOURS`, this node's own (`own`).
pub(crate) fn swarm_neighbours(doc: &crate::document::Document, own: usize) -> usize {
    match crate::document::store::read_path(doc, &neighbours_path()) {
        Ok(Some(value)) => match value.as_u64() {
            Some(k) if (2..=MAX_NEIGHBOURS).contains(&k) => k as usize,
            _ => {
                warn_once(&format!(
                    "The swarm's neighbours ({value}) isn't a number from 2 to {MAX_NEIGHBOURS}; using --neighbours ({own})"
                ));
                own
            }
        },
        _ => own,
    }
}

/// Logs `message` at info level, unless it was the last one logged so.
fn info_once(message: &str) {
    static LAST: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
    let mut last = LAST.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
    if *last != message {
        info!("{message}");
        *last = message.to_owned();
    }
}

/// Logs `message` at warn level, unless it was the last one logged so.
fn warn_once(message: &str) {
    static LAST: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
    let mut last = LAST.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
    if *last != message {
        tracing::warn!("{message}");
        *last = message.to_owned();
    }
}

/// The neighbours each member keeps, as the view was last computed with
/// (`swarm_neighbours`); this node's `--neighbours` before.
pub(crate) fn k(state: &AppState) -> usize {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map_or(state.neighbours_k, |view| view.k)
}

/// Recomputes the neighbours once ready, if the member list, the reachable members or
/// `k` changed.
pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
    if !state.is_ready().await {
        return None;
    }
    let (current, stable, transports, alive, k, recorded) = {
        let replica = state.doc.read().await;
        members::apply(state, &replica.doc);
        let k = swarm_neighbours(&replica.doc, state.neighbours_k);
        switch_ways(state, swarm_ways(&replica.doc, state.configured_ways));
        let current = members::members(&replica.doc);
        // Those that said they are alive within the electorate window.
        let alive: BTreeSet<String> = members::electorate(
            &replica.doc,
            state.clock.now_millis().0,
            state.electorate_window,
        )
        .into_iter()
        .collect();
        let stable: BTreeSet<String> = current
            .iter()
            .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
            .cloned()
            .collect();
        let transports: BTreeMap<String, BTreeSet<String>> = current
            .iter()
            .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
            .collect();
        let recorded: BTreeMap<String, members::RecordedLinks> = current
            .iter()
            .map(|m| (m.clone(), members::links(&replica.doc, m)))
            .filter(|(_, links)| *links != members::RecordedLinks::default())
            .collect();
        (current, stable, transports, alive, k, recorded)
    };
    if k != state.neighbours_k {
        info_once(&format!(
            "The swarm keeps {k} neighbours per member (recorded by its genesis node, or a protocol writer); --neighbours ({}) counts only on a genesis node",
            state.neighbours_k
        ));
    }
    let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
    // Those that don't answer, those on none of our transports, and those that haven't
    // said they are alive for longer than the electorate window, unless we heard from
    // them since we started (one that comes back says so to us first: its HELLO).
    let mut unreachable = BTreeSet::new();
    for member in &current {
        if member == state.local_id() {
            continue;
        }
        if !reaches(state, member).await {
            unreachable.insert(member.clone());
            continue;
        }
        let heard = match state.get_peer_state(member).await {
            Some(peer_state) => peer_state.lock().await.last_seen.is_some(),
            None => false,
        };
        if !heard && !alive.contains(member) {
            unreachable.insert(member.clone());
        }
    }
    // A node that joined while the genesis node hadn't registered yet has the genesis
    // document without the genesis node in it, and registers itself there: then neither
    // lists the other, and with each the other's neighbour only by the member list,
    // neither would ever sync with the other, and their registrations never pass. So a
    // peer that proved itself a member of our swarm counts until the list says otherwise:
    // syncing brings each the other's registration.
    let mut unlisted = BTreeSet::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        if current.contains(&peer) || peer == state.local_id() {
            continue;
        }
        let peer_state = peer_state.lock().await;
        if matches!(peer_state.standing, crate::state::Standing::Member { .. })
            && !peer_state.is_unreachable()
        {
            unlisted.insert(peer);
        }
    }
    // A member that lists no address on our transports, but reached us anyway, only
    // connects out: it isn't in the graph on our side (it's on none of our transports, as
    // far as the list goes), while on its side we are, so it would sync with us while we
    // never sent it our changes (only answers to its). It picks its neighbours; those it
    // picked keep it, as it says (STATUS).
    let mut outbound = BTreeSet::new();
    for member in &current {
        if member == state.local_id()
            || unreachable.contains(member)
            || transports
                .get(member)
                .is_some_and(|on| !on.is_disjoint(&local_transports))
        {
            continue;
        }
        if let Some(peer_state) = state.get_peer_state(member).await {
            let peer_state = peer_state.lock().await;
            if matches!(peer_state.standing, crate::state::Standing::Member { .. })
                && peer_state.syncs_with_us != Some(false)
            {
                outbound.insert(member.clone());
            }
        }
    }
    let unreachable_of = |m: &str| unreachable.contains(m);
    let stable_of = |m: &str| stable.contains(m);
    let inputs = Inputs {
        transports: &transports,
        local_transports: &local_transports,
        k,
        unreachable: &unreachable_of,
        stable: &stable_of,
    };
    let topology = state.topology();
    topology.maintain(state, &inputs).await;
    let version = (topology.name(), topology.version());
    let previous = {
        let view = state.neighbours.read_or_recover();
        if view.as_ref().is_some_and(|view| {
            view.version == version
                && view.members == current
                && view.unlisted == unlisted
                && view.outbound == outbound
                && view.k == k
                && view.unreachable == unreachable
                && view.stable == stable
                && view.transports == transports
                && view.local_transports == local_transports
                && view.recorded == recorded
        }) {
            return None;
        }
        view.as_ref().map(|view| {
            // The member list's graph stays while the member list does; with the
            // records, while they do too.
            let same_list = view.members == current
                && view.k == k
                && view.stable == stable
                && view.transports == transports;
            let whole = if same_list {
                view.whole.clone()
            } else {
                Default::default()
            };
            let working = if same_list && view.recorded == recorded {
                view.working.clone()
            } else {
                Default::default()
            };
            (view.neighbours.clone(), whole, working)
        })
    };
    let (previous, whole, working) = match previous {
        Some((previous, whole, working)) => (Some(previous), whole, working),
        None => (None, Default::default(), Default::default()),
    };

    // Before the first view everyone known counted as a neighbour.
    let previous = match previous {
        Some(previous) => previous,
        None => state.peer_states.read().await.keys().cloned().collect(),
    };
    let mut computed = topology.neighbours(state.local_id(), &inputs);
    computed.extend(outbound.iter().cloned());
    #[cfg(test)]
    let listed_neighbours = computed.len();
    computed.extend(unlisted.iter().cloned());
    let change = NeighbourChange {
        added: computed
            .iter()
            .filter(|m| !previous.contains(*m))
            .cloned()
            .collect(),
        removed: previous
            .iter()
            .filter(|m| !computed.contains(*m))
            .cloned()
            .collect(),
    };
    info!(
        members = current.len(),
        unreachable = unreachable.len(),
        stable = stable.len(),
        k,
        neighbours = computed.len(),
        unlisted = unlisted.len(),
        outbound = outbound.len(),
        added = change.added.len(),
        removed = change.removed.len(),
        "Neighbours computed from the member list"
    );
    #[cfg(test)]
    let reachable_others = current
        .iter()
        .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m))
        .count();
    *state.neighbours.write_or_recover() = Some(NeighbourView {
        #[cfg(test)]
        full_mesh: listed_neighbours >= reachable_others,
        members: current,
        k,
        unreachable,
        stable,
        transports,
        local_transports,
        unlisted,
        outbound,
        neighbours: computed,
        version,
        whole,
        recorded,
        working,
    });
    // New neighbours are kept alive and judged.
    state.peers_due.now();
    Some(change)
}

#[cfg(test)]
mod tests;