//! VSR cluster configuration.
//!
//! This module defines the configuration for a VSR cluster, including
//! cluster membership, timeouts, and protocol parameters.

use std::time::Duration;

use serde::{Deserialize, Serialize};

use crate::types::{MAX_REPLICAS, ReplicaId, quorum_size};

// ============================================================================
// Cluster Configuration
// ============================================================================

/// Configuration for a VSR cluster.
///
/// This configuration is immutable once a cluster is formed. Cluster
/// reconfiguration requires a separate protocol (TODO(v0.9.0): dynamic reconfiguration).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClusterConfig {
    /// The replica IDs in the cluster.
    ///
    /// Must be sorted and contain no duplicates.
    /// Length must be odd (2f+1) for proper quorum behavior.
    replicas: Vec<ReplicaId>,

    /// Protocol timeout configuration.
    pub timeouts: TimeoutConfig,

    /// Checkpoint configuration.
    pub checkpoint: CheckpointConfig,

    /// Maximum uncommitted operations in the pipeline.
    ///
    /// This bounds how far ahead the log can be from the commit point.
    /// Used during view changes to calculate conservative commit estimates.
    ///
    /// Default: 100
    pub max_pipeline_depth: u64,
}

impl ClusterConfig {
    /// Creates a new cluster configuration.
    ///
    /// Prefer [`ClusterConfig::try_new`] in new code — it returns a [`Result`]
    /// rather than panicking on invalid input.
    ///
    /// # Arguments
    ///
    /// * `replicas` - List of replica IDs in the cluster
    ///
    /// # Panics
    ///
    /// Panics if:
    /// - `replicas` is empty
    /// - `replicas` has an even number of elements (must be 2f+1)
    /// - `replicas` contains duplicates
    /// - `replicas` exceeds `MAX_REPLICAS`
    #[track_caller]
    pub fn new(replicas: Vec<ReplicaId>) -> Self {
        Self::try_new(replicas)
            .expect("ClusterConfig::new: invalid cluster configuration — use try_new for fallible construction")
    }

    /// Creates a new cluster configuration, validating invariants.
    ///
    /// # Errors
    ///
    /// Returns [`crate::VsrError::InvalidClusterConfig`] if replicas is empty,
    /// has even count, exceeds `MAX_REPLICAS`, or contains duplicates.
    pub fn try_new(mut replicas: Vec<ReplicaId>) -> crate::VsrResult<Self> {
        if replicas.is_empty() {
            return Err(crate::VsrError::InvalidClusterConfig(
                "cluster must have at least one replica",
            ));
        }
        if replicas.len() % 2 == 0 {
            return Err(crate::VsrError::InvalidClusterConfig(
                "cluster size must be odd (2f+1) for proper quorum behavior",
            ));
        }
        if replicas.len() > MAX_REPLICAS {
            return Err(crate::VsrError::InvalidClusterConfig(
                "cluster size exceeds MAX_REPLICAS",
            ));
        }

        // Sort and check for duplicates.
        replicas.sort();
        for i in 1..replicas.len() {
            if replicas[i - 1] == replicas[i] {
                return Err(crate::VsrError::InvalidClusterConfig(
                    "cluster contains duplicate replica IDs",
                ));
            }
        }

        Ok(Self {
            replicas,
            timeouts: TimeoutConfig::default(),
            checkpoint: CheckpointConfig::default(),
            max_pipeline_depth: 100,
        })
    }

    /// Creates a configuration for a single-node cluster.
    ///
    /// Single-node clusters are useful for development and testing.
    /// They provide no fault tolerance but maintain the same API.
    pub fn single_node(replica_id: ReplicaId) -> Self {
        Self {
            replicas: vec![replica_id],
            timeouts: TimeoutConfig::single_node(),
            checkpoint: CheckpointConfig::default(),
            max_pipeline_depth: 100,
        }
    }

    /// Returns the number of replicas in the cluster.
    pub fn cluster_size(&self) -> usize {
        self.replicas.len()
    }

    /// Returns the quorum size for this cluster.
    pub fn quorum_size(&self) -> usize {
        quorum_size(self.replicas.len())
    }

    /// Returns the maximum number of failures this cluster can tolerate.
    pub fn max_failures(&self) -> usize {
        self.replicas.len() / 2
    }

    /// Returns true if this is a single-node cluster.
    pub fn is_single_node(&self) -> bool {
        self.replicas.len() == 1
    }

    /// Returns an iterator over replica IDs.
    pub fn replicas(&self) -> impl Iterator<Item = ReplicaId> + '_ {
        self.replicas.iter().copied()
    }

    /// Returns the replica ID at the given index.
    ///
    /// # Panics
    ///
    /// Panics if `index` is out of bounds.
    pub fn replica_at(&self, index: usize) -> ReplicaId {
        self.replicas[index]
    }

    /// Returns the index of a replica ID in the cluster.
    ///
    /// Returns `None` if the replica is not in the cluster.
    pub fn replica_index(&self, id: ReplicaId) -> Option<usize> {
        self.replicas.iter().position(|&r| r == id)
    }

    /// Returns true if the replica is a member of this cluster.
    pub fn contains(&self, id: ReplicaId) -> bool {
        self.replicas.contains(&id)
    }

    /// Determines the leader for a given view.
    ///
    /// Leadership rotates through replicas based on view number.
    pub fn leader_for_view(&self, view: crate::types::ViewNumber) -> ReplicaId {
        let index = (view.as_u64() as usize) % self.replicas.len();
        self.replicas[index]
    }

    /// Returns the other replicas (excluding the given replica).
    pub fn others(&self, exclude: ReplicaId) -> impl Iterator<Item = ReplicaId> + '_ {
        self.replicas.iter().copied().filter(move |&r| r != exclude)
    }

    /// Sets the timeout configuration.
    pub fn with_timeouts(mut self, timeouts: TimeoutConfig) -> Self {
        self.timeouts = timeouts;
        self
    }

    /// Sets the checkpoint configuration.
    pub fn with_checkpoint(mut self, checkpoint: CheckpointConfig) -> Self {
        self.checkpoint = checkpoint;
        self
    }
}

// ============================================================================
// Timeout Configuration
// ============================================================================

/// Timeout configuration for VSR protocol operations.
///
/// These timeouts control how long a replica waits before taking action.
/// Timeouts should be tuned based on network latency and failure detection
/// requirements.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct TimeoutConfig {
    /// Time to wait for `PrepareOK` before retrying.
    ///
    /// If a leader doesn't receive enough `PrepareOKs` within this time,
    /// it retransmits the Prepare message.
    pub prepare_timeout: Duration,

    /// Time between heartbeats from the leader.
    ///
    /// Backups expect to receive heartbeats at this interval.
    /// Missing heartbeats trigger view change consideration.
    pub heartbeat_interval: Duration,

    /// Time without heartbeat before starting view change.
    ///
    /// Should be significantly larger than `heartbeat_interval` to
    /// avoid spurious view changes due to network jitter.
    pub view_change_timeout: Duration,

    /// Time to wait for `DoViewChange` messages during view change.
    ///
    /// If a replica doesn't receive enough `DoViewChange` messages,
    /// it may abandon the view change and try a higher view.
    pub view_change_wait: Duration,

    /// Time between recovery progress checks.
    ///
    /// During recovery, a replica periodically retries state transfer
    /// requests if they haven't been answered.
    pub recovery_retry: Duration,

    /// Time between repair requests for missing/corrupt entries.
    ///
    /// When a replica detects missing or corrupt log entries, it
    /// requests repairs at this interval.
    pub repair_interval: Duration,
}

impl TimeoutConfig {
    /// Creates timeout configuration suitable for single-node operation.
    ///
    /// Single-node clusters don't need network timeouts, so we use
    /// very short intervals.
    pub fn single_node() -> Self {
        Self {
            prepare_timeout: Duration::from_millis(1),
            heartbeat_interval: Duration::from_millis(10),
            view_change_timeout: Duration::from_millis(50),
            view_change_wait: Duration::from_millis(10),
            recovery_retry: Duration::from_millis(10),
            repair_interval: Duration::from_millis(10),
        }
    }

    /// Creates timeout configuration suitable for simulation testing.
    ///
    /// Uses very short intervals for fast test execution.
    pub fn simulation() -> Self {
        Self {
            prepare_timeout: Duration::from_micros(100),
            heartbeat_interval: Duration::from_micros(500),
            view_change_timeout: Duration::from_millis(2),
            view_change_wait: Duration::from_micros(500),
            recovery_retry: Duration::from_micros(500),
            repair_interval: Duration::from_micros(500),
        }
    }

    /// Creates timeout configuration suitable for local development.
    ///
    /// Uses moderate intervals for reasonable responsiveness.
    pub fn development() -> Self {
        Self {
            prepare_timeout: Duration::from_millis(100),
            heartbeat_interval: Duration::from_millis(250),
            view_change_timeout: Duration::from_secs(1),
            view_change_wait: Duration::from_millis(500),
            recovery_retry: Duration::from_millis(500),
            repair_interval: Duration::from_millis(500),
        }
    }

    /// Creates timeout configuration suitable for production.
    ///
    /// Uses longer intervals to handle network variability and
    /// avoid spurious failures.
    pub fn production() -> Self {
        Self {
            prepare_timeout: Duration::from_millis(500),
            heartbeat_interval: Duration::from_secs(1),
            view_change_timeout: Duration::from_secs(5),
            view_change_wait: Duration::from_secs(2),
            recovery_retry: Duration::from_secs(2),
            repair_interval: Duration::from_secs(1),
        }
    }
}

impl Default for TimeoutConfig {
    fn default() -> Self {
        Self::production()
    }
}

// ============================================================================
// Checkpoint Configuration
// ============================================================================

/// Configuration for VSR checkpoints.
///
/// Checkpoints provide consistent snapshots of the replicated state,
/// enabling efficient recovery and state transfer.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct CheckpointConfig {
    /// Number of operations between checkpoints.
    ///
    /// More frequent checkpoints reduce recovery time but increase
    /// storage and CPU overhead.
    pub checkpoint_interval: u64,

    /// Number of old checkpoints to retain.
    ///
    /// Old checkpoints enable state transfer to slow replicas.
    /// More retained checkpoints use more storage but allow recovery
    /// from longer outages.
    pub retain_count: usize,

    /// Whether to require Ed25519 signatures on checkpoints.
    ///
    /// Signed checkpoints provide cryptographic verification but
    /// add computational overhead.
    pub require_signatures: bool,
}

impl CheckpointConfig {
    /// Creates checkpoint configuration for testing.
    ///
    /// Uses frequent checkpoints for comprehensive test coverage.
    pub fn testing() -> Self {
        Self {
            checkpoint_interval: 10,
            retain_count: 3,
            require_signatures: false,
        }
    }
}

impl Default for CheckpointConfig {
    fn default() -> Self {
        Self {
            checkpoint_interval: 1000,
            retain_count: 5,
            require_signatures: true,
        }
    }
}

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

    #[test]
    fn cluster_config_creation() {
        let replicas = vec![ReplicaId::new(0), ReplicaId::new(1), ReplicaId::new(2)];
        let config = ClusterConfig::new(replicas);

        assert_eq!(config.cluster_size(), 3);
        assert_eq!(config.quorum_size(), 2);
        assert_eq!(config.max_failures(), 1);
        assert!(!config.is_single_node());
    }

    #[test]
    fn single_node_cluster() {
        let config = ClusterConfig::single_node(ReplicaId::new(0));

        assert_eq!(config.cluster_size(), 1);
        assert_eq!(config.quorum_size(), 1);
        assert_eq!(config.max_failures(), 0);
        assert!(config.is_single_node());
    }

    #[test]
    fn leader_rotation() {
        let replicas = vec![ReplicaId::new(0), ReplicaId::new(1), ReplicaId::new(2)];
        let config = ClusterConfig::new(replicas);

        assert_eq!(
            config.leader_for_view(ViewNumber::new(0)),
            ReplicaId::new(0)
        );
        assert_eq!(
            config.leader_for_view(ViewNumber::new(1)),
            ReplicaId::new(1)
        );
        assert_eq!(
            config.leader_for_view(ViewNumber::new(2)),
            ReplicaId::new(2)
        );
        assert_eq!(
            config.leader_for_view(ViewNumber::new(3)),
            ReplicaId::new(0)
        ); // wraps
    }

    #[test]
    fn replica_membership() {
        let replicas = vec![ReplicaId::new(1), ReplicaId::new(3), ReplicaId::new(5)];
        let config = ClusterConfig::new(replicas);

        assert!(config.contains(ReplicaId::new(1)));
        assert!(config.contains(ReplicaId::new(3)));
        assert!(config.contains(ReplicaId::new(5)));
        assert!(!config.contains(ReplicaId::new(0)));
        assert!(!config.contains(ReplicaId::new(2)));
    }

    #[test]
    fn others_excludes_self() {
        let replicas = vec![ReplicaId::new(0), ReplicaId::new(1), ReplicaId::new(2)];
        let config = ClusterConfig::new(replicas);

        let others: Vec<_> = config.others(ReplicaId::new(1)).collect();
        assert_eq!(others, vec![ReplicaId::new(0), ReplicaId::new(2)]);
    }

    #[test]
    #[should_panic(expected = "must be odd")]
    fn even_cluster_size_panics() {
        let replicas = vec![ReplicaId::new(0), ReplicaId::new(1)];
        let _ = ClusterConfig::new(replicas);
    }

    #[test]
    #[should_panic(expected = "at least one replica")]
    fn empty_cluster_panics() {
        let replicas: Vec<ReplicaId> = vec![];
        let _ = ClusterConfig::new(replicas);
    }

    #[test]
    #[should_panic(expected = "duplicate")]
    fn duplicate_replicas_panics() {
        let replicas = vec![ReplicaId::new(0), ReplicaId::new(0), ReplicaId::new(1)];
        let _ = ClusterConfig::new(replicas);
    }

    // ========================================================================
    // Property-Based Tests for Quorum Calculation (AUDIT-2026-03 M-5)
    // ========================================================================
    //
    // These properties ensure correctness of Byzantine fault tolerance:
    // 1. Quorum > f: Safety requires majority agreement
    // 2. No split-brain: Two disjoint quorums cannot both exist
    // 3. Quorum ≤ cluster_size: Validity check
    //
    // Run with: PROPTEST_CASES=10000 cargo test --package kimberlite-vsr
    //
    // **Security Context:** PCI-DSS Req 10.5.5, SOC 2 CC7.2, CWE-841

    #[cfg(test)]
    mod proptest_quorum {
        use super::*;
        use proptest::prelude::*;

        /// Generates valid odd cluster sizes (1, 3, 5, 7, ..., up to MAX_REPLICAS).
        fn odd_cluster_size() -> impl Strategy<Value = usize> {
            (0..=(crate::types::MAX_REPLICAS / 2)).prop_map(|f| 2 * f + 1)
        }

        proptest! {
            /// Property 1: Quorum size must be strictly greater than max failures.
            ///
            /// **Invariant:** `quorum > f` for `2f+1` replicas
            ///
            /// **Why it matters:** Byzantine fault tolerance requires that a quorum
            /// contains at least one honest replica. If `quorum <= f`, an all-Byzantine
            /// quorum could violate safety by committing conflicting operations.
            ///
            /// **Expected to catch:**
            /// - Off-by-one errors in quorum calculation (e.g., `n/2` instead of `n/2 + 1`)
            /// - Integer division truncation bugs
            #[test]
            fn quorum_strictly_greater_than_max_failures(cluster_size in odd_cluster_size()) {
                let replicas: Vec<ReplicaId> = (0..cluster_size as u8).map(ReplicaId::new).collect();
                let config = ClusterConfig::new(replicas);

                let quorum = config.quorum_size();
                let f = config.max_failures();

                // Core Byzantine fault tolerance invariant
                assert!(
                    quorum > f,
                    "Quorum ({quorum}) must be > f ({f}) for cluster_size={cluster_size}. \
                     Without this, an all-Byzantine quorum could compromise safety."
                );

                // Also verify the standard formula: quorum = floor(n/2) + 1
                let expected_quorum = (cluster_size / 2) + 1;
                assert_eq!(
                    quorum, expected_quorum,
                    "Quorum calculation incorrect for cluster_size={cluster_size}"
                );
            }

            /// Property 2: No split-brain possible (quorum intersection property).
            ///
            /// **Invariant:** Any two quorums must intersect in at least one replica.
            ///
            /// **Why it matters:** If two disjoint quorums could exist, the system could
            /// commit conflicting operations in different partitions (split-brain). The
            /// intersection guarantees at least one common replica that would detect conflicts.
            ///
            /// **Mathematical proof:** For `n = 2f+1` replicas:
            /// - Quorum size `q = f+1`
            /// - Two quorums: `|Q1| = f+1`, `|Q2| = f+1`
            /// - Maximum disjoint elements: `|Q1| + |Q2| - n = (f+1) + (f+1) - (2f+1) = 1`
            /// - Therefore, intersection size ≥ 1 (at least one shared replica)
            ///
            /// **Expected to catch:**
            /// - Quorum size too small (e.g., `n/2` instead of `n/2 + 1`)
            /// - Even cluster sizes (where split-brain is possible)
            #[test]
            fn no_split_brain_possible(cluster_size in odd_cluster_size()) {
                let replicas: Vec<ReplicaId> = (0..cluster_size as u8).map(ReplicaId::new).collect();
                let config = ClusterConfig::new(replicas);

                let quorum = config.quorum_size();

                // Two quorums of size `quorum` must share at least one replica
                // Proof: 2 * quorum > cluster_size implies overlap
                //
                // For split-brain to be impossible:
                //   If Q1 and Q2 are disjoint: |Q1| + |Q2| ≤ n
                //   But we have: |Q1| + |Q2| = 2*quorum = 2*(n/2 + 1) = n + 2 > n
                //   Contradiction! Therefore Q1 ∩ Q2 ≠ ∅
                assert!(
                    2 * quorum > cluster_size,
                    "Split-brain possible: two disjoint quorums of size {quorum} could exist in cluster_size={cluster_size}. \
                     This violates the fundamental quorum intersection property."
                );

                // Verify minimum intersection size
                let min_intersection = 2 * quorum - cluster_size;
                assert!(
                    min_intersection >= 1,
                    "Minimum quorum intersection must be ≥ 1, got {min_intersection} for cluster_size={cluster_size}"
                );
            }

            /// Property 3: Quorum size must not exceed cluster size.
            ///
            /// **Invariant:** `quorum ≤ cluster_size`
            ///
            /// **Why it matters:** A quorum larger than the cluster is impossible to achieve,
            /// making the system permanently stuck. This is a basic validity check.
            ///
            /// **Expected to catch:**
            /// - Arithmetic overflow in quorum calculation
            /// - Incorrect rounding (e.g., `n/2 + 2` instead of `n/2 + 1`)
            #[test]
            fn quorum_not_exceeds_cluster_size(cluster_size in odd_cluster_size()) {
                let replicas: Vec<ReplicaId> = (0..cluster_size as u8).map(ReplicaId::new).collect();
                let config = ClusterConfig::new(replicas);

                let quorum = config.quorum_size();

                assert!(
                    quorum <= cluster_size,
                    "Quorum ({quorum}) cannot exceed cluster_size ({cluster_size}). \
                     This would make consensus impossible."
                );
            }

            /// Property 4: Quorum calculation is deterministic and consistent.
            ///
            /// **Invariant:** Same cluster size always produces same quorum size,
            /// regardless of replica ID assignment.
            ///
            /// **Why it matters:** Non-deterministic quorum calculations would lead to
            /// replicas disagreeing on what constitutes a quorum, breaking consensus.
            #[test]
            fn quorum_calculation_deterministic(cluster_size in odd_cluster_size()) {
                // Create two configs with same size but different replica IDs
                let replicas1: Vec<ReplicaId> = (0..cluster_size as u8).map(ReplicaId::new).collect();

                // Use reversed IDs for second config (still within u8 range)
                let replicas2: Vec<ReplicaId> = (0..cluster_size as u8)
                    .rev()
                    .map(ReplicaId::new)
                    .collect();

                let config1 = ClusterConfig::new(replicas1);
                let config2 = ClusterConfig::new(replicas2);

                assert_eq!(
                    config1.quorum_size(), config2.quorum_size(),
                    "Quorum size must be deterministic for cluster_size={cluster_size}"
                );
            }

            /// Property 5: Standard quorum formulas hold for all cluster sizes.
            ///
            /// **Invariants:**
            /// - For `n = 2f+1`: quorum = f+1
            /// - quorum + f = n (complementary sizes)
            ///
            /// **Why it matters:** Validates implementation matches theoretical foundation.
            #[test]
            fn standard_quorum_formulas(cluster_size in odd_cluster_size()) {
                let replicas: Vec<ReplicaId> = (0..cluster_size as u8).map(ReplicaId::new).collect();
                let config = ClusterConfig::new(replicas);

                let quorum = config.quorum_size();
                let f = config.max_failures();

                // For n = 2f+1, quorum should be f+1
                assert_eq!(quorum, f + 1, "Quorum formula violated for cluster_size={cluster_size}");

                // Complementary sizes: quorum + f = n
                assert_eq!(
                    quorum + f, cluster_size,
                    "Complementary sizes property violated: {quorum} + {f} != {cluster_size}"
                );
            }
        }
    }
}
