sochdb_core/

txn.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! Transaction Manager for ACID Transactions
//!
//! Provides ACID guarantees using WAL-based transaction management:
//! - Atomicity: All writes in a transaction succeed or fail together
//! - Consistency: Transactions move database from valid state to valid state
//! - Isolation: MVCC snapshot isolation for concurrent transactions
//! - Durability: Committed transactions survive crashes via WAL

use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::sync::atomic::{AtomicU64, Ordering};

/// Transaction ID - monotonically increasing
pub type TxnId = u64;

/// Transaction states
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum TxnState {
    Active,
    Committed,
    Aborted,
}

/// WAL record types for ACID transactions (ARIES-style)
///
/// This is the **canonical** definition. All crates SHOULD use this type
/// via `use sochdb_core::txn::WalRecordType` rather than defining their own.
///
/// # Discriminant Scheme
///
/// Values use hex categories for logical grouping:
/// - `0x01-0x0F`: Data operations
/// - `0x10-0x1F`: Transaction lifecycle
/// - `0x20-0x2F`: Checkpoint/recovery
/// - `0x30-0x3F`: Schema operations  
/// - `0x40-0x4F`: ARIES-specific records
/// - `0x50-0x5F`: LSM/compaction operations
/// - `0x60-0x6F`: Atomic batch operations
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum WalRecordType {
    /// Data write within transaction
    Data = 0x01,
    /// Page update with before/after images
    PageUpdate = 0x02,
    /// Delete operation
    Delete = 0x03,
    /// Transaction begin marker
    TxnBegin = 0x10,
    /// Transaction commit marker
    TxnCommit = 0x11,
    /// Transaction abort marker
    TxnAbort = 0x12,
    /// Savepoint within a transaction
    Savepoint = 0x13,
    /// Rollback to savepoint
    RollbackToSavepoint = 0x14,
    /// Transaction end (resource cleanup after commit/abort)
    TxnEnd = 0x15,
    /// Checkpoint for recovery optimization
    Checkpoint = 0x20,
    /// End of checkpoint (contains active transactions and dirty pages)
    CheckpointEnd = 0x21,
    /// Schema change (DDL)
    SchemaChange = 0x30,
    /// Compensation Log Record (CLR) for ARIES undo operations
    CompensationLogRecord = 0x40,
    /// LSM compaction record
    Compaction = 0x50,
    /// LSM flush (memtable to SSTable)
    Flush = 0x51,
    /// Atomic batch begin
    BatchBegin = 0x60,
    /// Atomic batch commit  
    BatchCommit = 0x61,
}

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

    fn try_from(value: u8) -> Result<Self, Self::Error> {
        match value {
            0x01 => Ok(WalRecordType::Data),
            0x02 => Ok(WalRecordType::PageUpdate),
            0x03 => Ok(WalRecordType::Delete),
            0x10 => Ok(WalRecordType::TxnBegin),
            0x11 => Ok(WalRecordType::TxnCommit),
            0x12 => Ok(WalRecordType::TxnAbort),
            0x13 => Ok(WalRecordType::Savepoint),
            0x14 => Ok(WalRecordType::RollbackToSavepoint),
            0x15 => Ok(WalRecordType::TxnEnd),
            0x20 => Ok(WalRecordType::Checkpoint),
            0x21 => Ok(WalRecordType::CheckpointEnd),
            0x30 => Ok(WalRecordType::SchemaChange),
            0x40 => Ok(WalRecordType::CompensationLogRecord),
            0x50 => Ok(WalRecordType::Compaction),
            0x51 => Ok(WalRecordType::Flush),
            0x60 => Ok(WalRecordType::BatchBegin),
            0x61 => Ok(WalRecordType::BatchCommit),
            _ => Err(()),
        }
    }
}

/// Log Sequence Number (LSN) for ARIES recovery
///
/// LSN ordering guarantee: If LSN(A) < LSN(B), then A happened before B in the WAL.
/// This is critical for:
/// - Redo: Only redo operations where page_lsn < record_lsn
/// - Undo: Process undo in reverse LSN order
pub type Lsn = u64;

/// Page ID for tracking dirty pages
pub type PageId = u64;

/// ARIES transaction table entry for recovery
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AriesTransactionEntry {
    /// Transaction ID
    pub txn_id: TxnId,
    /// Transaction state during recovery
    pub state: TxnState,
    /// LSN of last log record for this transaction
    pub last_lsn: Lsn,
    /// LSN to undo next (for rollback)
    pub undo_next_lsn: Option<Lsn>,
}

/// ARIES dirty page table entry for recovery
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AriesDirtyPageEntry {
    /// Page ID
    pub page_id: PageId,
    /// Recovery LSN - first LSN that might have dirtied this page
    pub rec_lsn: Lsn,
}

/// Checkpoint data for ARIES recovery
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct AriesCheckpointData {
    /// Active transactions at checkpoint time
    pub active_transactions: Vec<AriesTransactionEntry>,
    /// Dirty pages at checkpoint time
    pub dirty_pages: Vec<AriesDirtyPageEntry>,
    /// LSN where checkpoint started
    pub begin_checkpoint_lsn: Lsn,
}

/// A write operation buffered in a transaction
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TxnWrite {
    /// Key being written
    pub key: Vec<u8>,
    /// Value being written (None for delete)
    pub value: Option<Vec<u8>>,
    /// Table/collection this write belongs to
    pub table: String,
}

/// A read operation recorded for conflict detection
#[derive(Debug, Clone, Hash, PartialEq, Eq)]
pub struct TxnRead {
    pub key: Vec<u8>,
    pub table: String,
}

/// WAL entry with ARIES transaction support
///
/// Extends standard WAL entries with ARIES-specific fields:
/// - LSN: Log Sequence Number for ordering and idempotent recovery
/// - prev_lsn: Previous LSN for this transaction (undo chain)
/// - undo_info: Before-image for undo operations
/// - page_id: Page affected by this operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TxnWalEntry {
    /// Type of this WAL record
    pub record_type: WalRecordType,
    /// Transaction ID
    pub txn_id: TxnId,
    /// Timestamp in microseconds
    pub timestamp_us: u64,
    /// Optional key for data records
    pub key: Option<Vec<u8>>,
    /// Optional value for data records (after-image)
    pub value: Option<Vec<u8>>,
    /// Optional table name
    pub table: Option<String>,
    /// CRC32 checksum
    pub checksum: u32,
    /// ARIES: Log Sequence Number (assigned when appended to WAL)
    #[serde(default)]
    pub lsn: Lsn,
    /// ARIES: Previous LSN in this transaction's chain (for undo)
    #[serde(default)]
    pub prev_lsn: Option<Lsn>,
    /// ARIES: Page ID affected by this record
    #[serde(default)]
    pub page_id: Option<PageId>,
    /// ARIES: Before-image for undo (original value before update)
    #[serde(default)]
    pub undo_info: Option<Vec<u8>>,
    /// ARIES: For CLRs, the next LSN to undo (skips compensated operations)
    #[serde(default)]
    pub undo_next_lsn: Option<Lsn>,
}

impl TxnWalEntry {
    pub fn new_begin(txn_id: TxnId, timestamp_us: u64) -> Self {
        Self {
            record_type: WalRecordType::TxnBegin,
            txn_id,
            timestamp_us,
            key: None,
            value: None,
            table: None,
            checksum: 0,
            lsn: 0,
            prev_lsn: None,
            page_id: None,
            undo_info: None,
            undo_next_lsn: None,
        }
    }

    pub fn new_commit(txn_id: TxnId, timestamp_us: u64) -> Self {
        Self {
            record_type: WalRecordType::TxnCommit,
            txn_id,
            timestamp_us,
            key: None,
            value: None,
            table: None,
            checksum: 0,
            lsn: 0,
            prev_lsn: None,
            page_id: None,
            undo_info: None,
            undo_next_lsn: None,
        }
    }

    pub fn new_abort(txn_id: TxnId, timestamp_us: u64) -> Self {
        Self {
            record_type: WalRecordType::TxnAbort,
            txn_id,
            timestamp_us,
            key: None,
            value: None,
            table: None,
            checksum: 0,
            lsn: 0,
            prev_lsn: None,
            page_id: None,
            undo_info: None,
            undo_next_lsn: None,
        }
    }

    pub fn new_data(
        txn_id: TxnId,
        timestamp_us: u64,
        table: String,
        key: Vec<u8>,
        value: Option<Vec<u8>>,
    ) -> Self {
        Self {
            record_type: WalRecordType::Data,
            txn_id,
            timestamp_us,
            key: Some(key),
            value,
            table: Some(table),
            checksum: 0,
            lsn: 0,
            prev_lsn: None,
            page_id: None,
            undo_info: None,
            undo_next_lsn: None,
        }
    }

    /// Create a new ARIES-style data record with before-image for undo
    #[allow(clippy::too_many_arguments)]
    pub fn new_aries_data(
        txn_id: TxnId,
        timestamp_us: u64,
        table: String,
        key: Vec<u8>,
        value: Option<Vec<u8>>,
        page_id: PageId,
        prev_lsn: Option<Lsn>,
        undo_info: Option<Vec<u8>>,
    ) -> Self {
        Self {
            record_type: WalRecordType::Data,
            txn_id,
            timestamp_us,
            key: Some(key),
            value,
            table: Some(table),
            checksum: 0,
            lsn: 0, // Assigned when appended to WAL
            prev_lsn,
            page_id: Some(page_id),
            undo_info,
            undo_next_lsn: None,
        }
    }

    /// Create a Compensation Log Record (CLR) for ARIES undo
    ///
    /// CLRs are redo-only records that describe undo operations.
    /// They include undo_next_lsn which points to the next record to undo,
    /// skipping the compensated operation.
    #[allow(clippy::too_many_arguments)]
    pub fn new_clr(
        txn_id: TxnId,
        timestamp_us: u64,
        table: String,
        key: Vec<u8>,
        value: Option<Vec<u8>>,
        page_id: PageId,
        prev_lsn: Lsn,
        undo_next_lsn: Lsn,
    ) -> Self {
        Self {
            record_type: WalRecordType::CompensationLogRecord,
            txn_id,
            timestamp_us,
            key: Some(key),
            value,
            table: Some(table),
            checksum: 0,
            lsn: 0,
            prev_lsn: Some(prev_lsn),
            page_id: Some(page_id),
            undo_info: None, // CLRs don't need undo info (redo-only)
            undo_next_lsn: Some(undo_next_lsn),
        }
    }

    /// Create a checkpoint end record with recovery data
    pub fn new_checkpoint_end(
        timestamp_us: u64,
        checkpoint_data: AriesCheckpointData,
    ) -> Result<Self, String> {
        let data = bincode::serialize(&checkpoint_data)
            .map_err(|e| format!("Failed to serialize checkpoint data: {}", e))?;
        Ok(Self {
            record_type: WalRecordType::CheckpointEnd,
            txn_id: 0,
            timestamp_us,
            key: None,
            value: Some(data),
            table: None,
            checksum: 0,
            lsn: 0,
            prev_lsn: None,
            page_id: None,
            undo_info: None,
            undo_next_lsn: None,
        })
    }

    /// Extract checkpoint data from a CheckpointEnd record
    pub fn get_checkpoint_data(&self) -> Option<AriesCheckpointData> {
        if self.record_type != WalRecordType::CheckpointEnd {
            return None;
        }
        self.value
            .as_ref()
            .and_then(|data| bincode::deserialize(data).ok())
    }

    /// Calculate and set checksum
    pub fn compute_checksum(&mut self) {
        let data = self.serialize_for_checksum();
        self.checksum = crc32fast::hash(&data);
    }

    /// Verify checksum
    pub fn verify_checksum(&self) -> bool {
        let data = self.serialize_for_checksum();
        crc32fast::hash(&data) == self.checksum
    }

    fn serialize_for_checksum(&self) -> Vec<u8> {
        // Serialize all fields except checksum for CRC32 computation.
        // IMPORTANT: This MUST include ALL ARIES fields to prevent silent
        // data corruption during crash recovery. If a field is not checksummed,
        // it can be corrupted without detection, breaking ARIES undo/redo chains.
        let mut buf = Vec::new();
        buf.push(self.record_type as u8);
        buf.extend(&self.txn_id.to_le_bytes());
        buf.extend(&self.timestamp_us.to_le_bytes());
        if let Some(ref key) = self.key {
            buf.extend(&(key.len() as u32).to_le_bytes());
            buf.extend(key);
        } else {
            buf.extend(&0u32.to_le_bytes());
        }
        if let Some(ref value) = self.value {
            buf.extend(&(value.len() as u32).to_le_bytes());
            buf.extend(value);
        } else {
            buf.extend(&0u32.to_le_bytes());
        }
        if let Some(ref table) = self.table {
            buf.extend(&(table.len() as u32).to_le_bytes());
            buf.extend(table.as_bytes());
        } else {
            buf.extend(&0u32.to_le_bytes());
        }
        // ARIES fields — critical for recovery correctness
        buf.extend(&self.lsn.to_le_bytes());
        // prev_lsn: 0 = None, 1 + value = Some(value)
        match self.prev_lsn {
            Some(lsn) => {
                buf.push(1u8);
                buf.extend(&lsn.to_le_bytes());
            }
            None => buf.push(0u8),
        }
        // page_id: same encoding
        match self.page_id {
            Some(pid) => {
                buf.push(1u8);
                buf.extend(&pid.to_le_bytes());
            }
            None => buf.push(0u8),
        }
        // undo_info: length-prefixed optional blob
        if let Some(ref undo) = self.undo_info {
            buf.extend(&(undo.len() as u32).to_le_bytes());
            buf.extend(undo);
        } else {
            buf.extend(&0u32.to_le_bytes());
        }
        // undo_next_lsn: same encoding as prev_lsn
        match self.undo_next_lsn {
            Some(lsn) => {
                buf.push(1u8);
                buf.extend(&lsn.to_le_bytes());
            }
            None => buf.push(0u8),
        }
        buf
    }

    /// Wire format version for forward/backward compatibility.
    ///
    /// - V0 (0x00): Legacy 6-field format (record_type, txn_id, timestamp_us, key, value, table, checksum).
    ///              ARIES fields are NOT serialized; from_bytes() defaults them to zero/None.
    /// - V1 (0x01): Full ARIES format. All fields serialized including lsn, prev_lsn,
    ///              page_id, undo_info, undo_next_lsn. Checksum covers ALL fields.
    pub const FORMAT_VERSION: u8 = 0x01;

    /// Serialize to bytes (versioned wire format)
    ///
    /// Format V1 layout:
    /// ```text
    /// [version:1] [serialize_for_checksum payload] [crc32:4]
    /// ```
    pub fn to_bytes(&self) -> Vec<u8> {
        let payload = self.serialize_for_checksum();
        let mut buf = Vec::with_capacity(1 + payload.len() + 4);
        buf.push(Self::FORMAT_VERSION);
        buf.extend(&payload);
        buf.extend(&self.checksum.to_le_bytes());
        buf
    }

    /// Deserialize from bytes with proper error propagation
    ///
    /// Supports both wire format versions:
    /// - V0 (legacy): No version byte; first byte is a valid WalRecordType discriminant.
    ///   ARIES fields default to zero/None.
    /// - V1: First byte is 0x01 (FORMAT_VERSION). All ARIES fields deserialized.
    ///
    /// Returns an error if:
    /// - Data is too short
    /// - Record type is invalid
    /// - Data is truncated mid-field
    /// - UTF-8 encoding is invalid for table name
    /// - Checksum validation fails
    pub fn from_bytes(data: &[u8]) -> Result<Self, String> {
        if data.is_empty() {
            return Err("WAL entry empty".to_string());
        }

        // Detect format version: V1 starts with 0x01 which is also WalRecordType::Data,
        // so we use a heuristic: V1 has version byte 0x01 followed by another record_type byte.
        // V0 has no version byte, first byte IS the record type.
        // Disambiguation: if first byte is 0x01 (FORMAT_VERSION/Data) AND data.len() >= 2
        // AND second byte is a valid WalRecordType, then it's V1.
        // Otherwise, treat as V0 for backward compatibility.
        let (version, payload_start) = if data.len() >= 2
            && data[0] == Self::FORMAT_VERSION
            && WalRecordType::try_from(data[1]).is_ok()
        {
            (1u8, 1usize) // V1: skip version byte
        } else {
            (0u8, 0usize) // V0: no version byte
        };

        let payload = &data[payload_start..];

        // Fixed header: 1 (type) + 8 (txn_id) + 8 (timestamp) = 17 minimum before variable fields
        if payload.len() < 21 {
            return Err(format!(
                "WAL entry too short: {} bytes, need at least 21",
                payload.len()
            ));
        }

        let record_type = WalRecordType::try_from(payload[0])
            .map_err(|_| format!("Invalid WAL record type: {}", payload[0]))?;

        let txn_id = u64::from_le_bytes(
            payload[1..9]
                .try_into()
                .map_err(|_| "Failed to parse txn_id: slice too short")?,
        );
        let timestamp_us = u64::from_le_bytes(
            payload[9..17]
                .try_into()
                .map_err(|_| "Failed to parse timestamp: slice too short")?,
        );

        let mut offset = 17;

        // Parse key with bounds checking
        if offset + 4 > payload.len() {
            return Err(format!(
                "WAL entry truncated at key_len: offset {} + 4 > {}",
                offset,
                payload.len()
            ));
        }
        let key_len = u32::from_le_bytes(
            payload[offset..offset + 4]
                .try_into()
                .map_err(|_| "Failed to parse key_len")?,
        ) as usize;
        offset += 4;

        if offset + key_len > payload.len() {
            return Err(format!(
                "WAL entry truncated at key: need {} bytes at offset {}, have {}",
                key_len,
                offset,
                payload.len()
            ));
        }
        let key = if key_len > 0 {
            Some(payload[offset..offset + key_len].to_vec())
        } else {
            None
        };
        offset += key_len;

        // Parse value with bounds checking
        if offset + 4 > payload.len() {
            return Err(format!(
                "WAL entry truncated at value_len: offset {} + 4 > {}",
                offset,
                payload.len()
            ));
        }
        let value_len = u32::from_le_bytes(
            payload[offset..offset + 4]
                .try_into()
                .map_err(|_| "Failed to parse value_len")?,
        ) as usize;
        offset += 4;

        if offset + value_len > payload.len() {
            return Err(format!(
                "WAL entry truncated at value: need {} bytes at offset {}, have {}",
                value_len,
                offset,
                payload.len()
            ));
        }
        let value = if value_len > 0 {
            Some(payload[offset..offset + value_len].to_vec())
        } else {
            None
        };
        offset += value_len;

        // Parse table name with bounds checking
        if offset + 4 > payload.len() {
            return Err(format!(
                "WAL entry truncated at table_len: offset {} + 4 > {}",
                offset,
                payload.len()
            ));
        }
        let table_len = u32::from_le_bytes(
            payload[offset..offset + 4]
                .try_into()
                .map_err(|_| "Failed to parse table_len")?,
        ) as usize;
        offset += 4;

        if offset + table_len > payload.len() {
            return Err(format!(
                "WAL entry truncated at table: need {} bytes at offset {}, have {}",
                table_len,
                offset,
                payload.len()
            ));
        }
        let table = if table_len > 0 {
            Some(
                String::from_utf8(payload[offset..offset + table_len].to_vec())
                    .map_err(|e| format!("Invalid UTF-8 in table name: {}", e))?,
            )
        } else {
            None
        };
        offset += table_len;

        // Parse ARIES fields for V1, default for V0
        let (lsn, prev_lsn, page_id, undo_info, undo_next_lsn) = if version >= 1 {
            // V1: ARIES fields follow table
            if offset + 8 > payload.len() {
                return Err(format!(
                    "WAL V1 entry truncated at lsn: offset {} + 8 > {}",
                    offset,
                    payload.len()
                ));
            }
            let lsn = u64::from_le_bytes(
                payload[offset..offset + 8]
                    .try_into()
                    .map_err(|_| "Failed to parse lsn")?,
            );
            offset += 8;

            // prev_lsn: tag byte + optional u64
            if offset >= payload.len() {
                return Err("WAL V1 entry truncated at prev_lsn tag".to_string());
            }
            let prev_lsn = if payload[offset] == 1 {
                offset += 1;
                if offset + 8 > payload.len() {
                    return Err("WAL V1 entry truncated at prev_lsn value".to_string());
                }
                let v = u64::from_le_bytes(
                    payload[offset..offset + 8]
                        .try_into()
                        .map_err(|_| "Failed to parse prev_lsn")?,
                );
                offset += 8;
                Some(v)
            } else {
                offset += 1;
                None
            };

            // page_id: tag byte + optional u64
            if offset >= payload.len() {
                return Err("WAL V1 entry truncated at page_id tag".to_string());
            }
            let page_id = if payload[offset] == 1 {
                offset += 1;
                if offset + 8 > payload.len() {
                    return Err("WAL V1 entry truncated at page_id value".to_string());
                }
                let v = u64::from_le_bytes(
                    payload[offset..offset + 8]
                        .try_into()
                        .map_err(|_| "Failed to parse page_id")?,
                );
                offset += 8;
                Some(v)
            } else {
                offset += 1;
                None
            };

            // undo_info: length-prefixed optional blob
            if offset + 4 > payload.len() {
                return Err("WAL V1 entry truncated at undo_info_len".to_string());
            }
            let undo_len = u32::from_le_bytes(
                payload[offset..offset + 4]
                    .try_into()
                    .map_err(|_| "Failed to parse undo_info_len")?,
            ) as usize;
            offset += 4;
            let undo_info = if undo_len > 0 {
                if offset + undo_len > payload.len() {
                    return Err("WAL V1 entry truncated at undo_info data".to_string());
                }
                let v = payload[offset..offset + undo_len].to_vec();
                offset += undo_len;
                Some(v)
            } else {
                None
            };

            // undo_next_lsn: tag byte + optional u64
            if offset >= payload.len() {
                return Err("WAL V1 entry truncated at undo_next_lsn tag".to_string());
            }
            let undo_next_lsn = if payload[offset] == 1 {
                offset += 1;
                if offset + 8 > payload.len() {
                    return Err("WAL V1 entry truncated at undo_next_lsn value".to_string());
                }
                let v = u64::from_le_bytes(
                    payload[offset..offset + 8]
                        .try_into()
                        .map_err(|_| "Failed to parse undo_next_lsn")?,
                );
                offset += 8;
                Some(v)
            } else {
                offset += 1;
                None
            };

            (lsn, prev_lsn, page_id, undo_info, undo_next_lsn)
        } else {
            // V0: ARIES fields default to zero/None for backward compatibility
            (0u64, None, None, None, None)
        };

        // Parse checksum with bounds checking
        if offset + 4 > payload.len() {
            return Err(format!(
                "WAL entry truncated at checksum: offset {} + 4 > {}",
                offset,
                payload.len()
            ));
        }
        let checksum = u32::from_le_bytes(
            payload[offset..offset + 4]
                .try_into()
                .map_err(|_| "Failed to parse checksum")?,
        );

        let entry = Self {
            record_type,
            txn_id,
            timestamp_us,
            key,
            value,
            table,
            checksum,
            lsn,
            prev_lsn,
            page_id,
            undo_info,
            undo_next_lsn,
        };

        // Verify checksum to detect corruption
        if !entry.verify_checksum() {
            return Err(format!(
                "WAL entry checksum mismatch for txn_id {}: expected valid checksum, got {}",
                entry.txn_id, entry.checksum
            ));
        }

        Ok(entry)
    }
}

/// Transaction handle for the user
#[derive(Debug)]
pub struct Transaction {
    /// Unique transaction ID
    pub id: TxnId,
    /// Transaction state
    pub state: TxnState,
    /// Start timestamp for MVCC
    pub start_ts: u64,
    /// Commit timestamp (set on commit)
    pub commit_ts: Option<u64>,
    /// Buffered writes
    pub writes: Vec<TxnWrite>,
    /// Read set for conflict detection
    pub read_set: HashSet<TxnRead>,
    /// Isolation level
    pub isolation: IsolationLevel,
}

/// Transaction isolation levels
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum IsolationLevel {
    /// Read committed - see committed changes
    ReadCommitted,
    /// Snapshot isolation - consistent point-in-time view
    #[default]
    SnapshotIsolation,
    /// Serializable - strongest isolation
    Serializable,
}

impl Transaction {
    /// Create a new transaction
    pub fn new(id: TxnId, start_ts: u64, isolation: IsolationLevel) -> Self {
        Self {
            id,
            state: TxnState::Active,
            start_ts,
            commit_ts: None,
            writes: Vec::new(),
            read_set: HashSet::new(),
            isolation,
        }
    }

    /// Buffer a write operation
    pub fn put(&mut self, table: &str, key: Vec<u8>, value: Vec<u8>) {
        self.writes.push(TxnWrite {
            key,
            value: Some(value),
            table: table.to_string(),
        });
    }

    /// Buffer a delete operation
    pub fn delete(&mut self, table: &str, key: Vec<u8>) {
        self.writes.push(TxnWrite {
            key,
            value: None,
            table: table.to_string(),
        });
    }

    /// Record a read for conflict detection
    pub fn record_read(&mut self, table: &str, key: Vec<u8>) {
        self.read_set.insert(TxnRead {
            key,
            table: table.to_string(),
        });
    }

    /// Check for read-your-writes
    pub fn get_local(&self, table: &str, key: &[u8]) -> Option<&TxnWrite> {
        self.writes
            .iter()
            .rev()
            .find(|w| w.table == table && w.key == key)
    }

    /// Check if transaction has any writes
    pub fn is_read_only(&self) -> bool {
        self.writes.is_empty()
    }
}

/// Transaction Manager stats
#[derive(Debug, Clone, Default)]
pub struct TxnStats {
    pub active_count: u64,
    pub committed_count: u64,
    pub aborted_count: u64,
    pub conflict_aborts: u64,
}

/// Transaction Manager (in-memory, no WAL durability)
///
/// Manages transaction lifecycle and provides ACID guarantees for in-memory
/// operations. This implementation does NOT include WAL integration.
/// 
/// For production workloads requiring durability, use [`sochdb_storage::MvccTransactionManager`]
/// which includes:
/// - Write-ahead logging for crash recovery  
/// - Serializable Snapshot Isolation (SSI)
/// - Group commit for high throughput
/// - Event-driven async architecture
pub struct TransactionManager {
    /// Next transaction ID
    next_txn_id: AtomicU64,
    /// Current timestamp counter
    timestamp_counter: AtomicU64,
    /// Committed transaction watermark
    committed_watermark: AtomicU64,
    /// Statistics
    stats: parking_lot::RwLock<TxnStats>,
}

impl TransactionManager {
    pub fn new() -> Self {
        Self {
            next_txn_id: AtomicU64::new(1),
            timestamp_counter: AtomicU64::new(1),
            committed_watermark: AtomicU64::new(0),
            stats: parking_lot::RwLock::new(TxnStats::default()),
        }
    }

    /// Begin a new transaction
    pub fn begin(&self) -> Transaction {
        self.begin_with_isolation(IsolationLevel::default())
    }

    /// Begin a transaction with specific isolation level
    pub fn begin_with_isolation(&self, isolation: IsolationLevel) -> Transaction {
        let txn_id = self.next_txn_id.fetch_add(1, Ordering::SeqCst);
        let start_ts = self.timestamp_counter.fetch_add(1, Ordering::SeqCst);

        {
            let mut stats = self.stats.write();
            stats.active_count += 1;
        }

        Transaction::new(txn_id, start_ts, isolation)
    }

    /// Get commit timestamp
    pub fn get_commit_ts(&self) -> u64 {
        self.timestamp_counter.fetch_add(1, Ordering::SeqCst)
    }

    /// Mark transaction as committed
    pub fn mark_committed(&self, txn: &mut Transaction) {
        txn.state = TxnState::Committed;
        txn.commit_ts = Some(self.get_commit_ts());

        let mut stats = self.stats.write();
        stats.active_count = stats.active_count.saturating_sub(1);
        stats.committed_count += 1;
    }

    /// Mark transaction as aborted
    pub fn mark_aborted(&self, txn: &mut Transaction) {
        txn.state = TxnState::Aborted;

        let mut stats = self.stats.write();
        stats.active_count = stats.active_count.saturating_sub(1);
        stats.aborted_count += 1;
    }

    /// Mark transaction as aborted due to conflict
    pub fn mark_conflict_abort(&self, txn: &mut Transaction) {
        self.mark_aborted(txn);

        let mut stats = self.stats.write();
        stats.conflict_aborts += 1;
    }

    /// Get the oldest active transaction timestamp
    pub fn oldest_active_ts(&self) -> u64 {
        self.committed_watermark.load(Ordering::SeqCst)
    }

    /// Update the committed watermark
    pub fn advance_watermark(&self, new_watermark: u64) {
        self.committed_watermark
            .fetch_max(new_watermark, Ordering::SeqCst);
    }

    /// Get current stats
    pub fn stats(&self) -> TxnStats {
        self.stats.read().clone()
    }
}

impl Default for TransactionManager {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn test_transaction_lifecycle() {
        let mgr = TransactionManager::new();

        let mut txn = mgr.begin();
        assert_eq!(txn.state, TxnState::Active);
        assert!(txn.is_read_only());

        txn.put("users", vec![1], vec![2, 3, 4]);
        assert!(!txn.is_read_only());

        mgr.mark_committed(&mut txn);
        assert_eq!(txn.state, TxnState::Committed);
        assert!(txn.commit_ts.is_some());
    }

    #[test]
    fn test_read_your_writes() {
        let mgr = TransactionManager::new();
        let mut txn = mgr.begin();

        txn.put("users", vec![1], vec![10, 20]);
        txn.put("users", vec![1], vec![30, 40]); // Overwrite

        let local = txn.get_local("users", &[1]);
        assert!(local.is_some());
        assert_eq!(local.unwrap().value, Some(vec![30, 40]));
    }

    #[test]
    fn test_wal_entry_serialization() {
        let mut entry = TxnWalEntry::new_data(
            42,
            1234567890,
            "users".to_string(),
            vec![1, 2, 3],
            Some(vec![4, 5, 6]),
        );
        entry.compute_checksum();

        let bytes = entry.to_bytes();
        let parsed = TxnWalEntry::from_bytes(&bytes).unwrap();

        assert_eq!(parsed.txn_id, 42);
        assert_eq!(parsed.timestamp_us, 1234567890);
        assert_eq!(parsed.table, Some("users".to_string()));
        assert_eq!(parsed.key, Some(vec![1, 2, 3]));
        assert_eq!(parsed.value, Some(vec![4, 5, 6]));
        assert!(parsed.verify_checksum());
    }

    #[test]
    fn test_wal_entry_aries_roundtrip() {
        // Verify ARIES fields survive serialization round-trip (T1-T2 fix)
        let mut entry = TxnWalEntry::new_aries_data(
            99,
            9999999,
            "orders".to_string(),
            vec![10, 20],
            Some(vec![30, 40, 50]),
            42,           // page_id
            Some(100),    // prev_lsn
            Some(vec![0xDE, 0xAD]), // undo_info
        );
        entry.lsn = 200;
        entry.undo_next_lsn = Some(50);
        entry.compute_checksum();

        let bytes = entry.to_bytes();
        let parsed = TxnWalEntry::from_bytes(&bytes).unwrap();

        assert_eq!(parsed.txn_id, 99);
        assert_eq!(parsed.lsn, 200);
        assert_eq!(parsed.prev_lsn, Some(100));
        assert_eq!(parsed.page_id, Some(42));
        assert_eq!(parsed.undo_info, Some(vec![0xDE, 0xAD]));
        assert_eq!(parsed.undo_next_lsn, Some(50));
        assert_eq!(parsed.key, Some(vec![10, 20]));
        assert_eq!(parsed.value, Some(vec![30, 40, 50]));
        assert!(parsed.verify_checksum());
    }

    #[test]
    fn test_wal_entry_clr_roundtrip() {
        let mut entry = TxnWalEntry::new_clr(
            77,
            5555555,
            "inventory".to_string(),
            vec![1],
            Some(vec![2]),
            10,     // page_id
            300,    // prev_lsn
            250,    // undo_next_lsn
        );
        entry.lsn = 400;
        entry.compute_checksum();

        let bytes = entry.to_bytes();
        let parsed = TxnWalEntry::from_bytes(&bytes).unwrap();

        assert_eq!(parsed.record_type, WalRecordType::CompensationLogRecord);
        assert_eq!(parsed.lsn, 400);
        assert_eq!(parsed.prev_lsn, Some(300));
        assert_eq!(parsed.page_id, Some(10));
        assert_eq!(parsed.undo_next_lsn, Some(250));
        assert!(parsed.undo_info.is_none()); // CLRs have no undo_info
        assert!(parsed.verify_checksum());
    }

    #[test]
    fn test_wal_entry_none_aries_fields_roundtrip() {
        // Entry with all ARIES fields as None/0
        let mut entry = TxnWalEntry::new_begin(1, 100);
        entry.compute_checksum();

        let bytes = entry.to_bytes();
        let parsed = TxnWalEntry::from_bytes(&bytes).unwrap();

        assert_eq!(parsed.lsn, 0);
        assert_eq!(parsed.prev_lsn, None);
        assert_eq!(parsed.page_id, None);
        assert_eq!(parsed.undo_info, None);
        assert_eq!(parsed.undo_next_lsn, None);
        assert!(parsed.verify_checksum());
    }

    #[test]
    fn test_transaction_stats() {
        let mgr = TransactionManager::new();

        let mut txn1 = mgr.begin();
        let mut txn2 = mgr.begin();

        assert_eq!(mgr.stats().active_count, 2);

        mgr.mark_committed(&mut txn1);
        assert_eq!(mgr.stats().committed_count, 1);

        mgr.mark_aborted(&mut txn2);
        assert_eq!(mgr.stats().aborted_count, 1);
        assert_eq!(mgr.stats().active_count, 0);
    }

    #[test]
    fn test_wal_entry_error_too_short() {
        // Less than minimum required bytes
        let short_data = vec![0u8; 10];
        let result = TxnWalEntry::from_bytes(&short_data);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("too short"));
    }

    #[test]
    fn test_wal_entry_error_invalid_record_type() {
        // Create data with invalid record type (255)
        let mut data = vec![0u8; 30];
        data[0] = 255; // Invalid record type
        let result = TxnWalEntry::from_bytes(&data);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("Invalid WAL record type"));
    }

    #[test]
    fn test_wal_entry_error_truncated_key() {
        // Create entry claiming 1000 byte key but data too short
        let mut entry =
            TxnWalEntry::new_data(1, 100, "test".to_string(), vec![1, 2], Some(vec![3, 4]));
        entry.compute_checksum();
        let mut bytes = entry.to_bytes();

        // Corrupt key_len to claim huge key
        let huge_len: u32 = 10000;
        bytes[17..21].copy_from_slice(&huge_len.to_le_bytes());

        let result = TxnWalEntry::from_bytes(&bytes);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("truncated at key"));
    }

    #[test]
    fn test_wal_entry_error_corrupted_checksum() {
        let mut entry = TxnWalEntry::new_data(
            42,
            1234567890,
            "users".to_string(),
            vec![1, 2, 3],
            Some(vec![4, 5, 6]),
        );
        entry.compute_checksum();

        let mut bytes = entry.to_bytes();
        // Corrupt the checksum (last 4 bytes)
        let len = bytes.len();
        bytes[len - 1] ^= 0xFF; // Flip bits

        let result = TxnWalEntry::from_bytes(&bytes);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("checksum mismatch"));
    }

    #[test]
    fn test_wal_entry_error_invalid_utf8_table() {
        let mut entry = TxnWalEntry::new_data(1, 100, "test".to_string(), vec![1], Some(vec![2]));
        entry.compute_checksum();
        let mut bytes = entry.to_bytes();

        // Find table offset and corrupt UTF-8
        // Header: 1 + 8 + 8 = 17, key_len: 4, key: 1, value_len: 4, value: 1, table_len: 4, table: 4
        let table_start = 17 + 4 + 1 + 4 + 1 + 4;
        bytes[table_start] = 0xFF; // Invalid UTF-8 byte

        let result = TxnWalEntry::from_bytes(&bytes);
        // Either checksum or UTF-8 error
        assert!(result.is_err());
    }
}