lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0

//! Transaction log (Write-Ahead Log / WAL) implementation.
//!
//! This module provides durable logging for transactions to ensure
//! crash recovery and atomicity.

use alloc::collections::BTreeMap;
use alloc::string::{String, ToString};
use alloc::vec::Vec;

use super::types::{
    RecoveryStats, TXN_LOG_MAGIC, TXN_LOG_VERSION, TxnError, TxnId, TxnLogEntry, TxnLogType,
    TxnOperation, TxnResultType, TxnState,
};

// ═══════════════════════════════════════════════════════════════════════════════
// LOG HEADER
// ═══════════════════════════════════════════════════════════════════════════════

/// Transaction log header.
#[derive(Debug, Clone)]
pub struct TxnLogHeader {
    /// Magic number.
    pub magic: u64,
    /// Version.
    pub version: u32,
    /// Log ID.
    pub log_id: u64,
    /// Creation timestamp.
    pub created_at: u64,
    /// Last LSN written.
    pub last_lsn: u64,
    /// Last checkpoint LSN.
    pub checkpoint_lsn: u64,
    /// Header checksum.
    pub checksum: u64,
}

impl TxnLogHeader {
    /// Create a new log header.
    pub fn new(log_id: u64, timestamp: u64) -> Self {
        let mut header = Self {
            magic: TXN_LOG_MAGIC,
            version: TXN_LOG_VERSION,
            log_id,
            created_at: timestamp,
            last_lsn: 0,
            checkpoint_lsn: 0,
            checksum: 0,
        };
        header.update_checksum();
        header
    }

    /// Update the checksum.
    pub fn update_checksum(&mut self) {
        self.checksum = self.magic
            ^ self.version as u64
            ^ self.log_id
            ^ self.created_at
            ^ self.last_lsn
            ^ self.checkpoint_lsn;
    }

    /// Verify the checksum.
    pub fn verify(&self) -> bool {
        let expected = self.magic
            ^ self.version as u64
            ^ self.log_id
            ^ self.created_at
            ^ self.last_lsn
            ^ self.checkpoint_lsn;
        expected == self.checksum
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// TRANSACTION LOG
// ═══════════════════════════════════════════════════════════════════════════════

/// Transaction log for a dataset.
#[derive(Debug)]
pub struct TxnLog {
    /// Dataset name.
    dataset: String,
    /// Log header.
    header: TxnLogHeader,
    /// Log entries (in-memory buffer).
    entries: Vec<TxnLogEntry>,
    /// Next LSN to assign.
    next_lsn: u64,
    /// Active transactions.
    active_txns: BTreeMap<TxnId, TxnState>,
    /// Pending operations by transaction.
    pending_ops: BTreeMap<TxnId, Vec<TxnOperation>>,
    /// Maximum entries to buffer before flush.
    max_buffer_entries: usize,
    /// Sync after each write.
    sync_on_write: bool,
}

impl TxnLog {
    /// Create a new transaction log.
    pub fn new(dataset: &str, log_id: u64, timestamp: u64) -> Self {
        Self {
            dataset: dataset.to_string(),
            header: TxnLogHeader::new(log_id, timestamp),
            entries: Vec::new(),
            next_lsn: 1,
            active_txns: BTreeMap::new(),
            pending_ops: BTreeMap::new(),
            max_buffer_entries: 1000,
            sync_on_write: true,
        }
    }

    /// Get the dataset name.
    pub fn dataset(&self) -> &str {
        &self.dataset
    }

    /// Get the log ID.
    pub fn log_id(&self) -> u64 {
        self.header.log_id
    }

    /// Get the current LSN.
    pub fn current_lsn(&self) -> u64 {
        self.next_lsn.saturating_sub(1)
    }

    /// Get the next LSN.
    pub fn next_lsn(&self) -> u64 {
        self.next_lsn
    }

    /// Get the checkpoint LSN.
    pub fn checkpoint_lsn(&self) -> u64 {
        self.header.checkpoint_lsn
    }

    /// Get number of buffered entries.
    pub fn buffered_entries(&self) -> usize {
        self.entries.len()
    }

    /// Get number of active transactions.
    pub fn active_txn_count(&self) -> usize {
        self.active_txns.len()
    }

    /// Check if a transaction exists.
    pub fn has_txn(&self, txn_id: TxnId) -> bool {
        self.active_txns.contains_key(&txn_id)
    }

    /// Get transaction state.
    pub fn txn_state(&self, txn_id: TxnId) -> Option<TxnState> {
        self.active_txns.get(&txn_id).copied()
    }

    /// Enable/disable sync on write.
    pub fn set_sync_on_write(&mut self, sync: bool) {
        self.sync_on_write = sync;
    }

    /// Set maximum buffer entries.
    pub fn set_max_buffer_entries(&mut self, max: usize) {
        self.max_buffer_entries = max;
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // TRANSACTION LIFECYCLE
    // ═══════════════════════════════════════════════════════════════════════════

    /// Begin a new transaction.
    pub fn begin(&mut self, txn_id: TxnId, timestamp: u64) -> TxnResultType<()> {
        if self.active_txns.contains_key(&txn_id) {
            return Err(TxnError::AlreadyExists(txn_id));
        }

        let lsn = self.allocate_lsn();
        let mut entry = TxnLogEntry::begin(txn_id, lsn, timestamp);
        entry.calculate_checksum();

        self.entries.push(entry);
        self.active_txns.insert(txn_id, TxnState::Active);
        self.pending_ops.insert(txn_id, Vec::new());

        Ok(())
    }

    /// Log an operation.
    pub fn log_operation(
        &mut self,
        txn_id: TxnId,
        op: TxnOperation,
        timestamp: u64,
    ) -> TxnResultType<u32> {
        // Check state first
        let state = self
            .active_txns
            .get(&txn_id)
            .copied()
            .ok_or(TxnError::NotFound(txn_id))?;
        if state != TxnState::Active {
            return Err(TxnError::InvalidState {
                txn_id,
                current: state,
                expected: &[TxnState::Active],
            });
        }

        // Get op_index
        let op_index = self
            .pending_ops
            .get(&txn_id)
            .map(|ops| ops.len() as u32)
            .unwrap_or(0);

        // Allocate LSN and create entry
        let lsn = self.allocate_lsn();
        let mut entry = TxnLogEntry::operation(txn_id, op_index, lsn, timestamp, op.clone());
        entry.calculate_checksum();

        self.entries.push(entry);
        if let Some(ops) = self.pending_ops.get_mut(&txn_id) {
            ops.push(op);
        }

        Ok(op_index)
    }

    /// Mark transaction as prepared.
    pub fn prepare(&mut self, txn_id: TxnId, timestamp: u64) -> TxnResultType<()> {
        // Check state first
        let state = self
            .active_txns
            .get(&txn_id)
            .copied()
            .ok_or(TxnError::NotFound(txn_id))?;
        if state != TxnState::Active {
            return Err(TxnError::InvalidState {
                txn_id,
                current: state,
                expected: &[TxnState::Active],
            });
        }

        // Allocate LSN and create entry
        let lsn = self.allocate_lsn();
        let mut entry = TxnLogEntry::prepare(txn_id, lsn, timestamp);
        entry.calculate_checksum();

        self.entries.push(entry);
        self.active_txns.insert(txn_id, TxnState::Prepared);

        Ok(())
    }

    /// Commit a transaction.
    pub fn commit(&mut self, txn_id: TxnId, timestamp: u64) -> TxnResultType<Vec<TxnOperation>> {
        // Check state first
        let state = self
            .active_txns
            .get(&txn_id)
            .copied()
            .ok_or(TxnError::NotFound(txn_id))?;
        if !matches!(state, TxnState::Active | TxnState::Prepared) {
            return Err(TxnError::InvalidState {
                txn_id,
                current: state,
                expected: &[TxnState::Active, TxnState::Prepared],
            });
        }

        // Allocate LSN and create entry
        let lsn = self.allocate_lsn();
        let mut entry = TxnLogEntry::commit(txn_id, lsn, timestamp);
        entry.calculate_checksum();

        self.entries.push(entry);
        self.active_txns.insert(txn_id, TxnState::Committed);

        // Return the operations to execute
        let ops = self.pending_ops.remove(&txn_id).unwrap_or_default();
        self.active_txns.remove(&txn_id);

        Ok(ops)
    }

    /// Abort a transaction.
    pub fn abort(&mut self, txn_id: TxnId, timestamp: u64) -> TxnResultType<Vec<TxnOperation>> {
        // Check state first
        let state = self
            .active_txns
            .get(&txn_id)
            .copied()
            .ok_or(TxnError::NotFound(txn_id))?;
        if state == TxnState::Committed {
            return Err(TxnError::InvalidState {
                txn_id,
                current: state,
                expected: &[TxnState::Active, TxnState::Prepared],
            });
        }

        // Allocate LSN and create entry
        let lsn = self.allocate_lsn();
        let mut entry = TxnLogEntry::abort(txn_id, lsn, timestamp);
        entry.calculate_checksum();

        self.entries.push(entry);
        self.active_txns.insert(txn_id, TxnState::Aborted);

        // Return the operations to rollback
        let ops = self.pending_ops.remove(&txn_id).unwrap_or_default();
        self.active_txns.remove(&txn_id);

        Ok(ops)
    }

    /// Log a rollback operation.
    pub fn log_rollback(
        &mut self,
        txn_id: TxnId,
        op_index: u32,
        op: TxnOperation,
        timestamp: u64,
    ) -> TxnResultType<()> {
        let lsn = self.allocate_lsn();
        let mut entry = TxnLogEntry::rollback(txn_id, op_index, lsn, timestamp, op);
        entry.calculate_checksum();

        self.entries.push(entry);
        Ok(())
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // CHECKPOINTING
    // ═══════════════════════════════════════════════════════════════════════════

    /// Create a checkpoint.
    pub fn checkpoint(&mut self, timestamp: u64) -> TxnResultType<u64> {
        // Don't checkpoint if there are active transactions
        // (In a real implementation, we'd handle this differently)
        if !self.active_txns.is_empty() {
            // Just update the checkpoint LSN to before active txns
            // For now, skip checkpointing
            return Ok(self.header.checkpoint_lsn);
        }

        let lsn = self.current_lsn();
        self.header.checkpoint_lsn = lsn;
        self.header.last_lsn = lsn;
        self.header.update_checksum();

        // Clear committed entries before checkpoint
        // In a real implementation, this would write to disk
        self.entries.retain(|e| e.lsn > lsn);

        Ok(lsn)
    }

    /// Flush buffered entries.
    pub fn flush(&mut self) -> TxnResultType<usize> {
        // In a real implementation, this would write to disk
        let count = self.entries.len();
        self.header.last_lsn = self.current_lsn();
        self.header.update_checksum();
        Ok(count)
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // RECOVERY
    // ═══════════════════════════════════════════════════════════════════════════

    /// Get transactions needing recovery.
    pub fn get_recovery_txns(&self) -> Vec<(TxnId, TxnState, Vec<TxnOperation>)> {
        let mut result = Vec::new();

        for (&txn_id, &state) in &self.active_txns {
            if state.is_recoverable() {
                let ops = self.pending_ops.get(&txn_id).cloned().unwrap_or_default();
                result.push((txn_id, state, ops));
            }
        }

        result
    }

    /// Replay entries from a given LSN.
    pub fn replay_from(&self, from_lsn: u64) -> impl Iterator<Item = &TxnLogEntry> {
        self.entries.iter().filter(move |e| e.lsn >= from_lsn)
    }

    /// Load entries from a byte buffer (for testing/recovery).
    pub fn load_entries(&mut self, entries: Vec<TxnLogEntry>) -> TxnResultType<RecoveryStats> {
        let mut stats = RecoveryStats::new();

        for entry in entries {
            // Verify checksum
            if !entry.verify_checksum() {
                stats.errors += 1;
                continue;
            }

            stats.log_entries_processed += 1;

            match entry.entry_type {
                TxnLogType::Begin => {
                    self.active_txns.insert(entry.txn_id, TxnState::Active);
                    self.pending_ops.insert(entry.txn_id, Vec::new());
                }
                TxnLogType::Operation => {
                    if let Some(op) = entry.operation {
                        if let Some(ops) = self.pending_ops.get_mut(&entry.txn_id) {
                            ops.push(op);
                        }
                    }
                }
                TxnLogType::Prepare => {
                    if let Some(state) = self.active_txns.get_mut(&entry.txn_id) {
                        *state = TxnState::Prepared;
                    }
                }
                TxnLogType::Commit => {
                    self.active_txns.remove(&entry.txn_id);
                    self.pending_ops.remove(&entry.txn_id);
                    stats.txns_recovered += 1;
                }
                TxnLogType::Abort => {
                    self.active_txns.remove(&entry.txn_id);
                    self.pending_ops.remove(&entry.txn_id);
                    stats.txns_rolled_back += 1;
                }
                TxnLogType::Rollback => {
                    stats.ops_undone += 1;
                }
                TxnLogType::Checkpoint => {
                    // Update checkpoint LSN
                    self.header.checkpoint_lsn = entry.lsn;
                }
            }

            // Update next LSN
            if entry.lsn >= self.next_lsn {
                self.next_lsn = entry.lsn + 1;
            }
        }

        Ok(stats)
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // INTERNAL HELPERS
    // ═══════════════════════════════════════════════════════════════════════════

    /// Allocate the next LSN.
    fn allocate_lsn(&mut self) -> u64 {
        let lsn = self.next_lsn;
        self.next_lsn += 1;
        lsn
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// TESTS
// ═══════════════════════════════════════════════════════════════════════════════

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

    fn test_log() -> TxnLog {
        TxnLog::new("test/pool", 1, 1000)
    }

    #[test]
    fn test_log_header() {
        let header = TxnLogHeader::new(1, 1000);
        assert_eq!(header.magic, TXN_LOG_MAGIC);
        assert_eq!(header.version, TXN_LOG_VERSION);
        assert!(header.verify());
    }

    #[test]
    fn test_begin_transaction() {
        let mut log = test_log();
        let txn_id = TxnId::new(1);

        log.begin(txn_id, 1000).unwrap();
        assert!(log.has_txn(txn_id));
        assert_eq!(log.txn_state(txn_id), Some(TxnState::Active));
    }

    #[test]
    fn test_begin_duplicate() {
        let mut log = test_log();
        let txn_id = TxnId::new(1);

        log.begin(txn_id, 1000).unwrap();
        assert!(log.begin(txn_id, 1001).is_err());
    }

    #[test]
    fn test_log_operation() {
        let mut log = test_log();
        let txn_id = TxnId::new(1);

        log.begin(txn_id, 1000).unwrap();

        let op = TxnOperation::Create {
            path: "/test.txt".into(),
            content: vec![1, 2, 3],
            mode: 0o644,
        };

        let op_index = log.log_operation(txn_id, op, 1001).unwrap();
        assert_eq!(op_index, 0);
    }

    #[test]
    fn test_prepare() {
        let mut log = test_log();
        let txn_id = TxnId::new(1);

        log.begin(txn_id, 1000).unwrap();
        log.prepare(txn_id, 1001).unwrap();
        assert_eq!(log.txn_state(txn_id), Some(TxnState::Prepared));
    }

    #[test]
    fn test_commit() {
        let mut log = test_log();
        let txn_id = TxnId::new(1);

        log.begin(txn_id, 1000).unwrap();

        let op = TxnOperation::Create {
            path: "/test.txt".into(),
            content: vec![1, 2, 3],
            mode: 0o644,
        };
        log.log_operation(txn_id, op, 1001).unwrap();

        let ops = log.commit(txn_id, 1002).unwrap();
        assert_eq!(ops.len(), 1);
        assert!(!log.has_txn(txn_id));
    }

    #[test]
    fn test_abort() {
        let mut log = test_log();
        let txn_id = TxnId::new(1);

        log.begin(txn_id, 1000).unwrap();

        let op = TxnOperation::Create {
            path: "/test.txt".into(),
            content: vec![1, 2, 3],
            mode: 0o644,
        };
        log.log_operation(txn_id, op, 1001).unwrap();

        let ops = log.abort(txn_id, 1002).unwrap();
        assert_eq!(ops.len(), 1);
        assert!(!log.has_txn(txn_id));
    }

    #[test]
    fn test_lsn_allocation() {
        let mut log = test_log();

        assert_eq!(log.next_lsn(), 1);

        log.begin(TxnId::new(1), 1000).unwrap();
        assert_eq!(log.next_lsn(), 2);

        log.log_operation(
            TxnId::new(1),
            TxnOperation::Mkdir {
                path: "/dir".into(),
                mode: 0o755,
            },
            1001,
        )
        .unwrap();
        assert_eq!(log.next_lsn(), 3);
    }

    #[test]
    fn test_checkpoint() {
        let mut log = test_log();
        let txn_id = TxnId::new(1);

        log.begin(txn_id, 1000).unwrap();
        log.commit(txn_id, 1001).unwrap();

        let cp_lsn = log.checkpoint(1002).unwrap();
        assert!(cp_lsn > 0);
        assert_eq!(log.checkpoint_lsn(), cp_lsn);
    }

    #[test]
    fn test_flush() {
        let mut log = test_log();

        log.begin(TxnId::new(1), 1000).unwrap();
        log.log_operation(
            TxnId::new(1),
            TxnOperation::Mkdir {
                path: "/dir".into(),
                mode: 0o755,
            },
            1001,
        )
        .unwrap();

        let count = log.flush().unwrap();
        assert_eq!(count, 2);
    }

    #[test]
    fn test_recovery_txns() {
        let mut log = test_log();

        // Active transaction
        log.begin(TxnId::new(1), 1000).unwrap();
        log.log_operation(
            TxnId::new(1),
            TxnOperation::Mkdir {
                path: "/dir".into(),
                mode: 0o755,
            },
            1001,
        )
        .unwrap();

        // Prepared transaction
        log.begin(TxnId::new(2), 1002).unwrap();
        log.prepare(TxnId::new(2), 1003).unwrap();

        let recovery = log.get_recovery_txns();
        assert_eq!(recovery.len(), 2);
    }

    #[test]
    fn test_load_entries() {
        let mut log = test_log();

        let entries = vec![
            {
                let mut e = TxnLogEntry::begin(TxnId::new(1), 1, 1000);
                e.calculate_checksum();
                e
            },
            {
                let mut e = TxnLogEntry::operation(
                    TxnId::new(1),
                    0,
                    2,
                    1001,
                    TxnOperation::Mkdir {
                        path: "/dir".into(),
                        mode: 0o755,
                    },
                );
                e.calculate_checksum();
                e
            },
            {
                let mut e = TxnLogEntry::commit(TxnId::new(1), 3, 1002);
                e.calculate_checksum();
                e
            },
        ];

        let stats = log.load_entries(entries).unwrap();
        assert_eq!(stats.log_entries_processed, 3);
        assert_eq!(stats.txns_recovered, 1);
    }

    #[test]
    fn test_replay_from() {
        let mut log = test_log();

        log.begin(TxnId::new(1), 1000).unwrap();
        log.log_operation(
            TxnId::new(1),
            TxnOperation::Mkdir {
                path: "/dir".into(),
                mode: 0o755,
            },
            1001,
        )
        .unwrap();
        log.commit(TxnId::new(1), 1002).unwrap();

        let entries: Vec<_> = log.replay_from(2).collect();
        assert_eq!(entries.len(), 2); // Operation and Commit
    }
}