//! 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 PING (a keepalive: heads, about 100
//! bytes, not the changes). 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 PING and one PONG per neighbour (plus the pulls), whatever the
//! write rate; the delay is the price per hop. 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).
//!
//! In a full mesh the writer reaches everyone itself, so nothing is announced.

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::send_ping;
use crate::swarm::sync::SyncEvent;
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. Added once per
/// hop to propagation; well below a round trip over Mycelium.
const ANNOUNCE_DELAY: 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) {
    // While the view is unknown everyone counts as a neighbour, but that isn't a full
    // mesh: the others may not hear from the writer.
    if !state.initial_sync_done.load(Ordering::Acquire)
        || (topology::view_known(state) && topology::is_full_mesh(state))
    {
        return;
    }
    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 {
        tokio::time::sleep(ANNOUNCE_DELAY).await;
        let senders = state
            .announce_pending
            .lock_or_recover()
            .take()
            .unwrap_or_default();
        announce_now(&state, &senders).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, senders: &HashSet<String>) {
    let skip = match senders.len() {
        1 => senders.iter().next().cloned(),
        _ => None,
    };
    let neighbours: Vec<String> = state
        .peer_states
        .read()
        .await
        .keys()
        .filter(|peer| {
            skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer)
        })
        .cloned()
        .collect();
    for peer in neighbours {
        if let Err(error) = send_ping(state, &peer).await {
            debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
        }
    }
}

/// 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::Keepalive>,
}

/// 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::Keepalive,
) -> 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: Vec<_> = {
        // AutoCommit's lookups take &mut self.
        let mut replica = state.doc.write().await;
        remote
            .heads
            .listed()
            .iter()
            .filter(|head| replica.doc.get_change_by_hash(head).is_none())
            .copied()
            .collect()
    };
    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 {
            pulling.waiting.insert(peer.to_owned(), remote.clone());
            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);
    debug!(peer = %peer, "Neighbour has changes we lack; pulling");
    let _ = state
        .sync_events
        .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
    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_some_and(|(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 });
}

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