//! # Cluster Module — Lithair Distributed Consensus
//!
//! Provides the building blocks for multi-node replication in Lithair.
//! Enable with `LithairServer::with_raft_cluster(node_id, peers)`.
//!
//! ## Architecture
//!
//! The replication pipeline processes writes through these components:
//!
//! ```text
//! HTTP Write
//!   → ConsensusLog (in-memory ordered log, atomic index assignment)
//!   → WAL (disk persistence, rkyv serialization, group commit every 5ms)
//!   + Replication (HTTP POST /_raft/append to followers, parallel)
//!   → Majority ack → Commit → Apply to state machine
//! ```
//!
//! ## Submodules
//!
//! - [`consensus_log`] — In-memory append-only log with term/index tracking
//! - [`wal`] — Write-Ahead Log for crash recovery (rkyv + group commit)
//! - [`replication_batcher`] — Per-follower health tracking and intelligent batching
//! - [`snapshot`] — Full-state snapshots for desynced follower resync
//! - [`upgrade`] — Rolling upgrade support with schema migration tracking
//!
//! ## Leader Election
//!
//! Currently uses static election: lowest `node_id` among alive peers becomes leader.
//! The [`RaftLeadershipState`] tracks leadership and sends periodic heartbeats
//! to prevent unnecessary elections during idle periods.
//!
//! ## Recovery
//!
//! On startup, the ConsensusLog is empty. The WAL is replayed to restore
//! all committed entries. See [`ConsensusLog::replay_from_wal_entries`].

use clap::Parser;
use reqwest::Client as HttpClient;
use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU64, Ordering};
use std::time::{Duration, Instant};

pub mod consensus_log;
pub mod replication_batcher;
pub mod snapshot;
pub mod upgrade;
pub mod wal;

pub use consensus_log::{ApplyResult, ConsensusLog, CrudOperation, LogEntry, LogId};
pub use replication_batcher::{BatcherConfig, FollowerHealth, FollowerStats, ReplicationBatcher};
pub use snapshot::{
    InstallSnapshotRequest, InstallSnapshotResponse, SnapshotData, SnapshotManager, SnapshotMeta,
};
pub use upgrade::{
    FieldDefinition, FieldType, MigrationContext, MigrationManager, MigrationStatus, ModelSchema,
    NodeMode, RollbackOp, SchemaChange, Version,
};
pub use wal::{GroupCommitConfig, WriteAheadLog};
// Resync stats are defined in this file, no re-export needed

/// Raft Node State for leader election and failover
#[derive(Debug, Clone, PartialEq)]
pub enum RaftNodeState {
    Follower,
    Candidate,
    Leader,
}

/// Raft Leadership State
///
/// Manages the Raft consensus state for a cluster node including:
/// - Leader election
/// - Heartbeat tracking
/// - State transitions (Follower -> Candidate -> Leader)
pub struct RaftLeadershipState {
    pub node_id: u64,
    pub self_port: u16,
    pub current_state: AtomicU64, // 0=Follower, 1=Candidate, 2=Leader
    pub is_leader: AtomicBool,
    pub current_leader_id: AtomicU64,
    pub leader_port: AtomicU16,
    pub peers: Vec<String>,
    pub last_heartbeat: std::sync::Mutex<Instant>,
    pub election_timeout: Duration,
}

impl RaftLeadershipState {
    /// Create a new RaftLeadershipState
    ///
    /// Uses static leader election: lowest node_id becomes leader initially.
    /// Dynamic election occurs when leader fails (heartbeat timeout).
    pub fn new(node_id: u64, self_port: u16, peers: Vec<String>) -> Self {
        // Simple leadership election: lowest node_id is leader
        // This is a static election - node_id 0 is always leader if present
        let is_leader = node_id == 0 || peers.is_empty();

        // Leader port: if we are the leader, it's our port.
        // For followers, we don't know the leader's port yet — it will be
        // discovered from the first heartbeat (AppendEntriesRequest contains
        // the leader_id, and the request comes from the leader's port).
        // Initialize to 0 for followers; updated by update_leader_port().
        let leader_port = if is_leader { self_port } else { 0 };

        // Find the leader node_id (always 0 in static election)
        let current_leader_id = 0u64;

        Self {
            node_id,
            self_port,
            current_state: AtomicU64::new(if is_leader { 2 } else { 0 }), // 2=Leader, 0=Follower
            is_leader: AtomicBool::new(is_leader),
            current_leader_id: AtomicU64::new(current_leader_id),
            leader_port: AtomicU16::new(leader_port),
            peers,
            last_heartbeat: std::sync::Mutex::new(Instant::now()),
            election_timeout: Duration::from_secs(5), // 5 second timeout
        }
    }

    /// Check if this node is currently the leader
    pub fn is_leader(&self) -> bool {
        self.is_leader.load(Ordering::Relaxed)
    }

    /// Get the current Raft state (Follower, Candidate, or Leader)
    pub fn get_current_state(&self) -> RaftNodeState {
        match self.current_state.load(Ordering::Relaxed) {
            0 => RaftNodeState::Follower,
            1 => RaftNodeState::Candidate,
            2 => RaftNodeState::Leader,
            _ => RaftNodeState::Follower,
        }
    }

    /// Get the current leader's port
    pub fn get_leader_port(&self) -> u16 {
        self.leader_port.load(Ordering::Relaxed)
    }

    /// Check if the provided id matches the current authoritative leader id
    #[allow(dead_code)]
    pub(crate) fn is_authoritative_leader_id(&self, leader_id: u64) -> bool {
        self.current_leader_id.load(Ordering::Relaxed) == leader_id
    }

    /// Update the last heartbeat timestamp (called when receiving heartbeat from leader)
    pub fn update_heartbeat(&self) {
        if let Ok(mut heartbeat) = self.last_heartbeat.lock() {
            *heartbeat = Instant::now();
        }
    }

    /// Update the leader port from AppendEntries RPC.
    ///
    /// Followers don't know the leader's port at startup (ports are dynamic).
    /// The leader's port is discovered from the first heartbeat or AppendEntries
    /// request, which includes the leader_id. The sender's address gives us the port.
    pub fn update_leader_port(&self, leader_id: u64, leader_port: u16) {
        self.current_leader_id.store(leader_id, Ordering::SeqCst);
        if leader_port > 0 {
            self.leader_port.store(leader_port, Ordering::SeqCst);
        }
    }

    /// Check if election should be started (heartbeat timeout exceeded)
    pub fn should_start_election(&self) -> bool {
        if self.is_leader() {
            return false;
        }

        if let Ok(heartbeat) = self.last_heartbeat.lock() {
            heartbeat.elapsed() > self.election_timeout
        } else {
            false
        }
    }

    /// Become the leader - called when this node wins an election
    pub fn become_leader(&self) {
        self.is_leader.store(true, Ordering::SeqCst);
        self.current_state.store(2, Ordering::SeqCst); // 2 = Leader
        self.current_leader_id.store(self.node_id, Ordering::SeqCst);
        self.leader_port.store(self.self_port, Ordering::SeqCst);
        self.update_heartbeat();
        log::info!("Node {} is now the LEADER", self.node_id);
    }

    /// Become a follower - called when a new leader is detected
    pub fn become_follower(&self, new_leader_id: u64, new_leader_port: u16) {
        self.is_leader.store(false, Ordering::SeqCst);
        self.current_state.store(0, Ordering::SeqCst); // 0 = Follower
        self.current_leader_id.store(new_leader_id, Ordering::SeqCst);
        self.leader_port.store(new_leader_port, Ordering::SeqCst);
        self.update_heartbeat();
        log::info!(
            "Node {} is now a FOLLOWER (leader: node {} on port {})",
            self.node_id,
            new_leader_id,
            new_leader_port
        );
    }

    /// Start election process - find the lowest available node_id to be leader
    ///
    /// Returns (should_become_leader, new_leader_id, new_leader_port)
    pub async fn start_election(&self) -> (bool, u64, u16) {
        log::debug!("Node {} starting election...", self.node_id);
        self.current_state.store(1, Ordering::SeqCst); // 1 = Candidate

        let client = HttpClient::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .unwrap_or_else(|_| HttpClient::new());

        // Check which peers are alive
        let mut alive_peers: Vec<(u64, u16)> = Vec::new();

        for peer in &self.peers {
            let url = format!("http://{}/status", peer);
            match client.get(&url).send().await {
                Ok(resp) if resp.status().is_success() => {
                    if let Ok(status) = resp.json::<serde_json::Value>().await {
                        if let Some(raft) = status.get("raft") {
                            let peer_id =
                                raft.get("node_id").and_then(|v| v.as_u64()).unwrap_or(u64::MAX);
                            let peer_port = peer
                                .split(':')
                                .nth(1)
                                .and_then(|p| p.parse::<u16>().ok())
                                .unwrap_or(0);
                            alive_peers.push((peer_id, peer_port));
                            log::debug!("Peer {} (node {}) is alive", peer, peer_id);
                        }
                    }
                }
                _ => {
                    log::warn!("Peer {} is not responding", peer);
                }
            }
        }

        // Find the lowest node_id among alive nodes (including self)
        let mut candidates: Vec<(u64, u16)> = alive_peers;
        candidates.push((self.node_id, self.self_port));
        candidates.sort_by_key(|(id, _)| *id);

        let (winner_id, winner_port) = candidates[0];
        let should_become_leader = winner_id == self.node_id;

        log::info!("Election result: node {} wins (port {})", winner_id, winner_port);

        (should_become_leader, winner_id, winner_port)
    }

    /// Get time since last heartbeat
    pub fn time_since_heartbeat(&self) -> Duration {
        if let Ok(heartbeat) = self.last_heartbeat.lock() {
            heartbeat.elapsed()
        } else {
            Duration::from_secs(0)
        }
    }
}

// ============================================================================
// RESYNC STATISTICS
// ============================================================================

/// Statistics for snapshot-based resync operations
///
/// Tracks all snapshot resync activity to provide observability into the
/// resync mechanism. Useful for debugging and validating that resync is working.
#[derive(Debug, Default)]
pub struct ResyncStats {
    /// Number of snapshots created by leader for resync
    pub snapshots_created: AtomicU64,
    /// Number of snapshot send attempts
    pub snapshot_send_attempts: AtomicU64,
    /// Number of successful snapshot sends
    pub snapshot_send_successes: AtomicU64,
    /// Number of failed snapshot sends
    pub snapshot_send_failures: AtomicU64,
    /// Number of snapshots received (follower side)
    pub snapshots_received: AtomicU64,
    /// Number of snapshots successfully applied (follower side)
    pub snapshots_applied: AtomicU64,
    /// Last snapshot index sent
    pub last_snapshot_index_sent: AtomicU64,
    /// Last snapshot index received
    pub last_snapshot_index_received: AtomicU64,
    /// Timestamp of last resync activity (unix millis)
    pub last_resync_timestamp_ms: AtomicU64,
}

impl ResyncStats {
    pub fn new() -> Self {
        Self::default()
    }

    /// Record a snapshot creation
    pub fn record_snapshot_created(&self) {
        self.snapshots_created.fetch_add(1, Ordering::Relaxed);
        self.update_timestamp();
    }

    /// Record a snapshot send attempt
    pub fn record_send_attempt(&self, index: u64) {
        self.snapshot_send_attempts.fetch_add(1, Ordering::Relaxed);
        self.last_snapshot_index_sent.store(index, Ordering::Relaxed);
        self.update_timestamp();
    }

    /// Record a successful snapshot send
    pub fn record_send_success(&self) {
        self.snapshot_send_successes.fetch_add(1, Ordering::Relaxed);
        self.update_timestamp();
    }

    /// Record a failed snapshot send
    pub fn record_send_failure(&self) {
        self.snapshot_send_failures.fetch_add(1, Ordering::Relaxed);
        self.update_timestamp();
    }

    /// Record a snapshot received (follower side)
    pub fn record_snapshot_received(&self, index: u64) {
        self.snapshots_received.fetch_add(1, Ordering::Relaxed);
        self.last_snapshot_index_received.store(index, Ordering::Relaxed);
        self.update_timestamp();
    }

    /// Record a snapshot applied (follower side)
    pub fn record_snapshot_applied(&self) {
        self.snapshots_applied.fetch_add(1, Ordering::Relaxed);
        self.update_timestamp();
    }

    fn update_timestamp(&self) {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;
        self.last_resync_timestamp_ms.store(now, Ordering::Relaxed);
    }

    /// Get stats as JSON-serializable struct
    pub fn to_json(&self) -> ResyncStatsJson {
        ResyncStatsJson {
            snapshots_created: self.snapshots_created.load(Ordering::Relaxed),
            snapshot_send_attempts: self.snapshot_send_attempts.load(Ordering::Relaxed),
            snapshot_send_successes: self.snapshot_send_successes.load(Ordering::Relaxed),
            snapshot_send_failures: self.snapshot_send_failures.load(Ordering::Relaxed),
            snapshots_received: self.snapshots_received.load(Ordering::Relaxed),
            snapshots_applied: self.snapshots_applied.load(Ordering::Relaxed),
            last_snapshot_index_sent: self.last_snapshot_index_sent.load(Ordering::Relaxed),
            last_snapshot_index_received: self.last_snapshot_index_received.load(Ordering::Relaxed),
            last_resync_timestamp_ms: self.last_resync_timestamp_ms.load(Ordering::Relaxed),
        }
    }
}

/// JSON-serializable version of ResyncStats
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ResyncStatsJson {
    pub snapshots_created: u64,
    pub snapshot_send_attempts: u64,
    pub snapshot_send_successes: u64,
    pub snapshot_send_failures: u64,
    pub snapshots_received: u64,
    pub snapshots_applied: u64,
    pub last_snapshot_index_sent: u64,
    pub last_snapshot_index_received: u64,
    pub last_resync_timestamp_ms: u64,
}

/// Standard command line arguments for Lithair cluster applications
///
/// Use with clap to parse cluster node configuration from CLI.
///
/// # Example
/// ```rust,ignore
/// use clap::Parser;
/// use lithair_core::cluster::ClusterArgs;
///
/// let args = ClusterArgs::parse();
/// let peers: Vec<String> = args.peers
///     .unwrap_or_default()
///     .iter()
///     .map(|p| format!("127.0.0.1:{}", p))
///     .collect();
///
/// LithairServer::new()
///     .with_port(args.port)
///     .with_raft_cluster(args.node_id, peers)
///     .build()?
///     .serve()
///     .await?;
/// ```
#[derive(Parser, Debug)]
#[command(name = "lithair-cluster")]
#[command(about = "Lithair Cluster Node")]
pub struct ClusterArgs {
    /// Unique node identifier (0 = initial leader)
    #[arg(long)]
    pub node_id: u64,

    /// Port to listen on
    #[arg(long)]
    pub port: u16,

    /// Comma-separated list of peer ports
    #[arg(long, value_delimiter = ',')]
    pub peers: Option<Vec<u16>>,
}

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

    #[test]
    fn test_raft_state_leader_election() {
        // Node 0 should be leader
        let state = RaftLeadershipState::new(0, 8080, vec!["127.0.0.1:8081".to_string()]);
        assert!(state.is_leader());
        assert_eq!(state.get_current_state(), RaftNodeState::Leader);
        assert_eq!(state.get_leader_port(), 8080);
    }

    #[test]
    fn test_raft_state_follower() {
        // Node 1 should be follower. Leader port is unknown until first heartbeat.
        let state = RaftLeadershipState::new(1, 8081, vec!["127.0.0.1:8080".to_string()]);
        assert!(!state.is_leader());
        assert_eq!(state.get_current_state(), RaftNodeState::Follower);
        assert_eq!(state.get_leader_port(), 0); // Unknown until heartbeat from leader

        // After receiving leader info (e.g. from heartbeat), port is updated
        state.update_leader_port(0, 8080);
        assert_eq!(state.get_leader_port(), 8080);
    }

    #[test]
    fn test_raft_state_single_node() {
        // Single node should be leader
        let state = RaftLeadershipState::new(5, 9000, vec![]);
        assert!(state.is_leader());
        assert_eq!(state.get_leader_port(), 9000);
    }

    #[test]
    fn test_become_leader() {
        let state = RaftLeadershipState::new(1, 8081, vec!["127.0.0.1:8080".to_string()]);
        assert!(!state.is_leader());

        state.become_leader();
        assert!(state.is_leader());
        assert_eq!(state.get_current_state(), RaftNodeState::Leader);
        assert_eq!(state.get_leader_port(), 8081);
    }

    #[test]
    fn test_become_follower() {
        let state = RaftLeadershipState::new(0, 8080, vec!["127.0.0.1:8081".to_string()]);
        assert!(state.is_leader());

        state.become_follower(1, 8081);
        assert!(!state.is_leader());
        assert_eq!(state.get_current_state(), RaftNodeState::Follower);
        assert_eq!(state.get_leader_port(), 8081);
    }

    #[test]
    fn test_heartbeat_timeout() {
        let state = RaftLeadershipState::new(1, 8081, vec!["127.0.0.1:8080".to_string()]);

        // Fresh state should not trigger election
        assert!(!state.should_start_election());

        // Update heartbeat
        state.update_heartbeat();
        assert!(!state.should_start_election());
    }

    #[test]
    fn test_leader_never_triggers_election() {
        let state = RaftLeadershipState::new(0, 8080, vec!["127.0.0.1:8081".to_string()]);
        assert!(state.is_leader());

        // Manually expire the heartbeat by setting it far in the past
        {
            let mut hb = state.last_heartbeat.lock().unwrap();
            *hb = Instant::now() - Duration::from_secs(60);
        }

        // Leader should never start election, even with expired heartbeat
        assert!(!state.should_start_election());
    }

    #[test]
    fn test_concurrent_state_transitions() {
        use std::sync::Arc;
        use std::thread;

        let state = Arc::new(RaftLeadershipState::new(1, 8081, vec!["127.0.0.1:8080".to_string()]));
        let mut handles = Vec::new();

        for i in 0..10 {
            let s = Arc::clone(&state);
            handles.push(thread::spawn(move || {
                if i % 2 == 0 {
                    s.become_leader();
                } else {
                    s.become_follower(0, 8080);
                }
            }));
        }

        for h in handles {
            h.join().unwrap();
        }

        // Final state must be consistent: either leader or follower, not torn
        let is_leader = state.is_leader();
        let node_state = state.get_current_state();
        if is_leader {
            assert_eq!(node_state, RaftNodeState::Leader);
            assert_eq!(state.get_leader_port(), 8081);
        } else {
            assert_eq!(node_state, RaftNodeState::Follower);
            assert_eq!(state.get_leader_port(), 8080);
        }
    }

    #[tokio::test]
    async fn test_election_with_no_alive_peers() {
        // No HTTP servers running on peer ports — self should win
        let state = RaftLeadershipState::new(
            3,
            9003,
            vec!["127.0.0.1:19999".to_string(), "127.0.0.1:19998".to_string()],
        );

        let (should_lead, winner_id, winner_port) = state.start_election().await;
        assert!(should_lead);
        assert_eq!(winner_id, 3);
        assert_eq!(winner_port, 9003);
    }

    #[test]
    fn test_resync_stats_concurrent_increments() {
        use std::sync::Arc;
        use std::thread;

        let stats = Arc::new(ResyncStats::new());
        let mut handles = Vec::new();

        for _ in 0..100 {
            let s = Arc::clone(&stats);
            handles.push(thread::spawn(move || {
                s.record_snapshot_created();
                s.record_send_attempt(1);
                s.record_send_success();
            }));
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(stats.snapshots_created.load(Ordering::Relaxed), 100);
        assert_eq!(stats.snapshot_send_attempts.load(Ordering::Relaxed), 100);
        assert_eq!(stats.snapshot_send_successes.load(Ordering::Relaxed), 100);
        assert!(stats.last_resync_timestamp_ms.load(Ordering::Relaxed) > 0);
    }
}
