//! Automerge synchronization: the sync worker, delivery with retries, inbound sync
//! messages, and the change publisher that feeds incremental streams.

use std::collections::HashSet;
use std::sync::Arc;
use std::sync::atomic::Ordering;

use anyhow::{Context, Result, anyhow, bail};
use tokio::sync::mpsc;
use tokio::time::Instant;
use tracing::{debug, info, trace, warn};

use crate::locks::MutexExt;
use crate::state::{AppState, PeerState, Standing};
use crate::swarm::peer::{peer_swarm, send_hello_once};
use crate::swarm::source;
use crate::swarm::throughput::SyncBytes;
use crate::transport::{self, ParsedMessage};

#[derive(Debug)]
pub(crate) enum SyncEvent {
    PeerValidated(String),
    DocumentChanged,
    MessageReceived(String),
    PeerNeedsSync(String),
}

async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
    if state.remake.fetching_snapshot() {
        // Rejoining: the snapshot comes as a download, not in a sync message. Sync
        // starts once it is here (`remake::catchup`).
        trace!("Waiting for the snapshot download; not syncing yet");
        return Ok(());
    }
    let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
        trace!("Peer was forgotten; skipping sync");
        return Ok(());
    };

    let we_ready = state.is_ready().await;

    let validated_swarm = {
        let peer_state = peer_state_mutex.lock().await;

        if peer_state.liveness.is_dormant() {
            // Reactivation re-handshakes, and PeerValidated then restarts sync.
            trace!("Skipping sync with dormant peer");
            return Ok(());
        }

        if peer_state
            .epoch
            .is_some_and(|epoch| epoch != state.local_epoch())
        {
            // Unrelated document histories must never be synced.
            trace!("Skipping sync with peer on another epoch");
            return Ok(());
        }

        let Standing::Member { swarm_id } = &peer_state.standing else {
            return Ok(());
        };

        if let Some(wait) = source::defer(state, peer, &peer_state) {
            trace!(
                wait_ms = wait.as_millis(),
                "Catching up from another peer first; not syncing with this one yet"
            );
            // Done earlier when the initial sync completes (every peer is enqueued
            // then); this only matters if the source goes quiet.
            state.request_sync_after(peer, wait);
            return Ok(());
        }

        if peer_state
            .sync
            .leave_opening(we_ready, peer_state.reports_ready)
        {
            // Its own opening (or anything else from it) brings us back here.
            trace!("Leaving the opening sync message to the peer, which isn't ready");
            return Ok(());
        }

        if !peer_state.sync.has_opened() && !crate::swarm::topology::is_neighbour(state, peer) {
            // We only answer peers that aren't our neighbours.
            trace!("Not opening sync with a peer that isn't our neighbour");
            return Ok(());
        }

        swarm_id.clone()
    };

    let swarm_id = state
        .swarm_id()
        .await
        .ok_or_else(|| anyhow!("Cannot sync without an established swarm ID"))?;

    if validated_swarm != swarm_id {
        bail!("Peer {peer} belongs to a different swarm");
    }

    {
        let mut peer_state = peer_state_mutex.lock().await;
        if let Some(sending) = &mut peer_state.sync.sending {
            sending.regenerate = true;
            return Ok(());
        }
        if peer_state
            .sync
            .retry_at
            .is_some_and(|at| Instant::now() < at)
        {
            // The retry timer enqueues the peer once it's due.
            return Ok(());
        }
        if peer_state.sync.delta_offered {
            // Its answer, or the reply timeout, brings us back here.
            trace!("A delta is offered; waiting for the peer");
            return Ok(());
        }
    }

    let (session, encoded) = {
        let doc = Arc::clone(&state.doc);
        let psm = Arc::clone(&peer_state_mutex);
        let transport = Arc::clone(&state.transport);
        let peer_owned = peer.to_owned();
        let ready = state.is_ready().await;
        // Before this message is counted.
        let bytes_before = SyncBytes::of(&state.traffic, peer);
        // Lower towards a peer a message that large wouldn't reach whole: a slow link
        // takes a delta chunk by chunk, where a message whole would fail and be sent
        // again, whole, every time.
        let rate = state
            .throughput
            .lock_or_recover()
            .get(peer)
            .map(|estimate| estimate.bytes_per_sec);
        let failed = peer_state_mutex.lock().await.failed_whole(Instant::now());
        let delta_threshold = state.blobs.delta_threshold_for(rate, failed);

        let stats = Arc::clone(&state.stats);
        let clock = Arc::clone(&state.clock);
        // The session the message comes from, for its send's outcome.
        let (session, generated) = tokio::task::spawn_blocking(
            move || -> (u64, Result<Option<transport::EncodedMessage>>) {
                // Label the message with the epoch of the document it was generated
                // from, read under the same lock.
                let (session, epoch, epoch_seq, body, writer, delivered) = {
                    let replica = doc.blocking_read();
                    let mut peer_state = psm.blocking_lock();
                    let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
                    // Whose change it passes on, if any: this message carries it.
                    let writer = peer_state.tree_writer.take();
                    // To a peer that had nothing we lack, with nothing awaited: taken
                    // as delivered (`SyncSession::assume_delivered`).
                    let level = peer_state.sync.reply.is_none()
                        && peer_state.sync.have_all_theirs(&replica.doc);
                    // Under the same locks as the epoch label: the message belongs to
                    // this document.
                    // The same delta as last time, the peer's heads and ours unchanged
                    // (it didn't answer in time): offered again without generating the
                    // message it replaces, which may take seconds when large.
                    let again = peer_state.sync.offer_again(
                        &replica.doc,
                        replica.heads(),
                        delta_threshold,
                        Instant::now(),
                    );
                    let body = if let Some(from) = again {
                        let spec = transport::DeltaSpec::new(from, replica.heads().to_vec());
                        debug!(
                            from = spec.from.len(),
                            to = spec.to.len(),
                            "Offering the same changes as a blob again"
                        );
                        Some(transport::SyncBody::Delta(spec))
                    } else {
                        let message =
                            peer_state
                                .sync
                                .generate(&replica.doc, Instant::now(), bytes_before);
                        message.and_then(|payload| {
                            if payload.0.len() <= delta_threshold {
                                return Some(transport::SyncBody::Message(payload));
                            }
                            // Too large for one message: all or nothing, and on a slow link
                            // within the transport's delivery limit. The peer fetches it in
                            // chunks.
                            let from = peer_state
                                .sync
                                .withdraw_for_delta(&replica.doc, Instant::now());
                            let spec = transport::DeltaSpec::new(from, replica.heads().to_vec());
                            if spec.to.iter().all(|head| spec.from.contains(head)) {
                                // It has our heads: nothing to offer (Automerge thought it
                                // lacked what it has, as from a view of it that was old).
                                debug!(
                                    message_bytes = payload.0.len(),
                                    "Sync message too large, but the peer has our heads; not sent"
                                );
                                peer_state.sync.nothing_to_send();
                                return None;
                            }
                            // And why it is too large here: the threshold, lowered by the
                            // throughput measured to the peer or a message that failed whole
                            // to it lately (`Blobs::delta_threshold_for`).
                            info!(
                                message_bytes = payload.0.len(),
                                threshold = delta_threshold,
                                throughput = ?rate.map(|rate| rate as u64),
                                failed_whole = ?failed,
                                from = spec.from.len(),
                                to = spec.to.len(),
                                "Sync message too large; offering the changes as a blob"
                            );
                            peer_state
                                .sync
                                .offered(&spec.from, &spec.to, payload.0.len());
                            Some(transport::SyncBody::Delta(spec))
                        })
                    };
                    let delivered = level && matches!(body, Some(transport::SyncBody::Message(_)));
                    if delivered {
                        peer_state.sync.assume_delivered(replica.heads());
                        if let Some(transport::SyncBody::Message(payload)) = &body {
                            stats.delivered_sent(payload.0.len());
                        }
                    }
                    (
                        peer_state.sync.id(),
                        epoch,
                        epoch_seq,
                        body,
                        writer,
                        delivered,
                    )
                };
                let Some(body) = body else {
                    return (session, Ok(None));
                };
                // What the envelope adds to an Automerge message (`swarm::stats`), and of
                // that, the epoch's tag (msgpack: 12 bytes, up to 20 for generations past
                // 127).
                let inner = match &body {
                    transport::SyncBody::Message(payload) => Some(payload.0.len()),
                    transport::SyncBody::Delta(_) => None,
                };
                let ids = rmp_serde::encode::to_vec(&transport::EpochTag::of(epoch))
                    .map_or(0, |tag| tag.len());
                let envelope = transport::SyncEnvelope {
                    epoch: transport::EpochTag::of(epoch),
                    epoch_seq,
                    body,
                    ready,
                    writer,
                    delivered,
                    sent: clock.now_millis(),
                };
                let encoded = transport
                    .encode(&peer_owned, &ParsedMessage::Sync(envelope))
                    .map(Some);
                if let (Ok(Some(encoded)), Some(inner)) = (&encoded, inner) {
                    stats.sync_framing(encoded.payload_len.saturating_sub(inner), ids);
                }
                (session, encoded)
            },
        )
        .await
        .context("Sync generation task panicked")?;
        let generated = match generated {
            Ok(generated) => generated,
            Err(error) => {
                // Not encodable, so nothing went out: take back what it would have carried.
                let _ = peer_state_mutex
                    .lock()
                    .await
                    .sync
                    .send_failed(session, Instant::now());
                return Err(error);
            }
        };

        let Some(encoded) = generated else {
            return Ok(());
        };
        // Its answer is due.
        let reply_due = peer_state_mutex
            .lock()
            .await
            .sync
            .reply
            .as_ref()
            .map(|wait| wait.deadline);
        if let Some(at) = reply_due {
            state.peers_due.by(at);
        }

        (session, encoded)
    };

    // Off the sync worker: a slow send (a large message, a busy transport) mustn't hold
    // up every other peer's. One message per peer is in flight (`sending`), so a peer's
    // messages still go out in order.
    let state = state.clone();
    let peer = peer.to_owned();
    tokio::spawn(async move {
        if let Err(error) = send_generated(&state, &peer, &peer_state_mutex, session, encoded).await
        {
            warn!(peer = %peer, error = %format!("{error:#}"), "Sync processing failed");
        }
    });
    Ok(())
}

/// Hands a message generated from session `session` to the transport and settles it.
async fn send_generated(
    state: &AppState,
    peer: &str,
    peer_state_mutex: &tokio::sync::Mutex<PeerState>,
    session: u64,
    encoded: transport::EncodedMessage,
) -> Result<()> {
    // No state locks are held across the send.
    match state.transport.send_encoded(&encoded).await {
        Ok(()) => {
            let regenerate = peer_state_mutex.lock().await.sync.send_succeeded(session);

            trace!(
                regenerate,
                "Outbound sync message accepted by the transport"
            );

            if regenerate {
                state.request_sync(peer);
            }

            Ok(())
        }
        Err(error) => {
            let delay = {
                let mut peer_state = peer_state_mutex.lock().await;
                let now = Instant::now();
                peer_state.whole_failed(encoded.payload_len, now);
                peer_state.sync.send_failed(session, now)
            };
            let Some(delay) = delay else {
                // The new session may not have opened yet.
                debug!("Sync message refused, but a reset already discarded it");
                state.request_sync(peer);
                return Ok(());
            };

            warn!(
                retry_ms = delay.as_millis(),
                error = %format!("{error:#}"),
                "Sync delivery failed; generating a fresh message later"
            );

            state.request_sync_after(peer, delay);

            Ok(())
        }
    }
}

pub(crate) async fn handle_sync(
    state: &AppState,
    peer: &str,
    envelope: transport::SyncEnvelope,
) -> Result<()> {
    debug!(peer = %peer, "Processing sync envelope");
    // This message is counted already (`transport::traffic`).
    let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer));
    // How long it took to come, where we know the peer's clock (`swarm::hop`).
    let hop_delay = state.transport.clock_offset(peer).map(|offset| {
        crate::swarm::hop::delay_ms(envelope.sent, state.clock.now_millis(), offset.offset_ms)
    });

    let peer_state_mutex = state.get_or_create_peer_state(peer).await;

    let needs_hello = {
        let mut peer_state = peer_state_mutex.lock().await;
        if let Some(delay_ms) = hop_delay {
            peer_state.hop_delays.record(delay_ms, arrived.0);
        }

        match &peer_state.standing {
            Standing::Stranger | Standing::Admitted => {
                // Not processed: our handshake reply resets the session anyway, and the
                // sender's reply timeout resends what still matters.
                debug!(peer = %peer, "Dropping sync message from a peer that isn't a member yet");
                true
            }
            // Its swarm was proven by the handshake (the envelope doesn't say it again).
            Standing::Member { swarm_id } => match state.swarm_id().await {
                Some(ours) if &ours == swarm_id => false,
                _ => bail!("Sync from {peer}, a member of swarm {swarm_id}, which isn't ours"),
            },
        }
    };

    if needs_hello {
        // The peer treats us as a member but we don't know it, typically because we
        // restarted. Our HELLO makes it reset its sync state and reply.
        if let Err(error) = send_hello_once(state, peer).await {
            warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to unknown sync sender failed; will retry");
        }
        return Ok(());
    }

    // The tag names our epoch, or the message is of no epoch we sync on (the sender
    // switched meanwhile, or we did).
    let Some(epoch) = envelope.epoch.resolve([state.local_epoch()]) else {
        debug!(peer = %peer, epoch = %envelope.epoch, "Dropped sync message from another epoch");
        return Ok(());
    };
    let epoch_label = epoch;
    let from_ready = envelope.ready;
    let writer = envelope.writer;
    let delivered = envelope.delivered;
    let payload = match envelope.body {
        transport::SyncBody::Message(payload) => payload,
        transport::SyncBody::Delta(spec) => {
            fetch_delta(state, peer, epoch, spec);
            return Ok(());
        }
    };
    if payload.0.len() > crate::swarm::blob::MAX_SYNC_MESSAGE {
        // Not decoded: a node of this version sends a delta instead.
        warn!(peer = %peer, bytes = payload.0.len(), "Dropping a sync message larger than any sender should send");
        return Ok(());
    }
    state.stats.sync_received(payload.0.len());

    let applied = {
        let doc = Arc::clone(&state.doc);
        let peer_state_mutex = Arc::clone(&peer_state_mutex);
        let stats = Arc::clone(&state.stats);
        let throughput = Arc::clone(&state.throughput);
        let rtt = state
            .lock_rtt()
            .effective(peer, Instant::now(), state.timing.rtt_fresh_for());
        let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
        // While catching up, everything arrives late; that says nothing about propagation.
        let measure_propagation = !check_caught_up;
        let message = payload;
        let peer = peer.to_owned();
        let clock = Arc::clone(&state.clock);
        let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);

        // Some((caught_up, heads changed, carried changes, the new changes' authors with
        // theirs, and whether any has none)) when applied, None when the message belongs
        // to another epoch.
        type Applied = (
            bool,
            bool,
            bool,
            (Vec<(String, Vec<crate::document::ChangeHash>)>, bool),
        );
        tokio::task::spawn_blocking(move || -> Result<Option<Applied>> {
            // Before the lock: decoding a large message takes a while.
            let payload_len = message.0.len();
            let message = crate::document::sync::decode(&message)?;
            let carried = message.carries_changes();
            let sizes = if carried { message.sizes() } else { (0, 0) };
            if carried {
                stats.carrying(sizes);
            }
            let mut replica = doc.blocking_write();

            // Authoritative check, under the same lock as the apply: the document may
            // have switched epochs since the message was accepted.
            if replica.epoch != epoch {
                return Ok(None);
            }

            let mut peer_state = peer_state_mutex.blocking_lock();
            peer_state.reports_ready = Some(from_ready);
            // What it says it has (`swarm::relay::unreached`).
            peer_state.last_heads =
                Some((epoch, transport::HeadsSummary::of(message.heads().to_vec())));
            let heads_before = replica.heads().to_vec();

            let sync = &mut peer_state.sync;
            replica
                .logged(|doc, log| (sync.receive_logged(doc, message, log), true))
                .context("Automerge sync failed")?;
            peer_state.last_sync_rx = Some(Instant::now());
            if let Some(wait) = &peer_state.sync.reply {
                throughput
                    .lock_or_recover()
                    .exchange(&peer, wait.started, arrived, rtt);
            } else if let Some(delay_ms) = hop_delay.filter(|ms| *ms > 0) {
                // Nothing awaited: as long as it took to come.
                throughput.lock_or_recover().message(
                    &peer,
                    payload_len as u64,
                    std::time::Duration::from_millis(delay_ms as u64),
                    arrived.0,
                    rtt,
                );
            }
            peer_state.sync.reply_settled();

            let asked = peer_state
                .sync
                .resend_if_needed(&replica.doc, &peer, Instant::now());
            stats.asked_known_elsewhere(asked);
            replica.refresh_heads();
            let heads_after = replica.heads().to_vec();
            if delivered {
                // Level with the sender now, which takes the message as delivered:
                // nothing to answer (`SyncSession::answered_as_delivered`).
                peer_state
                    .sync
                    .answered_as_delivered(&replica.doc, &heads_after);
            }
            stats.heads_seen(heads_after.len());
            let changed = heads_after != heads_before;
            if changed {
                // They came from it: none goes back (`SyncSession::came_from_peer`).
                let new = peer_state.sync.came_from_peer(&replica.doc, &heads_before);
                stats.new_changes(new);
            }
            if carried && !changed {
                stats.nothing_new(payload_len, writer.is_some(), delivered, sizes);
            }
            // Whose they are, to pass them on along their writers' trees, however they
            // came (along a tree, with an answer, pulled).
            let authors = if changed {
                crate::document::changes::authors_since(&replica.doc, &heads_before)
            } else {
                (Vec::new(), false)
            };
            if changed && measure_propagation {
                // Now by the peers' clocks (their median): this node's own clock skew
                // doesn't count; the writer's still does.
                let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms;
                let delays: Vec<i64> =
                    crate::document::changes::commit_times(&replica.doc, &heads_before)
                        .into_iter()
                        .map(|time| now_ms - time)
                        .collect();
                if writer.is_none() {
                    stats.off_tree(&delays);
                }
                stats.propagated(delays);
            }

            Ok(Some((
                // Only from a ready peer, and only if it has heads at all: empty heads are
                // "contained" in every document, so they prove nothing. A peer still
                // catching up has them, and so does a ready peer whose document is empty
                // (a misconfigured second swarm, or a genesis node in its first seconds,
                // before it registered itself).
                check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc),
                changed,
                carried,
                authors,
            )))
        })
        .await
        .context("Sync task panicked")??
    };

    let Some((caught_up, changed, carried, (authors, unauthored))) = applied else {
        debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
        return Ok(());
    };

    source::reported(state, peer, from_ready, None);

    let just_caught_up = caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel);
    if just_caught_up {
        info!(peer = %peer, "Initial sync complete");
        state
            .stats
            .caught_up(state.traffic.total().received.sync.bytes);
        source::finished(state, peer);
        state.join_progress.notify_one();
    }

    state.notify_store_changed();

    crate::swarm::relay::pulled(state, peer);
    // Caught up: our other neighbours may lack what the source had.
    if changed || just_caught_up {
        let writers: Vec<(u64, Vec<crate::document::ChangeHash>)> = authors
            .into_iter()
            .map(|(author, changes)| (crate::swarm::topology::pair_key(&author), changes))
            .collect();
        crate::swarm::relay::announce(state, peer, &writers, unauthored).await;
    } else if carried
        && let Some(writer) = writer
        && state.initial_sync_done.load(Ordering::Acquire)
    {
        // A writer's changes along its tree that we had already: they came another way
        // first, not saying whose (with an answer to one of our messages, say), so we
        // passed nothing on then. Our children in its tree may lack them still.
        crate::swarm::tree::forward(state, writer, peer, &[]).await;
    }

    state
        .sync_events
        .send(SyncEvent::MessageReceived(peer.to_owned()))
        .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;

    Ok(())
}

/// The peer offered the changes we lack as a blob (`SyncBody::Delta`): fetch them, from
/// it and any other peer that has them, apply them, and sync on. One download per peer
/// at a time: the offer is repeated while it runs.
fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: transport::DeltaSpec) {
    if !state.blobs.start_delta(peer) {
        debug!(peer = %peer, "Delta offered while another downloads; after it");
        return;
    }
    let state = state.clone();
    let peer = peer.to_owned();
    tokio::spawn(async move {
        let applied = download_and_apply_delta(&state, &peer, epoch, spec).await;
        // Those that offered meanwhile send what is still missing: our next message
        // says what we have now.
        for other in state.blobs.end_delta(&peer) {
            if other != peer {
                state.request_sync(&other);
            }
        }
        match applied {
            Ok(changed) => {
                state.notify_store_changed();
                if changed {
                    crate::swarm::relay::announce(&state, &peer, &[], true).await;
                }
            }
            Err(error) => {
                warn!(peer = %peer, error = %format!("{error:#}"), "Delta download failed; syncing on");
            }
        }
        // Our next message says what we have now; the peer answers with the rest.
        state.request_sync(&peer);
    });
}

/// Returns whether our heads changed.
async fn download_and_apply_delta(
    state: &AppState,
    peer: &str,
    epoch: transport::Epoch,
    spec: transport::DeltaSpec,
) -> Result<bool> {
    let mut peers = vec![peer.to_owned()];
    peers.extend(
        state
            .active_peers_on_local_epoch()
            .await
            .into_iter()
            .filter(|other| other != peer),
    );
    if let Some(peer_state) = state.get_peer_state(peer).await {
        let doc = Arc::clone(&state.doc);
        let from = spec.from.clone();
        let known = tokio::task::spawn_blocking(move || {
            let replica = doc.blocking_read();
            peer_state
                .blocking_lock()
                .sync
                .known_beyond(&replica.doc, &from)
        })
        .await
        .context("Delta check task panicked")?;
        if known > 0 {
            // Offered before it heard what we have: it would bring those again. Our
            // next message tells it (`fetch_delta`); it offers afresh if still needed.
            info!(peer = %peer, known, "Offered changes partly here already; not fetched");
            return Ok(false);
        }
    }
    info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob");
    let started = Instant::now();
    let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?;
    info!(
        peer = %peer,
        bytes = bytes.len(),
        took_ms = started.elapsed().as_millis() as u64,
        "Fetched changes offered as a blob"
    );

    let doc = Arc::clone(&state.doc);
    let peer_state = state.get_peer_state(peer).await;
    let stats = Arc::clone(&state.stats);
    // As for sync messages (`receive`): not while catching up.
    let measure_propagation = state.initial_sync_done.load(Ordering::Acquire);
    let clock = Arc::clone(&state.clock);
    let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
    tokio::task::spawn_blocking(move || -> Result<bool> {
        let mut replica = doc.blocking_write();
        if replica.epoch != epoch {
            bail!("the epoch changed during the download");
        }
        let heads_before = replica.heads().to_vec();
        crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
        replica.refresh_heads();
        let missing = crate::document::changes::missing(&replica.doc, &spec.to).len();
        if missing > 0 {
            bail!("{missing} of the offered heads are still missing after loading");
        }
        stats.heads_seen(replica.heads().len());
        let changed = replica.heads() != heads_before.as_slice();
        if changed && measure_propagation {
            // As in `receive`; a delta goes along no writer's tree.
            let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms;
            let delays: Vec<i64> =
                crate::document::changes::commit_times(&replica.doc, &heads_before)
                    .into_iter()
                    .map(|time| now_ms - time)
                    .collect();
            stats.off_tree(&delays);
            stats.propagated(delays);
        }
        if let Some(peer_state) = peer_state {
            let mut peer_state = peer_state.blocking_lock();
            // Sync from it made progress: no heal for a while (`keepalive`).
            peer_state.last_sync_rx = Some(Instant::now());
            if changed {
                // From its history: none goes back (`SyncSession::came_from_peer`), nor
                // is offered back as a delta.
                peer_state.sync.came_from_peer(&replica.doc, &heads_before);
            }
        }
        Ok(changed)
    })
    .await
    .context("Delta apply task panicked")?
}

pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
    let mut dirty_peers = HashSet::<String>::new();
    let mut document_dirty = false;

    info!("Starting the sync worker");

    loop {
        if dirty_peers.is_empty() && !document_dirty {
            match rx.recv().await {
                Some(event) => match event {
                    SyncEvent::PeerValidated(peer)
                    | SyncEvent::MessageReceived(peer)
                    | SyncEvent::PeerNeedsSync(peer) => {
                        dirty_peers.insert(peer);
                    }
                    SyncEvent::DocumentChanged => {
                        document_dirty = true;
                    }
                },
                None => {
                    warn!("Sync event channel closed");
                    return;
                }
            }
        }

        // Coalesce event bursts.
        while let Ok(event) = rx.try_recv() {
            match event {
                SyncEvent::PeerValidated(peer)
                | SyncEvent::MessageReceived(peer)
                | SyncEvent::PeerNeedsSync(peer) => {
                    dirty_peers.insert(peer);
                }
                SyncEvent::DocumentChanged => {
                    document_dirty = true;
                }
            }
        }

        // Let pending HTTP task locks wake up and complete their lifecycle transitions gracefully.
        tokio::task::yield_now().await;

        if document_dirty {
            document_dirty = false;

            // Lock and extract only the cloned string keys into a small transient vector
            let peer_keys: Vec<String> = {
                let peers_guard = state.peer_states.read().await;
                peers_guard.keys().cloned().collect()
            };

            // Our own changes go to our neighbours, saying they're ours: they pass them
            // on along our tree (`swarm::tree`). Others only get answers to what they
            // send us.
            let own = crate::swarm::topology::pair_key(state.local_id());
            for peer in peer_keys {
                if peer_swarm(&state, &peer).await.is_some()
                    && crate::swarm::topology::is_neighbour(&state, &peer)
                {
                    if let Some(peer_state) = state.get_peer_state(&peer).await {
                        peer_state.lock().await.tree_writer = Some(own);
                    }
                    dirty_peers.insert(peer);
                }
            }
        }

        let peers: Vec<String> = dirty_peers.drain().collect();

        for peer in peers {
            if let Err(error) = deliver_sync_to_peer(&state, &peer).await {
                warn!(
                    peer = %peer,
                    error = %format!("{error:#}"),
                    "Sync processing failed"
                );
            }
        }
    }
}