//! Passing a writer's changes on along its tree: every member gets them once, over a
//! link that works, and as soon as the member before it has them.
//!
//! The tree is the same on every node, computed from what they share: the graph the
//! member list gives, with the links members recorded as not working taken out and those
//! it lacks that do added (`topology::working_graph`, `swarm::working_links`), and what
//! each member tells its neighbours it reaches (`swarm::reach`). Every member's distance from the writer over that graph orders the
//! members: by distance, then whether it reaches a member a hop closer, then pair key;
//! a member's parent is the neighbour it reaches that comes before it, the closest to
//! the writer first (a link much slower than its fastest to one counting a hop more),
//! then one that reaches closer, then the one it has the shortest round trip to (as it
//! said, a bucket: `rtt::bucket`), then by pair score
//! (`topology::pair_score`). Coming before rules out cycles, and taking only a neighbour
//! it reaches routes around links that don't work: where a writer behind NAT reaches a
//! single member, that one comes first and the others take it.
//!
//! Whose a change is, its author says (every node writes as itself: its node ID, by
//! pair key here): whoever applies changes passes them on to its own children in each of
//! their writers' trees (`forward`), however they came; so does the writer, to all its
//! neighbours, with its own writes. A sync message along a tree also says whose
//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`); like any
//! sync message, it needs no answer where the child had nothing the parent lacks
//! (`SyncSession::assume_delivered`).
//! A member no member it reaches comes before (or whose parent didn't get the change)
//! gets it from the announcements (`swarm::relay`), as before.
//!
//! The graph is computed once per member list (tens of milliseconds for thousands of
//! members), the distances once per writer and graph (`MAX_WRITERS` kept).

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

use tracing::trace;

use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::{reach, rtt, topology};

/// The writers whose distances are kept at most.
const MAX_WRITERS: usize = 256;

type Graph = BTreeMap<String, BTreeSet<String>>;

/// The distances from each writer, over the graph they were computed for.
#[derive(Default)]
pub(crate) struct Trees {
    graph: Option<Arc<Graph>>,
    /// The members by pair key, of that graph.
    by_key: HashMap<u64, String>,
    distances: HashMap<u64, Arc<HashMap<String, u32>>>,
}

/// The graph, and every member's distance from `writer` (by pair key) over it; None
/// while the member list isn't known, or `writer` isn't in it.
fn distances(state: &AppState, writer: u64) -> Option<(Arc<Graph>, Arc<HashMap<String, u32>>)> {
    let graph = topology::working_graph(state)?;
    let mut trees = state.trees.lock_or_recover();
    if !trees
        .graph
        .as_ref()
        .is_some_and(|known| Arc::ptr_eq(known, &graph))
    {
        trees.by_key = graph
            .keys()
            .map(|member| (topology::pair_key(member), member.clone()))
            .collect();
        trees.distances.clear();
        trees.graph = Some(Arc::clone(&graph));
    }
    if let Some(known) = trees.distances.get(&writer) {
        return Some((graph, Arc::clone(known)));
    }
    let from = trees.by_key.get(&writer)?.clone();
    let mut distance = HashMap::from([(from.clone(), 0u32)]);
    let mut queue = VecDeque::from([from]);
    while let Some(member) = queue.pop_front() {
        let next = distance[&member] + 1;
        for neighbour in graph.get(&member).into_iter().flatten() {
            if !distance.contains_key(neighbour) {
                distance.insert(neighbour.clone(), next);
                queue.push_back(neighbour.clone());
            }
        }
    }
    if trees.distances.len() >= MAX_WRITERS {
        trees.distances.clear();
    }
    let distance = Arc::new(distance);
    trees.distances.insert(writer, Arc::clone(&distance));
    Some((graph, distance))
}

/// Our children in `writer`'s tree: our neighbours in the graph whose parent we are,
/// and whom we reach.
///
/// Members are ordered by distance from the writer, then whether they reach a member a
/// hop closer (as they say: else, while it hasn't the change from its own side, it is
/// as good as farther), then pair key; a member's parent is the neighbour it reaches
/// that comes before it, by the same (a much slower link counting a hop more), then
/// round trip, then pair score. So where the writer reaches few
/// (behind NAT), those it reaches come first among the members as far from it, and the
/// others take them as parents. What a member reaches we know of ourselves and our
/// neighbours, as they said; of others, the graph is taken as right (if we're wrong,
/// a change goes twice or by announcement).
#[cfg(test)]
pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
    plan(state, writer)
        .await
        .map(|plan| plan.children)
        .unwrap_or_default()
}

/// Where `member` comes in the writer's tree (by `distance` from it over `graph`):
/// its distance, then whether it reaches a member a hop closer (`reached(by, other)`:
/// whether `by` reaches `other`, as known; the writer does), then its pair key. None if
/// the graph doesn't have it.
fn rank(
    graph: &Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    member: &str,
) -> Option<(u32, bool, u64)> {
    let d = *distance.get(member)?;
    let closer =
        d == 0
            || graph.get(member).into_iter().flatten().any(|other| {
                distance.get(other.as_str()) == Some(&(d - 1)) && reached(member, other)
            });
    Some((d, !closer, topology::pair_key(member)))
}

/// A link to a possible parent this many round-trip buckets slower than the member's
/// fastest to one (`rtt::bucket`; with doubling bounds, at least twice the round trip)
/// counts as a hop more (`parent`): a member whose link to the writer is a slow circuit
/// takes one as close to the writer as itself, if it has a fast link to one, rather than
/// lag behind the others with every change.
const SLOW_BY_BUCKETS: u8 = 2;

/// `member`'s parent in the writer's tree (`rank`): the neighbour it reaches that comes
/// before it, the closest to the writer first (a link `SLOW_BY_BUCKETS` slower than its
/// fastest to one counting a hop more), then one that reaches closer, then the one it
/// has the shortest round trip to, as it said (`round_trip(member, candidate)`, a
/// bucket: `rtt::bucket`), then by pair score; None if none does (or it is the writer).
/// Its round trips are what its neighbours know too, so they compute the same parent.
fn parent<'g>(
    graph: &'g Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    round_trip: &dyn Fn(&str, &str) -> u8,
    member: &str,
) -> Option<&'g str> {
    let its_rank = rank(graph, distance, reached, member)?;
    let candidates: Vec<&str> = graph
        .get(member)
        .into_iter()
        .flatten()
        .map(String::as_str)
        .filter(|candidate| {
            rank(graph, distance, reached, candidate).is_some_and(|r| r < its_rank)
                && reached(member, candidate)
        })
        .collect();
    let fastest = candidates
        .iter()
        .map(|candidate| round_trip(member, candidate))
        .min()?;
    candidates.into_iter().min_by_key(|candidate| {
        let (d, far, _) = rank(graph, distance, reached, candidate).expect("ranked");
        let bucket = round_trip(member, candidate);
        let slow = u32::from(bucket >= fastest.saturating_add(SLOW_BY_BUCKETS));
        (
            d + slow,
            far,
            bucket,
            topology::pair_score(candidate, member),
        )
    })
}

/// `writer`'s whole tree, as far as `reached` knows whom each member reaches and
/// `round_trip` how far (the dashboard's map, from the links members report): every
/// member's parent, None for those without one (they get the writer's changes by
/// announcement). None while the member list isn't known, or the writer isn't in it.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) fn whole(
    state: &AppState,
    writer: &str,
    reached: &dyn Fn(&str, &str) -> bool,
    round_trip: &dyn Fn(&str, &str) -> u8,
) -> Option<BTreeMap<String, Option<String>>> {
    let (graph, distance) = distances(state, topology::pair_key(writer))?;
    Some(
        graph
            .keys()
            .filter(|member| distance.get(member.as_str()).is_some_and(|d| *d > 0))
            .map(|member| {
                let parent =
                    parent(&graph, &distance, reached, round_trip, member).map(str::to_owned);
                (member.clone(), parent)
            })
            .collect(),
    )
}

/// Whom a member reaches, by pair key, sorted (`reach::said`), and its round trip to
/// each (`rtt::bucket`), as it said.
struct Said {
    keys: Vec<u64>,
    round_trips: Option<Vec<u8>>,
}

/// Whom we and our neighbours in `graph` reach, and how far, as far as we know (our
/// own, and what our neighbours said). Of other members, nothing: the graph is taken as
/// right, and their round trips as alike.
struct Known<'g>(HashMap<&'g str, Said>);

impl Known<'_> {
    /// Whether `by` reaches `member`: as it said, or by the graph.
    fn reached(&self, by: &str, member: &str) -> bool {
        self.0
            .get(by)
            .is_none_or(|said| said.keys.binary_search(&topology::pair_key(member)).is_ok())
    }

    /// `by`'s round trip to `member`, as a bucket; `rtt::UNKNOWN_BUCKET` if not said.
    fn round_trip(&self, by: &str, member: &str) -> u8 {
        let key = topology::pair_key(member);
        self.0
            .get(by)
            .and_then(|said| {
                let index = said.keys.binary_search(&key).ok()?;
                said.round_trips.as_ref()?.get(index).copied()
            })
            .unwrap_or(rtt::UNKNOWN_BUCKET)
    }
}

async fn known_reaches<'g>(state: &AppState, graph: &'g Graph) -> Known<'g> {
    let me = state.local_id();
    let mut reaches: HashMap<&str, Said> = HashMap::new();
    if let Some((me, _)) = graph.get_key_value(me) {
        let (keys, round_trips) = reach::published_with_round_trips(state);
        reaches.insert(
            me.as_str(),
            Said {
                keys,
                round_trips: Some(round_trips),
            },
        );
    }
    for member in graph.get(me).into_iter().flatten() {
        if let Some(peer_state) = state.get_peer_state(member).await {
            let peer_state = peer_state.lock().await;
            if let Some(keys) = reach::of(peer_state.reaches.as_ref()) {
                let round_trips = peer_state.round_trips.clone();
                reaches.insert(member.as_str(), Said { keys, round_trips });
            }
        }
    }
    Known(reaches)
}

/// Our own tree, as our writes go along it (`Plan`, as every member computes its part).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct Shape {
    /// The other members in it.
    pub(crate) members: usize,
    /// The most hops from us to one of them.
    pub(crate) depth: u32,
    /// The hops to each, summed (for the mean).
    pub(crate) hops: u64,
    /// Members it doesn't reach: without a parent there (no member they reach comes
    /// before them), below one without, or not connected to us in the member list's
    /// graph. They get our writes by announcement.
    pub(crate) off: usize,
}

/// The shape of our own tree; None while the member list isn't known, or we aren't in
/// it. What members other than us and our neighbours reach is taken from the graph, as
/// in `plan`.
pub(crate) async fn own_shape(state: &AppState) -> Option<Shape> {
    let me = state.local_id();
    let (graph, distance) = distances(state, topology::pair_key(me))?;
    let known = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| known.reached(by, member);
    let round_trip = |by: &str, member: &str| known.round_trip(by, member);
    Some(shape(&graph, &distance, &reached, &round_trip, me))
}

/// `own_shape` over `graph`, from `me` at distance 0.
fn shape(
    graph: &Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    round_trip: &dyn Fn(&str, &str) -> u8,
    me: &str,
) -> Shape {
    let parents: HashMap<&str, Option<&str>> = graph
        .keys()
        .filter(|member| distance.get(member.as_str()).is_some_and(|d| *d > 0))
        .map(|member| {
            let parent = parent(graph, distance, reached, round_trip, member);
            (member.as_str(), parent)
        })
        .collect();
    // Hops from us along parents; None where the chain ends without us. Parents come
    // before their children, so chains end.
    let mut hops: HashMap<&str, Option<u32>> = HashMap::from([(me, Some(0))]);
    fn along<'g>(
        member: &'g str,
        parents: &HashMap<&'g str, Option<&'g str>>,
        hops: &mut HashMap<&'g str, Option<u32>>,
    ) -> Option<u32> {
        if let Some(known) = hops.get(member) {
            return *known;
        }
        let found = parents
            .get(member)
            .copied()
            .flatten()
            .and_then(|parent| along(parent, parents, hops))
            .map(|hops| hops + 1);
        hops.insert(member, found);
        found
    }
    let mut shape = Shape {
        members: 0,
        depth: 0,
        hops: 0,
        off: graph.len() - 1 - parents.len(),
    };
    for member in parents.keys() {
        match along(member, &parents, &mut hops) {
            Some(h) => {
                shape.members += 1;
                shape.depth = shape.depth.max(h);
                shape.hops += u64::from(h);
            }
            None => shape.off += 1,
        }
    }
    shape
}

/// How long our writes take to reach the members along our own tree, by link latency
/// alone (the bench's, which knows each link's): the mean over the members it reaches
/// (and the others do), against two other trees of ours. Our tree as the members chose
/// their parents (`parents`, from `own_parent` on each): of members beyond our
/// neighbours, we don't know the round trips their choice goes by. In one, each member's parent is
/// chosen as now, but among those as close to us and as well placed (`rank`) by the
/// exact latency, not the round-trip bucket it said; in the other, every member takes the fastest path over the links members
/// reach (no tree computed alike everywhere gets below it). In milliseconds, one way.
#[cfg(test)]
#[derive(Debug, Clone, Copy)]
pub(crate) struct Latencies {
    pub(crate) members: usize,
    pub(crate) tree: f64,
    pub(crate) by_latency: f64,
    pub(crate) fastest: f64,
}

/// Our parent in `writer`'s tree, as we choose it (`parent`, by what we and our
/// neighbours said); None if we have none, or no tree is known.
#[cfg(test)]
pub(crate) async fn own_parent(state: &AppState, writer: &str) -> Option<String> {
    our_parent(state, topology::pair_key(writer)).await
}

/// `own_parent`, of the writer by pair key.
async fn our_parent(state: &AppState, writer: u64) -> Option<String> {
    let (graph, distance) = distances(state, writer)?;
    let known = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| known.reached(by, member);
    let round_trip = |by: &str, member: &str| known.round_trip(by, member);
    parent(&graph, &distance, &reached, &round_trip, state.local_id()).map(str::to_owned)
}

/// Whether the trees work, from what arrives (`swarm::stats`, `/admin/stats`): a sync
/// message from `peer` along `writer`'s tree that `carried` changes, where we have a
/// parent in it and it isn't `peer` (a double send, or views of the tree that differ);
/// the changes of `authors` (by author) that came with a message along no tree, where
/// we have a parent in their writer's tree (the tree didn't bring them, or not first).
/// Not while catching up, when everything comes another way.
pub(crate) async fn note_arrival(
    state: &AppState,
    peer: &str,
    writer: Option<u64>,
    carried: bool,
    authors: &[(String, Vec<crate::document::ChangeHash>)],
) {
    match writer {
        Some(writer) if carried => {
            if our_parent(state, writer)
                .await
                .is_some_and(|parent| parent != peer)
            {
                state.stats.tree_from_other();
            }
        }
        None => {
            for (author, changes) in authors {
                if author != state.local_id()
                    && our_parent(state, topology::pair_key(author))
                        .await
                        .is_some()
                {
                    state.stats.tree_missed(changes.len());
                }
            }
        }
        Some(_) => {}
    }
}

/// Where we are in `writer`'s tree, as we see it: our distance from the writer, our
/// parent (`parent`), and the neighbours we reach that come before us (whom we could
/// take). None while the member list isn't known, or the writer or we aren't in it.
#[cfg(test)]
pub(crate) struct Place {
    pub(crate) distance: u32,
    pub(crate) parent: Option<String>,
    pub(crate) before: Vec<String>,
}

#[cfg(test)]
pub(crate) async fn own_place(state: &AppState, writer: &str) -> Option<Place> {
    let (graph, distance) = distances(state, topology::pair_key(writer))?;
    let known = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| known.reached(by, member);
    let round_trip = |by: &str, member: &str| known.round_trip(by, member);
    let me = state.local_id();
    let its_rank = rank(&graph, &distance, &reached, me)?;
    let before = graph
        .get(me)
        .into_iter()
        .flatten()
        .filter(|candidate| {
            rank(&graph, &distance, &reached, candidate).is_some_and(|r| r < its_rank)
                && reached(me, candidate)
        })
        .cloned()
        .collect();
    Some(Place {
        distance: distance[me],
        parent: parent(&graph, &distance, &reached, &round_trip, me).map(str::to_owned),
        before,
    })
}

/// `Latencies` of our own tree, by `latency(from, to)` in milliseconds; None while the
/// member list isn't known, or we aren't in it, or the tree reaches nobody.
#[cfg(test)]
pub(crate) async fn own_latencies(
    state: &AppState,
    parents: &HashMap<String, Option<String>>,
    latency: &(dyn Fn(&str, &str) -> f64 + Sync),
) -> Option<Latencies> {
    let me = state.local_id();
    let (graph, distance) = distances(state, topology::pair_key(me))?;
    let known = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| known.reached(by, member);
    let round_trip = |by: &str, member: &str| known.round_trip(by, member);
    let members: Vec<&str> = graph
        .keys()
        .map(String::as_str)
        .filter(|member| distance.get(*member).is_some_and(|d| *d > 0))
        .collect();
    let now: HashMap<&str, Option<&str>> = members
        .iter()
        .map(|member| {
            let chosen = match parents.get(*member) {
                Some(chosen) => chosen.as_deref(),
                None => parent(&graph, &distance, &reached, &round_trip, member),
            };
            (*member, chosen)
        })
        .collect();
    let faster: HashMap<&str, Option<&str>> = members
        .iter()
        .map(|member| {
            let its_rank = rank(&graph, &distance, &reached, member);
            let candidate = graph
                .get(*member)
                .into_iter()
                .flatten()
                .map(String::as_str)
                .filter(|candidate| {
                    its_rank.is_some_and(|its| {
                        rank(&graph, &distance, &reached, candidate).is_some_and(|r| r < its)
                    }) && reached(member, candidate)
                })
                .min_by(|a, b| {
                    let (da, fa, _) = rank(&graph, &distance, &reached, a).expect("ranked");
                    let (db, fb, _) = rank(&graph, &distance, &reached, b).expect("ranked");
                    (da, fa)
                        .cmp(&(db, fb))
                        .then(latency(a, member).total_cmp(&latency(b, member)))
                        .then(topology::pair_score(a, member).cmp(&topology::pair_score(b, member)))
                });
            (*member, candidate)
        })
        .collect();
    // Arrival along parents; None where the chain ends without us. Parents come
    // before their children, so chains end.
    fn arrival<'g>(
        member: &'g str,
        parents: &HashMap<&'g str, Option<&'g str>>,
        latency: &dyn Fn(&str, &str) -> f64,
        known: &mut HashMap<&'g str, Option<f64>>,
    ) -> Option<f64> {
        if let Some(at) = known.get(member) {
            return *at;
        }
        let at = parents.get(member).copied().flatten().and_then(|parent| {
            arrival(parent, parents, latency, known).map(|at| at + latency(parent, member))
        });
        known.insert(member, at);
        at
    }
    let arrivals = |parents: &HashMap<&str, Option<&str>>| -> HashMap<&str, f64> {
        let mut known = HashMap::from([(me, Some(0.0))]);
        members
            .iter()
            .filter_map(|member| {
                arrival(member, parents, latency, &mut known).map(|at| (*member, at))
            })
            .collect()
    };
    let (now, faster) = (arrivals(&now), arrivals(&faster));
    // The fastest paths: Dijkstra over the links members reach (a parent `p` of
    // `c` where `c` reaches `p`), from us.
    let mut fastest: HashMap<&str, f64> = HashMap::from([(me, 0.0)]);
    let mut done = std::collections::HashSet::new();
    while let Some((&next, &at)) = fastest
        .iter()
        .filter(|(member, _)| !done.contains(*member))
        .min_by(|a, b| a.1.total_cmp(b.1))
    {
        done.insert(next);
        for other in graph.get(next).into_iter().flatten().map(String::as_str) {
            if !reached(other, next) {
                continue;
            }
            let through = at + latency(next, other);
            if fastest.get(other).is_none_or(|known| through < *known) {
                fastest.insert(other, through);
            }
        }
    }
    // Over the members all three reach (the other tree may leave out others than ours).
    let common: Vec<&str> = now
        .keys()
        .copied()
        .filter(|member| faster.contains_key(member) && fastest.contains_key(member))
        .collect();
    if common.is_empty() {
        return None;
    }
    let mean = |times: &HashMap<&str, f64>| {
        common.iter().map(|member| times[member]).sum::<f64>() / common.len() as f64
    };
    Some(Latencies {
        members: common.len(),
        tree: mean(&now),
        by_latency: mean(&faster),
        fastest: mean(&fastest),
    })
}

/// What `writer`'s tree means for us: our children in it, and our neighbours it leaves
/// out (`uncovered`).
pub(crate) struct Plan {
    pub(crate) children: Vec<String>,
    /// Those of our children that reach no member but us, as they said: nobody else's
    /// slow round tells them of a change, should our message to them be lost (it is
    /// taken as delivered, `SyncSession::assume_delivered`), so ours does.
    pub(crate) alone: Vec<String>,
    /// Our neighbours that no member they reach comes before (the tree reaches them only
    /// by announcement), and those not in the member list's graph (whom no tree has).
    pub(crate) uncovered: Vec<String>,
    /// Our neighbours that get the writer's changes from another parent, and the writer
    /// itself: they have them, or will, without us (`forward`).
    pub(crate) covered: Vec<String>,
}

/// Our part in `writer`'s tree (`children`, `Plan`); None while the member list isn't
/// known, or the writer or we aren't in it.
pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
    let (graph, distance) = distances(state, writer)?;
    let me = state.local_id();
    if !distance.contains_key(me) {
        return None;
    }
    let known = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| known.reached(by, member);
    let round_trip = |by: &str, member: &str| known.round_trip(by, member);
    let my_rank = rank(&graph, &distance, &reached, me)?;
    let my_key = topology::pair_key(me);
    let mut children = Vec::new();
    let mut alone = Vec::new();
    let mut uncovered = Vec::new();
    let mut covered = Vec::new();
    for member in graph.get(me).into_iter().flatten() {
        let Some(its_rank) = rank(&graph, &distance, &reached, member) else {
            continue;
        };
        if distance[member.as_str()] == 0 {
            covered.push(member.clone());
            continue;
        }
        match parent(&graph, &distance, &reached, &round_trip, member) {
            None => uncovered.push(member.clone()),
            Some(parent)
                if parent == me && its_rank > my_rank && topology::reaches(state, member).await =>
            {
                if known
                    .0
                    .get(member.as_str())
                    .is_some_and(|said| said.keys.iter().all(|key| *key == my_key))
                {
                    alone.push(member.clone());
                }
                children.push(member.clone());
            }
            Some(parent) if parent != me => covered.push(member.clone()),
            // Ours by the graph, but we don't reach it, or it comes before us.
            Some(_) => {}
        }
    }
    // Neighbours the member list's graph doesn't give us (not listed yet, or only
    // connecting out): no tree has them. From the neighbours, not every peer known: this
    // runs for every change passed on.
    for peer in topology::our_neighbours(state) {
        if !graph.get(me).is_some_and(|ours| ours.contains(&peer))
            && state.get_peer_state(&peer).await.is_some()
        {
            uncovered.push(peer);
        }
    }
    Some(Plan {
        children,
        alone,
        uncovered,
        covered,
    })
}

/// We applied `changes` of `writer`'s that `from` passed on (none if we had them
/// already): to our children in its tree (but `from`), a sync message each, saying
/// whose. Returns our part in the tree (`Plan`), None if we have none (announcements
/// then do it all).
///
/// Our neighbours that get them from another parent (`Plan::covered`) count as having
/// them (`SyncSession::known_to_peer`): else, with writers in turn, each change coming
/// along another tree, the next message we pass on to one would carry all we got since
/// our last to it, mostly what its other parents brought it. Should one lack them after
/// all, it asks, and gets them (`resend_if_needed`). But to our children, those of
/// them that `changes` build on go along, if they were counted so lately
/// (`SyncSession::with_predecessors`).
pub(crate) async fn forward(
    state: &AppState,
    writer: u64,
    from: &str,
    changes: &[crate::document::ChangeHash],
) -> Option<Plan> {
    let mut plan = plan(state, writer).await?;
    if !changes.is_empty() {
        for neighbour in &plan.covered {
            if let Some(peer_state) = state.get_peer_state(neighbour).await {
                peer_state
                    .lock()
                    .await
                    .sync
                    .known_to_peer(changes, tokio::time::Instant::now());
            }
        }
    }
    plan.children.retain(|child| child != from);
    let children = &plan.children;
    let replica = if changes.is_empty() {
        None
    } else {
        Some(state.doc.read().await)
    };
    for child in children {
        if let Some(peer_state) = state.get_peer_state(child).await {
            let mut peer_state = peer_state.lock().await;
            peer_state.tree_writer = Some(writer);
            if let Some(replica) = &replica {
                peer_state.sync.with_predecessors(
                    &replica.doc,
                    changes,
                    tokio::time::Instant::now(),
                );
            }
        }
        state.stats.tree_forward();
        state.request_sync(child);
    }
    if !children.is_empty() {
        trace!(
            writer,
            children = children.len(),
            "Passing a change on along the writer's tree"
        );
    }
    Some(plan)
}

/// Writers' trees as the way changes reach every member
/// (`swarm::dissemination::Dissemination`): `plan`, `forward` and `note_arrival` above.
pub(crate) struct WritersTrees;

#[async_trait::async_trait]
impl crate::swarm::dissemination::Dissemination for WritersTrees {
    fn name(&self) -> &'static str {
        "writers-trees"
    }

    async fn plan(&self, state: &AppState, writer: u64) -> Option<Plan> {
        plan(state, writer).await
    }

    async fn forward(
        &self,
        state: &AppState,
        writer: u64,
        from: &str,
        changes: &[crate::document::ChangeHash],
    ) -> Option<Plan> {
        forward(state, writer, from, changes).await
    }

    async fn note_arrival(
        &self,
        state: &AppState,
        peer: &str,
        writer: Option<u64>,
        carried: bool,
        authors: &[(String, Vec<crate::document::ChangeHash>)],
    ) {
        note_arrival(state, peer, writer, carried, authors).await;
    }
}

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

    /// Of the members as close to the writer and as well placed, the one a member has
    /// the shorter round trip to is its parent, whatever the pair scores.
    #[test]
    fn the_nearer_parent_wins() {
        let mut graph = Graph::new();
        for (a, b) in [("w", "a"), ("w", "b"), ("a", "c"), ("b", "c")] {
            graph.entry(a.into()).or_default().insert(b.into());
            graph.entry(b.into()).or_default().insert(a.into());
        }
        let distance: HashMap<String, u32> = [("w", 0), ("a", 1), ("b", 1), ("c", 2)]
            .into_iter()
            .map(|(member, d)| (member.to_owned(), d))
            .collect();
        let reached = |_: &str, _: &str| true;
        for (near, far) in [("a", "b"), ("b", "a")] {
            let round_trip = |by: &str, member: &str| match (by, member) {
                ("c", m) if m == near => 1,
                ("c", m) if m == far => 5,
                _ => rtt::UNKNOWN_BUCKET,
            };
            assert_eq!(
                parent(&graph, &distance, &reached, &round_trip, "c"),
                Some(near)
            );
        }
    }

    /// A member whose link to the writer is much slower than to another member as close
    /// to the writer, one that comes before it, takes that one; one bucket slower, it
    /// keeps the writer.
    #[test]
    fn a_slow_link_counts_a_hop_more() {
        let mut graph = Graph::new();
        for (a, b) in [("w", "a"), ("w", "x"), ("a", "x")] {
            graph.entry(a.into()).or_default().insert(b.into());
            graph.entry(b.into()).or_default().insert(a.into());
        }
        let distance: HashMap<String, u32> = [("w", 0), ("a", 1), ("x", 1)]
            .into_iter()
            .map(|(member, d)| (member.to_owned(), d))
            .collect();
        let reached = |_: &str, _: &str| true;
        // Whichever of the two comes first (pair key) may be the other's parent.
        let (first, second) = if topology::pair_key("a") < topology::pair_key("x") {
            ("a", "x")
        } else {
            ("x", "a")
        };
        for (to_writer, expected) in [(6, first), (5, "w")] {
            let round_trip = |by: &str, member: &str| match (by, member) {
                (by, "w") if by == second => to_writer,
                (by, m) if by == second && m == first => 4,
                _ => rtt::UNKNOWN_BUCKET,
            };
            assert_eq!(
                parent(&graph, &distance, &reached, &round_trip, second),
                Some(expected),
                "{to_writer}"
            );
            assert_eq!(
                parent(&graph, &distance, &reached, &round_trip, first),
                Some("w")
            );
        }
    }

    /// Hops along parents from us; members without a parent, below one without, or
    /// not connected count as off the tree.
    #[test]
    fn our_tree_in_numbers() {
        let link = |graph: &mut Graph, a: &str, b: &str| {
            graph.entry(a.into()).or_default().insert(b.into());
            graph.entry(b.into()).or_default().insert(a.into());
        };
        let mut graph = Graph::new();
        link(&mut graph, "me", "a");
        link(&mut graph, "a", "b");
        link(&mut graph, "me", "c");
        link(&mut graph, "c", "e");
        graph.insert("d".into(), BTreeSet::new());
        let distance: HashMap<String, u32> = [("me", 0), ("a", 1), ("c", 1), ("b", 2), ("e", 2)]
            .into_iter()
            .map(|(member, d)| (member.to_owned(), d))
            .collect();
        // c doesn't reach us: no parent, nor e below it.
        let reached = |by: &str, member: &str| !(by == "c" && member == "me");
        assert_eq!(
            shape(
                &graph,
                &distance,
                &reached,
                &|_, _| rtt::UNKNOWN_BUCKET,
                "me"
            ),
            Shape {
                members: 2,
                depth: 2,
                hops: 3,
                off: 3,
            }
        );
    }
}