//! The peer worker: keepalives due, dormancy, HELLO retries and overdue sync replies,
//! when the earliest of them is due (or something may have made one due sooner,
//! `Due`); every `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready),
//! forget records, the member registry, stability, measurements and the state file;
//! forget records also at once when they changed.

use std::time::Duration;

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

use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
use super::handshake::retry_hello_if_due;
use super::keepalive::{send_ping, send_status};
use crate::locks::MutexExt;
use crate::state::{AppState, Keepalive};
use crate::swarm::topology;

/// The least time between two passes, however often something is due.
const MIN_PASS_GAP: Duration = Duration::from_millis(20);

const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);

/// When the peer worker is next due. What makes something of its due sooner than it
/// knows (a HELLO sent, a sync message waiting for its answer, a keepalive received, a
/// new peer, new neighbours) says so (`by`); what only puts things off needn't.
#[derive(Default)]
pub(crate) struct Due {
    /// Its next pass while it waits; None while it runs.
    next: std::sync::Mutex<Option<Instant>>,
    wake: tokio::sync::Notify,
}

impl Due {
    /// Something is due at `at`: a pass then at the latest.
    pub(crate) fn by(&self, at: Instant) {
        if self.next.lock_or_recover().is_none_or(|next| at < next) {
            self.wake.notify_one();
        }
    }

    /// A pass now.
    pub(crate) fn now(&self) {
        self.wake.notify_one();
    }

    fn passing(&self) {
        *self.next.lock_or_recover() = None;
    }

    /// Until `at`, or until something is due sooner.
    async fn until(&self, at: Instant) {
        *self.next.lock_or_recover() = Some(at);
        tokio::select! {
            () = self.wake.notified() => {}
            () = tokio::time::sleep_until(at) => {}
        }
    }
}

pub(crate) async fn run_peer_worker(state: AppState) {
    info!("Starting peer maintenance worker");

    let mut last_forget_check: Option<Instant> = None;
    let mut forgets_applied = AppliedForgets::new();
    let mut was_ready = false;

    loop {
        state.peers_due.passing();
        let mut next = Instant::now() + FORGET_CHECK_INTERVAL;

        // While joining, keep handshakes fast; dormancy would slow probes to minutes.
        let ready = state.is_ready().await;
        if ready && !was_ready {
            // Just became ready: register in the member list (and do the other periodic
            // checks) now rather than at the next check. Until it is registered, other
            // members don't count this node as a neighbour.
            last_forget_check = None;
        }
        was_ready = ready;

        let peers = state.peers_snapshot().await;

        for (peer, peer_state_mutex) in &peers {
            let now = Instant::now();
            // Only neighbours are kept alive and judged; others we only answer.
            let neighbour = topology::is_neighbour(&state, peer);

            let (became_dormant, keepalive, unanswered) = {
                let mut peer_state = peer_state_mutex.lock().await;
                let became_dormant = ready
                    && neighbour
                    && topology::change_peer(&state, &mut peer_state, |p| {
                        p.check_dormancy(&state.timing, now)
                    });
                let keepalive = if neighbour {
                    peer_state.take_keepalive_due(&state.timing, now)
                } else {
                    None
                };
                if !neighbour {
                    peer_state.stop_hello_retries();
                }
                let unanswered = peer_state.take_overdue_sync_reply(now);
                (became_dormant, keepalive, unanswered)
            };

            if let Some(changes) = unanswered {
                debug!(peer = %peer, changes, "No answer to our sync message; sending again");
                crate::swarm::source::reply_timed_out(&state, peer);
                state.request_sync(peer);
            }

            if became_dormant {
                warn!(
                    peer = %peer,
                    silent_secs = state.timing.dormant_after.as_secs(),
                    probe_secs = state.timing.dormant_probe.as_secs(),
                    "Peer went dormant"
                );
            }

            // The catch-up source gets STATUSes: we need its readiness, and whether it
            // syncs with us.
            let keepalive = match keepalive {
                None if crate::swarm::source::probe_due(&state, peer, Instant::now()) => {
                    Some(Keepalive::Status)
                }
                keepalive => keepalive,
            };
            let sent = match keepalive {
                Some(Keepalive::Ping) => send_ping(&state, peer).await,
                Some(Keepalive::Status) => send_status(&state, peer).await,
                None => Ok(()),
            };
            if let Err(error) = sent {
                debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed");
            }

            if let Err(error) = retry_hello_if_due(&state, peer).await {
                warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
            }

            let due = peer_state_mutex
                .lock()
                .await
                .next_due(&state.timing, neighbour, ready);
            let probe = crate::swarm::source::probe_at(&state, peer);
            next = [Some(next), due, probe]
                .into_iter()
                .flatten()
                .min()
                .unwrap_or(next);
        }

        let records_changed = state
            .forget_records_changed
            .swap(false, std::sync::atomic::Ordering::AcqRel);
        let periodic = last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL);
        if records_changed || periodic {
            apply_forget_records(&state, &mut forgets_applied).await;
        }
        if periodic {
            forget_strangers(&state).await;
            crate::swarm::members::ensure_registered(&state).await;
            crate::swarm::stability::tick(&state).await;
            crate::swarm::skew::check(&state).await;
            crate::swarm::stats::measure_saved_size(&state);
            crate::swarm::source::warn_if_stalled(&state);
            crate::swarm::state_file::update(&state, false).await;
            {
                let known = state.peer_states.read().await;
                // And their addresses, where the layers below the node count
                // (`Traffic::node`).
                let addresses: std::collections::HashSet<String> = known
                    .keys()
                    .flat_map(|peer| state.transport.identities_of(peer))
                    .collect();
                state
                    .traffic
                    .retain(|peer| known.contains_key(peer) || addresses.contains(peer));
                state
                    .throughput
                    .lock_or_recover()
                    .retain(|peer| known.contains_key(peer));
            }
            // Fallback for what doesn't notify (`topology::run_worker`).
            state.topology_changed.notify_one();
            last_forget_check = Some(Instant::now());
        }
        if let Some(checked) = last_forget_check {
            next = next.min(checked + FORGET_CHECK_INTERVAL);
        }

        let passed = Instant::now();
        state.peers_due.until(next).await;
        tokio::time::sleep_until(passed + MIN_PASS_GAP).await;
    }
}