//! Standby replica mode for disaster recovery and read scaling.
//!
//! This module implements standby replicas (Phase 4.3) - read-only followers
//! that receive log updates from active replicas but do NOT participate in
//! quorum decisions.
//!
//! # Architecture
//!
//! ```text
//! Active Cluster (3 replicas)          Standby Replicas
//! ┌────────────────────────┐          ┌──────────────────┐
//! │  Primary   Backup₁  Backup₂  │ ────► Standby₁ (DR)   │
//! │     └── Commit ──┘         │ ────► Standby₂ (Reads) │
//! └────────────────────────┘          └──────────────────┘
//!        Quorum = 2/3                 Not counted in quorum
//! ```
//!
//! # Use Cases
//!
//! 1. **Disaster Recovery**: Geographic redundancy without affecting quorum
//! 2. **Read Scaling**: Offload read queries from active replicas
//! 3. **Testing/Staging**: Mirror production data safely
//!
//! # Safety Properties
//!
//! - Standby replicas NEVER participate in quorum (enforced by Kani Proof #68)
//! - Promotion preserves log consistency (enforced by Kani Proof #69)
//! - Standby reads may lag behind committed operations (eventually consistent)
//!
//! # Promotion
//!
//! Standby replicas can be promoted to active replicas:
//! - **Automatic**: On active replica failure (if standby is up-to-date)
//! - **Manual**: Operator-initiated via `ReconfigCommand`
//!
//! Promotion requires:
//! 1. Standby log must be ⊆ active primary log (no divergence)
//! 2. Cluster reconfiguration (joint consensus)
//! 3. Quorum agreement from active replicas

use kimberlite_kernel::apply_committed;

use crate::config::ClusterConfig;
use crate::message::{Commit, Heartbeat, MessagePayload, Prepare};
use crate::types::{CommitNumber, OpNumber, ReplicaId, ReplicaStatus};

use super::{ReplicaOutput, ReplicaState, msg_broadcast};

// ============================================================================
// Standby State
// ============================================================================

/// State tracked for standby replicas.
///
/// Standby replicas maintain a simplified version of active replica state:
/// - They follow the log but don't participate in quorum
/// - They track `commit_number` to know which operations are committed
/// - They can be promoted to active status when needed
#[derive(Debug, Clone)]
pub struct StandbyState {
    /// Last operation number observed from active replicas.
    pub(crate) last_op_observed: OpNumber,

    /// Last commit number observed from active replicas.
    pub(crate) last_commit_observed: CommitNumber,

    /// Whether this standby is eligible for promotion.
    ///
    /// Promotion requires:
    /// - Log is up-to-date with active primary
    /// - No log divergence (all entries match primary)
    pub(crate) promotion_eligible: bool,
}

impl StandbyState {
    /// Creates a new standby state.
    pub fn new() -> Self {
        Self {
            last_op_observed: OpNumber::ZERO,
            last_commit_observed: CommitNumber::ZERO,
            promotion_eligible: true,
        }
    }

    /// Updates standby state based on observed operation.
    pub fn observe_operation(&mut self, op: OpNumber) {
        if op > self.last_op_observed {
            self.last_op_observed = op;
        }
    }

    /// Updates standby state based on observed commit.
    pub fn observe_commit(&mut self, commit: CommitNumber) {
        if commit > self.last_commit_observed {
            self.last_commit_observed = commit;
        }
    }

    /// Marks standby as ineligible for promotion (log divergence detected).
    pub fn mark_diverged(&mut self) {
        self.promotion_eligible = false;
    }
}

impl Default for StandbyState {
    fn default() -> Self {
        Self::new()
    }
}

// ============================================================================
// Standby Message Processing
// ============================================================================

#[allow(dead_code)] // Public API for standby replica integration (wired in server event loop)
impl ReplicaState {
    /// Processes a Prepare message as a standby replica.
    ///
    /// Standby replicas:
    /// - Append the log entry (if not already present)
    /// - Do NOT send `PrepareOK` (don't participate in quorum)
    /// - Track observed operation number
    ///
    /// Safety: Standby replicas NEVER participate in quorum.
    pub(crate) fn on_prepare_standby(&mut self, prepare: &Prepare) -> ReplicaOutput {
        // Standby replicas only accept Prepare messages
        assert!(
            self.status.is_standby(),
            "on_prepare_standby called on non-standby replica"
        );

        // Check if we already have this entry
        if prepare.op_number <= self.op_number {
            // Already have this operation, ignore
            return ReplicaOutput::empty();
        }

        // Append to log if consecutive
        if prepare.op_number == self.op_number.next() {
            // Use the entry from the Prepare message directly
            self.log.push(prepare.entry.clone());
            self.op_number = prepare.op_number;

            // Track observed operation (for promotion eligibility)
            if let Some(standby) = self.standby_state.as_mut() {
                standby.observe_operation(prepare.op_number);
            }
        } else {
            // Gap detected - mark as diverged, need repair
            if let Some(standby) = self.standby_state.as_mut() {
                standby.mark_diverged();
            }
        }

        // Standby replicas do NOT send PrepareOK
        ReplicaOutput::empty()
    }

    /// Processes a Commit message as a standby replica.
    ///
    /// Standby replicas:
    /// - Update `commit_number` based on active cluster
    /// - Apply committed operations to kernel state
    /// - Track observed commit number
    pub(crate) fn on_commit_standby(&mut self, commit: Commit) -> ReplicaOutput {
        assert!(
            self.status.is_standby(),
            "on_commit_standby called on non-standby replica"
        );

        // Update commit number if higher
        if commit.commit_number > self.commit_number {
            let old_commit = self.commit_number;

            // Track observed commit
            if let Some(standby) = self.standby_state.as_mut() {
                standby.observe_commit(commit.commit_number);
            }

            // Apply committed operations to kernel state (standby is read-only,
            // so we apply for state consistency but discard effects).
            let mut next_op = old_commit.as_op_number().next();
            while CommitNumber::new(next_op) <= commit.commit_number {
                if let Some(entry) = self.log_entry(next_op).cloned() {
                    match apply_committed(self.kernel_state.clone(), entry.command) {
                        Ok((new_state, _effects)) => {
                            self.kernel_state = new_state;
                            self.commit_number = CommitNumber::new(next_op);
                        }
                        Err(_) => {
                            // Standby cannot fix kernel errors — mark diverged
                            if let Some(standby) = self.standby_state.as_mut() {
                                standby.mark_diverged();
                            }
                            break;
                        }
                    }
                } else {
                    // Missing log entry — gap in the log, can't apply further
                    break;
                }
                next_op = next_op.next();
            }
        }

        ReplicaOutput::empty()
    }

    /// Promotes a standby replica to active status.
    ///
    /// Promotion process:
    /// 1. Verify standby is eligible (log up-to-date, no divergence)
    /// 2. Transition from Standby → Normal status
    /// 3. Participate in cluster reconfiguration (add self to active config)
    /// 4. Begin participating in quorum after reconfiguration complete
    ///
    /// Safety: Promotion preserves log consistency (Kani Proof #69).
    pub fn promote_to_active(&mut self, new_config: ClusterConfig) -> ReplicaOutput {
        assert!(
            self.status.is_standby(),
            "promote_to_active called on non-standby replica"
        );

        // Verify standby is eligible for promotion
        let standby = self.standby_state.as_ref().expect("standby state");
        assert!(
            standby.promotion_eligible,
            "standby not eligible for promotion (log diverged)"
        );

        // Verify new config includes this replica as active
        assert!(
            new_config.contains(self.replica_id),
            "promoted replica must be in new config"
        );

        // Transition to Normal status
        self.status = ReplicaStatus::Normal;
        self.config = new_config;
        self.standby_state = None;

        // Announce promotion to cluster via Heartbeat
        let announce = MessagePayload::Heartbeat(Heartbeat {
            view: self.view,
            commit_number: self.commit_number,
            monotonic_timestamp: 0, // Placeholder - real runtime provides actual timestamp
            wall_clock_timestamp: 0, // Placeholder - real runtime provides actual timestamp
            version: self.upgrade_state.self_version,
        });

        ReplicaOutput::with_messages(vec![
            self.sign_message(msg_broadcast(self.replica_id, announce)),
        ])
    }

    /// Checks if this replica is a standby.
    pub fn is_standby(&self) -> bool {
        self.status.is_standby()
    }

    /// Returns standby state if this is a standby replica.
    pub fn standby_state(&self) -> Option<&StandbyState> {
        self.standby_state.as_ref()
    }
}

// ============================================================================
// Kani Proofs (Phase 4.3)
// ============================================================================

#[cfg(kani)]
mod kani_proofs {
    use super::*;

    /// Proof #68: Standby replicas never participate in quorum
    ///
    /// Property: Standby replicas NEVER send PrepareOK messages.
    ///
    /// This is critical for safety - standby replicas must not affect
    /// quorum decisions. If they did, the cluster could commit operations
    /// without a true quorum of active replicas.
    ///
    /// Verification:
    /// - Process arbitrary Prepare message on standby replica
    /// - Verify output contains NO messages (especially no PrepareOK)
    #[kani::proof]
    #[kani::unwind(3)]
    fn proof_standby_never_participates_in_quorum() {
        use crate::types::{CommitNumber, LogEntry, ViewNumber};
        use kimberlite_kernel::Command;
        use kimberlite_types::{DataClass, Placement, StreamId, StreamName};

        let replica_id = kani::any();
        kani::assume(replica_id < 10); // Bounded replica count

        let mut state = ReplicaState::new_standby(ReplicaId::new(replica_id));

        // Verify status is standby
        assert!(state.status.is_standby());

        // Create log entry with CreateStream command
        let view_raw: u64 = kani::any();
        let op_raw: u64 = kani::any();
        let commit_raw: u64 = kani::any();
        kani::assume(view_raw < 100);
        kani::assume(op_raw < 100);
        kani::assume(commit_raw < op_raw);

        let command = Command::CreateStream {
            stream_id: StreamId::new(1),
            stream_name: StreamName::from("test".to_string()),
            data_class: DataClass::PHI,
            placement: Placement::Global,
        };

        let entry = LogEntry::new(
            OpNumber::new(op_raw),
            ViewNumber::new(view_raw),
            command,
            None, // idempotency_id
            None, // client_id
            None, // request_number
        );

        // Create arbitrary Prepare message
        let prepare = Prepare {
            view: ViewNumber::new(view_raw),
            op_number: OpNumber::new(op_raw),
            entry,
            commit_number: CommitNumber::new(OpNumber::new(commit_raw)),
            reconfig: None,
        };

        // Process as standby
        let output = state.on_prepare_standby(&prepare);

        // CRITICAL PROPERTY: Standby replicas never send messages
        // (especially never send PrepareOK)
        assert!(
            output.messages.is_empty(),
            "standby must not send any messages (no PrepareOK)"
        );
    }

    /// Proof #69: Promotion preserves log consistency
    ///
    /// Property: Promoting a standby to active preserves log consistency.
    ///
    /// Requirements for safe promotion:
    /// 1. Standby log must be ⊆ active primary log (no divergence)
    /// 2. Standby must be marked promotion_eligible
    /// 3. New config must include promoted replica
    ///
    /// Verification:
    /// - Create standby with eligible state
    /// - Promote to active
    /// - Verify status changed and config updated
    #[kani::proof]
    #[kani::unwind(3)]
    fn proof_promotion_preserves_log_consistency() {
        use crate::config::ClusterConfig;

        let replica_id_raw: u8 = kani::any();
        kani::assume(replica_id_raw < 10);

        let replica_id = ReplicaId::new(replica_id_raw);
        let mut state = ReplicaState::new_standby(replica_id);

        // Verify standby starts eligible
        let standby = state.standby_state.as_ref().unwrap();
        assert!(standby.promotion_eligible);

        // Create new config including this replica
        let mut replicas = vec![replica_id];
        for i in 0..2u8 {
            let other_id = ReplicaId::new(i);
            if other_id != replica_id {
                replicas.push(other_id);
            }
        }
        let new_config = ClusterConfig::new(replicas);

        // Promote to active
        let _output = state.promote_to_active(new_config.clone());

        // Verify promotion succeeded
        assert!(!state.status.is_standby());
        assert!(state.status == ReplicaStatus::Normal);
        assert!(state.standby_state.is_none());
        assert!(state.config.contains(replica_id));
    }
}

// ============================================================================
// Helper for creating standby replicas
// ============================================================================

impl ReplicaState {
    /// Creates a new standby replica.
    ///
    /// Standby replicas start with:
    /// - Status: Standby
    /// - Empty log
    /// - `StandbyState` tracking
    pub fn new_standby(replica_id: ReplicaId) -> Self {
        // Create minimal config (standby not part of active cluster)
        let config = ClusterConfig::new(vec![]);

        let mut state = Self::new(replica_id, config);
        state.status = ReplicaStatus::Standby;
        state.standby_state = Some(StandbyState::new());

        state
    }
}
