graph_d 1.3.2

//! Write-Ahead Logging (WAL) for crash recovery.
//!
//! Satisfies: RT-1 (Data MUST survive crashes and restarts)
//! Satisfies: T3 (Persistence must survive crash and recover)
//! Implements: TN1 resolution (Hybrid WAL + checkpoints)
//! Phase: A (Foundation) + B (Complete Durability)
//!
//! # Overview
//!
//! The WAL provides durability guarantees by writing all modifications to a
//! sequential log before applying them to the main storage. On crash recovery,
//! the log is replayed to restore the database to a consistent state.
//!
//! # Architecture
//!
//! ```text
//! ┌──────────────────────────────────────────────────────────────┐
//! │                      Write Path                               │
//! │  Transaction → WAL Entry → Sync to Disk → Apply to Storage   │
//! └──────────────────────────────────────────────────────────────┘
//!
//! ┌──────────────────────────────────────────────────────────────┐
//! │                    Recovery Path                              │
//! │  Open Storage → Read WAL → Replay Entries → Consistent State │
//! └──────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Example
//!
//! ```rust,ignore
//! use graph_d::storage::wal::{WriteAheadLog, WalEntry, WalOperation};
//!
//! // Create WAL
//! let wal = WriteAheadLog::open("data/graph.wal")?;
//!
//! // Log an operation before applying
//! let entry = WalEntry::new(WalOperation::CreateNode {
//!     id: 1,
//!     data: node_bytes,
//! });
//! wal.append(&entry)?;
//! wal.sync()?;
//!
//! // Now safe to apply to storage
//! storage.store_node(node)?;
//!
//! // On recovery
//! let entries = wal.recover()?;
//! for entry in entries {
//!     apply_entry_to_storage(&entry)?;
//! }
//! ```

use crate::error::{GraphError, Result};
use crate::graph::Id;
use std::fs::{File, OpenOptions};
use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

/// Magic bytes for WAL file identification.
/// Satisfies: Data integrity validation on recovery
const WAL_MAGIC: &[u8; 4] = b"GWAL";

/// Current WAL format version.
const WAL_VERSION: u32 = 1;

/// WAL file header size in bytes.
const HEADER_SIZE: usize = 4 + 4 + 8 + 8; // magic + version + sequence + checksum

/// Default sync mode for WAL operations.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum SyncMode {
    /// Sync after every write (safest, slowest)
    /// Satisfies: RT-1 with maximum durability
    Immediate,

    /// Sync after N operations
    /// Satisfies: TN4 resolution (ACID vs performance trade-off)
    Batched(usize),

    /// Let OS handle syncing (fastest, least durable)
    OsManaged,
}

impl Default for SyncMode {
    fn default() -> Self {
        SyncMode::Batched(100)
    }
}

/// WAL configuration options.
/// Satisfies: O1 (Support both in-memory and persistent storage)
#[derive(Debug, Clone)]
pub struct WalConfig {
    /// How often to sync WAL to disk
    pub sync_mode: SyncMode,

    /// Maximum WAL file size before rotation (Phase B feature)
    pub max_size_bytes: u64,

    /// Whether to use fsync (true) or fdatasync (false)
    pub use_fsync: bool,

    /// Checkpoint interval in number of entries (Phase B feature)
    pub checkpoint_interval: Option<u64>,
}

impl Default for WalConfig {
    fn default() -> Self {
        WalConfig {
            sync_mode: SyncMode::default(),
            max_size_bytes: 64 * 1024 * 1024, // 64MB
            use_fsync: true,
            checkpoint_interval: Some(10_000),
        }
    }
}

/// Operations that can be logged in the WAL.
/// Satisfies: B4 (ACID compliance - Atomicity through operation logging)
#[derive(Debug, Clone, PartialEq)]
pub enum WalOperation {
    /// Create a new node.
    /// Maps to: Graph::create_node()
    CreateNode {
        /// Unique identifier for the new node
        id: Id,
        /// Serialized node properties data
        data: Vec<u8>,
    },

    /// Update an existing node.
    UpdateNode {
        /// Identifier of the node to update
        id: Id,
        /// New serialized node properties data
        data: Vec<u8>,
    },

    /// Delete a node.
    DeleteNode {
        /// Identifier of the node to delete
        id: Id,
    },

    /// Create a new relationship.
    /// Maps to: Graph::create_relationship()
    CreateRelationship {
        /// Unique identifier for the new relationship
        id: Id,
        /// Source node identifier
        from_id: Id,
        /// Target node identifier
        to_id: Id,
        /// Relationship type label
        rel_type: String,
        /// Serialized relationship properties data
        data: Vec<u8>,
    },

    /// Update an existing relationship.
    UpdateRelationship {
        /// Identifier of the relationship to update
        id: Id,
        /// New serialized relationship properties data
        data: Vec<u8>,
    },

    /// Delete a relationship.
    DeleteRelationship {
        /// Identifier of the relationship to delete
        id: Id,
    },

    /// Transaction begin marker.
    /// Satisfies: B4 (Atomicity - transaction boundaries)
    BeginTransaction {
        /// Unique transaction identifier
        tx_id: u64,
    },

    /// Transaction commit marker.
    /// Satisfies: B4 (Durability - commit is durable after WAL sync)
    CommitTransaction {
        /// Transaction identifier being committed
        tx_id: u64,
    },

    /// Transaction rollback marker.
    RollbackTransaction {
        /// Transaction identifier being rolled back
        tx_id: u64,
    },

    /// Checkpoint marker (Phase B).
    /// Satisfies: TN1 resolution (Hybrid WAL + checkpoints)
    Checkpoint {
        /// WAL sequence number at checkpoint time
        sequence: u64,
        /// Timestamp when checkpoint was created (nanos since epoch)
        timestamp: u64,
    },
}

/// A single entry in the WAL.
/// Satisfies: RT-1 (each entry is self-contained for recovery)
#[derive(Debug, Clone)]
pub struct WalEntry {
    /// Monotonically increasing sequence number
    pub sequence: u64,

    /// Timestamp when entry was created (nanos since epoch)
    pub timestamp: u64,

    /// The operation being logged
    pub operation: WalOperation,

    /// CRC32 checksum for integrity validation
    pub checksum: u32,
}

impl WalEntry {
    /// Create a new WAL entry with the given operation.
    pub fn new(sequence: u64, operation: WalOperation) -> Self {
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos() as u64;

        let mut entry = WalEntry {
            sequence,
            timestamp,
            operation,
            checksum: 0,
        };

        entry.checksum = entry.calculate_checksum();
        entry
    }

    /// Calculate CRC32 checksum for integrity validation.
    /// Satisfies: Data integrity on recovery
    fn calculate_checksum(&self) -> u32 {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        self.sequence.hash(&mut hasher);
        self.timestamp.hash(&mut hasher);
        format!("{:?}", self.operation).hash(&mut hasher);
        hasher.finish() as u32
    }

    /// Validate entry integrity.
    pub fn is_valid(&self) -> bool {
        self.checksum == self.calculate_checksum()
    }

    /// Serialize entry to bytes.
    ///
    /// Format: `len:4 | sequence:8 | timestamp:8 | op_data:var | checksum:4`
    ///
    /// Note: op_data includes the op_type byte
    pub fn serialize(&self) -> Vec<u8> {
        let op_data = self.serialize_operation();
        let total_len = 8 + 8 + op_data.len() + 4;

        let mut data = Vec::with_capacity(4 + total_len);

        // Length prefix (excluding the length field itself)
        data.extend_from_slice(&(total_len as u32).to_le_bytes());

        // Sequence number
        data.extend_from_slice(&self.sequence.to_le_bytes());

        // Timestamp
        data.extend_from_slice(&self.timestamp.to_le_bytes());

        // Operation type and data
        data.extend_from_slice(&op_data);

        // Checksum
        data.extend_from_slice(&self.checksum.to_le_bytes());

        data
    }

    /// Serialize operation to bytes.
    fn serialize_operation(&self) -> Vec<u8> {
        let mut data = Vec::new();

        match &self.operation {
            WalOperation::CreateNode {
                id,
                data: node_data,
            } => {
                data.push(0x01);
                data.extend_from_slice(&id.to_le_bytes());
                data.extend_from_slice(&(node_data.len() as u32).to_le_bytes());
                data.extend_from_slice(node_data);
            }
            WalOperation::UpdateNode {
                id,
                data: node_data,
            } => {
                data.push(0x02);
                data.extend_from_slice(&id.to_le_bytes());
                data.extend_from_slice(&(node_data.len() as u32).to_le_bytes());
                data.extend_from_slice(node_data);
            }
            WalOperation::DeleteNode { id } => {
                data.push(0x03);
                data.extend_from_slice(&id.to_le_bytes());
            }
            WalOperation::CreateRelationship {
                id,
                from_id,
                to_id,
                rel_type,
                data: rel_data,
            } => {
                data.push(0x04);
                data.extend_from_slice(&id.to_le_bytes());
                data.extend_from_slice(&from_id.to_le_bytes());
                data.extend_from_slice(&to_id.to_le_bytes());
                let rel_type_bytes = rel_type.as_bytes();
                data.extend_from_slice(&(rel_type_bytes.len() as u16).to_le_bytes());
                data.extend_from_slice(rel_type_bytes);
                data.extend_from_slice(&(rel_data.len() as u32).to_le_bytes());
                data.extend_from_slice(rel_data);
            }
            WalOperation::UpdateRelationship { id, data: rel_data } => {
                data.push(0x05);
                data.extend_from_slice(&id.to_le_bytes());
                data.extend_from_slice(&(rel_data.len() as u32).to_le_bytes());
                data.extend_from_slice(rel_data);
            }
            WalOperation::DeleteRelationship { id } => {
                data.push(0x06);
                data.extend_from_slice(&id.to_le_bytes());
            }
            WalOperation::BeginTransaction { tx_id } => {
                data.push(0x10);
                data.extend_from_slice(&tx_id.to_le_bytes());
            }
            WalOperation::CommitTransaction { tx_id } => {
                data.push(0x11);
                data.extend_from_slice(&tx_id.to_le_bytes());
            }
            WalOperation::RollbackTransaction { tx_id } => {
                data.push(0x12);
                data.extend_from_slice(&tx_id.to_le_bytes());
            }
            WalOperation::Checkpoint {
                sequence,
                timestamp,
            } => {
                data.push(0x20);
                data.extend_from_slice(&sequence.to_le_bytes());
                data.extend_from_slice(&timestamp.to_le_bytes());
            }
        }

        data
    }

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

        let len = u32::from_le_bytes([data[0], data[1], data[2], data[3]]) as usize;
        if data.len() < 4 + len {
            return Err(GraphError::Storage("WAL entry truncated".to_string()));
        }

        let entry_data = &data[4..4 + len];

        let sequence = u64::from_le_bytes([
            entry_data[0],
            entry_data[1],
            entry_data[2],
            entry_data[3],
            entry_data[4],
            entry_data[5],
            entry_data[6],
            entry_data[7],
        ]);

        let timestamp = u64::from_le_bytes([
            entry_data[8],
            entry_data[9],
            entry_data[10],
            entry_data[11],
            entry_data[12],
            entry_data[13],
            entry_data[14],
            entry_data[15],
        ]);

        let (operation, op_len) = Self::deserialize_operation(&entry_data[16..])?;

        let checksum_start = 16 + op_len;
        let checksum = u32::from_le_bytes([
            entry_data[checksum_start],
            entry_data[checksum_start + 1],
            entry_data[checksum_start + 2],
            entry_data[checksum_start + 3],
        ]);

        let entry = WalEntry {
            sequence,
            timestamp,
            operation,
            checksum,
        };

        if !entry.is_valid() {
            return Err(GraphError::Storage(
                "WAL entry checksum mismatch".to_string(),
            ));
        }

        Ok((entry, 4 + len))
    }

    /// Deserialize operation from bytes.
    fn deserialize_operation(data: &[u8]) -> Result<(WalOperation, usize)> {
        if data.is_empty() {
            return Err(GraphError::Storage("Empty operation data".to_string()));
        }

        let op_type = data[0];
        let mut offset = 1;

        let operation = match op_type {
            0x01 => {
                // CreateNode
                let id = 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],
                ]);
                offset += 8;
                let data_len = u32::from_le_bytes([
                    data[offset],
                    data[offset + 1],
                    data[offset + 2],
                    data[offset + 3],
                ]) as usize;
                offset += 4;
                let node_data = data[offset..offset + data_len].to_vec();
                offset += data_len;
                WalOperation::CreateNode {
                    id,
                    data: node_data,
                }
            }
            0x03 => {
                // DeleteNode
                let id = 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],
                ]);
                offset += 8;
                WalOperation::DeleteNode { id }
            }
            0x10 => {
                // BeginTransaction
                let tx_id = 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],
                ]);
                offset += 8;
                WalOperation::BeginTransaction { tx_id }
            }
            0x11 => {
                // CommitTransaction
                let tx_id = 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],
                ]);
                offset += 8;
                WalOperation::CommitTransaction { tx_id }
            }
            0x20 => {
                // Checkpoint
                let seq = 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],
                ]);
                offset += 8;
                let ts = 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],
                ]);
                offset += 8;
                WalOperation::Checkpoint {
                    sequence: seq,
                    timestamp: ts,
                }
            }
            _ => {
                return Err(GraphError::Storage(format!(
                    "Unknown WAL operation type: {op_type}"
                )));
            }
        };

        Ok((operation, offset))
    }
}

/// Write-Ahead Log manager.
/// Satisfies: RT-1 (Data MUST survive crashes and restarts)
pub struct WriteAheadLog {
    /// Path to the WAL file
    path: PathBuf,

    /// File handle for writing
    writer: Option<BufWriter<File>>,

    /// Configuration
    config: WalConfig,

    /// Current sequence number
    sequence: AtomicU64,

    /// Number of unflushed entries (for batched sync)
    unflushed_count: AtomicU64,

    /// Last checkpoint sequence number
    last_checkpoint: AtomicU64,
}

impl WriteAheadLog {
    /// Open or create a WAL file.
    /// Satisfies: Recovery procedure on startup
    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
        Self::open_with_config(path, WalConfig::default())
    }

    /// Open or create a WAL file with custom configuration.
    pub fn open_with_config<P: AsRef<Path>>(path: P, config: WalConfig) -> Result<Self> {
        let path = path.as_ref().to_path_buf();

        let file = OpenOptions::new()
            .read(true)
            .create(true)
            .append(true)
            .open(&path)
            .map_err(|e| GraphError::Io(e.to_string()))?;

        let mut wal = WriteAheadLog {
            path,
            writer: Some(BufWriter::new(file)),
            config,
            sequence: AtomicU64::new(0),
            unflushed_count: AtomicU64::new(0),
            last_checkpoint: AtomicU64::new(0),
        };

        // Initialize or validate header
        wal.init_header()?;

        Ok(wal)
    }

    /// Initialize WAL file header.
    fn init_header(&mut self) -> Result<()> {
        let writer = self
            .writer
            .as_mut()
            .ok_or_else(|| GraphError::Storage("WAL writer not available".to_string()))?;

        let file = writer.get_ref();
        let file_len = file
            .metadata()
            .map_err(|e| GraphError::Io(e.to_string()))?
            .len();

        if file_len == 0 {
            // New file, write header
            writer
                .write_all(WAL_MAGIC)
                .map_err(|e| GraphError::Io(e.to_string()))?;
            writer
                .write_all(&WAL_VERSION.to_le_bytes())
                .map_err(|e| GraphError::Io(e.to_string()))?;
            writer
                .write_all(&0u64.to_le_bytes()) // sequence
                .map_err(|e| GraphError::Io(e.to_string()))?;
            writer
                .write_all(&0u64.to_le_bytes()) // checksum placeholder
                .map_err(|e| GraphError::Io(e.to_string()))?;
            writer.flush().map_err(|e| GraphError::Io(e.to_string()))?;
        } else {
            // Existing file, validate and recover sequence
            let recovered_seq = self.scan_for_sequence()?;
            self.sequence.store(recovered_seq, Ordering::SeqCst);
        }

        Ok(())
    }

    /// Scan WAL file to find the last sequence number.
    fn scan_for_sequence(&self) -> Result<u64> {
        let file = File::open(&self.path).map_err(|e| GraphError::Io(e.to_string()))?;
        let mut reader = BufReader::new(file);

        // Skip header
        reader
            .seek(SeekFrom::Start(HEADER_SIZE as u64))
            .map_err(|e| GraphError::Io(e.to_string()))?;

        let mut last_seq = 0u64;
        let mut buffer = vec![0u8; 4];

        // Read entries until EOF or error
        while reader.read_exact(&mut buffer).is_ok() {
            let len = u32::from_le_bytes([buffer[0], buffer[1], buffer[2], buffer[3]]) as usize;

            // Read rest of entry
            let mut entry_data = vec![0u8; len];
            if reader.read_exact(&mut entry_data).is_err() {
                break; // Truncated entry, stop here
            }

            // Extract sequence number (first 8 bytes after length)
            if len >= 8 {
                last_seq = u64::from_le_bytes([
                    entry_data[0],
                    entry_data[1],
                    entry_data[2],
                    entry_data[3],
                    entry_data[4],
                    entry_data[5],
                    entry_data[6],
                    entry_data[7],
                ]);
            }
        }

        Ok(last_seq)
    }

    /// Append an operation to the WAL.
    /// Satisfies: RT-1 (log before apply pattern)
    pub fn append(&mut self, operation: WalOperation) -> Result<u64> {
        let seq = self.sequence.fetch_add(1, Ordering::SeqCst) + 1;
        let entry = WalEntry::new(seq, operation);

        let writer = self
            .writer
            .as_mut()
            .ok_or_else(|| GraphError::Storage("WAL writer not available".to_string()))?;

        let data = entry.serialize();
        writer
            .write_all(&data)
            .map_err(|e| GraphError::Io(e.to_string()))?;

        let count = self.unflushed_count.fetch_add(1, Ordering::SeqCst) + 1;

        // Auto-sync based on config
        match self.config.sync_mode {
            SyncMode::Immediate => {
                self.sync()?;
            }
            SyncMode::Batched(batch_size) if count as usize >= batch_size => {
                self.sync()?;
            }
            _ => {}
        }

        Ok(seq)
    }

    /// Sync WAL to disk.
    /// Satisfies: RT-1 (durability guarantee after sync)
    pub fn sync(&mut self) -> Result<()> {
        let writer = self
            .writer
            .as_mut()
            .ok_or_else(|| GraphError::Storage("WAL writer not available".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()))?;
        } else {
            writer
                .get_ref()
                .sync_data()
                .map_err(|e| GraphError::Io(e.to_string()))?;
        }

        self.unflushed_count.store(0, Ordering::SeqCst);
        Ok(())
    }

    /// Recover entries from WAL file.
    /// Satisfies: RT-1 (recovery procedure on startup)
    pub fn recover(&self) -> Result<Vec<WalEntry>> {
        let file = File::open(&self.path).map_err(|e| GraphError::Io(e.to_string()))?;
        let mut reader = BufReader::new(file);

        // Skip header
        reader
            .seek(SeekFrom::Start(HEADER_SIZE as u64))
            .map_err(|e| GraphError::Io(e.to_string()))?;

        let mut entries = Vec::new();
        let mut buffer = Vec::new();

        reader
            .read_to_end(&mut buffer)
            .map_err(|e| GraphError::Io(e.to_string()))?;

        let mut offset = 0;
        while offset < buffer.len() {
            match WalEntry::deserialize(&buffer[offset..]) {
                Ok((entry, consumed)) => {
                    entries.push(entry);
                    offset += consumed;
                }
                Err(_) => {
                    // Truncated or corrupt entry, stop recovery here
                    // This is expected after a crash
                    break;
                }
            }
        }

        Ok(entries)
    }

    /// Create a checkpoint marker.
    /// Satisfies: TN1 resolution (Hybrid WAL + checkpoints)
    /// Phase B feature - placeholder for now
    pub fn checkpoint(&mut self) -> Result<u64> {
        let seq = self.sequence.load(Ordering::SeqCst);
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos() as u64;

        self.append(WalOperation::Checkpoint {
            sequence: seq,
            timestamp,
        })?;

        self.last_checkpoint.store(seq, Ordering::SeqCst);
        self.sync()?;

        Ok(seq)
    }

    /// Get current sequence number.
    pub fn current_sequence(&self) -> u64 {
        self.sequence.load(Ordering::SeqCst)
    }

    /// Get path to WAL file.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Truncate WAL after checkpoint (Phase B).
    /// Satisfies: O1 (storage efficiency)
    pub fn truncate_before(&mut self, _sequence: u64) -> Result<()> {
        // Phase B: Implement WAL truncation after checkpoint
        // For now, this is a no-op placeholder
        Ok(())
    }
}

impl Drop for WriteAheadLog {
    fn drop(&mut self) {
        // Best-effort sync on drop
        let _ = self.sync();
    }
}

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

    /// Test: WAL entry serialization round-trip
    /// Validates: Data integrity through serialization
    #[test]
    fn test_entry_serialization() {
        let entry = WalEntry::new(
            1,
            WalOperation::CreateNode {
                id: 42,
                data: vec![1, 2, 3, 4],
            },
        );

        let serialized = entry.serialize();
        let (deserialized, _) = WalEntry::deserialize(&serialized).unwrap();

        assert_eq!(deserialized.sequence, entry.sequence);
        assert!(deserialized.is_valid());
    }

    /// Test: WAL append and recover
    /// Validates: RT-1 (data survives writes and recovery)
    #[test]
    fn test_append_and_recover() {
        let temp_file = NamedTempFile::new().unwrap();
        let path = temp_file.path();

        // Write entries
        {
            let mut wal = WriteAheadLog::open(path).unwrap();
            wal.append(WalOperation::CreateNode {
                id: 1,
                data: vec![1, 2, 3],
            })
            .unwrap();
            wal.append(WalOperation::CreateNode {
                id: 2,
                data: vec![4, 5, 6],
            })
            .unwrap();
            wal.sync().unwrap();
        }

        // Recover entries
        {
            let wal = WriteAheadLog::open(path).unwrap();
            let entries = wal.recover().unwrap();
            assert_eq!(entries.len(), 2);
            assert_eq!(entries[0].sequence, 1);
            assert_eq!(entries[1].sequence, 2);
        }
    }

    /// Test: Transaction markers
    /// Validates: B4 (ACID transaction boundaries)
    #[test]
    fn test_transaction_markers() {
        let temp_file = NamedTempFile::new().unwrap();
        let path = temp_file.path();

        let mut wal = WriteAheadLog::open(path).unwrap();

        wal.append(WalOperation::BeginTransaction { tx_id: 100 })
            .unwrap();
        wal.append(WalOperation::CreateNode {
            id: 1,
            data: vec![],
        })
        .unwrap();
        wal.append(WalOperation::CommitTransaction { tx_id: 100 })
            .unwrap();
        wal.sync().unwrap();

        let entries = wal.recover().unwrap();
        assert_eq!(entries.len(), 3);

        match &entries[0].operation {
            WalOperation::BeginTransaction { tx_id } => assert_eq!(*tx_id, 100),
            _ => panic!("Expected BeginTransaction"),
        }

        match &entries[2].operation {
            WalOperation::CommitTransaction { tx_id } => assert_eq!(*tx_id, 100),
            _ => panic!("Expected CommitTransaction"),
        }
    }

    /// Test: Checkpoint creation
    /// Validates: TN1 (hybrid WAL + checkpoints)
    #[test]
    fn test_checkpoint() {
        let temp_file = NamedTempFile::new().unwrap();
        let path = temp_file.path();

        let mut wal = WriteAheadLog::open(path).unwrap();

        wal.append(WalOperation::CreateNode {
            id: 1,
            data: vec![],
        })
        .unwrap();

        let checkpoint_seq = wal.checkpoint().unwrap();
        assert!(checkpoint_seq > 0);

        let entries = wal.recover().unwrap();
        let checkpoint_entry = entries
            .iter()
            .find(|e| matches!(e.operation, WalOperation::Checkpoint { .. }));

        assert!(checkpoint_entry.is_some());
    }

    /// Test: Sequence number recovery after reopen
    /// Validates: RT-1 (state recovery after restart)
    #[test]
    fn test_sequence_recovery() {
        let temp_file = NamedTempFile::new().unwrap();
        let path = temp_file.path();

        // Write 5 entries
        {
            let mut wal = WriteAheadLog::open(path).unwrap();
            for i in 0..5 {
                wal.append(WalOperation::CreateNode {
                    id: i,
                    data: vec![],
                })
                .unwrap();
            }
            wal.sync().unwrap();
        }

        // Reopen and check sequence continues
        {
            let mut wal = WriteAheadLog::open(path).unwrap();
            let seq = wal
                .append(WalOperation::CreateNode {
                    id: 5,
                    data: vec![],
                })
                .unwrap();

            assert_eq!(seq, 6); // Should continue from 5 + 1
        }
    }
}