//! Passing changes on (`swarm::topology`): along writers' trees (`swarm::tree`), and by
//! announcement and pull where they don't reach.
//!
//! A node pushes its own changes to its neighbours, but nothing it receives goes any
//! further by itself but along a writer's tree. So after applying changes from one
//! neighbour, a node *announces*
//! its new heads to its other neighbours with a STATUS (heads, about 100 bytes, not the
//! changes; no keepalive, and answered only by a neighbour that lacks some of them). A
//! neighbour that sees heads it lacks *pulls* (those it still lacks after a round trip
//! to the announcer: while writes go on, the newest are usually on their way to it along
//! a tree, `pull_after_grace`): it enters the
//! announced heads as the peer's heads in its sync state, Automerge asks for what's
//! missing (`need`), and the announcer answers with exactly those changes. Regular
//! keepalives work the same way, so any gap is noticed within a keepalive interval.
//!
//! Announcements are coalesced: the first applied change schedules one announcement
//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
//! round costs one STATUS per neighbour (plus the pulls and their replies), whatever the
//! write rate; the delay is the price per hop. A neighbour that announced the same heads
//! to us first is skipped, so between two neighbours with the same news usually only one
//! STATUS goes. Without coalescing, every applied sync
//! message would announce to every neighbour, and announcements would outnumber the
//! data they announce (about 15 to 1 at 250 nodes).
//!
//! Changes that say whose they are (the writer's own, and what passes on along its tree)
//! go at once to the receiver's children in the writer's tree (`swarm::tree`). Then the
//! tree carries them, and the round after `ANNOUNCE_DELAY` goes only to the neighbours
//! the tree leaves out (`Plan::uncovered`: no member they reach comes before them). The
//! others are told in a slow round (`SLOW_ANNOUNCE_DELAY`): in case their parent missed
//! the change (it went away, or the nodes' views of the tree differed), coalescing all
//! that came meanwhile. Changes that don't say whose (pulled, a delta, from a node that
//! has no part in the writer's tree) are announced to all, as before.
//!
//! In a full mesh too: the member list makes every member the writer's neighbour, but
//! that says nothing about whether the writer reaches them. One behind NAT may reach a
//! single one of them, one on another transport (on Tor alone) none, and no node can
//! tell which from where it is. A neighbour that has the changes already does nothing
//! with an announcement but answer it.

use std::collections::{HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::time::Duration;

use tokio::time::Instant;
use tracing::debug;

use crate::locks::MutexExt;
use crate::state::{AppState, Standing};
use crate::swarm::peer::announce_heads;
use crate::swarm::topology;
use crate::transport::{self, HeadsSummary};

/// A pull that brought nothing for this long no longer blocks pulling from another peer
/// (sooner once its answer is overdue: `pull_unanswered`).
const PULL_TIMEOUT: Duration = Duration::from_secs(15);

/// The longest a STATUS's heads we lack wait before we pull them (`pull_after_grace`: a
/// round trip to its sender, at most this): over a slow transport (Tor), a change that
/// was missed waits no longer.
const PULL_GRACE_MAX: Duration = Duration::from_secs(1);

/// How long an announcement waits for further changes to ride along, and a random part
/// on top (`ANNOUNCE_JITTER`), so that of two neighbours with the same news one usually
/// announces first and the other, seeing the first's heads are its own, skips it
/// (`unreached`). Added once per hop to propagation; well below a round trip over the
/// transport.
const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
const ANNOUNCE_JITTER: Duration = Duration::from_millis(200);

/// How long the slow round after changes passed on along writers' trees waits: what a
/// tree misses arrives this much later, and while changes keep coming, every neighbour
/// is told at most this often.
const SLOW_ANNOUNCE_DELAY: Duration = Duration::from_secs(2);

/// An announcement round scheduled: for changes from these peers, and the neighbours
/// that were sent them along a writer's tree meanwhile (`swarm::tree`).
#[derive(Default)]
pub(crate) struct Pending {
    senders: HashSet<String>,
    sent: HashSet<String>,
    /// Some change had no tree to go along: every neighbour is told, not only `targets`.
    all: bool,
    /// The neighbours writers' trees leave out (`Plan::uncovered`).
    targets: HashSet<String>,
}

/// After changes from `from` were applied and our heads changed: passed on at once along
/// the trees of their `writers` (by their authors' pair keys); then announced, in the
/// next round to those the trees leave out, and to the others in the slow round. To all
/// in the next round if any change has no tree to go along: none known whose
/// (`unauthored`), a writer not in the member list, or we have no part in its tree.
pub(crate) async fn announce(state: &AppState, from: &str, writers: &[u64], unauthored: bool) {
    if !state.initial_sync_done.load(Ordering::Acquire) {
        return;
    }
    let mut plans = Vec::new();
    let mut treeless = unauthored || writers.is_empty();
    for writer in writers {
        match crate::swarm::tree::forward(state, *writer, from).await {
            Some(plan) => plans.push(plan),
            None => treeless = true,
        }
    }
    let tree = !plans.is_empty();
    let schedule = {
        let mut pending = state.announce_pending.lock_or_recover();
        let schedule = pending.is_none();
        let pending = pending.get_or_insert_with(Pending::default);
        pending.senders.insert(from.to_owned());
        for plan in &plans {
            pending.sent.extend(plan.children.iter().cloned());
            pending.targets.extend(plan.uncovered.iter().cloned());
        }
        pending.all |= treeless;
        schedule
    };
    if schedule {
        round(state, &state.announce_pending, ANNOUNCE_DELAY, false);
    }
    if !tree {
        return;
    }
    let schedule = {
        let mut slow = state.announce_slow.lock_or_recover();
        let schedule = slow.is_none();
        let slow = slow.get_or_insert_with(Pending::default);
        slow.senders.insert(from.to_owned());
        slow.all = true;
        for plan in &plans {
            slow.sent.extend(plan.children.iter().cloned());
        }
        schedule
    };
    if schedule {
        round(state, &state.announce_slow, SLOW_ANNOUNCE_DELAY, true);
    }
}

/// An announcement round of what `pending` collects, `delay` (and a jitter) from now:
/// the `slow` one, or the quick one.
fn round(
    state: &AppState,
    pending: &std::sync::Arc<std::sync::Mutex<Option<Pending>>>,
    delay: Duration,
    slow: bool,
) {
    let state = state.clone();
    let pending = std::sync::Arc::clone(pending);
    tokio::spawn(async move {
        let jitter = crate::transport::paths::random_nonce()
            % u64::try_from(ANNOUNCE_JITTER.as_millis())
                .unwrap_or(1)
                .max(1);
        tokio::time::sleep(delay + Duration::from_millis(jitter)).await;
        let pending = pending.lock_or_recover().take().unwrap_or_default();
        announce_now(&state, &pending, slow).await;
    });
}

/// One announcement round. The peer whose changes we announce has them already, so a
/// round with a single sender skips it; with several, each lacks the others' changes,
/// so every neighbour is told.
async fn announce_now(state: &AppState, pending: &Pending, slow: bool) {
    let mut neighbours = unreached(state, &pending.senders, &pending.sent).await;
    if !pending.all {
        neighbours.retain(|peer| pending.targets.contains(peer));
    }
    if neighbours.is_empty() {
        return;
    }
    state.stats.announce_round(pending.all, slow);
    for peer in neighbours {
        if let Err(error) = announce_heads(state, &peer, false).await {
            debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
        }
    }
}

/// The neighbours to tell of the changes `senders` sent us. The peer whose changes we
/// announce has them already, so with a single sender it is skipped; with several, each
/// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an
/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have; and
/// one `sent` the changes along a writer's tree, which has them already.
async fn unreached(
    state: &AppState,
    senders: &HashSet<String>,
    sent: &HashSet<String>,
) -> Vec<String> {
    let skip = match senders.len() {
        1 => senders.iter().next().cloned(),
        _ => None,
    };
    let ours = {
        let stamp = state.stamp.borrow();
        (stamp.epoch, HeadsSummary::of(stamp.heads.to_vec()))
    };
    let mut unreached = Vec::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        if skip.as_deref() == Some(peer.as_str())
            || sent.contains(&peer)
            || !topology::is_neighbour(state, &peer)
        {
            continue;
        }
        let has_ours = peer_state
            .lock()
            .await
            .last_heads
            .as_ref()
            .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1);
        if !has_ours {
            unreached.push(peer);
        }
    }
    unreached
}

/// Pulls in progress: one neighbour at a time, so each change arrives about once. The
/// neighbours that announced heads we lack meanwhile wait their turn, with what they
/// announced, instead of until their next keepalive.
#[derive(Default)]
pub(crate) struct Pulling {
    /// The neighbour we are pulling from, and since when.
    current: Option<(String, Instant)>,
    /// The latest keepalive of each neighbour waiting to be pulled from.
    waiting: HashMap<String, transport::Status>,
}

/// Whether we lack any of `heads` (with a digest, any of the sample it lists).
pub(crate) async fn lacks(state: &AppState, heads: &HeadsSummary) -> bool {
    let replica = state.doc.read().await;
    !crate::document::changes::missing(&replica.doc, heads.listed()).is_empty()
}

/// A STATUS from `peer` (a keepalive's, or an announcement): if it has heads we lack,
/// pull them, but only those we still lack after a round trip to it (`PULL_GRACE_MAX` at
/// most; at once if we know none). While writes go on, the newest of them are usually
/// on their way to us along their writers' trees, as fast as the STATUS: they come
/// meanwhile, and pulling them would only fetch them twice. What is still missing then
/// was missed.
pub(crate) async fn pull_after_grace(state: &AppState, peer: &str, remote: &transport::Status) {
    if HeadsSummary::of(state.stamp.borrow().heads.to_vec()) == remote.heads {
        return;
    }
    let grace = state
        .lock_rtt()
        .effective(peer, Instant::now(), state.timing.rtt_fresh_for())
        .map(|rtt| rtt.min(PULL_GRACE_MAX));
    if let Some(grace) = grace
        && lacks(state, &remote.heads).await
    {
        tokio::time::sleep(grace).await;
    }
    pull_if_behind(state, peer, remote).await;
}

/// If `peer` has heads we lack (as its STATUS `remote` says), pull them (one peer at a
/// time; a busy slot queues the peer). Returns whether a pull started.
pub(crate) async fn pull_if_behind(
    state: &AppState,
    peer: &str,
    remote: &transport::Status,
) -> bool {
    if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
        return false;
    }
    // The common case, the same heads, needs only the published heads.
    let ours = HeadsSummary::of(state.stamp.borrow().heads.to_vec());
    if ours == remote.heads {
        return false;
    }
    // Those of the listed heads we lack. With a digest only a sample is listed; if we
    // have all of it, a later keepalive lists others.
    let missing = {
        let replica = state.doc.read().await;
        crate::document::changes::missing(&replica.doc, remote.heads.listed())
    };
    if missing.is_empty() {
        return false;
    }
    // All heads, when we know them; otherwise those we lack.
    let theirs = match remote.heads.heads() {
        Some(heads) => heads.to_vec(),
        None => missing,
    };

    let Some(peer_state) = state.get_peer_state(peer).await else {
        return false;
    };
    if !matches!(peer_state.lock().await.standing, Standing::Member { .. }) {
        return false;
    }

    let now = Instant::now();
    {
        let mut pulling = state.pulling.lock_or_recover();
        let busy = matches!(
            &pulling.current,
            Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
        );
        if busy {
            if pulling
                .waiting
                .insert(peer.to_owned(), remote.clone())
                .is_none()
            {
                state.stats.pull_queued();
            }
            return false;
        }
        pulling.current = Some((peer.to_owned(), now));
        pulling.waiting.remove(peer);
    }

    // Automerge asks for whatever these heads need that we don't have.
    peer_state.lock().await.sync.pull_from(theirs);
    state.stats.pull_started();
    debug!(peer = %peer, "Neighbour has changes we lack; pulling");
    state.request_sync(peer);
    true
}

/// A sync message from `peer` was applied: a pull from it is answered, and the next
/// waiting neighbour's turn comes.
pub(crate) fn pulled(state: &AppState, peer: &str) {
    {
        let mut pulling = state.pulling.lock_or_recover();
        if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
            return;
        }
        pulling.current = None;
        if pulling.waiting.is_empty() {
            return;
        }
    }
    let state = state.clone();
    tokio::spawn(async move { pull_next(&state).await });
}

/// Our sync message to `peer` went unanswered for the reply timeout: a pull from it no
/// longer blocks the others, and the next waiting neighbour's turn comes. Otherwise
/// they would wait for `PULL_TIMEOUT`, and then for their next keepalive, as nothing
/// else brings them back.
pub(crate) fn pull_unanswered(state: &AppState, peer: &str) {
    {
        let mut pulling = state.pulling.lock_or_recover();
        if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
            return;
        }
        debug!(peer = %peer, "Pull unanswered; pulling from the next neighbour");
        pulling.current = None;
        if pulling.waiting.is_empty() {
            return;
        }
    }
    let state = state.clone();
    tokio::spawn(async move { pull_next(&state).await });
}

/// Pulls from the first waiting neighbour that still has heads we lack.
async fn pull_next(state: &AppState) {
    loop {
        let (peer, remote) = {
            let mut pulling = state.pulling.lock_or_recover();
            if pulling.current.is_some() {
                return;
            }
            let Some(peer) = pulling.waiting.keys().next().cloned() else {
                return;
            };
            let remote = pulling.waiting.remove(&peer).expect("just listed");
            (peer, remote)
        };
        if pull_if_behind(state, &peer, &remote).await {
            return;
        }
    }
}