//! What a node knows about each peer: standing, liveness, handshake, incidents.

use std::time::Duration;

use tokio::time::Instant;

use super::{SyncSession, Timing};
use crate::backoff::Backoff;
use crate::document::replica::Epoch;
use crate::transport;

/// 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 `PeerState::probe_backoff` while
    /// unreachable).
    AwaitingReply { next_attempt: Instant },
    /// HELLO_REPLY received. No more retries.
    Established,
}

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

#[derive(Clone)]
pub(crate) struct PeerState {
    /// Whether we lead the pair (`swarm::pair`): we send the first keepalive. 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,
    /// Wait between HELLOs to an unreachable peer (probes): `dormant_probe`, doubling
    /// while none is answered. Reset by any message from it.
    pub(crate) probe_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>,
    /// Our last keepalive to it (a PING or a STATUS).
    pub(crate) last_keepalive_at: Option<Instant>,
    /// Its last keepalive to us (a PING or a STATUS): after it, the next is ours.
    pub(crate) keepalive_received_at: Option<Instant>,
    /// The last STATUS between us, either way (`status_refresh`).
    pub(crate) last_status_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 (late ones aside: see
    /// `epoch_seq`).
    pub(crate) epoch: Option<Epoch>,
    /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run.
    /// An epoch with a lower one is older than what we have, delivered late.
    pub(crate) epoch_seq: Option<u32>,
    /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over.
    pub(crate) run_id: Option<crate::transport::RunId>,
    /// The newest handshake the peer started with us: its run and number
    /// (`HelloRequest::handshake`). A HELLO repeating it is a retry or a late copy.
    pub(crate) last_hello: Option<(crate::transport::RunId, u32)>,
    /// The number of our current handshake with the peer (`AppState::handshakes`).
    pub(crate) handshake: u32,

    /// 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). Only a message from it clears it (`mark_seen`): until then it is no
    /// neighbour candidate, and a member gets probes (`probe_backoff`) instead.
    pub(crate) unreachable_since: Option<Instant>,

    /// What went wrong with this peer so far (counted only, for /peers and /admin/stats).
    pub(crate) incidents: Incidents,
}

/// Trouble with a peer, counted since it was first listed. Lost keepalives and refused
/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING,
/// every STATUS_REPLY a STATUS.
#[derive(Debug, Default, Clone, Copy, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Incidents {
    /// Our sync messages left unanswered until the reply timeout (then sent again).
    pub(crate) reply_timeouts: u64,
    /// Re-handshakes because our documents differed while sync was idle.
    pub(crate) heals: u64,
    /// Times the peer went dormant.
    pub(crate) dormancies: u64,
    /// Time spent dormant, in milliseconds, up to its last return (the current
    /// dormancy, if any, isn't included).
    pub(crate) dormant_ms: u64,
}

impl Incidents {
    pub(crate) fn add(&mut self, other: &Incidents) {
        self.reply_timeouts += other.reply_timeouts;
        self.heals += other.heals;
        self.dormancies += other.dormancies;
        self.dormant_ms += other.dormant_ms;
    }
}

/// What a keepalive due is (`PeerState::take_keepalive_due`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Keepalive {
    /// A PING: the peer's transport answers it with a PONG.
    Ping,
    /// A STATUS: our state, answered with the peer's.
    Status,
}

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

impl PeerState {
    pub(crate) fn new(leads: bool, timing: &Timing) -> Self {
        Self {
            leads,
            standing: Standing::default(),
            hello_state: HelloState::default(),
            hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX),
            probe_backoff: timing.probe_backoff(),
            sync: SyncSession::new(timing.sync_reply_backoff()),
            liveness: Liveness::Active,
            added_at: Instant::now(),
            last_seen: None,
            last_keepalive_at: None,
            keepalive_received_at: None,
            last_status_at: None,
            last_sync_rx: None,
            last_resync_at: None,
            diverged_since: None,
            epoch: None,
            epoch_seq: None,
            run_id: None,
            last_hello: None,
            handshake: 0,
            reports_ready: None,
            syncs_with_us: None,
            reported_heads: None,
            hello_attempts: 0,
            unreachable_since: None,
            incidents: Incidents::default(),
        }
    }

    /// Start Automerge sync with this peer from scratch. A message still being handed to
    /// the transport was generated against the old state; its outcome is then ignored.
    pub(crate) fn reset_sync(&mut self) {
        // The same waits for answers, from the start.
        let mut reply_backoff = self.sync.reply_backoff;
        reply_backoff.reset();
        self.sync = SyncSession::new(reply_backoff);
    }

    /// 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;
        }
        let overdue = self.sync.take_overdue_reply(now);
        if overdue.is_some() {
            self.incidents.reply_timeouts += 1;
        }
        overdue
    }

    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;
        self.probe_backoff.reset();

        if let Liveness::Dormant { since } = self.liveness {
            self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64;
            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) -> bool {
        self.unreachable_since.is_some()
    }

    /// The wait before the next HELLO: a probe's if unreachable.
    pub(crate) fn next_hello_wait(&mut self) -> Duration {
        if self.is_unreachable() {
            self.probe_backoff.next()
        } else {
            self.hello_backoff.next()
        }
    }

    /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight
    /// finishes; a later handshake starts over. An unreachable member keeps its probes:
    /// it is no neighbour because it is unreachable, and only its answer can change that.
    pub(crate) fn stop_hello_retries(&mut self) {
        let probed = self.is_unreachable() && matches!(self.standing, Standing::Member { .. });
        if !probed && 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.incidents.dormancies += 1;
        self.unreachable_since.get_or_insert(now);
        self.reset_sync();

        if self.hello_state != HelloState::Sending {
            self.hello_state = HelloState::AwaitingReply {
                next_attempt: now + self.probe_backoff.next(),
            };
        }
        true
    }

    /// The keepalive an established, quiet peer needs, if any (and records it): a PING,
    /// or a STATUS if none went between us for `status_refresh`.
    ///
    /// A PING and its PONG (or a STATUS and its STATUS_REPLY) tell both sides that the
    /// other is alive, so one per interval does for the pair. The two sides take turns:
    /// after a keepalive from the peer, the next is ours (the one leading the pair,
    /// `leads`, sends the first), so both measure the round trip and each sends every
    /// other interval. On our turn we send once the pair was quiet for an interval; on its
    /// turn only after half an interval more, in case it doesn't (a keepalive or its
    /// answer was lost). Without turns, both sides would get due at about the same time
    /// (each one's quiet time starts with the same exchange) and their keepalives would
    /// often cross.
    ///
    /// When they do cross anyway (both sent before seeing the other's: the peer's came
    /// within half an interval after ours), each side would take the next turn for its
    /// own and they would keep crossing; then the next turn is the leader's.
    pub(crate) fn take_keepalive_due(
        &mut self,
        timing: &Timing,
        now: Instant,
    ) -> Option<Keepalive> {
        let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) {
            (None, None) => self.leads,
            (None, Some(_)) => true,
            (Some(_), None) => false,
            (Some(sent), Some(received)) if received > sent => {
                let crossed = received.saturating_duration_since(sent) < timing.keepalive / 2;
                !crossed || self.leads
            }
            (Some(_), Some(_)) => false,
        };
        let quiet_enough = if our_turn {
            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_keepalive_at
                .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
        if !due {
            return None;
        }
        self.last_keepalive_at = Some(now);
        let status = self
            .last_status_at
            .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
        Some(if status {
            Keepalive::Status
        } else {
            Keepalive::Ping
        })
    }

    /// Self-healing check on a STATUS or STATUS_REPLY. 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.incidents.heals += 1;

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

#[cfg(test)]
mod tests {
    use super::*;

    /// The two sides of a pair take turns: each sends every other interval.
    #[test]
    fn keepalives_take_turns() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut a = PeerState::new(true, &timing);
        let mut b = PeerState::new(false, &timing);
        a.hello_state = HelloState::Established;
        b.hello_state = HelloState::Established;
        b.added_at = a.added_at;
        let mut senders = Vec::new();
        let mut now = a.added_at;
        for _ in 0..40 {
            now += Duration::from_secs(5);
            // Whoever is due sends; the other one gets it at once, and both hear from
            // each other (the keepalive, its answer).
            if a.take_keepalive_due(&timing, now).is_some() {
                senders.push("a");
                b.keepalive_received_at = Some(now);
            } else if b.take_keepalive_due(&timing, now).is_some() {
                senders.push("b");
                a.keepalive_received_at = Some(now);
            } else {
                continue;
            }
            a.mark_seen(now);
            b.mark_seen(now);
        }
        // 200 s: one keepalive a minute, a first, then taking turns.
        assert_eq!(senders, ["a", "b", "a"]);
    }

    /// Keepalives that crossed (both sides sent before seeing the other's) leave the next
    /// turn to the leader, rather than both taking it and crossing again.
    #[test]
    fn crossed_keepalives_go_back_to_taking_turns() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut a = PeerState::new(true, &timing);
        let mut b = PeerState::new(false, &timing);
        let start = a.added_at;
        b.added_at = start;
        for peer in [&mut a, &mut b] {
            peer.hello_state = HelloState::Established;
        }
        // Both sent at 100 s and got the other's a moment later.
        let sent = start + Duration::from_secs(100);
        for peer in [&mut a, &mut b] {
            peer.last_keepalive_at = Some(sent);
            peer.keepalive_received_at = Some(sent + Duration::from_millis(50));
            peer.mark_seen(sent + Duration::from_millis(50));
        }
        let next = sent + timing.keepalive + Duration::from_secs(1);
        assert!(a.take_keepalive_due(&timing, next).is_some()); // The leader's turn.
        assert!(b.take_keepalive_due(&timing, next).is_none());
    }

    /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet.
    #[test]
    fn keepalives_are_pings_and_now_and_then_a_status() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        peer.hello_state = HelloState::Established;
        let start = peer.added_at;
        assert_eq!(peer.take_keepalive_due(&timing, start), None); // Too new.
        let mut kinds = Vec::new();
        for interval in 1..=6 {
            let now = start + timing.keepalive * interval;
            let due = peer.take_keepalive_due(&timing, now);
            if due == Some(Keepalive::Status) {
                peer.last_status_at = Some(now);
            }
            kinds.extend(due);
            // Not again within the interval.
            assert_eq!(peer.take_keepalive_due(&timing, now), None);
        }
        use Keepalive::{Ping, Status};
        assert_eq!(kinds, [Status, Ping, Ping, Status, Ping, Ping]);
        // Heard from just now: nothing due.
        let now = start + timing.keepalive * 7;
        peer.mark_seen(now);
        assert_eq!(peer.take_keepalive_due(&timing, now), None);
    }

    /// R41: unreachability lasts until the peer is heard from, not for a set time, and
    /// probes back off meanwhile.
    #[test]
    fn unreachable_until_heard_from() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        let start = Instant::now();
        for _ in 0..UNREACHABLE_AFTER_HELLOS {
            peer.hello_sent(start);
        }
        assert!(peer.is_unreachable());

        let probes: Vec<Duration> = (0..4).map(|_| peer.next_hello_wait()).collect();
        let probe = timing.dormant_probe;
        assert_eq!(probes, [probe, probe * 2, probe * 4, probe * 6]);
        peer.hello_sent(start + probe * 20);
        assert!(peer.is_unreachable());

        peer.mark_seen(start + probe * 21);
        assert!(!peer.is_unreachable());
        peer.hello_sent(start + probe * 22);
        peer.hello_sent(start + probe * 22);
        peer.hello_sent(start + probe * 22);
        assert_eq!(peer.next_hello_wait(), probe);
    }
}