featherdb_storage/

wal.rs

//! Write-Ahead Log (WAL) for durability with Group Commit support
//!
//! This module provides a WAL implementation with three sync modes:
//! - `Immediate`: Sync on every commit (safest, slowest)
//! - `GroupCommit`: Batch multiple commits into a single fsync (2-5x faster)
//! - `NoSync`: No sync - data may be lost on crash (fastest, least safe)
//!
//! Group commit works by buffering WAL records and periodically flushing them
//! to disk with a single fsync operation. This amortizes the cost of fsync
//! across multiple transactions, significantly improving throughput.

use bytes::{Buf, BufMut};
use featherdb_core::{Error, Lsn, Result, TransactionId, WalGroupCommitConfig, WalSyncMode};
use parking_lot::{Condvar, Mutex, RwLock};
use std::collections::VecDeque;
use std::fs::{File, OpenOptions};
use std::io::{BufReader, Read, Seek, SeekFrom, Write};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

/// WAL record types
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum WalRecordType {
    /// Transaction begin
    Begin = 0,
    /// Transaction commit
    Commit = 1,
    /// Transaction abort
    Abort = 2,
    /// Page write
    Write = 3,
    /// Checkpoint marker
    Checkpoint = 4,
}

impl TryFrom<u8> for WalRecordType {
    type Error = Error;

    fn try_from(value: u8) -> Result<Self> {
        match value {
            0 => Ok(WalRecordType::Begin),
            1 => Ok(WalRecordType::Commit),
            2 => Ok(WalRecordType::Abort),
            3 => Ok(WalRecordType::Write),
            4 => Ok(WalRecordType::Checkpoint),
            _ => Err(Error::CorruptedWal {
                message: format!("Invalid record type: {}", value),
            }),
        }
    }
}

/// A WAL record
#[derive(Debug, Clone)]
pub struct WalRecord {
    /// Log sequence number
    pub lsn: Lsn,
    /// Transaction ID
    pub txn_id: TransactionId,
    /// Record type
    pub record_type: WalRecordType,
    /// Previous LSN for this transaction (for undo chain)
    pub prev_lsn: Lsn,
    /// Page ID (for Write records)
    pub page_id: u64,
    /// Old page data (for undo)
    pub old_data: Vec<u8>,
    /// New page data (for redo)
    pub new_data: Vec<u8>,
}

impl WalRecord {
    /// Header size: lsn(8) + txn_id(8) + type(1) + prev_lsn(8) + page_id(8) + old_len(4) + new_len(4) + checksum(4)
    const HEADER_SIZE: usize = 45;

    /// Create a begin record
    pub fn begin(lsn: Lsn, txn_id: TransactionId) -> Self {
        WalRecord {
            lsn,
            txn_id,
            record_type: WalRecordType::Begin,
            prev_lsn: Lsn::ZERO,
            page_id: 0,
            old_data: vec![],
            new_data: vec![],
        }
    }

    /// Create a commit record
    pub fn commit(lsn: Lsn, txn_id: TransactionId, prev_lsn: Lsn) -> Self {
        WalRecord {
            lsn,
            txn_id,
            record_type: WalRecordType::Commit,
            prev_lsn,
            page_id: 0,
            old_data: vec![],
            new_data: vec![],
        }
    }

    /// Create an abort record
    pub fn abort(lsn: Lsn, txn_id: TransactionId, prev_lsn: Lsn) -> Self {
        WalRecord {
            lsn,
            txn_id,
            record_type: WalRecordType::Abort,
            prev_lsn,
            page_id: 0,
            old_data: vec![],
            new_data: vec![],
        }
    }

    /// Create a write record
    pub fn write(
        lsn: Lsn,
        txn_id: TransactionId,
        prev_lsn: Lsn,
        page_id: u64,
        old_data: Vec<u8>,
        new_data: Vec<u8>,
    ) -> Self {
        WalRecord {
            lsn,
            txn_id,
            record_type: WalRecordType::Write,
            prev_lsn,
            page_id,
            old_data,
            new_data,
        }
    }

    /// Create a checkpoint record
    pub fn checkpoint(lsn: Lsn) -> Self {
        WalRecord {
            lsn,
            txn_id: TransactionId::NONE,
            record_type: WalRecordType::Checkpoint,
            prev_lsn: Lsn::ZERO,
            page_id: 0,
            old_data: vec![],
            new_data: vec![],
        }
    }

    /// Serialize the record to bytes
    pub fn serialize(&self) -> Vec<u8> {
        let payload_size = self.old_data.len() + self.new_data.len();
        let mut buf = Vec::with_capacity(Self::HEADER_SIZE + payload_size);

        buf.put_u64_le(self.lsn.0);
        buf.put_u64_le(self.txn_id.0);
        buf.put_u8(self.record_type as u8);
        buf.put_u64_le(self.prev_lsn.0);
        buf.put_u64_le(self.page_id);
        buf.put_u32_le(self.old_data.len() as u32);
        buf.put_u32_le(self.new_data.len() as u32);
        buf.extend_from_slice(&self.old_data);
        buf.extend_from_slice(&self.new_data);

        // Compute CRC32 checksum
        let checksum = crc32(&buf);
        buf.put_u32_le(checksum);

        buf
    }

    /// Deserialize a record from bytes
    pub fn deserialize(data: &[u8]) -> Result<Self> {
        if data.len() < Self::HEADER_SIZE {
            return Err(Error::CorruptedWal {
                message: "Record too small".into(),
            });
        }

        let mut cursor = data;

        let lsn = Lsn(cursor.get_u64_le());
        let txn_id = TransactionId(cursor.get_u64_le());
        let record_type = WalRecordType::try_from(cursor.get_u8())?;
        let prev_lsn = Lsn(cursor.get_u64_le());
        let page_id = cursor.get_u64_le();
        let old_len = cursor.get_u32_le() as usize;
        let new_len = cursor.get_u32_le() as usize;

        if cursor.remaining() < old_len + new_len + 4 {
            return Err(Error::CorruptedWal {
                message: "Record data truncated".into(),
            });
        }

        let mut old_data = vec![0u8; old_len];
        cursor.copy_to_slice(&mut old_data);

        let mut new_data = vec![0u8; new_len];
        cursor.copy_to_slice(&mut new_data);

        let stored_checksum = cursor.get_u32_le();

        // Verify checksum
        let data_without_checksum = &data[..data.len() - 4];
        let computed_checksum = crc32(data_without_checksum);

        if stored_checksum != computed_checksum {
            return Err(Error::CorruptedWal {
                message: format!(
                    "Checksum mismatch: stored {}, computed {}",
                    stored_checksum, computed_checksum
                ),
            });
        }

        Ok(WalRecord {
            lsn,
            txn_id,
            record_type,
            prev_lsn,
            page_id,
            old_data,
            new_data,
        })
    }

    /// Get the total serialized size
    pub fn serialized_size(&self) -> usize {
        Self::HEADER_SIZE + self.old_data.len() + self.new_data.len()
    }
}

/// Simple CRC32 implementation
fn crc32(data: &[u8]) -> u32 {
    let mut crc: u32 = 0xFFFFFFFF;
    for byte in data {
        crc ^= *byte as u32;
        for _ in 0..8 {
            if crc & 1 != 0 {
                crc = (crc >> 1) ^ 0xEDB88320;
            } else {
                crc >>= 1;
            }
        }
    }
    !crc
}

/// Statistics for WAL operations
#[derive(Debug, Default)]
pub struct WalStats {
    /// Number of records currently buffered (for group commit)
    records_buffered: AtomicU64,
    /// Total number of group commits performed
    group_commits: AtomicU64,
    /// Total records written via group commit
    group_commit_records: AtomicU64,
    /// Total number of fsync operations
    fsync_count: AtomicU64,
    /// Total records written (all modes)
    total_records: AtomicU64,
}

impl WalStats {
    /// Create new stats
    pub fn new() -> Self {
        Self::default()
    }

    /// Get number of records currently buffered
    pub fn records_buffered(&self) -> u64 {
        self.records_buffered.load(Ordering::Relaxed)
    }

    /// Get total number of group commits
    pub fn group_commits(&self) -> u64 {
        self.group_commits.load(Ordering::Relaxed)
    }

    /// Get average batch size for group commits
    pub fn average_batch_size(&self) -> f64 {
        let commits = self.group_commits.load(Ordering::Relaxed);
        let records = self.group_commit_records.load(Ordering::Relaxed);
        if commits == 0 {
            0.0
        } else {
            records as f64 / commits as f64
        }
    }

    /// Get total fsync count
    pub fn fsync_count(&self) -> u64 {
        self.fsync_count.load(Ordering::Relaxed)
    }

    /// Get total records written
    pub fn total_records(&self) -> u64 {
        self.total_records.load(Ordering::Relaxed)
    }

    /// Get a snapshot of current statistics
    pub fn snapshot(&self) -> WalStatsSnapshot {
        WalStatsSnapshot {
            records_buffered: self.records_buffered.load(Ordering::Relaxed),
            group_commits: self.group_commits.load(Ordering::Relaxed),
            group_commit_records: self.group_commit_records.load(Ordering::Relaxed),
            fsync_count: self.fsync_count.load(Ordering::Relaxed),
            total_records: self.total_records.load(Ordering::Relaxed),
        }
    }

    /// Record a buffered record
    fn record_buffered(&self) {
        self.records_buffered.fetch_add(1, Ordering::Relaxed);
    }

    /// Record records flushed from buffer
    fn records_flushed(&self, count: u64) {
        self.records_buffered.fetch_sub(count, Ordering::Relaxed);
    }

    /// Record a group commit
    fn record_group_commit(&self, batch_size: u64) {
        self.group_commits.fetch_add(1, Ordering::Relaxed);
        self.group_commit_records
            .fetch_add(batch_size, Ordering::Relaxed);
    }

    /// Record an fsync
    fn record_fsync(&self) {
        self.fsync_count.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a written record
    fn record_written(&self) {
        self.total_records.fetch_add(1, Ordering::Relaxed);
    }
}

/// Immutable snapshot of WAL statistics
#[derive(Debug, Clone, Copy)]
pub struct WalStatsSnapshot {
    pub records_buffered: u64,
    pub group_commits: u64,
    pub group_commit_records: u64,
    pub fsync_count: u64,
    pub total_records: u64,
}

impl WalStatsSnapshot {
    /// Get average batch size for group commits
    pub fn average_batch_size(&self) -> f64 {
        if self.group_commits == 0 {
            0.0
        } else {
            self.group_commit_records as f64 / self.group_commits as f64
        }
    }
}

/// A pending commit waiting for group flush
struct PendingCommit {
    /// The LSN of this commit
    lsn: Lsn,
    /// Whether this commit has been synced
    synced: bool,
}

/// Internal buffer for group commit
struct WalBuffer {
    /// Buffered records waiting to be flushed
    records: VecDeque<WalRecord>,
    /// Pending commits waiting for sync notification
    pending_commits: VecDeque<PendingCommit>,
    /// LSN up to which data has been synced
    synced_lsn: Lsn,
}

impl WalBuffer {
    fn new() -> Self {
        WalBuffer {
            records: VecDeque::new(),
            pending_commits: VecDeque::new(),
            synced_lsn: Lsn::ZERO,
        }
    }

    fn is_empty(&self) -> bool {
        self.records.is_empty()
    }

    fn len(&self) -> usize {
        self.records.len()
    }
}

/// Background flusher thread state
struct FlushWorker {
    /// Shared buffer
    buffer: Arc<Mutex<WalBuffer>>,
    /// Condition variable for waking the flusher
    flush_cond: Arc<Condvar>,
    /// Condition variable for notifying waiters
    sync_cond: Arc<Condvar>,
    /// WAL file
    file: Arc<Mutex<File>>,
    /// Configuration
    config: WalGroupCommitConfig,
    /// Statistics
    stats: Arc<WalStats>,
    /// Shutdown flag
    shutdown: Arc<AtomicBool>,
}

impl FlushWorker {
    fn run(&self) {
        let interval = Duration::from_millis(self.config.group_commit_interval_ms);
        let mut last_flush = Instant::now();

        loop {
            // Check for shutdown
            if self.shutdown.load(Ordering::Relaxed) {
                // Flush any remaining records before shutting down
                self.flush_buffer();
                break;
            }

            // Wait for records or timeout
            {
                let mut buffer = self.buffer.lock();
                let should_flush = !buffer.is_empty()
                    && (buffer.len() >= self.config.group_commit_max_batch
                        || last_flush.elapsed() >= interval);

                if buffer.is_empty() {
                    // Wait for new records with timeout
                    self.flush_cond.wait_for(&mut buffer, interval);
                } else if !should_flush {
                    // Wait for more records or timeout
                    let remaining = interval.saturating_sub(last_flush.elapsed());
                    if !remaining.is_zero() {
                        self.flush_cond.wait_for(&mut buffer, remaining);
                    }
                }
            }

            // Flush if we have records
            let flushed = self.flush_buffer();
            if flushed {
                last_flush = Instant::now();
            }
        }
    }

    fn flush_buffer(&self) -> bool {
        let mut buffer = self.buffer.lock();

        if buffer.is_empty() {
            return false;
        }

        // Take all records from the buffer
        let records: Vec<_> = buffer.records.drain(..).collect();
        let record_count = records.len() as u64;

        // Update stats
        self.stats.records_flushed(record_count);

        // Serialize all records
        let mut data = Vec::new();
        let mut max_lsn = Lsn::ZERO;

        for record in &records {
            let serialized = record.serialize();
            let len = serialized.len() as u32;
            data.extend_from_slice(&len.to_le_bytes());
            data.extend_from_slice(&serialized);
            if record.lsn > max_lsn {
                max_lsn = record.lsn;
            }
        }

        // Release buffer lock before I/O
        drop(buffer);

        // Write all records at once
        {
            let mut file = self.file.lock();
            if let Err(e) = file.seek(SeekFrom::End(0)) {
                eprintln!("WAL seek error: {}", e);
                return false;
            }
            if let Err(e) = file.write_all(&data) {
                eprintln!("WAL write error: {}", e);
                return false;
            }
            // Single fsync for all records
            if let Err(e) = file.sync_all() {
                eprintln!("WAL sync error: {}", e);
                return false;
            }
        }

        // Update stats
        self.stats.record_fsync();
        self.stats.record_group_commit(record_count);

        // Update synced LSN and notify waiters
        {
            let mut buffer = self.buffer.lock();
            buffer.synced_lsn = max_lsn;

            // Mark all pending commits up to this LSN as synced
            for pending in buffer.pending_commits.iter_mut() {
                if pending.lsn <= max_lsn {
                    pending.synced = true;
                }
            }
        }

        // Notify all waiters
        self.sync_cond.notify_all();

        true
    }
}

/// Write-Ahead Log manager with group commit support
pub struct Wal {
    /// WAL file
    file: Arc<Mutex<File>>,
    /// Current LSN (protected by RwLock for concurrent reads)
    current_lsn: RwLock<Lsn>,
    /// File path (kept for diagnostics/logging)
    #[allow(dead_code)]
    path: std::path::PathBuf,
    /// Configuration
    config: WalGroupCommitConfig,
    /// Group commit buffer
    buffer: Arc<Mutex<WalBuffer>>,
    /// Condition variable for waking the flusher
    flush_cond: Arc<Condvar>,
    /// Condition variable for notifying commit waiters
    sync_cond: Arc<Condvar>,
    /// Background flush thread handle
    flush_thread: Option<JoinHandle<()>>,
    /// Shutdown flag
    shutdown: Arc<AtomicBool>,
    /// Statistics
    stats: Arc<WalStats>,
    /// Maximum WAL file size in bytes (None = unlimited)
    max_size: Option<u64>,
    /// Current WAL file size in bytes (approximate, tracked via writes)
    current_size: AtomicU64,
}

impl Wal {
    /// Open or create a WAL file with default configuration
    pub fn open(path: &Path) -> Result<Self> {
        Self::open_with_config(path, WalGroupCommitConfig::default(), None)
    }

    /// Open or create a WAL file with custom configuration
    pub fn open_with_config(
        path: &Path,
        config: WalGroupCommitConfig,
        max_size: Option<u64>,
    ) -> Result<Self> {
        let file = OpenOptions::new()
            .read(true)
            .create(true)
            .append(true)
            .open(path)?;

        // Find the last LSN by scanning the file
        let current_lsn = Self::find_last_lsn(&file)?;

        // Get current file size
        let current_size = file.metadata()?.len();

        let file = Arc::new(Mutex::new(file));
        let buffer = Arc::new(Mutex::new(WalBuffer::new()));
        let flush_cond = Arc::new(Condvar::new());
        let sync_cond = Arc::new(Condvar::new());
        let shutdown = Arc::new(AtomicBool::new(false));
        let stats = Arc::new(WalStats::new());

        // Start background flush thread for group commit mode
        let flush_thread = if config.sync_mode == WalSyncMode::GroupCommit {
            let worker = FlushWorker {
                buffer: buffer.clone(),
                flush_cond: flush_cond.clone(),
                sync_cond: sync_cond.clone(),
                file: file.clone(),
                config: config.clone(),
                stats: stats.clone(),
                shutdown: shutdown.clone(),
            };

            Some(thread::spawn(move || {
                worker.run();
            }))
        } else {
            None
        };

        Ok(Wal {
            file,
            current_lsn: RwLock::new(current_lsn),
            path: path.to_path_buf(),
            config,
            buffer,
            flush_cond,
            sync_cond,
            flush_thread,
            shutdown,
            stats,
            max_size,
            current_size: AtomicU64::new(current_size),
        })
    }

    /// Find the last LSN in the WAL file
    fn find_last_lsn(file: &File) -> Result<Lsn> {
        let mut reader = BufReader::new(file);
        let file_len = reader.seek(SeekFrom::End(0))?;

        if file_len == 0 {
            return Ok(Lsn(1)); // Start from 1
        }

        reader.seek(SeekFrom::Start(0))?;

        let mut last_lsn = Lsn(0);
        let mut buf = Vec::new();

        loop {
            // Read record length prefix (4 bytes)
            let mut len_buf = [0u8; 4];
            match reader.read_exact(&mut len_buf) {
                Ok(_) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
                Err(e) => return Err(e.into()),
            }

            let record_len = u32::from_le_bytes(len_buf) as usize;

            // Read record data
            buf.resize(record_len, 0);
            match reader.read_exact(&mut buf) {
                Ok(_) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
                Err(e) => return Err(e.into()),
            }

            // Parse to get LSN
            if let Ok(record) = WalRecord::deserialize(&buf) {
                last_lsn = record.lsn;
            }
        }

        Ok(Lsn(last_lsn.0 + 1))
    }

    /// Get the current LSN
    pub fn current_lsn(&self) -> Lsn {
        *self.current_lsn.read()
    }

    /// Check if WAL has records that need recovery (non-empty WAL)
    pub fn needs_recovery(&self) -> bool {
        self.current_size.load(Ordering::Relaxed) > 0
    }

    /// Get WAL statistics
    pub fn stats(&self) -> &Arc<WalStats> {
        &self.stats
    }

    /// Get the sync mode
    pub fn sync_mode(&self) -> WalSyncMode {
        self.config.sync_mode
    }

    /// Get current WAL file size in bytes
    pub fn current_size(&self) -> u64 {
        self.current_size.load(Ordering::Relaxed)
    }

    /// Get configured maximum WAL size (None if unlimited)
    pub fn max_size(&self) -> Option<u64> {
        self.max_size
    }

    /// Check if WAL write would exceed limit
    fn check_wal_limit(&self, write_size: u64) -> Result<()> {
        if let Some(limit) = self.max_size {
            let current = self.current_size.load(Ordering::Relaxed);
            if current.saturating_add(write_size) > limit {
                return Err(Error::WalLimitExceeded {
                    current_bytes: current,
                    limit_bytes: limit,
                });
            }
        }
        Ok(())
    }

    /// Append a record to the WAL (internal, doesn't sync)
    fn append_internal(&self, mut record: WalRecord) -> Result<Lsn> {
        let mut current_lsn = self.current_lsn.write();
        record.lsn = *current_lsn;
        let lsn = *current_lsn;
        *current_lsn = current_lsn.next();

        // Calculate the write size (length prefix + serialized data)
        let serialized = record.serialize();
        let write_size = (4 + serialized.len()) as u64; // 4 bytes for length prefix

        // Check WAL size limit before writing
        self.check_wal_limit(write_size)?;

        self.stats.record_written();

        match self.config.sync_mode {
            WalSyncMode::Immediate => {
                // Write directly to file
                let len = serialized.len() as u32;

                let mut file = self.file.lock();
                file.seek(SeekFrom::End(0))?;
                file.write_all(&len.to_le_bytes())?;
                file.write_all(&serialized)?;

                // Update size counter
                self.current_size.fetch_add(write_size, Ordering::Relaxed);
            }
            WalSyncMode::GroupCommit => {
                // Add to buffer
                let mut buffer = self.buffer.lock();
                buffer.records.push_back(record);
                self.stats.record_buffered();

                // Update size counter (estimate - actual write happens in flush)
                self.current_size.fetch_add(write_size, Ordering::Relaxed);

                // Wake flusher if batch is full
                if buffer.len() >= self.config.group_commit_max_batch {
                    self.flush_cond.notify_one();
                }
            }
            WalSyncMode::NoSync => {
                // Write directly to file without sync
                let len = serialized.len() as u32;

                let mut file = self.file.lock();
                file.seek(SeekFrom::End(0))?;
                file.write_all(&len.to_le_bytes())?;
                file.write_all(&serialized)?;

                // Update size counter
                self.current_size.fetch_add(write_size, Ordering::Relaxed);
            }
        }

        Ok(lsn)
    }

    /// Append a record to the WAL
    pub fn append(&mut self, record: WalRecord) -> Result<Lsn> {
        self.append_internal(record)
    }

    /// Sync the WAL to disk
    pub fn sync(&self) -> Result<()> {
        match self.config.sync_mode {
            WalSyncMode::Immediate | WalSyncMode::NoSync => {
                self.file.lock().sync_all()?;
                self.stats.record_fsync();
            }
            WalSyncMode::GroupCommit => {
                // Force a flush of the buffer
                self.flush_cond.notify_one();

                // Wait for sync to complete
                let current_lsn = self.current_lsn();
                let mut buffer = self.buffer.lock();
                while buffer.synced_lsn < current_lsn && !buffer.is_empty() {
                    self.sync_cond.wait(&mut buffer);
                }
            }
        }
        Ok(())
    }

    /// Flush the group commit buffer (for group commit mode)
    /// Returns the LSN up to which records were synced
    pub fn flush_group(&self) -> Result<Lsn> {
        if self.config.sync_mode != WalSyncMode::GroupCommit {
            // In non-group-commit mode, just sync
            self.sync()?;
            return Ok(self.current_lsn());
        }

        // Wake the flusher
        self.flush_cond.notify_one();

        // Wait for the flush to complete
        let target_lsn = {
            let current_lsn = self.current_lsn.read();
            Lsn(current_lsn.0.saturating_sub(1)) // Last written LSN
        };

        let mut buffer = self.buffer.lock();
        while buffer.synced_lsn < target_lsn {
            self.sync_cond.wait(&mut buffer);
        }

        Ok(buffer.synced_lsn)
    }

    /// Truncate the WAL (after checkpoint)
    pub fn truncate(&mut self) -> Result<()> {
        // Ensure all buffered records are flushed first
        if self.config.sync_mode == WalSyncMode::GroupCommit {
            self.flush_group()?;
        }

        let file = self.file.lock();
        file.set_len(0)?;
        file.sync_all()?;
        *self.current_lsn.write() = Lsn(1);

        // Reset size counter
        self.current_size.store(0, Ordering::Relaxed);

        // Clear the buffer
        let mut buffer = self.buffer.lock();
        buffer.records.clear();
        buffer.pending_commits.clear();
        buffer.synced_lsn = Lsn::ZERO;

        Ok(())
    }

    /// Iterate over all records in the WAL
    pub fn iter(&self) -> Result<WalIterator> {
        let file = self.file.lock();
        let mut reader = BufReader::new(file.try_clone()?);
        reader.seek(SeekFrom::Start(0))?;

        Ok(WalIterator { reader })
    }

    /// Log a transaction begin
    pub fn log_begin(&mut self, txn_id: TransactionId) -> Result<Lsn> {
        let record = WalRecord::begin(Lsn::ZERO, txn_id);
        self.append(record)
    }

    /// Log a transaction commit
    ///
    /// In GroupCommit mode, this adds the commit to the buffer and waits
    /// for the group flush to complete before returning.
    pub fn log_commit(&mut self, txn_id: TransactionId, prev_lsn: Lsn) -> Result<Lsn> {
        let record = WalRecord::commit(Lsn::ZERO, txn_id, prev_lsn);
        let lsn = self.append(record)?;

        match self.config.sync_mode {
            WalSyncMode::Immediate => {
                self.file.lock().sync_all()?;
                self.stats.record_fsync();
            }
            WalSyncMode::GroupCommit => {
                // Add to pending commits and wait for sync
                {
                    let mut buffer = self.buffer.lock();
                    buffer
                        .pending_commits
                        .push_back(PendingCommit { lsn, synced: false });
                }

                // Wake the flusher
                self.flush_cond.notify_one();

                // Wait for this commit to be synced
                let mut buffer = self.buffer.lock();
                loop {
                    // Check if our commit is synced
                    if buffer.synced_lsn >= lsn {
                        // Remove our pending commit
                        buffer.pending_commits.retain(|p| p.lsn != lsn);
                        break;
                    }
                    self.sync_cond.wait(&mut buffer);
                }
            }
            WalSyncMode::NoSync => {
                // No sync needed
            }
        }

        Ok(lsn)
    }

    /// Log a transaction abort
    pub fn log_abort(&mut self, txn_id: TransactionId, prev_lsn: Lsn) -> Result<Lsn> {
        let record = WalRecord::abort(Lsn::ZERO, txn_id, prev_lsn);
        self.append(record)
    }

    /// Log a page write
    pub fn log_write(
        &mut self,
        txn_id: TransactionId,
        prev_lsn: Lsn,
        page_id: u64,
        old_data: Vec<u8>,
        new_data: Vec<u8>,
    ) -> Result<Lsn> {
        let record = WalRecord::write(Lsn::ZERO, txn_id, prev_lsn, page_id, old_data, new_data);
        self.append(record)
    }

    /// Log a checkpoint
    pub fn log_checkpoint(&mut self) -> Result<Lsn> {
        let record = WalRecord::checkpoint(Lsn::ZERO);
        let lsn = self.append(record)?;

        // Always sync on checkpoint
        match self.config.sync_mode {
            WalSyncMode::GroupCommit => {
                self.flush_group()?;
            }
            _ => {
                self.file.lock().sync_all()?;
                self.stats.record_fsync();
            }
        }

        Ok(lsn)
    }

    /// Shutdown the WAL gracefully
    pub fn shutdown(&mut self) -> Result<()> {
        // Signal shutdown
        self.shutdown.store(true, Ordering::Relaxed);

        // Wake the flusher so it can flush remaining records and exit
        self.flush_cond.notify_one();

        // Wait for the flush thread to finish
        if let Some(handle) = self.flush_thread.take() {
            handle
                .join()
                .map_err(|_| Error::Internal("Flush thread panicked".into()))?;
        }

        // Final sync
        self.file.lock().sync_all()?;

        Ok(())
    }
}

impl Drop for Wal {
    fn drop(&mut self) {
        // Try to shutdown gracefully
        let _ = self.shutdown();
    }
}

/// Iterator over WAL records
pub struct WalIterator {
    reader: BufReader<File>,
}

impl Iterator for WalIterator {
    type Item = Result<WalRecord>;

    fn next(&mut self) -> Option<Self::Item> {
        // Read length prefix
        let mut len_buf = [0u8; 4];
        match self.reader.read_exact(&mut len_buf) {
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return None,
            Err(e) => return Some(Err(e.into())),
        }

        let record_len = u32::from_le_bytes(len_buf) as usize;

        // Read record data
        let mut buf = vec![0u8; record_len];
        match self.reader.read_exact(&mut buf) {
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return None,
            Err(e) => return Some(Err(e.into())),
        }

        Some(WalRecord::deserialize(&buf))
    }
}

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

    #[test]
    fn test_wal_record_serialization() {
        let record = WalRecord::write(
            Lsn(1),
            TransactionId(100),
            Lsn::ZERO,
            42,
            vec![1, 2, 3],
            vec![4, 5, 6],
        );

        let serialized = record.serialize();
        let deserialized = WalRecord::deserialize(&serialized).unwrap();

        assert_eq!(record.lsn, deserialized.lsn);
        assert_eq!(record.txn_id, deserialized.txn_id);
        assert_eq!(record.record_type, deserialized.record_type);
        assert_eq!(record.page_id, deserialized.page_id);
        assert_eq!(record.old_data, deserialized.old_data);
        assert_eq!(record.new_data, deserialized.new_data);
    }

    #[test]
    fn test_wal_append_iterate() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test.wal");

        let mut wal = Wal::open(&wal_path).unwrap();

        // Append some records
        let lsn1 = wal.log_begin(TransactionId(1)).unwrap();
        let lsn2 = wal
            .log_write(TransactionId(1), lsn1, 10, vec![0], vec![1])
            .unwrap();
        let _lsn3 = wal.log_commit(TransactionId(1), lsn2).unwrap();

        // Iterate
        let records: Vec<_> = wal.iter().unwrap().collect();
        assert_eq!(records.len(), 3);

        assert_eq!(
            records[0].as_ref().unwrap().record_type,
            WalRecordType::Begin
        );
        assert_eq!(
            records[1].as_ref().unwrap().record_type,
            WalRecordType::Write
        );
        assert_eq!(
            records[2].as_ref().unwrap().record_type,
            WalRecordType::Commit
        );
    }

    #[test]
    fn test_wal_recovery() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test.wal");

        // Write some records
        {
            let mut wal = Wal::open(&wal_path).unwrap();
            wal.log_begin(TransactionId(1)).unwrap();
            wal.sync().unwrap();
        }

        // Reopen and check
        {
            let wal = Wal::open(&wal_path).unwrap();
            let records: Vec<_> = wal.iter().unwrap().collect();
            assert_eq!(records.len(), 1);
        }
    }

    #[test]
    fn test_wal_group_commit() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test_group.wal");

        let config = WalGroupCommitConfig {
            sync_mode: WalSyncMode::GroupCommit,
            group_commit_interval_ms: 50,
            group_commit_max_batch: 10,
        };

        let mut wal = Wal::open_with_config(&wal_path, config, None).unwrap();

        // Write multiple transactions
        for i in 1..=5 {
            let txn_id = TransactionId(i);
            let lsn1 = wal.log_begin(txn_id).unwrap();
            let lsn2 = wal
                .log_write(txn_id, lsn1, i * 10, vec![0], vec![1])
                .unwrap();
            wal.log_commit(txn_id, lsn2).unwrap();
        }

        // Verify all records are written
        let records: Vec<_> = wal.iter().unwrap().collect();
        assert_eq!(records.len(), 15); // 5 transactions * 3 records each

        // Check stats
        let stats = wal.stats();
        assert!(stats.group_commits() >= 1);
        assert_eq!(stats.total_records(), 15);

        // Shutdown gracefully
        wal.shutdown().unwrap();
    }

    #[test]
    fn test_wal_no_sync_mode() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test_nosync.wal");

        let config = WalGroupCommitConfig {
            sync_mode: WalSyncMode::NoSync,
            ..Default::default()
        };

        let mut wal = Wal::open_with_config(&wal_path, config, None).unwrap();

        // Write some records
        let lsn1 = wal.log_begin(TransactionId(1)).unwrap();
        let lsn2 = wal
            .log_write(TransactionId(1), lsn1, 10, vec![0], vec![1])
            .unwrap();
        wal.log_commit(TransactionId(1), lsn2).unwrap();

        // No fsyncs should have been done for commit
        let stats = wal.stats();
        assert_eq!(stats.fsync_count(), 0);

        // Manual sync
        wal.sync().unwrap();
        assert_eq!(stats.fsync_count(), 1);
    }

    #[test]
    fn test_wal_stats() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test_stats.wal");

        let mut wal = Wal::open(&wal_path).unwrap();

        let lsn1 = wal.log_begin(TransactionId(1)).unwrap();
        let lsn2 = wal
            .log_write(TransactionId(1), lsn1, 10, vec![0], vec![1])
            .unwrap();
        wal.log_commit(TransactionId(1), lsn2).unwrap();

        let stats = wal.stats();
        assert_eq!(stats.total_records(), 3);
        // In immediate mode, we sync on every commit
        assert!(stats.fsync_count() >= 1);
    }

    #[test]
    fn test_wal_batch_flush() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test_batch.wal");

        let config = WalGroupCommitConfig {
            sync_mode: WalSyncMode::GroupCommit,
            group_commit_interval_ms: 1000, // Long interval
            group_commit_max_batch: 5,      // Small batch size
        };

        let mut wal = Wal::open_with_config(&wal_path, config, None).unwrap();

        // Write enough records to trigger batch flush
        for i in 1..=3 {
            let txn_id = TransactionId(i);
            let lsn1 = wal.log_begin(txn_id).unwrap();
            wal.log_commit(txn_id, lsn1).unwrap();
        }

        // Records should be flushed due to commits
        let records: Vec<_> = wal.iter().unwrap().collect();
        assert_eq!(records.len(), 6); // 3 transactions * 2 records each

        wal.shutdown().unwrap();
    }

    #[test]
    fn test_wal_size_limit() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test_limited.wal");

        // Create WAL with a small size limit (10KB)
        let max_size = Some(10 * 1024);
        let mut wal =
            Wal::open_with_config(&wal_path, WalGroupCommitConfig::default(), max_size).unwrap();

        // Write records until we hit the limit
        let mut count = 0;
        loop {
            let txn_id = TransactionId(count + 1);
            match wal.log_begin(txn_id) {
                Ok(_) => {
                    count += 1;
                    // Safety check
                    if count > 1000 {
                        panic!("Should have hit WAL limit by now");
                    }
                }
                Err(Error::WalLimitExceeded { .. }) => {
                    // Expected error
                    break;
                }
                Err(e) => {
                    panic!("Unexpected error: {}", e);
                }
            }
        }

        assert!(count > 0, "Should have written at least some records");
        assert!(count < 1000, "Should have hit limit before 1000 records");

        wal.shutdown().unwrap();
    }

    #[test]
    fn test_wal_unlimited() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test_unlimited.wal");

        // Create WAL with no size limit
        let mut wal =
            Wal::open_with_config(&wal_path, WalGroupCommitConfig::default(), None).unwrap();

        // Write many records - should not hit limit
        for i in 1..=100 {
            let txn_id = TransactionId(i);
            let lsn1 = wal.log_begin(txn_id).unwrap();
            wal.log_commit(txn_id, lsn1).unwrap();
        }

        wal.shutdown().unwrap();
    }

    #[test]
    fn test_wal_truncate_resets_size() {
        let tmp = TempDir::new().unwrap();
        let wal_path = tmp.path().join("test_truncate.wal");

        let max_size = Some(100 * 1024); // 100KB
        let mut wal =
            Wal::open_with_config(&wal_path, WalGroupCommitConfig::default(), max_size).unwrap();

        // Write some records
        for i in 1..=10 {
            let txn_id = TransactionId(i);
            let lsn1 = wal.log_begin(txn_id).unwrap();
            wal.log_commit(txn_id, lsn1).unwrap();
        }

        let size_before = wal.current_size.load(Ordering::Relaxed);
        assert!(size_before > 0);

        // Truncate
        wal.truncate().unwrap();

        let size_after = wal.current_size.load(Ordering::Relaxed);
        assert_eq!(size_after, 0);

        // Should be able to write again after truncate
        let lsn1 = wal.log_begin(TransactionId(100)).unwrap();
        wal.log_commit(TransactionId(100), lsn1).unwrap();

        wal.shutdown().unwrap();
    }
}