#![allow(dead_code)] // Many fields/methods are infrastructure for future tests

//! VSR simulation testing module.
//!
//! This module provides deterministic simulation testing for the VSR protocol,
//! enabling comprehensive testing of:
//!
//! - Normal operation (prepare/commit flow)
//! - View changes (leader failure, election)
//! - Network faults (swizzle-clogging, partitions)
//! - Gray failures (slow/intermittent nodes)
//! - Recovery and repair scenarios
//! - Byte-identical replica consistency
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────┐
//! │                    VSR Simulation Harness                        │
//! │  ┌─────────────┐   ┌──────────────┐   ┌─────────────────────┐   │
//! │  │ SimClock    │   │ MessageQueue │   │ SimRng              │   │
//! │  └─────────────┘   └──────────────┘   └─────────────────────┘   │
//! │                                                                   │
//! │  ┌─────────────────────────────────────────────────────────────┐ │
//! │  │                    Simulated Replicas                        │ │
//! │  │  Replica₀     Replica₁     Replica₂     ...                 │ │
//! │  └─────────────────────────────────────────────────────────────┘ │
//! │                                                                   │
//! │  ┌─────────────────────────────────────────────────────────────┐ │
//! │  │                    Fault Injectors                           │ │
//! │  │  SwizzleClogger    GrayFailureInjector    StorageFaults     │ │
//! │  └─────────────────────────────────────────────────────────────┘ │
//! │                                                                   │
//! │  ┌─────────────────────────────────────────────────────────────┐ │
//! │  │                    Invariant Checkers                        │ │
//! │  │  ReplicaConsistency    LinearizabilityChecker               │ │
//! │  └─────────────────────────────────────────────────────────────┘ │
//! └─────────────────────────────────────────────────────────────────┘
//! ```

use std::collections::HashMap;

use kimberlite_kernel::Command;
use kimberlite_types::{DataClass, IdempotencyId, Placement};
use rand::prelude::*;
use rand::rngs::SmallRng;

use crate::config::ClusterConfig;
use crate::message::Message;
use crate::replica::{ReplicaEvent, ReplicaOutput, ReplicaState, TimeoutKind};
use crate::types::{CommitNumber, OpNumber, ReplicaId, ReplicaStatus};

// ============================================================================
// Constants
// ============================================================================

/// Default heartbeat interval in simulation ticks.
const HEARTBEAT_INTERVAL: u64 = 100;

/// Default prepare timeout in simulation ticks.
#[allow(dead_code)]
const PREPARE_TIMEOUT: u64 = 50;

/// Default view change timeout in simulation ticks.
const VIEW_CHANGE_TIMEOUT: u64 = 200;

// ============================================================================
// Simulation Events
// ============================================================================

/// Events in the VSR simulation.
#[derive(Debug, Clone)]
#[allow(clippy::large_enum_variant)] // MessageDelivery contains Message (416 bytes) - acceptable for simulation
pub enum SimEvent {
    /// A message is delivered to a replica.
    MessageDelivery { to: ReplicaId, message: Message },
    /// A timeout fires for a replica.
    Timeout {
        replica: ReplicaId,
        kind: TimeoutKind,
    },
    /// A client submits a request.
    ClientRequest {
        command: Command,
        idempotency_id: Option<IdempotencyId>,
    },
    /// Periodic tick for housekeeping.
    Tick,
}

// ============================================================================
// Scheduled Event
// ============================================================================

/// An event scheduled to occur at a specific time.
#[derive(Debug, Clone)]
struct ScheduledEvent {
    /// Time when the event should occur.
    time: u64,
    /// Sequence number for FIFO ordering of same-time events.
    seq: u64,
    /// The event to execute.
    event: SimEvent,
}

impl PartialEq for ScheduledEvent {
    fn eq(&self, other: &Self) -> bool {
        self.time == other.time && self.seq == other.seq
    }
}

impl Eq for ScheduledEvent {}

impl PartialOrd for ScheduledEvent {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for ScheduledEvent {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        // Reverse order for min-heap behavior (earliest time first)
        // For same time, use sequence number (lower seq = earlier)
        match other.time.cmp(&self.time) {
            std::cmp::Ordering::Equal => other.seq.cmp(&self.seq),
            ord => ord,
        }
    }
}

// ============================================================================
// Network Fault Configuration
// ============================================================================

/// Configuration for network fault injection.
#[derive(Debug, Clone)]
pub struct NetworkFaultConfig {
    /// Probability of dropping a message (0.0 - 1.0).
    pub drop_probability: f64,
    /// Probability of duplicating a message.
    pub duplicate_probability: f64,
    /// Probability of reordering messages.
    pub reorder_probability: f64,
    /// Maximum delay to add to messages (in ticks).
    pub max_delay: u64,
    /// Minimum delay for messages (in ticks).
    pub min_delay: u64,
}

impl Default for NetworkFaultConfig {
    fn default() -> Self {
        Self {
            drop_probability: 0.0,
            duplicate_probability: 0.0,
            reorder_probability: 0.0,
            max_delay: 10,
            min_delay: 1,
        }
    }
}

impl NetworkFaultConfig {
    /// Creates a configuration for swizzle-clogging (intermittent network).
    pub fn swizzle_clogging() -> Self {
        Self {
            drop_probability: 0.1,
            duplicate_probability: 0.05,
            reorder_probability: 0.2,
            max_delay: 50,
            min_delay: 1,
        }
    }

    /// Creates a configuration for high-latency network.
    pub fn high_latency() -> Self {
        Self {
            drop_probability: 0.0,
            duplicate_probability: 0.0,
            reorder_probability: 0.1,
            max_delay: 100,
            min_delay: 20,
        }
    }

    /// Creates a configuration for lossy network.
    pub fn lossy() -> Self {
        Self {
            drop_probability: 0.2,
            duplicate_probability: 0.0,
            reorder_probability: 0.0,
            max_delay: 10,
            min_delay: 1,
        }
    }
}

// ============================================================================
// Gray Failure Configuration
// ============================================================================

/// Configuration for gray failure injection.
#[derive(Debug, Clone)]
pub struct GrayFailureConfig {
    /// Probability of a replica becoming slow (0.0 - 1.0).
    pub slow_probability: f64,
    /// How much slower a slow replica processes events (multiplier).
    pub slow_factor: u64,
    /// Probability of a replica becoming unresponsive temporarily.
    pub unresponsive_probability: f64,
    /// How long a replica stays unresponsive (in ticks).
    pub unresponsive_duration: u64,
}

impl Default for GrayFailureConfig {
    fn default() -> Self {
        Self {
            slow_probability: 0.0,
            slow_factor: 1,
            unresponsive_probability: 0.0,
            unresponsive_duration: 0,
        }
    }
}

impl GrayFailureConfig {
    /// Creates configuration for simulating slow replicas.
    pub fn slow_replicas() -> Self {
        Self {
            slow_probability: 0.2,
            slow_factor: 5,
            unresponsive_probability: 0.0,
            unresponsive_duration: 0,
        }
    }

    /// Creates configuration for intermittent failures.
    pub fn intermittent() -> Self {
        Self {
            slow_probability: 0.1,
            slow_factor: 3,
            unresponsive_probability: 0.05,
            unresponsive_duration: 20,
        }
    }
}

// ============================================================================
// VSR Simulation Harness
// ============================================================================

/// Simulation harness for VSR protocol testing.
///
/// This harness runs multiple replicas in a simulated environment with
/// controlled time, deterministic randomness, and fault injection.
pub struct VsrSimulation {
    /// Configuration.
    config: VsrSimConfig,
    /// Simulated replicas.
    replicas: HashMap<ReplicaId, ReplicaState>,
    /// Event queue (priority queue by time).
    events: std::collections::BinaryHeap<ScheduledEvent>,
    /// Current simulation time.
    time: u64,
    /// Random number generator (seeded for reproducibility).
    rng: SmallRng,
    /// Network fault configuration.
    network_faults: NetworkFaultConfig,
    /// Gray failure configuration.
    gray_failures: GrayFailureConfig,
    /// Replicas that are currently "down" (crashed).
    crashed_replicas: std::collections::HashSet<ReplicaId>,
    /// Replicas that are temporarily unresponsive.
    unresponsive_replicas: HashMap<ReplicaId, u64>,
    /// Statistics tracking.
    stats: SimStats,
    /// Committed operations (for linearizability checking).
    committed_ops: Vec<(OpNumber, Command)>,
    /// Last time each replica received a heartbeat from the leader.
    last_heartbeat_received: HashMap<ReplicaId, u64>,
    /// Sequence counter for event ordering.
    event_seq: u64,
}

/// Configuration for VSR simulation.
#[derive(Debug, Clone)]
pub struct VsrSimConfig {
    /// Number of replicas in the cluster.
    pub cluster_size: usize,
    /// Random seed for reproducibility.
    pub seed: u64,
    /// Maximum simulation time.
    pub max_time: u64,
    /// Maximum number of events to process.
    pub max_events: u64,
}

impl Default for VsrSimConfig {
    fn default() -> Self {
        Self {
            cluster_size: 3,
            seed: 0,
            max_time: 100_000,
            max_events: 100_000,
        }
    }
}

impl VsrSimConfig {
    /// Creates a new configuration with the specified seed.
    pub fn with_seed(mut self, seed: u64) -> Self {
        self.seed = seed;
        self
    }

    /// Sets the cluster size.
    pub fn with_cluster_size(mut self, size: usize) -> Self {
        self.cluster_size = size;
        self
    }

    /// Sets the maximum simulation time.
    pub fn with_max_time(mut self, max_time: u64) -> Self {
        self.max_time = max_time;
        self
    }

    /// Sets the maximum number of events.
    pub fn with_max_events(mut self, max_events: u64) -> Self {
        self.max_events = max_events;
        self
    }
}

/// Statistics from a simulation run.
#[derive(Debug, Clone, Default)]
pub struct SimStats {
    /// Total events processed.
    pub events_processed: u64,
    /// Messages sent.
    pub messages_sent: u64,
    /// Messages dropped (due to faults).
    pub messages_dropped: u64,
    /// Messages delayed.
    pub messages_delayed: u64,
    /// Operations committed.
    pub ops_committed: u64,
    /// View changes occurred.
    pub view_changes: u64,
    /// Timeouts fired.
    pub timeouts_fired: u64,
}

impl VsrSimulation {
    /// Creates a new VSR simulation.
    pub fn new(config: VsrSimConfig) -> Self {
        let rng = SmallRng::seed_from_u64(config.seed);

        // Create cluster configuration
        let replica_ids: Vec<ReplicaId> = (0..config.cluster_size)
            .map(|i| ReplicaId::new(i as u8))
            .collect();
        let cluster_config = ClusterConfig::new(replica_ids.clone())
            .with_timeouts(crate::TimeoutConfig::simulation());

        // Create replicas
        let replicas: HashMap<ReplicaId, ReplicaState> = replica_ids
            .iter()
            .map(|&id| (id, ReplicaState::new(id, cluster_config.clone())))
            .collect();

        let mut sim = Self {
            config,
            replicas,
            events: std::collections::BinaryHeap::new(),
            time: 0,
            rng,
            network_faults: NetworkFaultConfig::default(),
            gray_failures: GrayFailureConfig::default(),
            crashed_replicas: std::collections::HashSet::new(),
            unresponsive_replicas: HashMap::new(),
            stats: SimStats::default(),
            committed_ops: Vec::new(),
            last_heartbeat_received: HashMap::new(),
            event_seq: 0,
        };

        // Schedule initial heartbeat timeouts for all backups
        for &id in replica_ids.iter().skip(1) {
            sim.schedule_timeout(id, TimeoutKind::Heartbeat, HEARTBEAT_INTERVAL);
        }

        // Schedule initial leader heartbeat (to prevent backup timeouts)
        // Leader sends heartbeats at half the heartbeat interval
        sim.schedule(HEARTBEAT_INTERVAL / 2, SimEvent::Tick);

        sim
    }

    /// Sets the network fault configuration.
    pub fn with_network_faults(mut self, config: NetworkFaultConfig) -> Self {
        self.network_faults = config;
        self
    }

    /// Sets the gray failure configuration.
    pub fn with_gray_failures(mut self, config: GrayFailureConfig) -> Self {
        self.gray_failures = config;
        self
    }

    /// Schedules an event at the specified time.
    fn schedule(&mut self, time: u64, event: SimEvent) {
        let seq = self.event_seq;
        self.event_seq += 1;
        self.events.push(ScheduledEvent { time, seq, event });
    }

    /// Schedules a timeout for a replica.
    fn schedule_timeout(&mut self, replica: ReplicaId, kind: TimeoutKind, delay: u64) {
        let time = self.time + delay;
        self.schedule(time, SimEvent::Timeout { replica, kind });
    }

    /// Schedules a message delivery with network fault injection.
    fn schedule_message(&mut self, to: ReplicaId, message: Message) {
        use rand::distributions::{Distribution, Standard, Uniform};

        // Check if message should be dropped
        let drop_roll: f64 = Standard.sample(&mut self.rng);
        if drop_roll < self.network_faults.drop_probability {
            self.stats.messages_dropped += 1;
            return;
        }

        // Calculate delay
        // If reorder_probability is 0, use fixed delay to preserve message ordering
        let delay = if self.network_faults.reorder_probability == 0.0 {
            self.network_faults.min_delay
        } else {
            let delay_dist = Uniform::new_inclusive(
                self.network_faults.min_delay,
                self.network_faults.max_delay,
            );
            delay_dist.sample(&mut self.rng)
        };
        self.stats.messages_delayed += 1;

        let time = self.time + delay;
        self.schedule(
            time,
            SimEvent::MessageDelivery {
                to,
                message: message.clone(),
            },
        );

        // Check if message should be duplicated
        let dup_roll: f64 = Standard.sample(&mut self.rng);
        if dup_roll < self.network_faults.duplicate_probability {
            let extra_delay_dist = Uniform::new_inclusive(1u64, 10u64);
            let extra_delay = extra_delay_dist.sample(&mut self.rng);
            self.schedule(
                time + extra_delay,
                SimEvent::MessageDelivery { to, message },
            );
        }

        self.stats.messages_sent += 1;
    }

    /// Submits a client request to the leader.
    pub fn submit_request(&mut self, command: Command) {
        let idempotency_id = Some(IdempotencyId::generate());
        self.schedule(
            self.time + 1,
            SimEvent::ClientRequest {
                command,
                idempotency_id,
            },
        );
    }

    /// Crashes a replica (stops processing events).
    pub fn crash_replica(&mut self, id: ReplicaId) {
        self.crashed_replicas.insert(id);
    }

    /// Recovers a crashed replica.
    pub fn recover_replica(&mut self, id: ReplicaId) {
        self.crashed_replicas.remove(&id);
        // Schedule recovery timeout
        self.schedule_timeout(id, TimeoutKind::Recovery, 1);
    }

    /// Runs the simulation until completion.
    pub fn run(&mut self) -> SimResult {
        while let Some(event) = self.events.pop() {
            // Check limits
            if self.time >= self.config.max_time {
                break;
            }
            if self.stats.events_processed >= self.config.max_events {
                break;
            }

            // Advance time
            self.time = event.time;

            // Process event
            self.process_event(event.event);
            self.stats.events_processed += 1;

            // Check invariants
            if let Err(violation) = self.check_invariants() {
                return SimResult::InvariantViolation(violation);
            }
        }

        SimResult::Success(self.stats.clone())
    }

    /// Runs the simulation for a specific number of events.
    pub fn run_events(&mut self, count: u64) -> SimResult {
        for _ in 0..count {
            if let Some(event) = self.events.pop() {
                self.time = event.time;
                self.process_event(event.event);
                self.stats.events_processed += 1;

                if let Err(violation) = self.check_invariants() {
                    return SimResult::InvariantViolation(violation);
                }
            } else {
                break;
            }
        }
        SimResult::Success(self.stats.clone())
    }

    /// Processes a single event.
    fn process_event(&mut self, event: SimEvent) {
        match event {
            SimEvent::MessageDelivery { to, message } => {
                self.deliver_message(to, message);
            }
            SimEvent::Timeout { replica, kind } => {
                self.handle_timeout(replica, kind);
            }
            SimEvent::ClientRequest {
                command,
                idempotency_id,
            } => {
                self.handle_client_request(command, idempotency_id);
            }
            SimEvent::Tick => {
                self.handle_tick();
            }
        }
    }

    /// Delivers a message to a replica.
    fn deliver_message(&mut self, to: ReplicaId, message: Message) {
        // Check if replica is crashed or unresponsive
        if self.crashed_replicas.contains(&to) {
            return;
        }
        if let Some(&until) = self.unresponsive_replicas.get(&to) {
            if self.time < until {
                // Re-schedule for later
                self.schedule(until, SimEvent::MessageDelivery { to, message });
                return;
            }
            self.unresponsive_replicas.remove(&to);
        }

        // Track heartbeat reception for backups
        if matches!(
            message.payload,
            crate::message::MessagePayload::Heartbeat(_)
        ) {
            self.last_heartbeat_received.insert(to, self.time);
        }

        // Get replica state
        if let Some(state) = self.replicas.remove(&to) {
            // Process message
            let (new_state, output) = state.process(ReplicaEvent::Message(Box::new(message)));
            self.replicas.insert(to, new_state);

            // Track committed operations
            if output.committed_op.is_some() {
                self.stats.ops_committed += 1;
            }

            // Handle output
            self.handle_output(to, output);
        }
    }

    /// Handles a timeout for a replica.
    fn handle_timeout(&mut self, replica: ReplicaId, kind: TimeoutKind) {
        // Check if replica is crashed
        if self.crashed_replicas.contains(&replica) {
            return;
        }

        // For heartbeat timeouts, check if we received a heartbeat recently
        if matches!(kind, TimeoutKind::Heartbeat) {
            if let Some(&last_hb) = self.last_heartbeat_received.get(&replica) {
                // If we received a heartbeat within the heartbeat interval, don't trigger view change
                if self.time.saturating_sub(last_hb) < HEARTBEAT_INTERVAL {
                    // Just reschedule and return
                    self.schedule_timeout(replica, TimeoutKind::Heartbeat, HEARTBEAT_INTERVAL);
                    return;
                }
            }
        }

        self.stats.timeouts_fired += 1;

        // Get replica state
        if let Some(state) = self.replicas.remove(&replica) {
            let (new_state, output) = state.process(ReplicaEvent::Timeout(kind));

            // Check for view change
            if new_state.status() == ReplicaStatus::ViewChange
                && self
                    .replicas
                    .get(&replica)
                    .is_none_or(|s| s.status() != ReplicaStatus::ViewChange)
            {
                self.stats.view_changes += 1;
            }

            self.replicas.insert(replica, new_state);

            // Handle output
            self.handle_output(replica, output);
        }

        // Reschedule heartbeat timeout
        if matches!(kind, TimeoutKind::Heartbeat) {
            self.schedule_timeout(replica, TimeoutKind::Heartbeat, HEARTBEAT_INTERVAL);
        }
    }

    /// Handles a client request.
    fn handle_client_request(&mut self, command: Command, idempotency_id: Option<IdempotencyId>) {
        // Find the leader
        let leader = self
            .replicas
            .values()
            .find(|r| r.is_leader() && r.status() == ReplicaStatus::Normal)
            .map(ReplicaState::replica_id);

        if let Some(leader_id) = leader {
            if let Some(state) = self.replicas.remove(&leader_id) {
                // TODO(v0.7.0): Add client session management (client_id, request_number)
                let (new_state, output) = state.process(ReplicaEvent::ClientRequest {
                    command: command.clone(),
                    idempotency_id,
                    client_id: None,
                    request_number: None,
                });

                // Track committed op
                if let Some(op) = output.committed_op {
                    self.committed_ops.push((op, command));
                    self.stats.ops_committed += 1;
                }

                self.replicas.insert(leader_id, new_state);
                self.handle_output(leader_id, output);
            }
        } else {
            // No leader available, retry later
            self.schedule(
                self.time + VIEW_CHANGE_TIMEOUT,
                SimEvent::ClientRequest {
                    command,
                    idempotency_id,
                },
            );
        }
    }

    /// Handles a periodic tick (leader heartbeat).
    fn handle_tick(&mut self) {
        // Find the leader and have it send heartbeats
        let leader = self
            .replicas
            .values()
            .find(|r| r.is_leader() && r.status() == ReplicaStatus::Normal)
            .map(ReplicaState::replica_id);

        if let Some(leader_id) = leader {
            if let Some(state) = self.replicas.get(&leader_id) {
                // Generate heartbeat message
                if let Some(heartbeat_msg) = state.generate_heartbeat() {
                    // Broadcast to all backups
                    let replica_ids: Vec<ReplicaId> = self.replicas.keys().copied().collect();
                    for to in replica_ids {
                        if to != leader_id {
                            self.schedule_message(to, heartbeat_msg.clone());
                        }
                    }
                }
            }
        }

        // Reschedule next tick
        self.schedule(self.time + HEARTBEAT_INTERVAL / 2, SimEvent::Tick);
    }

    /// Handles output from replica processing.
    fn handle_output(&mut self, from: ReplicaId, output: ReplicaOutput) {
        // Collect replica IDs first to avoid borrow conflict
        let replica_ids: Vec<ReplicaId> = self.replicas.keys().copied().collect();

        // Schedule message deliveries
        for msg in output.messages {
            if msg.is_broadcast() {
                // Deliver to all replicas except sender
                for &to in &replica_ids {
                    if to != from {
                        self.schedule_message(to, msg.clone());
                    }
                }
            } else if let Some(to) = msg.to {
                self.schedule_message(to, msg);
            }
        }

        // Committed operations are tracked in handle_client_request
        let _ = output.committed_op;
    }

    /// Checks simulation invariants.
    fn check_invariants(&self) -> Result<(), InvariantViolation> {
        // Check 1: All replicas in the same view have consistent commit numbers
        // (committed entries must be identical)
        self.check_commit_consistency()?;

        // Check 2: Log entries with the same op number in the same view are identical
        self.check_log_consistency()?;

        // Check 3: Committed entries are never rolled back
        Self::check_no_rollback();

        Ok(())
    }

    /// Checks that committed entries are consistent across replicas.
    fn check_commit_consistency(&self) -> Result<(), InvariantViolation> {
        let normal_replicas: Vec<_> = self
            .replicas
            .values()
            .filter(|r| r.status() == ReplicaStatus::Normal)
            .collect();

        if normal_replicas.len() < 2 {
            return Ok(());
        }

        // Find the minimum commit number among all replicas
        let min_commit = normal_replicas
            .iter()
            .map(|r| r.commit_number().as_u64())
            .min()
            .unwrap_or(0);

        // All replicas should have identical logs up to min_commit
        for i in 1..=min_commit {
            let op = OpNumber::new(i);
            let entries: Vec<_> = normal_replicas
                .iter()
                .filter_map(|r| r.log_entry(op))
                .collect();

            if entries.len() > 1 {
                let first = entries[0];
                for entry in entries.iter().skip(1) {
                    if entry.checksum != first.checksum {
                        return Err(InvariantViolation::CommitInconsistency {
                            op_number: i,
                            details: format!(
                                "checksum mismatch: {:08x} vs {:08x}",
                                first.checksum, entry.checksum
                            ),
                        });
                    }
                }
            }
        }

        Ok(())
    }

    /// Checks that log entries are consistent (same op+view = same entry).
    fn check_log_consistency(&self) -> Result<(), InvariantViolation> {
        let mut entries_by_op_view: HashMap<(u64, u64), u32> = HashMap::new();

        for replica in self.replicas.values() {
            for op in 1..=replica.op_number().as_u64() {
                if let Some(entry) = replica.log_entry(OpNumber::new(op)) {
                    let key = (op, entry.view.as_u64());
                    if let Some(&existing_checksum) = entries_by_op_view.get(&key) {
                        if existing_checksum != entry.checksum {
                            return Err(InvariantViolation::LogInconsistency {
                                op_number: op,
                                view: entry.view.as_u64(),
                                details: format!(
                                    "same op/view but different checksum: {:08x} vs {:08x}",
                                    existing_checksum, entry.checksum
                                ),
                            });
                        }
                    } else {
                        entries_by_op_view.insert(key, entry.checksum);
                    }
                }
            }
        }

        Ok(())
    }

    /// Checks that committed entries are never rolled back.
    fn check_no_rollback() {
        // This is implicitly checked by commit_consistency,
        // but we could add explicit tracking here
    }

    // ========================================================================
    // Accessors for Testing
    // ========================================================================

    /// Returns the current simulation time.
    pub fn time(&self) -> u64 {
        self.time
    }

    /// Returns the statistics.
    pub fn stats(&self) -> &SimStats {
        &self.stats
    }

    /// Returns the replica states.
    pub fn replicas(&self) -> &HashMap<ReplicaId, ReplicaState> {
        &self.replicas
    }

    /// Returns a specific replica state.
    pub fn replica(&self, id: ReplicaId) -> Option<&ReplicaState> {
        self.replicas.get(&id)
    }

    /// Returns the current leader (if in normal operation).
    pub fn leader(&self) -> Option<ReplicaId> {
        self.replicas
            .values()
            .find(|r| r.is_leader() && r.status() == ReplicaStatus::Normal)
            .map(ReplicaState::replica_id)
    }

    /// Returns the highest commit number across all replicas.
    pub fn max_commit_number(&self) -> CommitNumber {
        self.replicas
            .values()
            .map(ReplicaState::commit_number)
            .max()
            .unwrap_or(CommitNumber::ZERO)
    }

    /// Checks if all normal replicas have byte-identical logs up to commit.
    pub fn check_byte_identical_logs(&self) -> bool {
        let normal_replicas: Vec<_> = self
            .replicas
            .values()
            .filter(|r| r.status() == ReplicaStatus::Normal)
            .collect();

        if normal_replicas.len() < 2 {
            return true;
        }

        let min_commit = normal_replicas
            .iter()
            .map(|r| r.commit_number().as_u64())
            .min()
            .unwrap_or(0);

        for i in 1..=min_commit {
            let op = OpNumber::new(i);
            let checksums: Vec<_> = normal_replicas
                .iter()
                .filter_map(|r| r.log_entry(op).map(|e| e.checksum))
                .collect();

            if checksums.len() > 1 && !checksums.iter().all(|&c| c == checksums[0]) {
                return false;
            }
        }

        true
    }
}

// ============================================================================
// Simulation Result
// ============================================================================

/// Result of a simulation run.
#[derive(Debug)]
pub enum SimResult {
    /// Simulation completed successfully.
    Success(SimStats),
    /// An invariant was violated.
    InvariantViolation(InvariantViolation),
}

impl SimResult {
    /// Returns true if the simulation succeeded.
    pub fn is_success(&self) -> bool {
        matches!(self, SimResult::Success(_))
    }
}

/// Types of invariant violations.
#[derive(Debug, Clone)]
pub enum InvariantViolation {
    /// Committed entries are inconsistent across replicas.
    CommitInconsistency { op_number: u64, details: String },
    /// Log entries are inconsistent.
    LogInconsistency {
        op_number: u64,
        view: u64,
        details: String,
    },
    /// A committed entry was rolled back.
    CommitRollback { op_number: u64, details: String },
}

// ============================================================================
// Test Helpers
// ============================================================================

/// Creates a test command for simulation.
pub fn test_command(name: &str) -> Command {
    Command::create_stream_with_auto_id(name.into(), DataClass::Public, Placement::Global)
}

// ============================================================================
// Tests
// ============================================================================

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

    #[test]
    fn simulation_basic_lifecycle() {
        let config = VsrSimConfig::default()
            .with_seed(42)
            .with_cluster_size(3)
            .with_max_time(10_000);

        let mut sim = VsrSimulation::new(config);

        // Submit a request
        sim.submit_request(test_command("test-stream-1"));

        // Run simulation
        let result = sim.run();
        assert!(result.is_success(), "simulation should succeed: {result:?}");

        // Verify commit happened
        let stats = sim.stats();
        assert!(
            stats.ops_committed >= 1,
            "should have committed at least 1 op"
        );
    }

    #[test]
    fn simulation_multiple_operations() {
        let config = VsrSimConfig::default()
            .with_seed(123)
            .with_cluster_size(3)
            .with_max_time(50_000);

        let mut sim = VsrSimulation::new(config);

        // Submit multiple requests
        for i in 0..10 {
            sim.submit_request(test_command(&format!("stream-{i}")));
        }

        let result = sim.run();
        assert!(result.is_success(), "simulation should succeed");

        // All operations should eventually commit
        let stats = sim.stats();
        assert!(stats.ops_committed >= 10, "should commit all operations");
    }

    #[test]
    fn simulation_with_swizzle_clogging() {
        let config = VsrSimConfig::default()
            .with_seed(456)
            .with_cluster_size(3)
            .with_max_time(100_000);

        let mut sim =
            VsrSimulation::new(config).with_network_faults(NetworkFaultConfig::swizzle_clogging());

        // Submit requests
        for i in 0..5 {
            sim.submit_request(test_command(&format!("clog-stream-{i}")));
        }

        let result = sim.run();
        assert!(
            result.is_success(),
            "should handle swizzle clogging: {result:?}"
        );

        // Verify logs are consistent
        assert!(sim.check_byte_identical_logs(), "logs should be identical");
    }

    #[test]
    fn simulation_with_lossy_network() {
        let config = VsrSimConfig::default()
            .with_seed(789)
            .with_cluster_size(3)
            .with_max_time(100_000);

        let mut sim = VsrSimulation::new(config).with_network_faults(NetworkFaultConfig::lossy());

        // Submit requests
        for i in 0..3 {
            sim.submit_request(test_command(&format!("lossy-stream-{i}")));
        }

        let result = sim.run();
        assert!(result.is_success(), "should handle lossy network");
    }

    #[test]
    fn simulation_five_node_cluster() {
        let config = VsrSimConfig::default()
            .with_seed(999)
            .with_cluster_size(5)
            .with_max_time(50_000);

        let mut sim = VsrSimulation::new(config);

        // Submit requests
        for i in 0..5 {
            sim.submit_request(test_command(&format!("five-node-{i}")));
        }

        let result = sim.run();
        assert!(result.is_success(), "5-node cluster should work");

        // Verify consistency
        assert!(sim.check_byte_identical_logs());
    }

    #[test]
    fn simulation_replica_consistency() {
        let config = VsrSimConfig::default()
            .with_seed(111)
            .with_cluster_size(3)
            .with_max_time(30_000);

        let mut sim = VsrSimulation::new(config);

        // Submit a few operations
        for i in 0..3 {
            sim.submit_request(test_command(&format!("consistency-{i}")));
        }

        let result = sim.run();
        assert!(result.is_success());

        // Check byte-identical logs
        assert!(
            sim.check_byte_identical_logs(),
            "committed logs must be byte-identical"
        );

        // Verify all replicas agree on commit number
        let replicas = sim.replicas();
        let normal_replicas: Vec<_> = replicas
            .values()
            .filter(|r| r.status() == ReplicaStatus::Normal)
            .collect();

        if normal_replicas.len() >= 2 {
            let first_commit = normal_replicas[0].commit_number();
            // All should have same commit (eventually)
            for r in &normal_replicas {
                // Allow some difference during execution
                let diff = first_commit.as_u64().abs_diff(r.commit_number().as_u64());
                assert!(diff <= 1, "commit numbers should be close");
            }
        }
    }

    #[test]
    fn simulation_extended_run() {
        // Extended run with many operations and light network faults.
        // Reordering is kept low to avoid gaps that require the full
        // write reorder repair integration into the prepare handler.
        let config = VsrSimConfig::default()
            .with_seed(222)
            .with_cluster_size(3)
            .with_max_time(200_000)
            .with_max_events(50_000);

        let mut sim = VsrSimulation::new(config).with_network_faults(NetworkFaultConfig {
            drop_probability: 0.01, // Light drops
            duplicate_probability: 0.02,
            reorder_probability: 0.0, // No reordering (prepare handler integration pending)
            max_delay: 20,
            min_delay: 1,
        });

        // Submit many operations
        for i in 0..50 {
            sim.submit_request(test_command(&format!("extended-{i}")));
        }

        let result = sim.run();
        assert!(
            result.is_success(),
            "extended run should succeed: {result:?}"
        );

        let stats = sim.stats();
        println!(
            "Extended run: {} events, {} ops committed, {} messages sent, {} dropped",
            stats.events_processed,
            stats.ops_committed,
            stats.messages_sent,
            stats.messages_dropped
        );

        // Should have committed most operations
        assert!(
            stats.ops_committed >= 40,
            "should commit most operations in extended run"
        );

        // Logs must be consistent
        assert!(sim.check_byte_identical_logs());
    }

    #[test]
    fn simulation_high_latency_network() {
        let config = VsrSimConfig::default()
            .with_seed(333)
            .with_cluster_size(3)
            .with_max_time(200_000);

        let mut sim =
            VsrSimulation::new(config).with_network_faults(NetworkFaultConfig::high_latency());

        for i in 0..5 {
            sim.submit_request(test_command(&format!("latency-{i}")));
        }

        let result = sim.run();
        assert!(result.is_success(), "should handle high latency");
    }

    #[test]
    fn simulation_deterministic_with_same_seed() {
        // Two runs with the same seed should produce identical results
        let run_simulation = |seed: u64| {
            let config = VsrSimConfig::default()
                .with_seed(seed)
                .with_cluster_size(3)
                .with_max_time(10_000);

            let mut sim = VsrSimulation::new(config);
            sim.submit_request(test_command("deterministic-test"));
            sim.run();
            sim.stats().clone()
        };

        let stats1 = run_simulation(444);
        let stats2 = run_simulation(444);

        assert_eq!(stats1.events_processed, stats2.events_processed);
        assert_eq!(stats1.messages_sent, stats2.messages_sent);
        assert_eq!(stats1.ops_committed, stats2.ops_committed);
    }

    #[test]
    fn simulation_different_seeds_different_results() {
        let run_simulation = |seed: u64| {
            let config = VsrSimConfig::default()
                .with_seed(seed)
                .with_cluster_size(3)
                .with_max_time(20_000);

            let mut sim = VsrSimulation::new(config)
                .with_network_faults(NetworkFaultConfig::swizzle_clogging());

            for i in 0..5 {
                sim.submit_request(test_command(&format!("seed-test-{i}")));
            }
            sim.run();
            sim.stats().clone()
        };

        let stats1 = run_simulation(555);
        let stats2 = run_simulation(666);

        // With different seeds and network faults, results should differ
        // (though both should succeed)
        let different = stats1.messages_dropped != stats2.messages_dropped
            || stats1.messages_delayed != stats2.messages_delayed;

        assert!(
            different,
            "different seeds should produce different fault patterns"
        );
    }

    #[test]
    fn simulation_with_write_reordering() {
        // Validate that the simulation completes without invariant violations
        // when write reordering is enabled. With reordering, some operations
        // may not commit (out-of-order prepares create gaps), but the protocol
        // must never violate safety invariants (commit consistency, log consistency).
        //
        // The WriteReorderGapRequest/Response protocol infrastructure is tested
        // separately. Full integration into the prepare handler will increase
        // commit throughput under reordering.
        let config = VsrSimConfig::default()
            .with_seed(777)
            .with_cluster_size(3)
            .with_max_time(100_000)
            .with_max_events(30_000);

        let mut sim = VsrSimulation::new(config).with_network_faults(NetworkFaultConfig {
            drop_probability: 0.01,
            duplicate_probability: 0.01,
            reorder_probability: 0.05,
            max_delay: 30,
            min_delay: 1,
        });

        // Submit a batch of operations
        for i in 0..20 {
            sim.submit_request(test_command(&format!("reorder-{i}")));
        }

        let result = sim.run();
        assert!(
            result.is_success(),
            "simulation with write reordering must not violate safety invariants: {result:?}"
        );

        // Committed logs must be consistent despite reordering.
        // Some operations may not commit due to reorder gaps, but whatever
        // does commit must be byte-identical across replicas.
        assert!(
            sim.check_byte_identical_logs(),
            "committed logs must be byte-identical even with write reordering"
        );

        let stats = sim.stats();
        // At least some operations should commit (the first batch before
        // reordering kicks in)
        assert!(
            stats.ops_committed >= 1,
            "at least one operation should commit (committed: {})",
            stats.ops_committed
        );
    }
}
