//! Background log scrubber for proactive corruption detection.
//!
//! This module implements background scrubbing of the replicated log to detect
//! latent sector errors and silent corruption before they cause double-fault data loss.
//!
//! # Design
//!
//! The scrubber tours the entire log periodically, validating checksums on every entry.
//! Key features:
//!
//! - **Tour-based**: Scrubs entire log from start to end, then begins new tour
//! - **PRNG-based origin**: Randomized start position prevents thundering herd
//! - **Rate-limited**: Reserves IOPS budget for production traffic
//! - **Proactive**: Detects corruption before it's accessed by reads
//!
//! # Background
//!
//! Google's 2007 study found that >60% of latent sector errors are discovered
//! by scrubbers, not active reads. Background scrubbing is critical for:
//!
//! - Detecting silent corruption before it causes data loss
//! - Finding errors while replicas are healthy (before double-fault)
//! - Triggering repair proactively rather than reactively
//!
//! # Inspiration
//!
//! Based on `TigerBeetle`'s `grid_scrubber.zig` implementation.

use crate::types::{LogEntry, OpNumber, ReplicaId};
use rand::{Rng as RandRng, SeedableRng};
use rand_chacha::ChaCha8Rng;

// ============================================================================
// Configuration Constants
// ============================================================================

/// Maximum reads per tick (IOPS budget).
///
/// This reserves ~90% of IOPS for production traffic, using ~10% for scrubbing.
/// `TigerBeetle` uses similar reservation to prevent scrubbing from impacting latency.
const MAX_SCRUB_READS_PER_TICK: usize = 10;

// ============================================================================
// Scrub Result
// ============================================================================

/// Result of a scrub operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScrubResult {
    /// Entry scrubbed successfully, checksum valid.
    Ok,

    /// Corruption detected (checksum mismatch).
    Corruption,

    /// Tour complete, no more entries to scrub.
    TourComplete,

    /// Budget exhausted, cannot scrub now.
    BudgetExhausted,
}

// ============================================================================
// Log Scrubber
// ============================================================================

/// Background log scrubber for proactive corruption detection.
///
/// The scrubber tours the entire log, validating checksums on every entry.
/// Tours start at randomized origins to prevent synchronized load spikes.
#[derive(Debug, Clone)]
pub struct LogScrubber {
    /// Replica ID (for origin randomization).
    replica_id: ReplicaId,

    /// Current position in the tour.
    current_position: OpNumber,

    /// Where this tour started (PRNG-based origin).
    tour_start: OpNumber,

    /// Where this tour will end (log head when tour started).
    tour_end: OpNumber,

    /// Number of completed tours.
    tour_count: u64,

    /// IOPS budget for rate limiting.
    scrub_budget: ScrubBudget,

    /// Detected corruptions (`op_number`, tour when detected).
    corruptions: Vec<(OpNumber, u64)>,
}

impl LogScrubber {
    /// Creates a new log scrubber.
    ///
    /// The scrubber starts at a randomized origin to prevent all replicas
    /// from scrubbing the same region simultaneously (thundering herd).
    pub fn new(replica_id: ReplicaId, log_head: OpNumber) -> Self {
        let tour_count = 0;
        let origin = Self::randomize_origin(replica_id, tour_count, log_head);

        Self {
            replica_id,
            current_position: origin,
            tour_start: origin,
            tour_end: log_head,
            tour_count,
            scrub_budget: ScrubBudget::new(MAX_SCRUB_READS_PER_TICK),
            corruptions: Vec::new(),
        }
    }

    /// Randomizes the tour origin using PRNG.
    ///
    /// Uses `ChaCha8Rng` seeded with `replica_id` + `tour_count` for determinism.
    /// Same seed always produces same origin (important for reproducibility).
    ///
    /// This prevents thundering herd: if all replicas started at op 0,
    /// they would all scrub the same region simultaneously, causing load spikes.
    fn randomize_origin(replica_id: ReplicaId, tour_count: u64, log_head: OpNumber) -> OpNumber {
        if log_head.as_u64() == 0 {
            return OpNumber::new(0);
        }

        // Seed with replica_id and tour_count for deterministic randomness
        let seed = u64::from(replica_id.as_u8()) << 32 | tour_count;
        let mut rng = ChaCha8Rng::seed_from_u64(seed);

        // Generate random offset within log range
        let offset = RandRng::r#gen::<u64>(&mut rng) % (log_head.as_u64() + 1);

        OpNumber::new(offset)
    }

    /// Returns the next operation number to scrub.
    ///
    /// Tours wrap around: if we started at op 50 and the log ends at op 100,
    /// we scrub [50..100], then [0..50).
    pub fn next_op_to_scrub(&self) -> Option<OpNumber> {
        if self.is_tour_complete() {
            return None;
        }

        Some(self.current_position)
    }

    /// Checks if the current tour is complete.
    pub fn is_tour_complete(&self) -> bool {
        // Tour complete if we've wrapped back to start
        if self.tour_start.as_u64() == 0 {
            // Simple case: started at 0, tour ends at log head
            self.current_position > self.tour_end
        } else {
            // Wrapped case: check if we've returned to origin
            self.current_position >= self.tour_start && self.current_position > self.tour_end
        }
    }

    /// Advances to the next position.
    pub(crate) fn advance(&mut self) {
        self.current_position = OpNumber::new(self.current_position.as_u64() + 1);

        // PRODUCTION ASSERTION: Tour progress bounds
        // Ensures tour position never exceeds log head (prevents infinite loops)
        assert!(
            self.current_position.as_u64() <= self.tour_end.as_u64() + 1,
            "tour position {} exceeded tour end {} + 1",
            self.current_position.as_u64(),
            self.tour_end.as_u64()
        );
    }

    /// Starts a new tour.
    ///
    /// Called when the current tour completes. Randomizes the origin for
    /// the new tour and resets position.
    pub fn start_new_tour(&mut self, new_log_head: OpNumber) {
        self.tour_count += 1;
        let origin = Self::randomize_origin(self.replica_id, self.tour_count, new_log_head);

        self.tour_start = origin;
        self.tour_end = new_log_head;
        self.current_position = origin;

        tracing::debug!(
            replica = %self.replica_id,
            tour = self.tour_count,
            origin = %origin,
            end = %new_log_head,
            "starting new scrub tour"
        );
    }

    /// Records a detected corruption.
    pub fn record_corruption(&mut self, op: OpNumber) {
        tracing::warn!(
            replica = %self.replica_id,
            op = %op,
            tour = self.tour_count,
            "scrubber detected corruption"
        );

        self.corruptions.push((op, self.tour_count));
    }

    /// Returns detected corruptions.
    pub fn corruptions(&self) -> &[(OpNumber, u64)] {
        &self.corruptions
    }

    /// Returns the current tour count.
    pub fn tour_count(&self) -> u64 {
        self.tour_count
    }

    /// Returns the scrub budget.
    pub fn budget(&self) -> &ScrubBudget {
        &self.scrub_budget
    }

    /// Returns mutable scrub budget.
    pub fn budget_mut(&mut self) -> &mut ScrubBudget {
        &mut self.scrub_budget
    }

    /// Sets the tour position (for testing).
    #[cfg(test)]
    pub fn set_tour_position_for_test(
        &mut self,
        current: OpNumber,
        start: OpNumber,
        end: OpNumber,
    ) {
        self.current_position = current;
        self.tour_start = start;
        self.tour_end = end;
    }

    /// Updates the log head (called when log grows).
    ///
    /// If the log head advances beyond `tour_end`, we need to update `tour_end`
    /// to ensure we scrub new entries in this tour.
    pub fn update_log_head(&mut self, new_head: OpNumber) {
        if new_head > self.tour_end {
            self.tour_end = new_head;
        }
    }

    /// Scrubs the next log entry.
    ///
    /// This is the main scrubbing method called on every scrub timeout.
    ///
    /// Returns:
    /// - `ScrubResult::Ok` - Entry validated successfully
    /// - `ScrubResult::Corruption` - Checksum mismatch detected
    /// - `ScrubResult::TourComplete` - No more entries in this tour
    /// - `ScrubResult::BudgetExhausted` - IOPS budget depleted
    ///
    /// The caller is responsible for:
    /// - Providing the log to read from
    /// - Triggering repair on corruption detection
    /// - Starting a new tour when complete
    pub fn scrub_next(&mut self, log: &[LogEntry]) -> ScrubResult {
        // Check budget first
        if !self.scrub_budget.can_scrub() {
            // PRODUCTION ASSERTION: Rate limit enforcement
            // Ensures scrubbing respects IOPS budget (prevents production impact)
            assert!(
                self.scrub_budget.reads_this_tick() >= self.scrub_budget.max_reads_per_tick(),
                "budget exhausted: reads {} >= max {}",
                self.scrub_budget.reads_this_tick(),
                self.scrub_budget.max_reads_per_tick()
            );
            return ScrubResult::BudgetExhausted;
        }

        // Check if tour complete
        if self.is_tour_complete() {
            return ScrubResult::TourComplete;
        }

        // Get next op to scrub
        let Some(op) = self.next_op_to_scrub() else {
            return ScrubResult::TourComplete;
        };

        // Find entry in log
        let Some(entry) = log.iter().find(|e| e.op_number == op) else {
            // Entry not in log yet (might be beyond current log head)
            // This can happen if tour_end was set based on a future log head
            // Just skip this entry and move to next
            self.advance();
            return ScrubResult::Ok;
        };

        // Consume budget
        self.scrub_budget.record_scrub();

        // Validate checksum
        let is_valid = entry.verify_checksum();

        // Advance position
        self.advance();

        if is_valid {
            ScrubResult::Ok
        } else {
            // Corruption detected!
            let corruptions_before = self.corruptions.len();
            self.record_corruption(op);

            // PRODUCTION ASSERTION: Corruption tracking
            // Ensures corruption detection is recorded (triggers repair)
            assert!(
                self.corruptions.len() == corruptions_before + 1,
                "corruption must be recorded: {} corruptions before, {} after",
                corruptions_before,
                self.corruptions.len()
            );

            ScrubResult::Corruption
        }
    }
}

// ============================================================================
// Scrub Budget (IOPS Rate Limiting)
// ============================================================================

/// IOPS budget for scrubbing to prevent impacting production traffic.
///
/// The budget resets every tick, allowing a fixed number of scrub reads
/// per time window. This reserves the majority of IOPS for production workloads.
#[derive(Debug, Clone)]
pub struct ScrubBudget {
    /// Maximum reads allowed per tick.
    max_reads_per_tick: usize,

    /// Reads consumed this tick.
    reads_this_tick: usize,
}

impl ScrubBudget {
    /// Creates a new scrub budget.
    pub fn new(max_reads_per_tick: usize) -> Self {
        Self {
            max_reads_per_tick,
            reads_this_tick: 0,
        }
    }

    /// Checks if scrubbing is allowed (budget available).
    pub fn can_scrub(&self) -> bool {
        self.reads_this_tick < self.max_reads_per_tick
    }

    /// Records a scrub read (consumes budget).
    pub fn record_scrub(&mut self) {
        debug_assert!(
            self.can_scrub(),
            "scrub budget exceeded: {} >= {}",
            self.reads_this_tick,
            self.max_reads_per_tick
        );
        self.reads_this_tick += 1;
    }

    /// Resets the budget for a new tick.
    pub fn reset_tick(&mut self) {
        self.reads_this_tick = 0;
    }

    /// Returns the number of reads consumed this tick.
    pub fn reads_this_tick(&self) -> usize {
        self.reads_this_tick
    }

    /// Returns the maximum reads per tick.
    pub fn max_reads_per_tick(&self) -> usize {
        self.max_reads_per_tick
    }
}

// ============================================================================
// Kani Verification Helpers
// ============================================================================

#[cfg(kani)]
impl LogScrubber {
    /// Returns the current scrub position for verification.
    ///
    /// Used in bounded model checking to verify scrub progress.
    pub(crate) fn current_position(&self) -> OpNumber {
        self.current_position
    }

    /// Sets the tour range for testing scrub logic.
    ///
    /// Used in bounded model checking to set up specific test scenarios.
    pub(crate) fn set_tour_range(&mut self, start: OpNumber, end: OpNumber) {
        self.tour_start = start;
        self.tour_end = end;
        self.current_position = start;
    }

    /// Resets the tour for testing.
    ///
    /// Used in bounded model checking to test multiple tour scenarios.
    pub(crate) fn reset_tour_for_test(&mut self, log_head: OpNumber) {
        let origin = Self::randomize_origin(self.replica_id, self.tour_count + 1, log_head);
        self.current_position = origin;
        self.tour_start = origin;
        self.tour_end = log_head;
        self.tour_count += 1;
    }
}

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

    #[test]
    fn scrubber_creation() {
        let scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(100));

        assert!(scrubber.current_position <= OpNumber::new(100));
        assert_eq!(scrubber.tour_count, 0);
        assert!(scrubber.corruptions.is_empty());
    }

    #[test]
    fn origin_randomization_deterministic() {
        let origin1 = LogScrubber::randomize_origin(ReplicaId::new(0), 0, OpNumber::new(100));
        let origin2 = LogScrubber::randomize_origin(ReplicaId::new(0), 0, OpNumber::new(100));

        // Same seed should produce same origin
        assert_eq!(origin1, origin2);
    }

    #[test]
    fn origin_randomization_varies_by_replica() {
        let origin0 = LogScrubber::randomize_origin(ReplicaId::new(0), 0, OpNumber::new(100));
        let origin1 = LogScrubber::randomize_origin(ReplicaId::new(1), 0, OpNumber::new(100));

        // Different replicas should (likely) have different origins
        // Note: Small chance they're equal due to randomness, but very unlikely
        assert_ne!(origin0, origin1);
    }

    #[test]
    fn origin_randomization_varies_by_tour() {
        let origin_tour0 = LogScrubber::randomize_origin(ReplicaId::new(0), 0, OpNumber::new(100));
        let origin_tour1 = LogScrubber::randomize_origin(ReplicaId::new(0), 1, OpNumber::new(100));

        // Different tours should have different origins
        assert_ne!(origin_tour0, origin_tour1);
    }

    #[test]
    fn tour_advances_correctly() {
        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(10));

        // Force start at 0 for simplicity
        scrubber.current_position = OpNumber::new(0);
        scrubber.tour_start = OpNumber::new(0);
        scrubber.tour_end = OpNumber::new(10);

        // Scrub from 0 to 10
        for i in 0..=10 {
            assert!(!scrubber.is_tour_complete());
            assert_eq!(scrubber.next_op_to_scrub(), Some(OpNumber::new(i)));
            scrubber.advance();
        }

        // Tour should be complete
        assert!(scrubber.is_tour_complete());
        assert_eq!(scrubber.next_op_to_scrub(), None);
    }

    #[test]
    fn start_new_tour_increments_count() {
        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(100));

        assert_eq!(scrubber.tour_count, 0);

        scrubber.start_new_tour(OpNumber::new(200));
        assert_eq!(scrubber.tour_count, 1);

        scrubber.start_new_tour(OpNumber::new(300));
        assert_eq!(scrubber.tour_count, 2);
    }

    #[test]
    fn corruption_tracking() {
        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(100));

        assert_eq!(scrubber.corruptions().len(), 0);

        scrubber.record_corruption(OpNumber::new(42));
        assert_eq!(scrubber.corruptions().len(), 1);
        assert_eq!(scrubber.corruptions()[0], (OpNumber::new(42), 0));

        scrubber.record_corruption(OpNumber::new(99));
        assert_eq!(scrubber.corruptions().len(), 2);
    }

    #[test]
    fn scrub_budget_respects_limit() {
        let mut budget = ScrubBudget::new(10);

        // Can scrub up to limit
        for _ in 0..10 {
            assert!(budget.can_scrub());
            budget.record_scrub();
        }

        // Budget exhausted
        assert!(!budget.can_scrub());
        assert_eq!(budget.reads_this_tick(), 10);

        // Reset allows more scrubs
        budget.reset_tick();
        assert!(budget.can_scrub());
        assert_eq!(budget.reads_this_tick(), 0);
    }

    #[test]
    fn update_log_head_extends_tour() {
        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(100));

        scrubber.tour_end = OpNumber::new(100);

        // Log grows
        scrubber.update_log_head(OpNumber::new(150));

        // Tour end should extend
        assert_eq!(scrubber.tour_end, OpNumber::new(150));
    }

    #[test]
    fn scrub_next_validates_checksum() {
        use crate::types::{LogEntry, ViewNumber};
        use kimberlite_kernel::Command;
        use kimberlite_types::{DataClass, Placement};

        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(10));
        scrubber.current_position = OpNumber::new(0);
        scrubber.tour_start = OpNumber::new(0);
        scrubber.tour_end = OpNumber::new(2);

        // Create log with valid entries
        let cmd = Command::create_stream_with_auto_id(
            "test".into(),
            DataClass::Public,
            Placement::Global,
        );
        let entry = LogEntry::new(OpNumber::new(0), ViewNumber::ZERO, cmd, None, None, None);

        let log = vec![entry];

        // Scrub should succeed
        let result = scrubber.scrub_next(&log);
        assert_eq!(result, ScrubResult::Ok);
        assert_eq!(scrubber.current_position, OpNumber::new(1));
    }

    #[test]
    fn scrub_next_detects_corruption() {
        use crate::types::{LogEntry, ViewNumber};
        use kimberlite_kernel::Command;
        use kimberlite_types::{DataClass, Placement};

        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(10));
        scrubber.current_position = OpNumber::new(0);
        scrubber.tour_start = OpNumber::new(0);
        scrubber.tour_end = OpNumber::new(2);

        // Create entry with invalid checksum
        let cmd = Command::create_stream_with_auto_id(
            "test".into(),
            DataClass::Public,
            Placement::Global,
        );
        let mut entry = LogEntry::new(OpNumber::new(0), ViewNumber::ZERO, cmd, None, None, None);

        // Corrupt the checksum
        entry.checksum = 0xDEAD_BEEF;

        let log = vec![entry];

        // Scrub should detect corruption
        let result = scrubber.scrub_next(&log);
        assert_eq!(result, ScrubResult::Corruption);
        assert_eq!(scrubber.corruptions().len(), 1);
        assert_eq!(scrubber.corruptions()[0].0, OpNumber::new(0));
    }

    #[test]
    fn scrub_next_respects_budget() {
        use crate::types::{LogEntry, ViewNumber};
        use kimberlite_kernel::Command;
        use kimberlite_types::{DataClass, Placement};

        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(100));
        scrubber.current_position = OpNumber::new(0);
        scrubber.tour_start = OpNumber::new(0);
        scrubber.tour_end = OpNumber::new(20);

        // Create log
        let cmd = Command::create_stream_with_auto_id(
            "test".into(),
            DataClass::Public,
            Placement::Global,
        );
        let mut log = Vec::new();
        for i in 0..20 {
            let entry = LogEntry::new(
                OpNumber::new(i),
                ViewNumber::ZERO,
                cmd.clone(),
                None,
                None,
                None,
            );
            log.push(entry);
        }

        // Exhaust budget (default is 10)
        for _ in 0..10 {
            let result = scrubber.scrub_next(&log);
            assert_eq!(result, ScrubResult::Ok);
        }

        // Next scrub should fail due to budget
        let result = scrubber.scrub_next(&log);
        assert_eq!(result, ScrubResult::BudgetExhausted);
    }

    #[test]
    fn scrub_next_returns_tour_complete() {
        let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(10));
        scrubber.current_position = OpNumber::new(11); // Beyond tour_end
        scrubber.tour_start = OpNumber::new(0);
        scrubber.tour_end = OpNumber::new(10);

        let log = Vec::new();

        let result = scrubber.scrub_next(&log);
        assert_eq!(result, ScrubResult::TourComplete);
    }

    // ========================================================================
    // Property-Based Tests (Phase 3)
    // ========================================================================

    use proptest::prelude::*;

    proptest! {
        /// Property: Tour always completes within bounded iterations.
        #[test]
        fn prop_tour_always_completes(
            log_size in 1_u64..100,
        ) {
            let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(log_size));
            scrubber.current_position = OpNumber::new(0);
            scrubber.tour_start = OpNumber::new(0);
            scrubber.tour_end = OpNumber::new(log_size);

            // Tour must complete within log_size + 1 iterations
            let mut iterations = 0;
            while !scrubber.is_tour_complete() && iterations <= log_size + 1 {
                scrubber.advance();
                iterations += 1;
            }

            prop_assert!(scrubber.is_tour_complete(), "tour should complete within {} iterations", log_size + 1);
            prop_assert!(iterations <= log_size + 1, "tour took {} iterations (max {})", iterations, log_size + 1);
        }

        /// Property: Every op visited exactly once per tour.
        #[test]
        fn prop_every_op_visited_once(
            log_size in 1_u64..50,
        ) {
            let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(log_size));
            scrubber.current_position = OpNumber::new(0);
            scrubber.tour_start = OpNumber::new(0);
            scrubber.tour_end = OpNumber::new(log_size);

            let mut visited = std::collections::HashSet::new();

            while !scrubber.is_tour_complete() {
                if let Some(op) = scrubber.next_op_to_scrub() {
                    prop_assert!(!visited.contains(&op), "op {} visited twice", op.as_u64());
                    visited.insert(op);
                    scrubber.advance();
                }
            }

            // All ops from 0 to log_size should have been visited
            for i in 0..=log_size {
                prop_assert!(visited.contains(&OpNumber::new(i)), "op {} not visited", i);
            }
        }

        /// Property: Origin randomization distributes uniformly.
        #[test]
        fn prop_origin_distribution_varies(
            tour_count in 0_u64..20,
        ) {
            let log_head = OpNumber::new(100);
            let origins: Vec<_> = (0..10)
                .map(|replica_id| {
                    LogScrubber::randomize_origin(
                        ReplicaId::new(replica_id),
                        tour_count,
                        log_head,
                    )
                })
                .collect();

            // At least some origins should be different
            // (Very unlikely all 10 replicas get same origin)
            let unique_origins: std::collections::HashSet<_> = origins.iter().collect();
            prop_assert!(unique_origins.len() > 1, "origins should vary across replicas");

            // All origins should be within valid range
            for origin in &origins {
                prop_assert!(origin.as_u64() <= log_head.as_u64());
            }
        }

        /// Property: Rate limit never exceeded.
        #[test]
        fn prop_rate_limit_never_exceeded(
            scrub_attempts in 1_usize..50,
        ) {
            let max_reads = 10;
            let mut budget = ScrubBudget::new(max_reads);

            let mut successful_scrubs = 0;

            for _ in 0..scrub_attempts {
                if budget.can_scrub() {
                    budget.record_scrub();
                    successful_scrubs += 1;
                }
            }

            // Should never exceed limit
            prop_assert!(successful_scrubs <= max_reads);
            prop_assert!(budget.reads_this_tick() <= max_reads);
            prop_assert_eq!(budget.reads_this_tick(), successful_scrubs);
        }

        /// Property: Detected corruptions are recorded correctly.
        #[test]
        fn prop_corruptions_recorded_correctly(
            corruption_count in 0_usize..20,
        ) {
            let mut scrubber = LogScrubber::new(ReplicaId::new(0), OpNumber::new(100));

            for i in 0..corruption_count {
                scrubber.record_corruption(OpNumber::new(i as u64));
            }

            prop_assert_eq!(scrubber.corruptions().len(), corruption_count);

            // Verify all corruption records
            for (i, (op, tour)) in scrubber.corruptions().iter().enumerate() {
                prop_assert_eq!(op.as_u64(), i as u64);
                prop_assert_eq!(*tour, 0); // Initial tour
            }
        }
    }
}
