//! Joining a swarm (or creating one as the genesis node).
//!
//! A node starts a new swarm only with `--genesis`. Otherwise it joins through its
//! seeds and keeps trying until it is ready: seeds that don't resolve are resolved
//! again with backoff, and HELLOs to resolved ones are retried by the peer worker. A
//! node without seeds waits (nothing to join yet) and says so now and then.

use std::sync::atomic::Ordering;
use std::time::Duration;

use anyhow::Result;
use tokio::time::Instant;
use tracing::{debug, info, warn};
use uuid::Uuid;

use crate::state::AppState;
use crate::swarm::peer::{remember_peer, send_hello_once};

const BOOTSTRAP_POLL: Duration = Duration::from_millis(500);
/// How often a node that is still joining says why.
const BOOTSTRAP_REPORT: Duration = Duration::from_secs(300);
/// Retrying seeds that didn't resolve: first after this, doubling up to the maximum.
const RESOLVE_RETRY: Duration = Duration::from_secs(5);
const RESOLVE_RETRY_MAX: Duration = Duration::from_secs(300);

pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool) -> Result<()> {
    if genesis {
        info!("Starting a new swarm as its genesis node (--genesis)");
        state.initial_sync_done.store(true, Ordering::Release);
        state.mark_ready(Uuid::new_v4().to_string()).await?;
        return Ok(());
    }

    if seeds.is_empty() {
        warn!(
            "No seeds: nothing to join. Add --seeds to join a swarm, or --genesis to start a new one"
        );
    } else {
        info!(seeds = seeds.len(), "Joining existing swarm");
    }

    let mut unresolved: Vec<String> = seeds.to_vec();
    let mut resolve_retry = RESOLVE_RETRY;
    let mut next_resolve = Instant::now();
    let mut next_report = Instant::now() + BOOTSTRAP_REPORT;
    loop {
        if let Some(swarm_id) = state.swarm_id().await {
            // A node joining a compacted swarm rejoins the newer epoch first; it is
            // ready only once that rejoin verified the epoch.
            if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() {
                state.mark_ready(swarm_id.clone()).await?;
                info!(swarm_id = %swarm_id, "Node is ready");
                return Ok(());
            }
        }

        let now = Instant::now();
        if !unresolved.is_empty() && now >= next_resolve {
            resolve_seeds(state, &mut unresolved).await;
            if !unresolved.is_empty() {
                next_resolve = now + resolve_retry;
                resolve_retry = (resolve_retry * 2).min(RESOLVE_RETRY_MAX);
            }
        }
        if now >= next_report {
            next_report = now + BOOTSTRAP_REPORT;
            report_joining(state, seeds, &unresolved).await;
        }
        tokio::time::sleep(BOOTSTRAP_POLL).await;
    }
}

/// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the
/// rest for later.
async fn resolve_seeds(state: &AppState, unresolved: &mut Vec<String>) {
    let mut still = Vec::new();
    for seed in unresolved.drain(..) {
        match state.transport.resolve_seed(&seed).await {
            Ok(peer) => {
                debug!(seed = %seed, peer = %peer, "Resolved seed");
                remember_peer(state, &peer).await;
                if let Err(error) = send_hello_once(state, &peer).await {
                    warn!(peer = %peer, error = %format!("{error:#}"), "Initial HELLO failed; will retry");
                }
            }
            Err(error) => {
                warn!(seed = %seed, error = %error, "Seed doesn't resolve; will retry");
                still.push(seed);
            }
        }
    }
    *unresolved = still;
}

/// Why the node is still joining, for the log.
async fn report_joining(state: &AppState, seeds: &[String], unresolved: &[String]) {
    if seeds.is_empty() {
        warn!(
            "Still no seeds: nothing to join. Add --seeds to join a swarm, or --genesis to start a new one"
        );
        return;
    }
    let mut peer_diagnostics: Vec<String> = Vec::new();
    let peer_states = state.peer_states.read().await.clone();
    for (peer, peer_state_mutex) in &peer_states {
        let peer_state = peer_state_mutex.lock().await;
        peer_diagnostics.push(format!(
            "{peer}: validated_swarm={:?}, epoch={:?}",
            peer_state.standing.swarm_id(),
            peer_state.epoch,
        ));
    }
    warn!(
        local_swarm = ?state.swarm_id().await,
        local_epoch = %state.local_epoch(),
        initial_sync_done = state.initial_sync_done.load(Ordering::Acquire),
        unresolved_seeds = ?unresolved,
        peers = %peer_diagnostics.join("; "),
        "Still joining the swarm; trying on"
    );
}