//! One Automerge sync with a peer, from a reset to the next.
//!
//! Automerge's sync state (`automerge::sync::State`) is private to this module: the
//! rest of the node generates, receives and inspects through `SyncSession`'s methods.
//! The workarounds for its assumption of reliable delivery (resending after a timeout,
//! forgetting what was sent when the peer still needs it) live here with it, so an
//! Automerge upgrade that changes the state touches this file only.

use std::time::Duration;

use automerge::sync::{self, SyncDoc};
use automerge::{AutoCommit, AutomergeError};
use tokio::time::Instant;
use tracing::debug;

use crate::backoff::Backoff;
use crate::swarm::throughput::SyncBytes;

/// A sync message on its way to our Mycelium (the HTTP request is in progress).
#[derive(Clone)]
pub(crate) struct Sending {
    /// Changes it carries for the first time; taken back if the send fails.
    pub(crate) sent: Vec<automerge::ChangeHash>,
    /// A sync pass was requested meanwhile. Once this one is handed over, the peer must
    /// be enqueued again so the deferred changes go out.
    pub(crate) regenerate: bool,
}

/// First wait before generating again after our Mycelium refused a message; doubles
/// while it keeps refusing.
const SEND_RETRY: Duration = Duration::from_millis(500);
const SEND_RETRY_MAX: Duration = Duration::from_secs(30);

/// One Automerge sync with a peer, from a reset to the next. Everything a reset must
/// forget lives here, so a reset replaces it whole and can't miss a field.
#[derive(Clone)]
pub(crate) struct SyncSession {
    state: sync::State,

    /// A message being handed to our Mycelium.
    pub(crate) sending: Option<Sending>,

    /// After our Mycelium refused a message: no new one before this. The refused one is
    /// not kept; the next one is generated afresh and carries everything pending.
    pub(crate) retry_at: Option<Instant>,
    /// The wait after the next refusal.
    pub(crate) send_backoff: Backoff,

    /// The answer we expect to our last sync message(s), if any.
    pub(crate) reply: Option<SyncReplyWait>,
    /// How long to wait for it; doubles while answers stay out.
    pub(crate) reply_backoff: Backoff,
}

/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
/// a lost message leaves both sides waiting (`in_flight`), and changes it carried are
/// never sent again (`sent_hashes`). We resend after a timeout instead.
#[derive(Clone, Debug)]
pub(crate) struct SyncReplyWait {
    pub(crate) deadline: Instant,
    /// When the first of the unanswered messages was generated, and the sync bytes
    /// exchanged with the peer by then (`swarm::throughput`).
    pub(crate) started: (Instant, SyncBytes),
    /// Changes first sent in the unanswered messages.
    pub(crate) sent: Vec<automerge::ChangeHash>,
}

impl SyncSession {
    /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`).
    pub(crate) fn new(reply_backoff: Backoff) -> Self {
        Self {
            state: sync::State::new(),
            sending: None,
            retry_at: None,
            send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
            reply: None,
            reply_backoff,
        }
    }

    /// Whether the peer said anything in this session yet (its heads are known).
    pub(crate) fn has_opened(&self) -> bool {
        self.state.their_heads.is_some()
    }

    /// The next sync message for the peer from `doc`, if there is anything to say. A
    /// message is marked as being handed over (`sending`), and an answer to it is
    /// expected, unless the peer's heads are ours already: it has everything, and
    /// Automerge answers only when it has something to say, so waiting for an answer
    /// would only end in a pointless resend. `bytes`: the sync bytes exchanged with the
    /// peer so far (`swarm::throughput`).
    pub(crate) fn generate(
        &mut self,
        doc: &mut AutoCommit,
        now: Instant,
        bytes: SyncBytes,
    ) -> Option<sync::Message> {
        let sent_before = self.state.sent_hashes.clone();
        let message = doc.sync().generate_sync_message(&mut self.state);
        let in_step = self.state.their_heads.as_ref() == Some(&doc.get_heads());
        let sent: Vec<_> = self
            .state
            .sent_hashes
            .difference(&sent_before)
            .copied()
            .collect();
        if in_step {
            self.reply_settled();
        } else if message.is_some() {
            self.expect_reply(sent.clone(), now, bytes);
        }
        if message.is_some() {
            // Marked together with the generation, so no second message is generated
            // while this one is handed over.
            self.sending = Some(Sending {
                sent,
                regenerate: false,
            });
        }
        message
    }

    /// Applies a sync message from the peer to `doc`. Doesn't settle the awaited
    /// answer: the caller measures the exchange first (`reply`).
    pub(crate) fn receive(
        &mut self,
        doc: &mut AutoCommit,
        message: sync::Message,
    ) -> Result<(), AutomergeError> {
        doc.sync().receive_sync_message(&mut self.state, message)
    }

    /// Whether `doc` holds everything the peer had when it last told us its heads: they
    /// are known, not empty (empty heads are "contained" in every document, so they
    /// prove nothing), and all in `doc`.
    pub(crate) fn have_all_theirs(&self, doc: &mut AutoCommit) -> bool {
        self.state.their_heads.as_ref().is_some_and(|heads| {
            !heads.is_empty()
                && heads
                    .iter()
                    .all(|head| doc.get_change_by_hash(head).is_some())
        })
    }

    /// The peer has heads we lack (its keepalive said so): ask for what they need,
    /// even though Automerge thinks a message is still in flight.
    pub(crate) fn pull_from(&mut self, heads: Vec<automerge::ChangeHash>) {
        self.state.their_heads = Some(heads);
        self.state.in_flight = false;
    }

    /// Automerge never resends a change it believes the peer has received
    /// (`sent_hashes`). If the message carrying it was lost while later ones arrive, the
    /// peer keeps asking for it (`need`, from the heads we advertise) and we keep not
    /// sending it: a livelock the reply timeout can't see (answers keep coming) and the
    /// keepalive heal can't either (sync never goes idle). So when the peer needs a
    /// change we think it has, what we think it has is wrong: forget all of it, and the
    /// next message carries everything its bloom filter lacks. (A reordered message can
    /// make that send a change twice; cheap.) Call after `receive`.
    pub(crate) fn resend_if_needed(&mut self, peer: &str) {
        let Some(need) = &self.state.their_need else {
            return;
        };
        if need
            .iter()
            .any(|hash| self.state.sent_hashes.contains(hash))
        {
            debug!(
                peer = %peer,
                sent = self.state.sent_hashes.len(),
                "Peer needs changes we had sent; forgetting what was sent"
            );
            self.state.sent_hashes.clear();
        }
    }

    /// For the simulator's dump: the peer's heads as Automerge knows them, and whether
    /// it thinks a message is in flight.
    #[cfg(test)]
    pub(crate) fn automerge_view(&self) -> (Option<&[automerge::ChangeHash]>, bool) {
        (self.state.their_heads.as_deref(), self.state.in_flight)
    }

    /// Whether we leave the first message of this session to the peer.
    ///
    /// After a reset, both sides usually open at once, and the openings cross. Between
    /// peers that share history that costs nothing. But a peer with an empty document
    /// answers our opening with empty heads, which Automerge takes for "lost
    /// everything", so the whole document would go out a second time. So a ready node
    /// never opens towards a peer that isn't ready (one that is still catching up); that
    /// peer opens itself.
    pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool {
        self.state.their_heads.is_none() && we_ready && peer_ready == Some(false)
    }

    /// Our Mycelium took the message.
    pub(crate) fn send_succeeded(&mut self) -> bool {
        self.retry_at = None;
        self.send_backoff.reset();
        self.sending
            .take()
            .is_some_and(|sending| sending.regenerate)
    }

    /// Our Mycelium refused the message: as far as the peer is concerned, it was never
    /// sent. Takes back its changes (they go out with the next message), clears
    /// `in_flight`, and returns how long to wait before generating again. None if a reset
    /// already discarded what was being sent.
    pub(crate) fn send_failed(&mut self, now: Instant) -> Option<Duration> {
        let sending = self.sending.take()?;
        for hash in &sending.sent {
            self.state.sent_hashes.remove(hash);
        }
        if let Some(wait) = &mut self.reply {
            wait.sent.retain(|hash| !sending.sent.contains(hash));
        }
        self.state.in_flight = false;
        let delay = self.send_backoff.next();
        self.retry_at = Some(now + delay);
        Some(delay)
    }

    /// A sync message went out; `sent` are the changes it carried for the first time.
    /// `bytes` are the sync bytes exchanged with the peer before it.
    pub(crate) fn expect_reply(
        &mut self,
        sent: Vec<automerge::ChangeHash>,
        now: Instant,
        bytes: SyncBytes,
    ) {
        match &mut self.reply {
            // Still waiting for an earlier one: keep its deadline, remember these too.
            Some(wait) => wait.sent.extend(sent),
            None => {
                self.reply = Some(SyncReplyWait {
                    deadline: now + self.reply_backoff.current(),
                    started: (now, bytes),
                    sent,
                });
            }
        }
    }

    /// No answer is due any more: the peer answered, or keepalives show both documents
    /// are the same.
    pub(crate) fn reply_settled(&mut self) {
        self.reply = None;
        self.reply_backoff.reset();
    }

    /// If the answer is overdue: forgets that the unanswered changes were sent and that
    /// a message is in flight, so the next sync pass sends them again. Returns how many
    /// changes that concerns.
    pub(super) fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> {
        if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) {
            return None;
        }
        let wait = self.reply.take()?;
        for hash in &wait.sent {
            self.state.sent_hashes.remove(hash);
        }
        self.state.in_flight = false;
        self.reply_backoff.grow();
        Some(wait.sent.len())
    }
}