//! Joining a cluster (or creating one as the genesis node).

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

use anyhow::{bail, Result};
use tracing::{debug, error, info, instrument, warn};
use uuid::Uuid;

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

const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300);
const BOOTSTRAP_RETRY: Duration = Duration::from_millis(500);

#[instrument(skip(state, seeds))]
pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> {
    if seeds.is_empty() {
        info!("Initializing genesis node (no seed peers configured)");
        state.initial_sync_done.store(true, Ordering::Release);
        state.mark_ready(Uuid::new_v4().to_string()).await?;
        return Ok(());
    }

    info!(
        timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(),
        "Joining existing cluster"
    );

    let mut peers = Vec::new();

    for seed in seeds {
        match state.transport.resolve_peer(seed).await {
            Ok(peer) => {
                debug!(seed = %seed, peer = %peer, "Resolved seed to peer");
                peers.push(peer)
            }
            Err(e) => warn!(seed = %seed, error = %e, "Ignoring seed argument"),
        }
    }

    if peers.is_empty() {
        bail!("None of the configured seeds could be resolved");
    }

    for peer in peers {
        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");
        }
    }

    let deadline = tokio::time::Instant::now() + BOOTSTRAP_TIMEOUT;
    loop {
        if let Some(cluster_id) = state.cluster_id().await {
            if state.initial_sync_done.load(Ordering::Acquire) {
                state.mark_ready(cluster_id.clone()).await?;
                info!(cluster_id = %cluster_id, "Node is ready");
                return Ok(());
            }
        }

        if tokio::time::Instant::now() >= deadline {
            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;
                let validated_cluster = match &peer_state.membership {
                    PeerMembership::Unknown => None,
                    PeerMembership::Member { cluster_id } => Some(cluster_id.as_str()),
                };
                peer_diagnostics.push(format!(
                    "{peer}: validated_cluster={:?}, epoch={:?}, pending_inbound_sync={}",
                    validated_cluster,
                    peer_state.epoch,
                    peer_state.pending_inbound_sync.len()
                ))
            }
            let local_cluster = state.cluster_id().await;
            let local_epoch = state.local_epoch();
            let initial_sync_done = state.initial_sync_done.load(Ordering::Acquire);
            error!(
                local_cluster = ?local_cluster,
                local_epoch = %local_epoch,
                initial_sync_done,
                peers = %peer_diagnostics.join("; "),
                timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(),
                "Bootstrap timed out"
            );
            bail!(
                "Bootstrap timed out after {:?}; local_cluster={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.",
                BOOTSTRAP_TIMEOUT,
                local_cluster,
                local_epoch,
                initial_sync_done,
                peer_diagnostics.join("; ")
            );
        }
        tokio::time::sleep(BOOTSTRAP_RETRY).await;
    }
}