graph_d 1.3.2

//! Checkpoint system for efficient crash recovery.
//!
//! Satisfies: RT-1 (Data MUST survive crashes and restarts)
//! Satisfies: TN1 resolution (Hybrid WAL + periodic checkpoints)
//! Phase: B (Complete Durability)
//!
//! # Overview
//!
//! Checkpoints provide a known-good snapshot of the database state,
//! allowing WAL truncation and faster recovery. Instead of replaying
//! the entire WAL from the beginning, recovery starts from the last
//! checkpoint and replays only subsequent entries.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────┐
//! │                    Checkpoint Strategy                          │
//! │                                                                 │
//! │  WAL: [E1][E2][E3][CP1][E4][E5][E6][E7][CP2][E8][E9]            │
//! │                    ↑                    ↑                       │
//! │                Checkpoint 1         Checkpoint 2                │
//! │                                                                 │
//! │  Recovery from CP2: Only replay E8, E9 (not E1-E7)             │
//! └─────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Checkpoint Strategies
//!
//! - **Time-based**: Checkpoint every N seconds
//! - **Entry-based**: Checkpoint every N WAL entries
//! - **Size-based**: Checkpoint when WAL exceeds N bytes
//! - **Manual**: Explicit checkpoint calls
//!
//! # Example
//!
//! ```rust,ignore
//! use graph_d::storage::checkpoint::{CheckpointManager, CheckpointConfig};
//!
//! let config = CheckpointConfig {
//!     interval_entries: 10_000,
//!     interval_seconds: Some(300), // 5 minutes
//!     ..Default::default()
//! };
//!
//! let checkpoint_mgr = CheckpointManager::new(storage, wal, config)?;
//!
//! // Automatic checkpoint based on config
//! checkpoint_mgr.maybe_checkpoint()?;
//!
//! // Force checkpoint
//! checkpoint_mgr.force_checkpoint()?;
//!
//! // Recovery
//! let (last_checkpoint, entries_to_replay) = checkpoint_mgr.recover()?;
//! ```

use crate::error::{GraphError, Result};
use crate::graph::{Id, Node, Relationship};
use std::collections::HashMap;
use std::fs::{self, File, OpenOptions};
use std::io::{BufReader, BufWriter, Read, Write};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

/// Checkpoint file magic bytes.
const CHECKPOINT_MAGIC: &[u8; 4] = b"GCPT";

/// Checkpoint file format version.
const CHECKPOINT_VERSION: u32 = 1;

/// Checkpoint configuration.
/// Satisfies: Configurable checkpoint intervals (anchor phase B)
#[derive(Debug, Clone)]
pub struct CheckpointConfig {
    /// Checkpoint after this many WAL entries.
    pub interval_entries: u64,

    /// Checkpoint after this many seconds (optional).
    pub interval_seconds: Option<u64>,

    /// Checkpoint when WAL exceeds this size in bytes (optional).
    pub max_wal_size_bytes: Option<u64>,

    /// Keep this many old checkpoints for safety.
    pub keep_old_checkpoints: usize,

    /// Use fsync for checkpoint files.
    pub use_fsync: bool,
}

impl Default for CheckpointConfig {
    fn default() -> Self {
        CheckpointConfig {
            interval_entries: 10_000,
            interval_seconds: Some(300),                // 5 minutes
            max_wal_size_bytes: Some(64 * 1024 * 1024), // 64MB
            keep_old_checkpoints: 2,
            use_fsync: true,
        }
    }
}

/// Checkpoint metadata header.
#[derive(Debug, Clone)]
pub struct CheckpointHeader {
    /// Magic bytes for validation
    pub magic: [u8; 4],

    /// Format version
    pub version: u32,

    /// WAL sequence number at checkpoint
    pub wal_sequence: u64,

    /// Timestamp when checkpoint was created
    pub timestamp: u64,

    /// Number of nodes in checkpoint
    pub node_count: u64,

    /// Number of relationships in checkpoint
    pub relationship_count: u64,

    /// CRC32 checksum of the checkpoint data
    pub checksum: u32,
}

impl CheckpointHeader {
    /// Header size in bytes: magic(4) + version(4) + sequence(8) + timestamp(8) +
    /// node_count(8) + rel_count(8) + checksum(4) = 44 bytes
    pub const SIZE: usize = 44;

    /// Create a new checkpoint header.
    pub fn new(wal_sequence: u64, node_count: u64, relationship_count: u64) -> Self {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        CheckpointHeader {
            magic: *CHECKPOINT_MAGIC,
            version: CHECKPOINT_VERSION,
            wal_sequence,
            timestamp,
            node_count,
            relationship_count,
            checksum: 0, // Will be computed during serialization
        }
    }

    /// Validate the header.
    pub fn is_valid(&self) -> bool {
        self.magic == *CHECKPOINT_MAGIC && self.version == CHECKPOINT_VERSION
    }

    /// Serialize header to bytes.
    pub fn serialize(&self) -> Vec<u8> {
        let mut data = Vec::with_capacity(Self::SIZE);
        data.extend_from_slice(&self.magic);
        data.extend_from_slice(&self.version.to_le_bytes());
        data.extend_from_slice(&self.wal_sequence.to_le_bytes());
        data.extend_from_slice(&self.timestamp.to_le_bytes());
        data.extend_from_slice(&self.node_count.to_le_bytes());
        data.extend_from_slice(&self.relationship_count.to_le_bytes());
        data.extend_from_slice(&self.checksum.to_le_bytes());
        data
    }

    /// Deserialize header from bytes.
    pub fn deserialize(data: &[u8]) -> Result<Self> {
        if data.len() < Self::SIZE {
            return Err(GraphError::Storage(
                "Checkpoint header too short".to_string(),
            ));
        }

        let mut magic = [0u8; 4];
        magic.copy_from_slice(&data[0..4]);

        let version = u32::from_le_bytes([data[4], data[5], data[6], data[7]]);
        let wal_sequence = u64::from_le_bytes([
            data[8], data[9], data[10], data[11], data[12], data[13], data[14], data[15],
        ]);
        let timestamp = u64::from_le_bytes([
            data[16], data[17], data[18], data[19], data[20], data[21], data[22], data[23],
        ]);
        let node_count = u64::from_le_bytes([
            data[24], data[25], data[26], data[27], data[28], data[29], data[30], data[31],
        ]);
        let relationship_count = u64::from_le_bytes([
            data[32], data[33], data[34], data[35], data[36], data[37], data[38], data[39],
        ]);
        let checksum = u32::from_le_bytes([data[40], data[41], data[42], data[43]]);

        let header = CheckpointHeader {
            magic,
            version,
            wal_sequence,
            timestamp,
            node_count,
            relationship_count,
            checksum,
        };

        if !header.is_valid() {
            return Err(GraphError::Storage("Invalid checkpoint header".to_string()));
        }

        Ok(header)
    }
}

/// A complete checkpoint snapshot.
/// Satisfies: RT-1 (known-good state for recovery)
#[derive(Debug)]
pub struct Checkpoint {
    /// Checkpoint header with metadata
    pub header: CheckpointHeader,

    /// All nodes at checkpoint time
    pub nodes: HashMap<Id, Node>,

    /// All relationships at checkpoint time
    pub relationships: HashMap<Id, Relationship>,

    /// Node relationship index
    pub node_relationships: HashMap<Id, Vec<Id>>,
}

impl Checkpoint {
    /// Create a new checkpoint from current storage state.
    pub fn from_storage(
        wal_sequence: u64,
        nodes: HashMap<Id, Node>,
        relationships: HashMap<Id, Relationship>,
        node_relationships: HashMap<Id, Vec<Id>>,
    ) -> Self {
        let header =
            CheckpointHeader::new(wal_sequence, nodes.len() as u64, relationships.len() as u64);

        Checkpoint {
            header,
            nodes,
            relationships,
            node_relationships,
        }
    }

    /// Serialize checkpoint to bytes.
    ///
    /// Format: `header | nodes_data | relationships_data | index_data`
    pub fn serialize(&self) -> Result<Vec<u8>> {
        let mut data = Vec::new();

        // Serialize nodes
        let nodes_json = serde_json::to_vec(&self.nodes)
            .map_err(|e| GraphError::Serialization(e.to_string()))?;

        // Serialize relationships
        let rels_json = serde_json::to_vec(&self.relationships)
            .map_err(|e| GraphError::Serialization(e.to_string()))?;

        // Serialize index
        let index_json = serde_json::to_vec(&self.node_relationships)
            .map_err(|e| GraphError::Serialization(e.to_string()))?;

        // Calculate checksum over all data
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        use std::hash::{Hash, Hasher};
        nodes_json.hash(&mut hasher);
        rels_json.hash(&mut hasher);
        index_json.hash(&mut hasher);
        let checksum = hasher.finish() as u32;

        // Update header with checksum
        let mut header = self.header.clone();
        header.checksum = checksum;

        // Write header
        data.extend_from_slice(&header.serialize());

        // Write nodes length + data
        data.extend_from_slice(&(nodes_json.len() as u64).to_le_bytes());
        data.extend_from_slice(&nodes_json);

        // Write relationships length + data
        data.extend_from_slice(&(rels_json.len() as u64).to_le_bytes());
        data.extend_from_slice(&rels_json);

        // Write index length + data
        data.extend_from_slice(&(index_json.len() as u64).to_le_bytes());
        data.extend_from_slice(&index_json);

        Ok(data)
    }

    /// Deserialize checkpoint from bytes.
    pub fn deserialize(data: &[u8]) -> Result<Self> {
        if data.len() < CheckpointHeader::SIZE {
            return Err(GraphError::Storage("Checkpoint data too short".to_string()));
        }

        let header = CheckpointHeader::deserialize(data)?;
        let mut offset = CheckpointHeader::SIZE;

        // Read nodes
        let nodes_len = u64::from_le_bytes([
            data[offset],
            data[offset + 1],
            data[offset + 2],
            data[offset + 3],
            data[offset + 4],
            data[offset + 5],
            data[offset + 6],
            data[offset + 7],
        ]) as usize;
        offset += 8;

        let nodes: HashMap<Id, Node> = serde_json::from_slice(&data[offset..offset + nodes_len])
            .map_err(|e| GraphError::Serialization(e.to_string()))?;
        offset += nodes_len;

        // Read relationships
        let rels_len = u64::from_le_bytes([
            data[offset],
            data[offset + 1],
            data[offset + 2],
            data[offset + 3],
            data[offset + 4],
            data[offset + 5],
            data[offset + 6],
            data[offset + 7],
        ]) as usize;
        offset += 8;

        let relationships: HashMap<Id, Relationship> =
            serde_json::from_slice(&data[offset..offset + rels_len])
                .map_err(|e| GraphError::Serialization(e.to_string()))?;
        offset += rels_len;

        // Read index
        let index_len = u64::from_le_bytes([
            data[offset],
            data[offset + 1],
            data[offset + 2],
            data[offset + 3],
            data[offset + 4],
            data[offset + 5],
            data[offset + 6],
            data[offset + 7],
        ]) as usize;
        offset += 8;

        let node_relationships: HashMap<Id, Vec<Id>> =
            serde_json::from_slice(&data[offset..offset + index_len])
                .map_err(|e| GraphError::Serialization(e.to_string()))?;

        Ok(Checkpoint {
            header,
            nodes,
            relationships,
            node_relationships,
        })
    }
}

/// Checkpoint manager coordinates checkpoint creation and recovery.
/// Satisfies: TN1 resolution (Hybrid WAL + checkpoints)
pub struct CheckpointManager {
    /// Directory for checkpoint files
    checkpoint_dir: PathBuf,

    /// Configuration
    config: CheckpointConfig,

    /// Last checkpoint time
    last_checkpoint_time: Instant,

    /// WAL entries since last checkpoint
    entries_since_checkpoint: u64,

    /// Last checkpoint sequence number
    last_checkpoint_sequence: u64,
}

impl CheckpointManager {
    /// Create a new checkpoint manager.
    pub fn new<P: AsRef<Path>>(checkpoint_dir: P, config: CheckpointConfig) -> Result<Self> {
        let checkpoint_dir = checkpoint_dir.as_ref().to_path_buf();

        // Create checkpoint directory if needed
        fs::create_dir_all(&checkpoint_dir).map_err(|e| GraphError::Io(e.to_string()))?;

        Ok(CheckpointManager {
            checkpoint_dir,
            config,
            last_checkpoint_time: Instant::now(),
            entries_since_checkpoint: 0,
            last_checkpoint_sequence: 0,
        })
    }

    /// Check if a checkpoint should be created.
    /// Satisfies: Configurable checkpoint intervals
    pub fn should_checkpoint(&self, current_wal_sequence: u64, wal_size_bytes: u64) -> bool {
        // Entry-based trigger
        let entries_since = current_wal_sequence.saturating_sub(self.last_checkpoint_sequence);
        if entries_since >= self.config.interval_entries {
            return true;
        }

        // Time-based trigger
        if let Some(interval_secs) = self.config.interval_seconds {
            if self.last_checkpoint_time.elapsed() >= Duration::from_secs(interval_secs) {
                return true;
            }
        }

        // Size-based trigger
        if let Some(max_size) = self.config.max_wal_size_bytes {
            if wal_size_bytes >= max_size {
                return true;
            }
        }

        false
    }

    /// Create a checkpoint.
    /// Satisfies: RT-1 (durable snapshot)
    pub fn create_checkpoint(&mut self, checkpoint: &Checkpoint) -> Result<PathBuf> {
        let filename = format!(
            "checkpoint_{:016x}_{}.gcpt",
            checkpoint.header.wal_sequence, checkpoint.header.timestamp
        );
        let checkpoint_path = self.checkpoint_dir.join(&filename);

        // Serialize checkpoint
        let data = checkpoint.serialize()?;

        // Write to temporary file first
        let temp_path = checkpoint_path.with_extension("tmp");
        {
            let file = OpenOptions::new()
                .write(true)
                .create(true)
                .truncate(true)
                .open(&temp_path)
                .map_err(|e| GraphError::Io(e.to_string()))?;

            let mut writer = BufWriter::new(file);
            writer
                .write_all(&data)
                .map_err(|e| GraphError::Io(e.to_string()))?;
            writer.flush().map_err(|e| GraphError::Io(e.to_string()))?;

            if self.config.use_fsync {
                writer
                    .get_ref()
                    .sync_all()
                    .map_err(|e| GraphError::Io(e.to_string()))?;
            }
        }

        // Atomic rename
        fs::rename(&temp_path, &checkpoint_path).map_err(|e| GraphError::Io(e.to_string()))?;

        // Update state
        self.last_checkpoint_time = Instant::now();
        self.last_checkpoint_sequence = checkpoint.header.wal_sequence;
        self.entries_since_checkpoint = 0;

        // Cleanup old checkpoints
        self.cleanup_old_checkpoints()?;

        Ok(checkpoint_path)
    }

    /// Find the latest valid checkpoint.
    /// Satisfies: RT-1 (recovery from checkpoint)
    pub fn find_latest_checkpoint(&self) -> Result<Option<PathBuf>> {
        let mut checkpoints: Vec<PathBuf> = fs::read_dir(&self.checkpoint_dir)
            .map_err(|e| GraphError::Io(e.to_string()))?
            .filter_map(|entry| entry.ok())
            .map(|entry| entry.path())
            .filter(|path| path.extension().map(|ext| ext == "gcpt").unwrap_or(false))
            .collect();

        // Sort by name (which includes sequence number) in descending order
        checkpoints.sort_by(|a, b| b.cmp(a));

        for checkpoint_path in checkpoints {
            // Validate checkpoint
            if self.validate_checkpoint(&checkpoint_path).is_ok() {
                return Ok(Some(checkpoint_path));
            }
        }

        Ok(None)
    }

    /// Load a checkpoint from file.
    pub fn load_checkpoint<P: AsRef<Path>>(&self, path: P) -> Result<Checkpoint> {
        let file = File::open(path.as_ref()).map_err(|e| GraphError::Io(e.to_string()))?;
        let mut reader = BufReader::new(file);

        let mut data = Vec::new();
        reader
            .read_to_end(&mut data)
            .map_err(|e| GraphError::Io(e.to_string()))?;

        Checkpoint::deserialize(&data)
    }

    /// Validate a checkpoint file.
    fn validate_checkpoint<P: AsRef<Path>>(&self, path: P) -> Result<()> {
        let checkpoint = self.load_checkpoint(path)?;

        if !checkpoint.header.is_valid() {
            return Err(GraphError::Storage("Invalid checkpoint header".to_string()));
        }

        // Verify counts match
        if checkpoint.nodes.len() as u64 != checkpoint.header.node_count {
            return Err(GraphError::Storage("Node count mismatch".to_string()));
        }

        if checkpoint.relationships.len() as u64 != checkpoint.header.relationship_count {
            return Err(GraphError::Storage(
                "Relationship count mismatch".to_string(),
            ));
        }

        Ok(())
    }

    /// Cleanup old checkpoints, keeping only the configured number.
    fn cleanup_old_checkpoints(&self) -> Result<()> {
        let mut checkpoints: Vec<PathBuf> = fs::read_dir(&self.checkpoint_dir)
            .map_err(|e| GraphError::Io(e.to_string()))?
            .filter_map(|entry| entry.ok())
            .map(|entry| entry.path())
            .filter(|path| path.extension().map(|ext| ext == "gcpt").unwrap_or(false))
            .collect();

        // Sort by name in descending order (newest first)
        checkpoints.sort_by(|a, b| b.cmp(a));

        // Remove checkpoints beyond the keep limit
        for checkpoint in checkpoints.iter().skip(self.config.keep_old_checkpoints) {
            let _ = fs::remove_file(checkpoint);
        }

        Ok(())
    }

    /// Get the WAL sequence number to start recovery from.
    /// Returns the sequence after the last checkpoint, or 0 if no checkpoint.
    pub fn get_recovery_start_sequence(&self) -> Result<u64> {
        match self.find_latest_checkpoint()? {
            Some(path) => {
                let checkpoint = self.load_checkpoint(&path)?;
                Ok(checkpoint.header.wal_sequence + 1)
            }
            None => Ok(0),
        }
    }

    /// Increment the entries counter (called after each WAL append).
    pub fn record_wal_entry(&mut self) {
        self.entries_since_checkpoint += 1;
    }

    /// Get statistics about checkpoints.
    pub fn get_stats(&self) -> CheckpointStats {
        CheckpointStats {
            entries_since_checkpoint: self.entries_since_checkpoint,
            last_checkpoint_sequence: self.last_checkpoint_sequence,
            time_since_checkpoint: self.last_checkpoint_time.elapsed(),
        }
    }
}

/// Checkpoint statistics.
#[derive(Debug, Clone)]
pub struct CheckpointStats {
    /// Number of WAL entries since last checkpoint
    pub entries_since_checkpoint: u64,

    /// Sequence number of last checkpoint
    pub last_checkpoint_sequence: u64,

    /// Time elapsed since last checkpoint
    pub time_since_checkpoint: Duration,
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;
    use tempfile::TempDir;

    /// Test: Checkpoint header serialization
    /// Validates: Data integrity
    #[test]
    fn test_header_serialization() {
        let header = CheckpointHeader::new(100, 50, 25);
        let serialized = header.serialize();
        let deserialized = CheckpointHeader::deserialize(&serialized).unwrap();

        assert_eq!(deserialized.wal_sequence, 100);
        assert_eq!(deserialized.node_count, 50);
        assert_eq!(deserialized.relationship_count, 25);
        assert!(deserialized.is_valid());
    }

    /// Test: Checkpoint creation and recovery
    /// Validates: RT-1 (data survives checkpoint cycle)
    #[test]
    fn test_checkpoint_roundtrip() {
        let temp_dir = TempDir::new().unwrap();
        let mut manager =
            CheckpointManager::new(temp_dir.path(), CheckpointConfig::default()).unwrap();

        // Create test data
        let mut nodes = HashMap::new();
        nodes.insert(
            1,
            Node {
                id: 1,
                properties: HashMap::new(),
            },
        );
        nodes.insert(
            2,
            Node {
                id: 2,
                properties: HashMap::new(),
            },
        );

        let relationships = HashMap::new();
        let node_relationships = HashMap::new();

        let checkpoint = Checkpoint::from_storage(100, nodes, relationships, node_relationships);

        // Create checkpoint
        let path = manager.create_checkpoint(&checkpoint).unwrap();
        assert!(path.exists());

        // Load checkpoint
        let loaded = manager.load_checkpoint(&path).unwrap();
        assert_eq!(loaded.header.wal_sequence, 100);
        assert_eq!(loaded.nodes.len(), 2);
    }

    /// Test: Checkpoint trigger conditions
    /// Validates: Configurable checkpoint intervals
    #[test]
    fn test_should_checkpoint() {
        let temp_dir = TempDir::new().unwrap();
        let config = CheckpointConfig {
            interval_entries: 100,
            interval_seconds: None,
            max_wal_size_bytes: Some(1000),
            ..Default::default()
        };

        let manager = CheckpointManager::new(temp_dir.path(), config).unwrap();

        // Not enough entries
        assert!(!manager.should_checkpoint(50, 0));

        // Enough entries
        assert!(manager.should_checkpoint(100, 0));

        // Size trigger
        assert!(manager.should_checkpoint(0, 1000));
    }

    /// Test: Find latest checkpoint
    /// Validates: Recovery finds correct checkpoint
    #[test]
    fn test_find_latest_checkpoint() {
        let temp_dir = TempDir::new().unwrap();
        let mut manager =
            CheckpointManager::new(temp_dir.path(), CheckpointConfig::default()).unwrap();

        // Create multiple checkpoints
        for seq in [100, 200, 300] {
            let checkpoint =
                Checkpoint::from_storage(seq, HashMap::new(), HashMap::new(), HashMap::new());
            manager.create_checkpoint(&checkpoint).unwrap();
        }

        // Should find the latest (seq=300)
        let latest = manager.find_latest_checkpoint().unwrap().unwrap();
        let loaded = manager.load_checkpoint(&latest).unwrap();
        assert_eq!(loaded.header.wal_sequence, 300);
    }
}