//! The epochs peers advertise: recorded per peer, late messages aside, and compared
//! with ours.

use tokio::time::Instant;
use tracing::{debug, info, warn};

use crate::document::replica::{Epoch, EpochRelation};
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
use crate::transport;

/// Records the epoch a peer advertised and compares it with ours. Returns true when the
/// message's epoch is ours, the only case in which its documents may be synced.
///
/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was
/// sent before the peer's last epoch change and delivered late: its epoch isn't
/// recorded, so it can't flip our view of the peer back. The count starts over when the
/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`).
pub(super) async fn observe_peer_epoch(
    state: &AppState,
    peer: &str,
    remote: Epoch,
    seq: u32,
    run: Option<transport::RunId>,
) -> bool {
    let local = state.local_epoch();
    let relation = local.relation_to(&remote);

    let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
        return relation == EpochRelation::Same;
    };

    let previous = {
        let mut peer_state = peer_state_mutex.lock().await;
        if let Some(run) = run.filter(|run| peer_state.run_id != Some(*run)) {
            peer_state.run_id = Some(run);
            peer_state.epoch_seq = None;
        }
        if peer_state.epoch_seq.is_some_and(|newest| seq < newest) {
            debug!(
                peer = %peer,
                epoch = %remote,
                seq,
                newest = ?peer_state.epoch_seq,
                "Ignoring a late message's epoch"
            );
            return relation == EpochRelation::Same;
        }
        peer_state.epoch_seq = Some(seq);
        let previous = peer_state.epoch.replace(remote);
        if previous.is_some_and(|previous| previous != remote) {
            // The peer switched documents; our sync state describes its old one. Count
            // the restart as sync activity, so the keepalive heal doesn't re-handshake
            // (resetting both sides again) while this fresh sync is under way.
            peer_state.reset_sync();
            peer_state.last_sync_rx = Some(Instant::now());
            // Whether it would sync with us was said about its old epoch.
            peer_state.syncs_with_us = None;
        }
        previous
    };

    // Log transitions only, not every keepalive.
    if previous != Some(remote) {
        match relation {
            EpochRelation::Same => {
                if previous.is_some() {
                    info!(peer = %peer, epoch = %remote, "Peer is on our epoch again; resuming sync");
                    let _ = state
                        .sync_events
                        .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
                }
            }
            // Expected after a remake: either side switches or catches up.
            EpochRelation::PeerAhead | EpochRelation::PeerBehind => {
                info!(
                    peer = %peer,
                    local_epoch = %local,
                    peer_epoch = %remote,
                    relation = ?relation,
                    "Peer is on another epoch; not syncing with it until both are on the same"
                );
            }
            EpochRelation::Diverged => {
                warn!(
                    peer = %peer,
                    local_epoch = %local,
                    peer_epoch = %remote,
                    "Peer is on a different epoch of the same generation; not syncing with it"
                );
            }
        }
    }

    relation == EpochRelation::Same
}