//! The peer worker: every `PEER_TICK`, keepalives due, dormancy, HELLO retries and
//! overdue sync replies; every `FORGET_CHECK_INTERVAL` (and at once when the node
//! becomes ready), forget records, the member registry, stability, measurements and
//! the state file.

use std::time::Duration;

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

use super::forget::apply_forget_records;
use super::handshake::retry_hello_if_due;
use super::keepalive::send_ping;
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;

const PEER_TICK: Duration = Duration::from_millis(500);

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

pub(crate) async fn run_peer_worker(state: AppState) {
    let mut tick = tokio::time::interval(PEER_TICK);
    tick.set_missed_tick_behavior(MissedTickBehavior::Delay);

    info!("Starting peer maintenance worker");

    let mut last_forget_check: Option<Instant> = None;
    let mut was_ready = false;

    loop {
        tick.tick().await;

        // 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, ping_due, unanswered) = {
                let mut peer_state = peer_state_mutex.lock().await;
                let became_dormant =
                    ready && neighbour && peer_state.check_dormancy(&state.timing, now);
                let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now);
                if !neighbour {
                    peer_state.stop_hello_retries();
                }
                let unanswered = peer_state.take_overdue_sync_reply(now);
                (became_dormant, ping_due, 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);
                let _ = state
                    .sync_events
                    .send(SyncEvent::PeerNeedsSync(peer.clone()));
            }

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

            let probe_due =
                !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
            if ping_due || probe_due {
                if let Err(error) = send_ping(&state, peer).await {
                    debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed");
                }
            }

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

        if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
            apply_forget_records(&state).await;
            crate::swarm::members::ensure_registered(&state).await;
            crate::swarm::stability::tick(&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;
                state.traffic.retain(|peer| known.contains_key(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());
        }
    }
}