//! Shared node state: the document, lifecycle, per-peer state and timing.

use std::collections::HashMap;
use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;

use anyhow::{Result, bail};
use automerge::{AutoCommit, ChangeHash, sync};
use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
use tokio::time::Instant;
use tracing::{debug, error, warn};

use crate::backoff::Backoff;
use crate::document::replica::{Epoch, Replica};
use crate::remake::{Host, PeerSummary, RemakeState};
use crate::swarm::sync::{ChangeSet, SyncEvent};
use crate::transport;
use crate::transport::firewall::Firewall;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum NodeLifecycle {
    // TODO maybe rename discovered_swarm_id to swarm_id if that doesn't impair readability.
    Joining { discovered_swarm_id: Option<String> },
    Ready { swarm_id: String },
}

/// How far a peer got into the swarm, as far as we're concerned.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) enum Standing {
    /// Nothing proven yet. With a swarm key, everything but a handshake from it is
    /// ignored (`swarm::peer::admit`).
    #[default]
    Stranger,
    /// Proved the swarm key in a HELLO; no HELLO_REPLY yet. Only reached with a key.
    Admitted,
    /// Its HELLO_REPLY named our swarm (and, with a key, proved it): we sync with it.
    Member { swarm_id: String },
}

impl Standing {
    /// The swarm a member belongs to; None before its HELLO_REPLY.
    pub(crate) fn swarm_id(&self) -> Option<&str> {
        match self {
            Self::Member { swarm_id } => Some(swarm_id),
            Self::Stranger | Self::Admitted => None,
        }
    }

    /// Proved the swarm key (meaningful only with one).
    pub(crate) fn is_admitted(&self) -> bool {
        !matches!(self, Self::Stranger)
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Liveness {
    Active,
    /// No inbound message for Timing::dormant_after. Kept indefinitely, probed slowly.
    Dormant {
        since: Instant,
    },
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) enum HelloState {
    #[default]
    NotSent,
    /// A send is in flight.
    Sending,
    /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at
    /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant).
    AwaitingReply { next_attempt: Instant },
    /// HELLO_REPLY received. No more retries.
    Established,
}

/// 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 for a HELLO_REPLY before sending the HELLO again; doubles while none
/// comes.
const HELLO_RETRY: Duration = Duration::from_secs(1);
const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);

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

#[derive(Clone)]
pub(crate) struct PeerState {
    /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for
    /// the pair, so computed once.
    pub(crate) leads: bool,

    pub(crate) standing: Standing,
    pub(crate) hello_state: HelloState,
    /// Wait between unanswered HELLOs; reset whenever a new handshake starts.
    pub(crate) hello_backoff: Backoff,

    /// Everything about the current Automerge sync with this peer; a reset replaces it.
    pub(crate) sync: SyncSession,

    pub(crate) liveness: Liveness,
    pub(crate) added_at: Instant,
    pub(crate) last_seen: Option<Instant>,
    pub(crate) last_ping_at: Option<Instant>,

    /// Last time a sync message from this peer was applied (survives resets: the
    /// keepalive heal looks at it).
    pub(crate) last_sync_rx: Option<Instant>,

    /// Last self-healing re-handshake (cooldown).
    pub(crate) last_resync_at: Option<Instant>,

    /// Since when keepalives show the peer's heads differ from ours with no sync message
    /// from it in between (None while they match).
    pub(crate) diverged_since: Option<Instant>,

    /// Epoch the peer advertised in its most recent message.
    pub(crate) epoch: Option<Epoch>,

    /// Whether the peer was ready according to its most recent sync message or keepalive.
    pub(crate) reports_ready: Option<bool>,

    /// Whether the peer said, in its most recent keepalive, that it would sync with us.
    pub(crate) syncs_with_us: Option<bool>,

    /// How many heads its most recent keepalive listed (None if it sent a digest).
    pub(crate) reported_heads: Option<usize>,

    /// HELLOs sent in the current handshake attempt without hearing anything back.
    pub(crate) hello_attempts: u32,
    /// Since when we consider the peer unreachable (three unanswered HELLOs, or
    /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the
    /// peer becomes a neighbour candidate again and gets tried once more.
    pub(crate) unreachable_since: Option<Instant>,
}

/// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart).
const UNREACHABLE_AFTER_HELLOS: u32 = 3;

/// 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 {
    pub(crate) 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,
    /// Changes first sent in the unanswered messages.
    pub(crate) sent: Vec<automerge::ChangeHash>,
}

/// First wait for a sync answer; generous, since Mycelium delivers large messages in
/// many small packets.
const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
/// Upper bound (Mycelium gives up on a message after five minutes).
const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);

impl SyncSession {
    pub(crate) fn new() -> Self {
        Self {
            state: sync::State::new(),
            sending: None,
            retry_at: None,
            send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
            reply: None,
            reply_backoff: Backoff::new(SYNC_REPLY_TIMEOUT, SYNC_REPLY_TIMEOUT_MAX),
        }
    }

    /// 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.
    pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) {
        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(),
                    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.
    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())
    }
}

impl PeerState {
    pub(crate) fn new(leads: bool) -> Self {
        Self {
            leads,
            standing: Standing::default(),
            hello_state: HelloState::default(),
            hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX),
            sync: SyncSession::new(),
            liveness: Liveness::Active,
            added_at: Instant::now(),
            last_seen: None,
            last_ping_at: None,
            last_sync_rx: None,
            last_resync_at: None,
            diverged_since: None,
            epoch: None,
            reports_ready: None,
            syncs_with_us: None,
            reported_heads: None,
            hello_attempts: 0,
            unreachable_since: None,
        }
    }

    /// Start Automerge sync with this peer from scratch. A message still being handed to
    /// Mycelium was generated against the old state; its outcome is then ignored.
    pub(crate) fn reset_sync(&mut self) {
        self.sync = SyncSession::new();
    }

    /// An overdue sync answer from an established, active peer (see
    /// `SyncSession::take_overdue_reply`).
    pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> {
        if self.liveness != Liveness::Active || self.hello_state != HelloState::Established {
            return None;
        }
        self.sync.take_overdue_reply(now)
    }

    fn quiet_for(&self, now: Instant) -> Duration {
        now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at))
    }

    /// Records inbound activity. Returns true if the peer was dormant.
    pub(crate) fn mark_seen(&mut self, now: Instant) -> bool {
        self.last_seen = Some(now);
        self.hello_attempts = 0;
        self.unreachable_since = None;

        if matches!(self.liveness, Liveness::Dormant { .. }) {
            self.liveness = Liveness::Active;
            // Force a fresh handshake; its reply resets sync on both sides.
            if self.hello_state != HelloState::Sending {
                self.hello_state = HelloState::NotSent;
            }
            return true;
        }
        false
    }

    /// A HELLO went out (or failed to). After the third unanswered one, the peer counts
    /// as unreachable.
    pub(crate) fn hello_sent(&mut self, now: Instant) {
        self.hello_attempts += 1;
        if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS {
            self.unreachable_since.get_or_insert(now);
        }
    }

    /// Whether to leave the peer out when choosing neighbours (`swarm::topology`).
    pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool {
        self.unreachable_since
            .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe)
    }

    /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight
    /// finishes; a later handshake starts over.
    pub(crate) fn stop_hello_retries(&mut self) {
        if matches!(self.hello_state, HelloState::AwaitingReply { .. }) {
            self.hello_state = HelloState::NotSent;
        }
    }

    /// Moves a silent peer to Dormant. Returns true on the transition.
    pub(crate) fn check_dormancy(&mut self, timing: &Timing, now: Instant) -> bool {
        if self.liveness != Liveness::Active || self.quiet_for(now) < timing.dormant_after {
            return false;
        }

        self.liveness = Liveness::Dormant { since: now };
        self.unreachable_since.get_or_insert(now);
        self.reset_sync();

        if self.hello_state != HelloState::Sending {
            self.hello_state = HelloState::AwaitingReply {
                next_attempt: now + timing.dormant_probe,
            };
        }
        true
    }

    /// True (and records the ping) if an established, quiet peer needs a keepalive.
    ///
    /// One PING and its PONG tell both sides that the other is alive, so only one side of
    /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs
    /// makes the other side's peer look recently heard from, so the other side never
    /// pings while it does. If the pings stop, the other side pings once it has heard
    /// nothing for half an interval more. Without that, both sides would get due at about
    /// the same time (each one's quiet time starts with the same exchange) and their
    /// PINGs would often cross.
    pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
        let quiet_enough = if self.leads {
            timing.keepalive
        } else {
            timing.keepalive + timing.keepalive / 2
        };
        let due = self.liveness == Liveness::Active
            && self.hello_state == HelloState::Established
            && now.saturating_duration_since(self.added_at) >= timing.keepalive
            && self.quiet_for(now) >= quiet_enough
            && self
                .last_ping_at
                .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);

        if due {
            self.last_ping_at = Some(now);
        }
        due
    }

    /// Self-healing check on PING/PONG. Returns true if a re-handshake should start.
    pub(crate) fn needs_heal(
        &mut self,
        local: &transport::HeadsSummary,
        remote: &transport::HeadsSummary,
        timing: &Timing,
        now: Instant,
    ) -> bool {
        if local == remote {
            self.diverged_since = None;
            return false;
        }

        // How long the heads have differed without sync making progress. Measured from
        // when we noticed, not from the last sync message: in a quiet swarm that one is
        // long past, and the first keepalive after a lost message would heal at once,
        // before the sender's reply timeout could resend it. Any sync message restarts
        // the clock.
        let noticed = match self.diverged_since {
            Some(since) if self.last_sync_rx.is_none_or(|rx| rx <= since) => since,
            _ => *self.diverged_since.insert(now),
        };

        // While an answer to our last sync message is due, a lost message is the reply
        // timeout's to resend; healing is for when neither side has anything outstanding.
        if self.liveness != Liveness::Active
            || self.hello_state != HelloState::Established
            || self.sync.reply.is_some()
        {
            return false;
        }

        let persisted = now.saturating_duration_since(noticed) >= timing.resync_after;
        let cooled_down = self
            .last_resync_at
            .is_none_or(|t| now.saturating_duration_since(t) >= timing.resync_after);

        if !(persisted && cooled_down) {
            return false;
        }
        self.diverged_since = None;

        self.last_resync_at = Some(now);
        self.hello_state = HelloState::NotSent; // establish_peer resets sync on the reply
        true
    }

    /// Heard from recently. A returning node counts as alive from its first message,
    /// before the handshake finishes, so a swarm-wide forget can't make it flap.
    pub(crate) fn is_alive(&self, timing: &Timing, now: Instant) -> bool {
        self.liveness == Liveness::Active
            && self
                .last_seen
                .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after)
    }
}

#[derive(Clone, Copy, Debug)]
pub(crate) struct Timing {
    pub(crate) keepalive: Duration,
    pub(crate) dormant_after: Duration,
    pub(crate) resync_after: Duration,
    pub(crate) dormant_probe: Duration,
    /// While catching up: PING the source after this much silence (`swarm::source`).
    pub(crate) catch_up_probe: Duration,
}

impl Timing {
    /// How long a round-trip sample stays meaningful (`swarm::rtt`): a few keepalive
    /// intervals, since nodes move.
    pub(crate) fn rtt_fresh_for(&self) -> Duration {
        self.keepalive * 5
    }

    // Recommended values
    //  Environment                                 | Keepalive | Notes
    //  --------------------------------------------+-----------+------------------------------------------------------------
    //  Test harness                                | 1-2 s     | Dormancy after about 4–8 s and healing within seconds keep tests short.
    //  LAN or small swarm (up to ~50 nodes)      | 15-30 s   | Under 3.5 messages/s per node at 50 nodes and 30 s. Current setting.
    //  WAN, mobile or metered links, or 100+ nodes | 60-120 s  | At 100 nodes and 30 s it’s about 6–7 messages/s per node. Fine, but rarely worth it on battery or metered links.
    //  Several hundred nodes or more               | -         | The full mesh becomes the problem in itself, not the interval. Message counts grow O(N²) across the swarm, which calls for partial views (SWIM-style random probing and syncing with a subset of peers).
    pub(crate) fn from_keepalive(keepalive: Duration) -> Self {
        Self {
            keepalive,
            dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row
            // >= 2 keepalives, so a second exchange confirms the difference; and >= twice
            // the first sync reply timeout, so a lost message's sender resends it first.
            resync_after: (keepalive * 2).max(SYNC_REPLY_TIMEOUT * 2),
            dormant_probe: keepalive * 10, // >= 10*: dormant peers cost almost nothing, but still rejoin within minutes
            // Seconds, not a keepalive interval: a dead source should cost little time.
            // Only the source gets these, and only while catching up.
            catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)),
        }
    }
}

#[derive(Clone)]
pub(crate) struct AppState {
    /// The document and its epoch. Derefs to `AutoCommit`.
    pub(crate) doc: Arc<RwLock<Replica>>,

    /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers).
    /// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`.
    pub(crate) epoch_tx: watch::Sender<Epoch>,

    pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>,

    // Serializes swarm-ID establishment, peer membership, and readiness transitions.
    //[concurrent]swarm_validation: Arc<Mutex<()>>,
    pub(crate) peer_states: Arc<RwLock<HashMap<String, Arc<Mutex<PeerState>>>>>,
    pub(crate) transport: Arc<dyn transport::Transport>,
    pub(crate) firewall: Arc<Firewall>,
    /// Messages and bytes exchanged, per peer (`transport::traffic`).
    pub(crate) traffic: Arc<transport::traffic::Traffic>,
    /// Node-wide figures for /admin/stats (`swarm::stats`).
    pub(crate) stats: Arc<crate::swarm::stats::NodeStats>,

    pub(crate) sync_events: mpsc::UnboundedSender<SyncEvent>,

    // TODO see key_watchers.md ("Use a per-key notification registry instead of the global `watch<u64>`")
    pub(crate) change_tx: watch::Sender<u64>,

    pub(crate) initial_sync_done: Arc<AtomicBool>,

    pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>,

    pub(crate) timing: Timing,

    /// Per-node remake state (prebuilt snapshots, switch status, archive).
    pub(crate) remake: Arc<RemakeState>,

    /// Pre-shared key every member must prove in the handshake; None admits anyone who
    /// knows the swarm ID.
    pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>,

    /// Random per run of this node; our HELLOs carry it and replies must echo it.
    pub(crate) run_id: transport::RunId,

    /// Target number of neighbours (`--neighbours`); up to this many + 1 members the
    /// swarm is a full mesh.
    pub(crate) neighbours_k: usize,
    /// Our own connectivity record and stability tier (`swarm::stability`).
    pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>,
    /// Round-trip times to peers (`swarm::rtt`).
    pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>,
    /// The neighbour we are pulling changes from, and since when (`swarm::relay`).
    pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>,
    /// Our neighbours, once known (`swarm::topology`).
    pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,

    /// While rejoining a newer epoch, the previous document, served read-only (and
    /// marked stale) instead of answering 503.
    pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>,

    /// While the initial sync isn't done: the one peer we catch up from.
    pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>,
    /// Since when a closer peer has had a head start to become the source.
    pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>,
}

/// The peer a catching-up node syncs with first (`swarm::source`).
#[derive(Clone, Debug)]
pub(crate) struct CatchUpSource {
    pub(crate) peer: String,
    /// When it was chosen or last sent us anything.
    pub(crate) last_heard: Instant,
    /// Our last probe PING to it.
    pub(crate) last_probe: Option<Instant>,
    /// Its keepalives in a row saying it wouldn't sync with us.
    pub(crate) refusals: u32,
}

impl AppState {
    pub(crate) fn local_id(&self) -> &str {
        self.transport.local_id()
    }

    pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> {
        self.rtt
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// The local epoch. Code that labels or applies sync messages must use the epoch
    /// of the locked `Replica` instead, so label and document can't disagree.
    pub(crate) fn local_epoch(&self) -> Epoch {
        *self.epoch_tx.borrow()
    }

    pub(crate) async fn is_ready(&self) -> bool {
        matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. })
    }

    pub(crate) async fn swarm_id(&self) -> Option<String> {
        match &*self.lifecycle.read().await {
            NodeLifecycle::Ready { swarm_id } => Some(swarm_id.clone()),
            NodeLifecycle::Joining {
                discovered_swarm_id,
            } => discovered_swarm_id.clone(),
        }
    }

    // Record the expected swarm without making the node Ready. Bootstrap owns readiness.
    pub(crate) async fn set_expected_swarm_id(&self, swarm_id: String) -> Result<()> {
        //[concurrent]let _guard = self.swarm_validation.lock().await;
        let mut lifecycle = self.lifecycle.write().await;
        match &*lifecycle {
            NodeLifecycle::Joining {
                discovered_swarm_id: Some(existing),
            } if existing != &swarm_id => {
                bail!(
                    "Conflicting expected swarm IDs: expected {}, received {}",
                    existing,
                    swarm_id
                );
            }
            NodeLifecycle::Joining {
                discovered_swarm_id: None,
            } => {
                let old_lifecycle = lifecycle.clone();
                *lifecycle = NodeLifecycle::Joining {
                    discovered_swarm_id: Some(swarm_id),
                };
                debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
                Ok(())
            }
            NodeLifecycle::Joining {
                discovered_swarm_id: Some(_),
            } => {
                warn!("Node already Joining with the same identity");
                Ok(())
            }
            NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
                warn!("Node already Ready with the same identity");
                Ok(())
            }
            NodeLifecycle::Ready { swarm_id: existing } => {
                bail!(
                    "Cannot change established swarm ID: existing {}, received {}",
                    existing,
                    swarm_id
                );
            }
        }
    }

    // Only bootstrap should call this, after validating document metadata.
    pub(crate) async fn mark_ready(&self, swarm_id: String) -> Result<()> {
        //[concurrent]let _guard = self.swarm_validation.lock().await;
        let mut lifecycle = self.lifecycle.write().await;
        match &*lifecycle {
            NodeLifecycle::Joining {
                discovered_swarm_id: Some(expected),
            } if expected != &swarm_id => {
                bail!(
                    "Cannot mark node ready: expected swarm {}, received {}",
                    expected,
                    swarm_id
                );
            }
            NodeLifecycle::Joining { .. } => {
                let old_lifecycle = lifecycle.clone();
                *lifecycle = NodeLifecycle::Ready { swarm_id };
                debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
            }
            NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
                debug!("Node already ready with the same identity");
            }
            NodeLifecycle::Ready { swarm_id: existing } => {
                bail!(
                    "Cannot change established swarm ID: existing {}, received {}",
                    existing,
                    swarm_id
                );
            }
        }
        Ok(())
    }

    /// Looks up a peer without creating it. Background work (sync, retries,
    /// keepalives) must use this so a forgotten peer isn't resurrected.
    pub(crate) async fn get_peer_state(&self, pk: &str) -> Option<Arc<Mutex<PeerState>>> {
        self.peer_states.read().await.get(pk).cloned()
    }

    pub(crate) async fn get_or_create_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
        let mut peers = self.peer_states.write().await;

        Arc::clone(
            peers
                .entry(pk.to_owned())
                .or_insert_with(|| self.new_peer_state(pk)),
        )
    }

    /// State for a peer we haven't known so far (not inserted).
    pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
        let leads = crate::swarm::pair::leads(self.local_id(), pk);
        Arc::new(Mutex::new(PeerState::new(leads)))
    }

    pub(crate) fn notify_store_changed(&self) {
        self.change_tx.send_modify(|version| {
            *version = version.wrapping_add(1);
        });
    }

    /// Reads the document on the blocking pool; `finish` runs after the lock is released.
    pub(crate) async fn render<T, R>(
        &self,
        read: impl FnOnce(&AutoCommit) -> T + Send + 'static,
        finish: impl FnOnce(T) -> R + Send + 'static,
    ) -> R
    where
        R: Send + 'static,
    {
        let doc = Arc::clone(&self.doc);
        tokio::task::spawn_blocking(move || {
            let data = {
                let guard = doc.blocking_read();
                read(&guard.doc)
            };
            finish(data)
        })
        .await
        .expect("document read task panicked")
    }

    /// Mutates the document on the blocking pool.
    pub(crate) async fn write_doc<T>(
        &self,
        f: impl FnOnce(&mut AutoCommit) -> T + Send + 'static,
    ) -> T
    where
        T: Send + 'static,
    {
        let doc = Arc::clone(&self.doc);
        tokio::task::spawn_blocking(move || f(&mut doc.blocking_write().doc))
            .await
            .expect("document write task panicked")
    }

    /// Mutates the document on the blocking pool as one Automerge change, then
    /// publishes it. Write and notification run in the same blocking task, so a
    /// cancelled request can't leave a committed write unannounced.
    pub(crate) async fn mutate_doc<T, E>(
        &self,
        f: impl FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
    ) -> Result<T, E>
    where
        T: Send + 'static,
        E: Send + 'static,
    {
        let state = self.clone();

        tokio::task::spawn_blocking(move || {
            let result = {
                let mut replica = state.doc.blocking_write();
                let result = f(&mut replica.doc);
                match &result {
                    // One API write = one Automerge change.
                    Ok(_) => {
                        replica.doc.commit();
                    }
                    // Discard partial ops (e.g. old value deleted, new one not written).
                    Err(_) => {
                        replica.doc.rollback();
                    }
                }
                result
            }; // lock released before notifying

            if result.is_ok() {
                state.publish_change();
            }
            result
        })
        .await
        .expect("document write task panicked")
    }

    pub(crate) fn publish_change(&self) {
        self.notify_store_changed();
        if let Err(error) = self.sync_events.send(SyncEvent::DocumentChanged) {
            // The write is committed; failing the request would misreport it.
            error!(error = %error, "Sync worker is gone; change will not be replicated");
        }
    }
}

impl AppState {
    /// After the document was replaced: every peer's sync state refers to the old
    /// document, so start over with all of them. Peers still on the old epoch are
    /// skipped by the sync worker until they switch too.
    async fn after_epoch_switch(&self) {
        let peers: Vec<(String, Arc<Mutex<PeerState>>)> = self
            .peer_states
            .read()
            .await
            .iter()
            .map(|(id, peer)| (id.clone(), Arc::clone(peer)))
            .collect();

        // Sync states were reset under the document lock by the swap itself.
        self.publish_change();

        // Tell peers about the new epoch now instead of with the next keepalive (up to a
        // keepalive interval later). Ready peers already on it learn it from our sync
        // opening, which goes out right away; everyone else gets a PING, so peers that
        // never saw the commit start catching up (we don't sync with them, so nothing
        // else would tell them).
        let local = self.local_epoch();
        let mut announce = Vec::new();
        for (id, peer_state_mutex) in &peers {
            // Others learn it from their own neighbours.
            if !crate::swarm::topology::is_neighbour(self, id) {
                continue;
            }
            let mut peer_state = peer_state_mutex.lock().await;
            // What it said about syncing with us was about our old epoch.
            peer_state.syncs_with_us = None;
            let gets_our_opening = peer_state.liveness == Liveness::Active
                && matches!(peer_state.standing, Standing::Member { .. })
                && peer_state.epoch == Some(local)
                && peer_state.reports_ready == Some(true);
            if !gets_our_opening {
                announce.push(id);
            }
        }

        for id in announce {
            if let Err(error) = crate::swarm::peer::send_ping(self, id).await {
                debug!(peer = %id, error = %format!("{error:#}"), "Epoch announcement failed");
            }
        }

        for (id, _) in peers {
            let _ = self.sync_events.send(SyncEvent::PeerNeedsSync(id));
        }
    }
}

/// A read-only view of the document the node served before it started rejoining.
pub(crate) struct StaleView {
    pub(crate) epoch: Epoch,
    pub(crate) doc: Arc<AutoCommit>,
}

/// Where a read was served from.
pub(crate) enum ReadSource {
    Live,
    /// The pre-rejoin document of this epoch.
    Stale(Epoch),
}

impl AppState {
    /// Renders a read for the store API: from the live document when the node is ready,
    /// from the stale view while it rejoins, and otherwise `503`, or the live document
    /// anyway with `unready_live`.
    ///
    /// With `with_heads`, a live read also returns the document's heads at the moment of
    /// the read (for clients that base later requests on what they read). That needs the
    /// write lock, so it is opt-in. Stale reads have no heads.
    pub(crate) async fn render_read<T, R>(
        &self,
        unready_live: bool,
        with_heads: bool,
        read: impl FnOnce(&AutoCommit) -> T + Send + 'static,
        finish: impl FnOnce(T) -> R + Send + 'static,
    ) -> Result<(R, ReadSource, Option<Vec<ChangeHash>>), axum::http::StatusCode>
    where
        T: Send + 'static,
        R: Send + 'static,
    {
        if self.is_ready().await {
            let (rendered, heads) = self.render_live(with_heads, read, finish).await;
            return Ok((rendered, ReadSource::Live, heads));
        }

        let stale = self
            .stale
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .as_ref()
            .map(|view| (view.epoch, Arc::clone(&view.doc)));
        let Some((epoch, doc)) = stale else {
            if unready_live {
                let (rendered, heads) = self.render_live(with_heads, read, finish).await;
                return Ok((rendered, ReadSource::Live, heads));
            }
            return Err(axum::http::StatusCode::SERVICE_UNAVAILABLE);
        };

        let rendered = tokio::task::spawn_blocking(move || finish(read(&doc)))
            .await
            .expect("document read task panicked");
        Ok((rendered, ReadSource::Stale(epoch), None))
    }

    async fn render_live<T, R>(
        &self,
        with_heads: bool,
        read: impl FnOnce(&AutoCommit) -> T + Send + 'static,
        finish: impl FnOnce(T) -> R + Send + 'static,
    ) -> (R, Option<Vec<ChangeHash>>)
    where
        T: Send + 'static,
        R: Send + 'static,
    {
        if !with_heads {
            return (self.render(read, finish).await, None);
        }
        // AutoCommit::get_heads takes &mut self (it closes a pending transaction).
        let (data, mut heads) = self
            .write_doc(move |doc| {
                let heads = doc.get_heads();
                (read(doc), heads)
            })
            .await;
        heads.sort();
        let rendered = tokio::task::spawn_blocking(move || finish(data))
            .await
            .expect("document read task panicked");
        (rendered, Some(heads))
    }

    fn set_stale(&self, view: Option<StaleView>) {
        *self
            .stale
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = view;
    }
}

pub(crate) fn unix_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// The node as seen by the remake module.
impl Host for AppState {
    fn remake(&self) -> &RemakeState {
        &self.remake
    }

    fn node_id(&self) -> String {
        self.local_id().to_owned()
    }

    fn current_epoch(&self) -> Epoch {
        self.local_epoch()
    }

    fn dormant_after(&self) -> Duration {
        self.timing.dormant_after
    }

    fn is_node_ready(&self) -> impl Future<Output = bool> + Send {
        let state = self.clone();
        async move { state.is_ready().await }
    }

    fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send {
        let state = self.clone();
        async move {
            let peers: Vec<(String, Arc<Mutex<PeerState>>)> = state
                .peer_states
                .read()
                .await
                .iter()
                .map(|(id, peer)| (id.clone(), Arc::clone(peer)))
                .collect();

            let mut summaries = Vec::with_capacity(peers.len());
            for (id, peer) in peers {
                let peer = peer.lock().await;
                summaries.push(PeerSummary {
                    id,
                    active: peer.liveness == Liveness::Active,
                    epoch: peer.epoch,
                });
            }
            summaries
        }
    }

    fn suspend(&self) -> impl Future<Output = bool> + Send {
        let state = self.clone();
        async move {
            let mut lifecycle = state.lifecycle.write().await;
            let NodeLifecycle::Ready { swarm_id } = &*lifecycle else {
                return false;
            };
            let swarm_id = swarm_id.clone();
            debug!(swarm_id = %swarm_id, "Node suspended (rejoining)");
            *lifecycle = NodeLifecycle::Joining {
                discovered_swarm_id: Some(swarm_id),
            };
            true
        }
    }

    fn resume(&self) -> impl Future<Output = ()> + Send {
        let state = self.clone();
        async move {
            let mut lifecycle = state.lifecycle.write().await;
            if let NodeLifecycle::Joining {
                discovered_swarm_id: Some(swarm_id),
            } = &*lifecycle
            {
                let swarm_id = swarm_id.clone();
                debug!(swarm_id = %swarm_id, "Node resumed");
                *lifecycle = NodeLifecycle::Ready { swarm_id };
            }
        }
    }

    fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>) {
        self.set_stale(view.map(|(epoch, doc)| StaleView {
            epoch,
            doc: Arc::new(doc),
        }));
    }

    fn initial_sync_done(&self) -> bool {
        self.initial_sync_done
            .load(std::sync::atomic::Ordering::Acquire)
    }

    fn reset_initial_sync(&self) {
        self.initial_sync_done
            .store(false, std::sync::atomic::Ordering::Release);
        *self
            .catch_up_source
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
        *self
            .catch_up_preference
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
    }

    fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> {
        self.change_events.subscribe()
    }

    fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
    where
        F: FnOnce(&AutoCommit) -> T + Send + 'static,
        T: Send + 'static,
    {
        let state = self.clone();
        async move { state.render(f, |value| value).await }
    }

    fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
    where
        F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
        T: Send + 'static,
        E: Send + 'static,
    {
        let state = self.clone();
        async move { state.mutate_doc(f).await }
    }

    fn switch_epoch<E, F>(
        &self,
        epoch: Epoch,
        f: F,
    ) -> impl Future<Output = Result<AutoCommit, E>> + Send
    where
        F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static,
        E: Send + 'static,
    {
        let state = self.clone();
        async move {
            let swapped = {
                let state = state.clone();
                tokio::task::spawn_blocking(move || -> Result<AutoCommit, E> {
                    let mut replica = state.doc.blocking_write();
                    let new_doc = f(&mut replica.doc)?;
                    let old_doc = std::mem::replace(&mut replica.doc, new_doc);
                    replica.epoch = epoch;
                    // Updated under the same lock, so the cached epoch never disagrees
                    // with the document for anyone holding the lock.
                    state.epoch_tx.send_replace(epoch);
                    // Reset every peer's sync state before the lock is released, so no
                    // message is generated from or applied to a state that describes the
                    // old document. (Lock order doc -> peer, as in sync generation.)
                    let now = Instant::now();
                    for peer in state.peer_states.blocking_read().values() {
                        let mut peer = peer.blocking_lock();
                        peer.reset_sync();
                        // A fresh start, not an idle sync: no heal re-handshake for it.
                        peer.last_sync_rx = Some(now);
                    }
                    Ok(old_doc)
                })
                .await
                .expect("epoch switch task panicked")
            };

            if swapped.is_ok() {
                state.after_epoch_switch().await;
            }
            swapped
        }
    }
}