//! 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.
//!
//! A sync message that passes a writer's change on says whose (`SyncEnvelope::writer`).
//! Whoever applies one passes it on to its own children in that writer's tree
//! (`forward`), however it got it; so does the writer, to all its neighbours, with its
//! own writes. 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).
pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
    let Some((graph, distance)) = distances(state, writer) else {
        return Vec::new();
    };
    let me = state.local_id();
    if !distance.contains_key(me) {
        return Vec::new();
    }
    // Whom we and our neighbours reach, by pair key.
    let mut reaches: HashMap<&str, Vec<u64>> = HashMap::new();
    reaches.insert(me, 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(state, peer_state.lock().await.reaches.as_ref())
        {
            reaches.insert(member.as_str(), keys);
        }
    }
    let reached = |by: &str, member: &str| {
        reaches
            .get(by)
            .is_none_or(|keys| keys.contains(&topology::pair_key(member)))
    };
    // Whether a member reaches one a hop closer to the writer (the writer: yes).
    let closer = |member: &str| {
        let d = distance[member];
        d == 0
            || graph.get(member).into_iter().flatten().any(|other| {
                distance.get(other.as_str()) == Some(&(d - 1)) && reached(member, other)
            })
    };
    let rank = |member: &str| {
        distance
            .get(member)
            .map(|d| (*d, !closer(member), topology::pair_key(member)))
    };
    let Some(my_rank) = rank(me) else {
        return Vec::new();
    };
    let mut children = Vec::new();
    for member in graph.get(me).into_iter().flatten() {
        let Some(its_rank) = rank(member) else {
            continue;
        };
        if its_rank <= my_rank {
            continue;
        }
        let parent = graph
            .get(member)
            .into_iter()
            .flatten()
            .filter(|candidate| rank(candidate).is_some_and(|r| r < its_rank))
            .filter(|candidate| reached(member, candidate))
            .min_by_key(|candidate| {
                let (d, far, _) = rank(candidate).expect("ranked");
                (d, far, topology::pair_score(candidate, member))
            });
        if parent.is_some_and(|parent| parent == me) && topology::reaches(state, member).await {
            children.push(member.clone());
        }
    }
    children
}

/// We applied a change of `writer`'s that `from` passed on: to our children in its
/// tree (but `from`), a sync message each, saying whose. Returns them.
pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Vec<String> {
    let children: Vec<String> = children(state, writer)
        .await
        .into_iter()
        .filter(|child| child != from)
        .collect();
    for child in &children {
        if let Some(peer_state) = state.get_peer_state(child).await {
            peer_state.lock().await.tree_writer = Some(writer);
        }
        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"
        );
    }
    children
}