//! 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::HashSet;
use std::sync::atomic::Ordering;
use std::time::Duration;

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

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) {
    if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
        return;
    }
    let schedule = {
        let mut pending = state
            .announce_pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        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()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .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");
        }
    }
}

/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time).
pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transport::Keepalive) {
    if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
        return;
    }
    // The common case, the same heads, needs only the cached heads (read lock).
    if HeadsSummary::of(state.doc.read().await.heads().to_vec()) == remote.heads {
        return;
    }
    let missing = match &remote.heads {
        HeadsSummary::Heads(heads) => {
            // AutoCommit's lookups take &mut self.
            let mut replica = state.doc.write().await;
            let missing = heads
                .iter()
                .any(|head| replica.doc.get_change_by_hash(head).is_none());
            if !missing {
                return;
            }
            Some(heads.clone())
        }
        // Can't tell which heads are missing; the digest differs from ours, though.
        HeadsSummary::Digest(_) => None,
    };

    let now = Instant::now();
    {
        let mut pulling = state
            .pulling
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let busy = matches!(
            &*pulling,
            Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
        );
        if busy {
            return;
        }
        *pulling = Some((peer.to_owned(), now));
    }

    let Some(peer_state) = state.get_peer_state(peer).await else {
        return;
    };
    {
        let mut peer_state = peer_state.lock().await;
        if !matches!(peer_state.standing, Standing::Member { .. }) {
            return;
        }
        match missing {
            Some(heads) => {
                // Automerge asks for whatever these heads need that we don't have.
                peer_state.sync.state.their_heads = Some(heads);
                peer_state.sync.state.in_flight = false;
            }
            // A fresh session's opening carries our heads and bloom filter; the answer
            // brings what we lack.
            None => peer_state.reset_sync(),
        }
    }
    debug!(peer = %peer, "Neighbour has changes we lack; pulling");
    let _ = state
        .sync_events
        .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
}

/// A sync message from `peer` was applied: a pull from it is answered.
pub(crate) fn pulled(state: &AppState, peer: &str) {
    let mut pulling = state
        .pulling
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    if pulling.as_ref().is_some_and(|(p, _)| p == peer) {
        *pulling = None;
    }
}