//! 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.
//!
//! In a full mesh the writer reaches everyone itself, so nothing is announced.

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

/// After changes from `from` were applied and our heads changed: tell our other
/// neighbours.
pub(crate) async fn announce(state: &AppState, from: &str) {
    if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
        return;
    }
    let neighbours: Vec<String> = state
        .peer_states
        .read()
        .await
        .keys()
        .filter(|peer| peer.as_str() != from && 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;
    }
    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;
    }
}