feltdb 0.6.0

/// Crash-Atomic Boundary Testing (Gate 3)
///
/// Verifies the critical invariant: State mutation and OperationId persistence
/// are crash-atomic with respect to recovery.
///
/// The dangerous boundary:
///
///   apply state mutation
///           ↓
///       CRASH 💥
///           ↓
///   persist OperationId to dedup set
///
/// After recovery, MUST have either:
///   - state reflects operation AND OperationId in dedup set, OR
///   - state doesn't reflect operation AND OperationId not in dedup set
///
/// Never:
///   - state reflects op but dedup missing (double-apply on retry)
///   - dedup has op but state wasn't mutated (phantom operation)
use crate::state_hash::CanonicalState;
use std::collections::HashSet;

/// Simulates crash points in state/dedup boundary
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CrashBoundaryPoint {
    /// Crash before state mutation begins
    BeforeStateMutation,
    /// Crash after state mutation, before dedup persistence
    AfterStateMutationBeforeDedupPersist,
    /// Crash after dedup persistence complete
    AfterDedupPersist,
}

/// Simulates operation application with recovery points
#[derive(Clone, Debug)]
pub struct CrashAtomicExecutor {
    state: CanonicalState,
    dedup_set: HashSet<String>, // OperationId set
}

impl CrashAtomicExecutor {
    pub fn new() -> Self {
        Self {
            state: CanonicalState::new(),
            dedup_set: HashSet::new(),
        }
    }

    /// Apply operation with potential crash at specified point
    pub fn apply_with_crash(
        &mut self,
        operation_id: &str,
        record_id: &str,
        value: i32,
    ) -> Result<(), String> {
        // Check for duplicate before applying
        if self.dedup_set.contains(operation_id) {
            return Err(format!(
                "Operation {} already applied (dedup hit)",
                operation_id
            ));
        }

        // Simulate state mutation
        self.mutate_state(record_id, value);

        // Simulate dedup persistence
        self.dedup_set.insert(operation_id.to_string());

        Ok(())
    }

    /// Apply operation and simulate crash at specified point
    /// Returns the state at the time of crash (for recovery testing)
    pub fn apply_with_simulated_crash(
        &mut self,
        operation_id: &str,
        record_id: &str,
        value: i32,
        crash_point: CrashBoundaryPoint,
    ) -> CrashRecoveryState {
        match crash_point {
            CrashBoundaryPoint::BeforeStateMutation => {
                // Crash before anything happens
                CrashRecoveryState {
                    state_mutated: false,
                    dedup_persisted: false,
                    operation_id: operation_id.to_string(),
                }
            }
            CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist => {
                // Mutate state but crash before dedup
                self.mutate_state(record_id, value);
                CrashRecoveryState {
                    state_mutated: true,
                    dedup_persisted: false,
                    operation_id: operation_id.to_string(),
                }
            }
            CrashBoundaryPoint::AfterDedupPersist => {
                // Mutate state and persist dedup
                self.mutate_state(record_id, value);
                self.dedup_set.insert(operation_id.to_string());
                CrashRecoveryState {
                    state_mutated: true,
                    dedup_persisted: true,
                    operation_id: operation_id.to_string(),
                }
            }
        }
    }

    /// Recover from crash state
    /// Returns whether the operation should be reapplied
    pub fn recover(&self, crash_state: &CrashRecoveryState) -> RecoveryDecision {
        let in_dedup = self.dedup_set.contains(&crash_state.operation_id);

        match (
            crash_state.state_mutated,
            crash_state.dedup_persisted,
            in_dedup,
        ) {
            // State mutated, dedup persisted → fully applied, don't retry
            (true, true, true) => RecoveryDecision::AlreadyApplied,

            // State mutated, dedup not persisted, not in dedup set → partially applied, needs recovery
            (true, false, false) => RecoveryDecision::PartiallyApplied {
                reason: "State mutated but dedup lost in crash".to_string(),
            },

            // State mutated, dedup not persisted, but in dedup set → inconsistent
            (true, false, true) => RecoveryDecision::InconsistentState {
                reason: "State mutated but dedup signal false, yet in dedup set".to_string(),
            },

            // State not mutated, dedup persisted → phantom state, needs investigation
            (false, true, true) => RecoveryDecision::PhantomOperation {
                reason: "Dedup persisted but state mutation didn't complete".to_string(),
            },

            // Nothing persisted → safe to retry
            (false, false, false) => RecoveryDecision::SafeToRetry,

            // Inconsistent states
            (true, true, false) => RecoveryDecision::InconsistentState {
                reason: "State shows applied but dedup missing".to_string(),
            },
            (false, false, true) => RecoveryDecision::InconsistentState {
                reason: "Dedup has operation but state not applied".to_string(),
            },
            (false, true, false) => RecoveryDecision::InconsistentState {
                reason: "Dedup persisted signal true but not in set".to_string(),
            },
        }
    }

    /// Verify the crash-atomic invariant
    /// After recovery, either (state AND dedup) or (neither)
    pub fn verify_invariant(&self, crash_state: &CrashRecoveryState) -> InvariantCheckResult {
        let in_dedup = self.dedup_set.contains(&crash_state.operation_id);

        // Both true or both false → invariant satisfied
        if crash_state.state_mutated == in_dedup {
            InvariantCheckResult::Satisfied
        } else {
            InvariantCheckResult::Violated {
                state_mutated: crash_state.state_mutated,
                dedup_persisted: in_dedup,
            }
        }
    }

    fn mutate_state(&mut self, record_id: &str, value: i32) {
        self.state.set_record(
            "test_collection".to_string(),
            record_id.to_string(),
            serde_json::json!({ "value": value }),
        );
    }

    pub fn state(&self) -> &CanonicalState {
        &self.state
    }

    pub fn dedup_set(&self) -> &HashSet<String> {
        &self.dedup_set
    }
}

/// State at time of crash (what we recover from)
#[derive(Clone, Debug)]
pub struct CrashRecoveryState {
    pub state_mutated: bool,
    pub dedup_persisted: bool,
    pub operation_id: String,
}

/// Decision about what to do after recovery
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RecoveryDecision {
    /// Operation fully applied, don't retry
    AlreadyApplied,
    /// Operation partially applied, needs recovery action
    PartiallyApplied { reason: String },
    /// Operation marked as applied but state missing (data corruption)
    PhantomOperation { reason: String },
    /// Safe to retry the operation
    SafeToRetry,
    /// Inconsistent state detected
    InconsistentState { reason: String },
}

/// Result of verifying the crash-atomic invariant
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum InvariantCheckResult {
    /// Invariant satisfied: state and dedup agree (both true or both false)
    Satisfied,
    /// Invariant violated: state and dedup disagree
    Violated {
        state_mutated: bool,
        dedup_persisted: bool,
    },
}

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

    #[test]
    fn test_crash_before_state_mutation() {
        let mut executor = CrashAtomicExecutor::new();
        let crash_state = executor.apply_with_simulated_crash(
            "op1",
            "rec1",
            10,
            CrashBoundaryPoint::BeforeStateMutation,
        );

        assert!(!crash_state.state_mutated);
        assert!(!crash_state.dedup_persisted);

        let decision = executor.recover(&crash_state);
        assert_eq!(decision, RecoveryDecision::SafeToRetry);

        // Invariant must be satisfied
        assert_eq!(
            executor.verify_invariant(&crash_state),
            InvariantCheckResult::Satisfied
        );
    }

    #[test]
    fn test_crash_after_state_mutation_before_dedup() {
        let mut executor = CrashAtomicExecutor::new();
        let crash_state = executor.apply_with_simulated_crash(
            "op1",
            "rec1",
            10,
            CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
        );

        assert!(crash_state.state_mutated);
        assert!(!crash_state.dedup_persisted);

        // State was mutated, but dedup wasn't persisted
        // This violates the invariant!
        let invariant_result = executor.verify_invariant(&crash_state);
        assert!(matches!(
            invariant_result,
            InvariantCheckResult::Violated { .. }
        ));

        let decision = executor.recover(&crash_state);
        assert!(matches!(
            decision,
            RecoveryDecision::PartiallyApplied { .. }
        ));
    }

    #[test]
    fn test_crash_after_dedup_persist() {
        let mut executor = CrashAtomicExecutor::new();
        let crash_state = executor.apply_with_simulated_crash(
            "op1",
            "rec1",
            10,
            CrashBoundaryPoint::AfterDedupPersist,
        );

        assert!(crash_state.state_mutated);
        assert!(crash_state.dedup_persisted);

        let decision = executor.recover(&crash_state);
        assert_eq!(decision, RecoveryDecision::AlreadyApplied);

        // Invariant satisfied
        assert_eq!(
            executor.verify_invariant(&crash_state),
            InvariantCheckResult::Satisfied
        );
    }

    #[test]
    fn test_duplicate_prevention_after_recovery() {
        let mut executor = CrashAtomicExecutor::new();

        // First application with crash after dedup
        let crash_state = executor.apply_with_simulated_crash(
            "op1",
            "rec1",
            10,
            CrashBoundaryPoint::AfterDedupPersist,
        );

        // After recovery, operation is in dedup
        assert!(executor.dedup_set().contains("op1"));

        // Try to apply the same operation again
        let result = executor.apply_with_crash("op1", "rec1", 20);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("already applied"));
    }

    #[test]
    fn test_exactly_once_under_all_crash_points() {
        let crash_points = vec![
            CrashBoundaryPoint::BeforeStateMutation,
            CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
            CrashBoundaryPoint::AfterDedupPersist,
        ];

        for crash_point in crash_points {
            let mut executor = CrashAtomicExecutor::new();
            let crash_state =
                executor.apply_with_simulated_crash("op1", "rec1", 10, crash_point.clone());

            let decision = executor.recover(&crash_state);

            // The decision should allow recovery without double-apply
            match decision {
                RecoveryDecision::AlreadyApplied => {
                    // Safe: already fully applied
                }
                RecoveryDecision::SafeToRetry => {
                    // Safe: not yet applied, can retry
                }
                RecoveryDecision::PartiallyApplied { .. } => {
                    // This is the dangerous case - but we detected it
                }
                _ => panic!("Unexpected recovery decision: {:?}", decision),
            }
        }
    }

    #[test]
    fn test_invariant_satisfaction_across_crash_points() {
        let crash_points = vec![
            CrashBoundaryPoint::BeforeStateMutation,
            CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
            CrashBoundaryPoint::AfterDedupPersist,
        ];

        for crash_point in crash_points {
            let mut executor = CrashAtomicExecutor::new();
            let crash_state =
                executor.apply_with_simulated_crash("op1", "rec1", 10, crash_point.clone());

            // After crash, verify what we find
            let in_dedup = executor.dedup_set().contains("op1");

            // For consistency: state_mutated should equal in_dedup
            // (they're either both true or both false, never mixed)
            if crash_point == CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist {
                // This is the violation case: state true, dedup false
                assert_ne!(crash_state.state_mutated, in_dedup);
            } else {
                // All other cases should agree
                assert_eq!(crash_state.state_mutated, in_dedup);
            }
        }
    }

    #[test]
    fn test_recovery_state_tracking() {
        let mut executor = CrashAtomicExecutor::new();

        // Apply operation normally
        let _ = executor.apply_with_crash("op1", "rec1", 10);
        assert!(executor.dedup_set().contains("op1"));

        // Simulate crash recovery state
        let crash_state = CrashRecoveryState {
            state_mutated: true,
            dedup_persisted: true,
            operation_id: "op1".to_string(),
        };

        let decision = executor.recover(&crash_state);
        assert_eq!(decision, RecoveryDecision::AlreadyApplied);
    }

    #[test]
    fn test_multiple_operations_crash_atomic() {
        let mut executor = CrashAtomicExecutor::new();

        // Apply first operation fully
        let _ = executor.apply_with_crash("op1", "rec1", 10);
        assert_eq!(executor.dedup_set().len(), 1);

        // Apply second operation with crash after state mutation
        let crash_state = executor.apply_with_simulated_crash(
            "op2",
            "rec2",
            20,
            CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
        );

        // op1 is still in dedup, op2 is not
        assert!(executor.dedup_set().contains("op1"));
        assert!(!executor.dedup_set().contains("op2"));

        // op2's crash state shows state mutated but not persisted to dedup
        let invariant = executor.verify_invariant(&crash_state);
        assert!(matches!(invariant, InvariantCheckResult::Violated { .. }));
    }
}