//! 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 (`topology::member_list_graph`: every member's neighbours, as if
//! all reached all) 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, then one that reaches closer, 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, 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::member_list_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, 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)))
}

/// `member`'s parent in the writer's tree (`rank`): the neighbour it reaches that comes
/// before it, the closest to the writer first, then one that reaches closer, then by
/// pair score; None if none does (or it is the writer).
fn parent<'g>(
    graph: &'g Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    member: &str,
) -> Option<&'g str> {
    let its_rank = rank(graph, distance, reached, member)?;
    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)
        })
        .min_by_key(|candidate| {
            let (d, far, _) = rank(graph, distance, reached, candidate).expect("ranked");
            (d, far, topology::pair_score(candidate, member))
        })
}

/// `writer`'s whole tree, as far as `reached` knows whom each member reaches (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,
) -> 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, member).map(str::to_owned);
                (member.clone(), parent)
            })
            .collect(),
    )
}

/// Whom we and our neighbours in `graph` reach, by pair key, as far as we know (our
/// own, and what our neighbours said).
async fn known_reaches<'g>(state: &AppState, graph: &'g Graph) -> HashMap<&'g str, Vec<u64>> {
    let me = state.local_id();
    let mut reaches: HashMap<&str, Vec<u64>> = HashMap::new();
    if let Some((me, _)) = graph.get_key_value(me) {
        reaches.insert(me.as_str(), reach::published(state));
    }
    for member in graph.get(me).into_iter().flatten() {
        if let Some(peer_state) = state.get_peer_state(member).await
            && let Some(keys) = reach::of(peer_state.lock().await.reaches.as_ref())
        {
            reaches.insert(member.as_str(), keys);
        }
    }
    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 reaches = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| {
        reaches
            .get(by)
            .is_none_or(|keys| keys.contains(&topology::pair_key(member)))
    };
    Some(shape(&graph, &distance, &reached, me))
}

/// `own_shape` over `graph`, from `me` at distance 0.
fn shape(
    graph: &Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    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| (member.as_str(), parent(graph, distance, reached, member)))
        .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. In one, each member's parent is chosen as now, but
/// among those as close to us and as well placed (`rank`) by latency before pair
/// score; 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,
}

/// `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,
    latency: &(dyn Fn(&str, &str) -> f64 + Sync),
) -> Option<Latencies> {
    let me = state.local_id();
    let (graph, distance) = distances(state, topology::pair_key(me))?;
    let reaches = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| {
        reaches
            .get(by)
            .is_none_or(|keys| keys.contains(&topology::pair_key(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| (*member, parent(&graph, &distance, &reached, member)))
        .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 reaches = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| {
        reaches
            .get(by)
            .is_none_or(|keys| keys.contains(&topology::pair_key(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, member) {
            None => uncovered.push(member.clone()),
            Some(parent)
                if parent == me && its_rank > my_rank && topology::reaches(state, member).await =>
            {
                if reaches
                    .get(member.as_str())
                    .is_some_and(|keys| 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.
    for (peer, _) in state.peers_snapshot().await {
        if topology::is_neighbour(state, &peer)
            && !graph.get(me).is_some_and(|ours| ours.contains(&peer))
        {
            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)
}

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

    /// 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, "me"),
            Shape {
                members: 2,
                depth: 2,
                hops: 3,
                off: 3,
            }
        );
    }
}