//! Passing changes on in a partial mesh (`swarm::topology`).
//!
//! A node pushes its own changes to its neighbours, but nothing it receives goes any
//! further by itself. 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 whose heads differ). A
//! neighbour that sees heads it lacks *pulls*: 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. Each node
//! receives a change about once; the rest is announcements. 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).
//!
//! Links start *lazy*: changes are announced over them and pulled. A pull that brought
//! changes we lacked makes the link *eager* in that direction (Plumtree's GRAFT): our
//! STATUS asks the neighbour for its changes as soon as it applies them (`eager`), and
//! it sends them right away as sync messages, without announcing or waiting
//! (`forward_eagerly`). Changes we had already over an eager link make it lazy again
//! (Plumtree's PRUNE): another path brought them first. So eager links form where
//! changes actually flow, over links that deliver, and the announcements on the others
//! are what catches whatever they miss. A link starts lazy again whenever its sync does
//! (`PeerState::reset_sync`: a new neighbour, dormancy, a new run of the peer).
//!
//! 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.
const PULL_TIMEOUT: Duration = Duration::from_secs(15);

/// 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);

/// After changes from `from` were applied and our heads changed: tell our other
/// neighbours, in the next announcement round.
pub(crate) async fn announce(state: &AppState, from: &str) {
    if !state.initial_sync_done.load(Ordering::Acquire) {
        return;
    }
    forward_eagerly(state, from).await;
    let schedule = {
        let mut pending = state.announce_pending.lock_or_recover();
        match pending.as_mut() {
            Some(senders) => {
                senders.insert(from.to_owned());
                false
            }
            None => {
                *pending = Some(HashSet::from([from.to_owned()]));
                true
            }
        }
    };
    if !schedule {
        return;
    }
    let state = state.clone();
    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(ANNOUNCE_DELAY + Duration::from_millis(jitter)).await;
        let senders = state
            .announce_pending
            .lock_or_recover()
            .take()
            .unwrap_or_default();
        announce_now(&state, &senders).await;
    });
}

/// What `from` sent us, at once to the neighbours that asked for our changes as we apply
/// them (eager links): a sync message each, with what it lacks (the sync worker's).
async fn forward_eagerly(state: &AppState, from: &str) {
    for (peer, peer_state) in state.peers_snapshot().await {
        if peer == from || !topology::is_neighbour(state, &peer) {
            continue;
        }
        if peer_state.lock().await.eager_to {
            state.stats.eager_forward();
            state.request_sync(&peer);
        }
    }
}

/// 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, senders: &HashSet<String>) {
    let neighbours = unreached(state, senders).await;
    if neighbours.is_empty() {
        return;
    }
    state.stats.announce_round();
    for peer in neighbours {
        if let Err(error) = announce_heads(state, &peer).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 we forward to eagerly, which was sent the changes already.
async fn unreached(state: &AppState, senders: &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()) || !topology::is_neighbour(state, &peer) {
            continue;
        }
        let peer_state = peer_state.lock().await;
        // Sent what we have as we applied it (`forward_eagerly`).
        let eager = peer_state.eager_to;
        let has_ours = peer_state
            .last_heads
            .as_ref()
            .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1);
        drop(peer_state);
        if !eager && !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>,
}

/// A keepalive from `peer`: if it has heads we lack, 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 (`changed`: it brought changes we lacked): a
/// pull from it is answered, and the next waiting neighbour's turn comes. A pull that
/// brought changes makes the link eager (`graft`).
pub(crate) async fn pulled(state: &AppState, peer: &str, changed: bool) {
    let next = {
        let mut pulling = state.pulling.lock_or_recover();
        if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
            return;
        }
        pulling.current = None;
        !pulling.waiting.is_empty()
    };
    if changed {
        graft(state, peer).await;
    }
    if next {
        let state = state.clone();
        tokio::spawn(async move { pull_next(&state).await });
    }
}

/// `peer` brought changes we lacked when we pulled: from now on we want them as it
/// applies them, and our STATUS says so (Plumtree's GRAFT).
async fn graft(state: &AppState, peer: &str) {
    if !set_eager_from(state, peer, true).await {
        return;
    }
    state.stats.link_grafted();
    debug!(peer = %peer, "Link eager: we want its changes as it applies them");
    if let Err(error) = announce_heads(state, peer).await {
        debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is eager failed");
    }
}

/// Changes we had already came from `peer` over an eager link: another path brings them
/// first, so announcements will do; our STATUS says so (Plumtree's PRUNE).
pub(crate) async fn prune(state: &AppState, peer: &str) {
    if !set_eager_from(state, peer, false).await {
        return;
    }
    state.stats.link_pruned();
    debug!(peer = %peer, "Link lazy: its changes reach us first another way");
    if let Err(error) = announce_heads(state, peer).await {
        debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed");
    }
}

/// Sets whether we want `peer`'s changes eagerly; whether that changed it.
async fn set_eager_from(state: &AppState, peer: &str, eager: bool) -> bool {
    let Some(peer_state) = state.get_peer_state(peer).await else {
        return false;
    };
    let mut peer_state = peer_state.lock().await;
    let changed = peer_state.eager_from != eager;
    peer_state.eager_from = eager;
    changed
}

/// 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;
        }
    }
}