graph_d 1.3.2

//! Transaction management for ACID compliance.
//!
//! This module provides transaction support to ensure ACID (Atomicity, Consistency,
//! Isolation, Durability) properties for graph database operations. It implements
//! concurrent transaction management with configurable isolation levels.
//!
//! # ACID Properties
//!
//! - **Atomicity**: All operations in a transaction succeed or all fail
//! - **Consistency**: Transactions maintain database invariants
//! - **Isolation**: Concurrent transactions don't interfere with each other
//! - **Durability**: Committed changes persist across system failures
//!
//! # Isolation Levels
//!
//! The module supports standard SQL isolation levels:
//! - **Read Uncommitted**: Allows dirty reads (lowest isolation)
//! - **Read Committed**: Prevents dirty reads
//! - **Repeatable Read**: Prevents dirty and non-repeatable reads
//! - **Serializable**: Prevents all read anomalies (highest isolation)
//!
//! # Concurrency Control
//!
//! The system uses a combination of:
//! - **Multi-Version Concurrency Control (MVCC)**: Timestamp-based versioning
//! - **Two-Phase Locking**: For write operations requiring exclusive access
//! - **Deadlock Detection**: Automatic detection and resolution of deadlocks
//!
//! # Example
//!
//! ```rust
//! use graph_d::transaction::{TransactionManager, IsolationLevel};
//!
//! let manager = TransactionManager::new(IsolationLevel::ReadCommitted);
//!
//! // Begin a transaction
//! let mut tx = manager.begin();
//! assert!(tx.is_active());
//!
//! // Perform operations...
//! tx.mark_write(); // Indicate write operations
//!
//! // Commit the transaction
//! tx.commit().unwrap();
//! ```
//!
//! # Concurrent Transactions
//!
//! ```rust
//! use graph_d::transaction::TransactionManager;
//!
//! let manager = TransactionManager::default();
//!
//! // Begin concurrent transactions with locking support
//! let tx1 = manager.begin_concurrent();
//! let tx2 = manager.begin_concurrent();
//!
//! // Transactions can safely operate concurrently
//! // Lock manager handles conflicts automatically
//! ```

pub mod concurrency;
pub mod mvcc;

use crate::error::{GraphError, Result};
use crate::graph::{Node, Relationship};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

pub use concurrency::{
    ConcurrentTransaction, LockInfo, LockManager, LockStatistics, LockType, LockableResource,
};
pub use mvcc::{MvccManager, MvccStats, MvccTransaction};

/// Unique transaction identifier.
pub type TransactionId = u64;

/// Transaction isolation levels.
///
/// Isolation levels control the degree to which transactions are isolated from
/// each other. Higher isolation levels provide stronger consistency guarantees
/// but may reduce concurrency and performance.
///
/// # Anomalies Prevented
///
/// | Level | Dirty Read | Non-Repeatable Read | Phantom Read | Serialization Anomaly |
/// |-------|------------|---------------------|--------------|------------------------|
/// | Read Uncommitted | ❌ | ❌ | ❌ | ❌ |
/// | Read Committed | ✅ | ❌ | ❌ | ❌ |
/// | Repeatable Read | ✅ | ✅ | ❌ | ❌ |
/// | Serializable | ✅ | ✅ | ✅ | ✅ |
///
/// # Performance Impact
///
/// - **Read Uncommitted**: Highest performance, lowest consistency
/// - **Read Committed**: Good balance of performance and consistency (default)
/// - **Repeatable Read**: Moderate performance impact
/// - **Serializable**: Highest consistency, potential performance impact
///
/// # Example
///
/// ```rust
/// use graph_d::transaction::{TransactionManager, IsolationLevel};
///
/// // Conservative approach for critical data
/// let manager = TransactionManager::new(IsolationLevel::Serializable);
///
/// // Balanced approach for most applications
/// let manager = TransactionManager::new(IsolationLevel::ReadCommitted);
///
/// // High-performance approach for read-heavy workloads
/// let manager = TransactionManager::new(IsolationLevel::ReadUncommitted);
/// ```
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum IsolationLevel {
    /// Read uncommitted - allows dirty reads.
    ///
    /// Transactions can read uncommitted changes from other transactions.
    /// This provides the highest performance but weakest consistency guarantees.
    /// Use only when data consistency is not critical.
    ReadUncommitted,

    /// Read committed - prevents dirty reads.
    ///
    /// Transactions can only read committed data, but may see different values
    /// when reading the same data multiple times within the transaction.
    /// This is the default isolation level for most applications.
    ReadCommitted,

    /// Repeatable read - prevents dirty and non-repeatable reads.
    ///
    /// Once a transaction reads data, subsequent reads within the same transaction
    /// will see the same values. However, phantom reads may still occur.
    RepeatableRead,

    /// Serializable - prevents all anomalies.
    ///
    /// Provides the strongest isolation guarantees. Transactions appear to execute
    /// serially, preventing all read anomalies. May impact performance due to
    /// increased locking and potential for serialization failures.
    Serializable,
}

/// Transaction state.
///
/// Represents the current lifecycle state of a transaction. Transactions
/// progress through these states during their lifetime:
///
/// ```text
/// Active → Committed
///    ↓
/// RolledBack
/// ```
///
/// # State Transitions
///
/// - **Active → Committed**: Normal transaction completion
/// - **Active → RolledBack**: Transaction aborted due to error or explicit rollback
/// - **Committed/RolledBack**: Terminal states (no further transitions)
///
/// # Example
///
/// ```rust
/// use graph_d::transaction::{TransactionManager, TransactionState};
///
/// let manager = TransactionManager::default();
/// let mut tx = manager.begin();
///
/// assert_eq!(tx.state(), TransactionState::Active);
///
/// tx.commit().unwrap();
/// assert_eq!(tx.state(), TransactionState::Committed);
/// ```
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum TransactionState {
    /// Transaction is active and can be used.
    ///
    /// The transaction is in progress and can perform read and write operations.
    /// This is the initial state when a transaction is created.
    Active,

    /// Transaction has been committed.
    ///
    /// All changes made by the transaction have been permanently applied
    /// to the database. This is a terminal state.
    Committed,

    /// Transaction has been rolled back.
    ///
    /// All changes made by the transaction have been discarded and the
    /// database state is restored to before the transaction began.
    /// This is a terminal state.
    RolledBack,
}

/// A database transaction.
///
/// A [`Transaction`] represents a unit of work that is executed atomically.
/// It maintains timestamps for concurrency control and tracks its current state.
///
/// # Timestamps
///
/// - **Read Timestamp**: When the transaction started reading data
/// - **Write Timestamp**: When the transaction first performed a write (if any)
///
/// These timestamps are used for Multi-Version Concurrency Control (MVCC)
/// to determine which version of data a transaction should see.
///
/// # Lifecycle
///
/// 1. **Creation**: Transaction starts in Active state
/// 2. **Operations**: Read and write operations can be performed
/// 3. **Completion**: Transaction is either committed or rolled back
///
/// # Example
///
/// ```rust
/// use graph_d::transaction::{TransactionManager, IsolationLevel};
///
/// let manager = TransactionManager::new(IsolationLevel::ReadCommitted);
/// let mut tx = manager.begin();
///
/// // Transaction is active
/// assert!(tx.is_active());
/// assert!(tx.write_timestamp().is_none());
///
/// // Perform write operations
/// tx.mark_write();
/// assert!(tx.write_timestamp().is_some());
///
/// // Commit the transaction
/// tx.commit().unwrap();
/// assert!(!tx.is_active());
/// ```
pub struct Transaction {
    id: TransactionId,
    isolation_level: IsolationLevel,
    state: TransactionState,
    read_timestamp: u64,
    write_timestamp: Option<u64>,
}

impl Transaction {
    /// Create a new transaction with the given isolation level.
    pub fn new(id: TransactionId, isolation_level: IsolationLevel) -> Self {
        Transaction {
            id,
            isolation_level,
            state: TransactionState::Active,
            read_timestamp: current_timestamp(),
            write_timestamp: None,
        }
    }

    /// Get the transaction ID.
    pub fn id(&self) -> TransactionId {
        self.id
    }

    /// Get the isolation level.
    pub fn isolation_level(&self) -> IsolationLevel {
        self.isolation_level
    }

    /// Get the current state.
    pub fn state(&self) -> TransactionState {
        self.state
    }

    /// Check if the transaction is active.
    pub fn is_active(&self) -> bool {
        self.state == TransactionState::Active
    }

    /// Get the read timestamp.
    pub fn read_timestamp(&self) -> u64 {
        self.read_timestamp
    }

    /// Get the write timestamp.
    pub fn write_timestamp(&self) -> Option<u64> {
        self.write_timestamp
    }

    /// Mark the transaction as having performed writes.
    pub fn mark_write(&mut self) {
        if self.write_timestamp.is_none() {
            self.write_timestamp = Some(current_timestamp());
        }
    }

    /// Commit the transaction.
    pub fn commit(&mut self) -> Result<()> {
        if self.state != TransactionState::Active {
            return Err(GraphError::Transaction(format!(
                "Cannot commit transaction {} in state {:?}",
                self.id, self.state
            )));
        }

        self.state = TransactionState::Committed;
        Ok(())
    }

    /// Roll back the transaction.
    pub fn rollback(&mut self) -> Result<()> {
        if self.state != TransactionState::Active {
            return Err(GraphError::Transaction(format!(
                "Cannot rollback transaction {} in state {:?}",
                self.id, self.state
            )));
        }

        self.state = TransactionState::RolledBack;
        Ok(())
    }
}

/// Transaction manager for coordinating concurrent transactions.
///
/// The [`TransactionManager`] is responsible for creating and coordinating
/// transactions in a multi-user environment. It provides transaction isolation,
/// manages unique transaction IDs, and integrates with the locking system
/// for concurrency control.
///
/// # Features
///
/// - **Transaction Creation**: Begin transactions with configurable isolation levels
/// - **Concurrency Support**: Integration with lock manager for concurrent access
/// - **MVCC Integration**: Multi-Version Concurrency Control for snapshot isolation
/// - **ID Generation**: Atomic generation of unique transaction identifiers
/// - **Statistics**: Monitor lock contention and transaction performance
///
/// # Thread Safety
///
/// The transaction manager is thread-safe and can be shared across multiple
/// threads. Internal state is protected by atomic operations and the lock
/// manager handles concurrent access to shared resources.
///
/// # Example
///
/// ```rust
/// use graph_d::transaction::{TransactionManager, IsolationLevel};
/// use std::sync::Arc;
/// use std::thread;
///
/// let manager = Arc::new(TransactionManager::new(IsolationLevel::ReadCommitted));
///
/// // Spawn multiple threads with concurrent transactions
/// let handles: Vec<_> = (0..4).map(|i| {
///     let manager = manager.clone();
///     thread::spawn(move || {
///         let mut tx = manager.begin_concurrent();
///         // Perform operations...
///         tx.commit().unwrap();
///     })
/// }).collect();
///
/// // Wait for all transactions to complete
/// for handle in handles {
///     handle.join().unwrap();
/// }
/// ```
///
/// # MVCC Example
///
/// ```rust,ignore
/// use graph_d::transaction::{TransactionManager, IsolationLevel};
/// use graph_d::graph::Node;
///
/// let manager = TransactionManager::new_with_mvcc(IsolationLevel::RepeatableRead);
///
/// // Begin MVCC-enabled transaction
/// let tx_id = manager.begin_mvcc();
///
/// // Read with snapshot isolation
/// let node = manager.mvcc_read_node(tx_id, 1)?;
///
/// // Write creates new version
/// manager.mvcc_write_node(tx_id, updated_node)?;
///
/// // Commit with conflict detection
/// manager.mvcc_commit(tx_id)?;
/// ```
///
/// # Performance Considerations
///
/// - Use appropriate isolation levels for your use case
/// - Monitor lock statistics to identify contention
/// - Consider transaction scope - shorter transactions reduce lock contention
/// - Batch operations within transactions when possible
/// - Use MVCC for read-heavy workloads with RepeatableRead or Serializable isolation
pub struct TransactionManager {
    next_transaction_id: AtomicU64,
    default_isolation_level: IsolationLevel,
    lock_manager: Arc<LockManager>,
    /// MVCC manager for snapshot isolation (optional, enabled with new_with_mvcc)
    mvcc_manager: Option<Arc<MvccManager>>,
}

impl TransactionManager {
    /// Create a new transaction manager without MVCC.
    pub fn new(default_isolation_level: IsolationLevel) -> Self {
        TransactionManager {
            next_transaction_id: AtomicU64::new(1),
            default_isolation_level,
            lock_manager: Arc::new(LockManager::new()),
            mvcc_manager: None,
        }
    }

    /// Create a new transaction manager with MVCC enabled.
    /// Satisfies: RT-2 (ACID compliance through MVCC)
    /// Satisfies: B4 (ACID transactions)
    pub fn new_with_mvcc(default_isolation_level: IsolationLevel) -> Self {
        TransactionManager {
            next_transaction_id: AtomicU64::new(1),
            default_isolation_level,
            lock_manager: Arc::new(LockManager::new()),
            mvcc_manager: Some(Arc::new(MvccManager::new())),
        }
    }

    /// Check if MVCC is enabled.
    pub fn has_mvcc(&self) -> bool {
        self.mvcc_manager.is_some()
    }

    /// Get MVCC manager reference.
    pub fn mvcc(&self) -> Option<&Arc<MvccManager>> {
        self.mvcc_manager.as_ref()
    }

    /// Begin a new transaction with the default isolation level.
    pub fn begin(&self) -> Transaction {
        self.begin_with_isolation(self.default_isolation_level)
    }

    /// Begin a new transaction with the specified isolation level.
    pub fn begin_with_isolation(&self, isolation_level: IsolationLevel) -> Transaction {
        let id = self.next_transaction_id.fetch_add(1, Ordering::SeqCst);
        Transaction::new(id, isolation_level)
    }

    /// Begin a new concurrent transaction with locking support.
    pub fn begin_concurrent(&self) -> ConcurrentTransaction {
        let transaction = self.begin();
        ConcurrentTransaction::new(transaction, self.lock_manager.clone())
    }

    /// Begin a new concurrent transaction with the specified isolation level.
    pub fn begin_concurrent_with_isolation(
        &self,
        isolation_level: IsolationLevel,
    ) -> ConcurrentTransaction {
        let transaction = self.begin_with_isolation(isolation_level);
        ConcurrentTransaction::new(transaction, self.lock_manager.clone())
    }

    /// Get the default isolation level.
    pub fn default_isolation_level(&self) -> IsolationLevel {
        self.default_isolation_level
    }

    /// Get lock manager statistics.
    pub fn lock_statistics(&self) -> LockStatistics {
        self.lock_manager.get_lock_statistics()
    }

    /// Get the lock manager reference.
    pub fn lock_manager(&self) -> &Arc<LockManager> {
        &self.lock_manager
    }

    // ========================================================================
    // MVCC Transaction Methods
    // Satisfies: RT-2 (ACID compliance), B4 (ACID transactions), TN4 (optimistic concurrency)
    // ========================================================================

    /// Begin an MVCC-enabled transaction.
    /// Returns transaction ID for use with other mvcc_* methods.
    pub fn begin_mvcc(&self) -> Result<TransactionId> {
        let mvcc = self.mvcc_manager.as_ref().ok_or_else(|| {
            GraphError::Transaction("MVCC not enabled. Use new_with_mvcc()".to_string())
        })?;
        Ok(mvcc.begin(self.default_isolation_level))
    }

    /// Begin an MVCC transaction with specific isolation level.
    pub fn begin_mvcc_with_isolation(
        &self,
        isolation_level: IsolationLevel,
    ) -> Result<TransactionId> {
        let mvcc = self.mvcc_manager.as_ref().ok_or_else(|| {
            GraphError::Transaction("MVCC not enabled. Use new_with_mvcc()".to_string())
        })?;
        Ok(mvcc.begin(isolation_level))
    }

    /// Read a node within an MVCC transaction.
    /// Returns the version visible at the transaction's start timestamp.
    pub fn mvcc_read_node(
        &self,
        tx_id: TransactionId,
        node_id: crate::graph::Id,
    ) -> Result<Option<Node>> {
        let mvcc = self
            .mvcc_manager
            .as_ref()
            .ok_or_else(|| GraphError::Transaction("MVCC not enabled".to_string()))?;
        mvcc.read_node(tx_id, node_id)
    }

    /// Read a relationship within an MVCC transaction.
    pub fn mvcc_read_relationship(
        &self,
        tx_id: TransactionId,
        rel_id: crate::graph::Id,
    ) -> Result<Option<Relationship>> {
        let mvcc = self
            .mvcc_manager
            .as_ref()
            .ok_or_else(|| GraphError::Transaction("MVCC not enabled".to_string()))?;
        mvcc.read_relationship(tx_id, rel_id)
    }

    /// Write a node within an MVCC transaction.
    /// Creates a new version that becomes visible on commit.
    pub fn mvcc_write_node(&self, tx_id: TransactionId, node: Node) -> Result<()> {
        let mvcc = self
            .mvcc_manager
            .as_ref()
            .ok_or_else(|| GraphError::Transaction("MVCC not enabled".to_string()))?;
        mvcc.write_node(tx_id, node)
    }

    /// Write a relationship within an MVCC transaction.
    pub fn mvcc_write_relationship(&self, tx_id: TransactionId, rel: Relationship) -> Result<()> {
        let mvcc = self
            .mvcc_manager
            .as_ref()
            .ok_or_else(|| GraphError::Transaction("MVCC not enabled".to_string()))?;
        mvcc.write_relationship(tx_id, rel)
    }

    /// Commit an MVCC transaction with conflict detection.
    /// Fails if write-write or read-write conflicts are detected.
    pub fn mvcc_commit(&self, tx_id: TransactionId) -> Result<()> {
        let mvcc = self
            .mvcc_manager
            .as_ref()
            .ok_or_else(|| GraphError::Transaction("MVCC not enabled".to_string()))?;
        mvcc.commit(tx_id)
    }

    /// Rollback an MVCC transaction.
    /// Discards all uncommitted changes.
    pub fn mvcc_rollback(&self, tx_id: TransactionId) -> Result<()> {
        let mvcc = self
            .mvcc_manager
            .as_ref()
            .ok_or_else(|| GraphError::Transaction("MVCC not enabled".to_string()))?;
        mvcc.rollback(tx_id)
    }

    /// Get MVCC statistics.
    pub fn mvcc_stats(&self) -> Option<MvccStats> {
        self.mvcc_manager.as_ref().map(|m| m.get_stats())
    }

    /// Run MVCC garbage collection.
    pub fn mvcc_gc(&self) {
        if let Some(mvcc) = &self.mvcc_manager {
            mvcc.gc();
        }
    }
}

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

/// Get the current timestamp for transaction ordering.
///
/// In a production system, this would be a high-resolution timestamp
/// or a logical clock for proper ordering of concurrent operations.
fn current_timestamp() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos() as u64
}

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

    #[test]
    fn test_transaction_lifecycle() {
        let manager = TransactionManager::new(IsolationLevel::ReadCommitted);
        let mut tx = manager.begin();

        assert_eq!(tx.id(), 1);
        assert_eq!(tx.isolation_level(), IsolationLevel::ReadCommitted);
        assert_eq!(tx.state(), TransactionState::Active);
        assert!(tx.is_active());
        assert!(tx.write_timestamp().is_none());

        // Mark as having writes
        tx.mark_write();
        assert!(tx.write_timestamp().is_some());

        // Commit
        tx.commit().unwrap();
        assert_eq!(tx.state(), TransactionState::Committed);
        assert!(!tx.is_active());

        // Cannot commit again
        assert!(tx.commit().is_err());
    }

    #[test]
    fn test_transaction_rollback() {
        let manager = TransactionManager::new(IsolationLevel::ReadCommitted);
        let mut tx = manager.begin();

        assert!(tx.is_active());

        // Rollback
        tx.rollback().unwrap();
        assert_eq!(tx.state(), TransactionState::RolledBack);
        assert!(!tx.is_active());

        // Cannot rollback again
        assert!(tx.rollback().is_err());
    }

    #[test]
    fn test_transaction_manager() {
        let manager = TransactionManager::new(IsolationLevel::Serializable);

        let tx1 = manager.begin();
        let tx2 = manager.begin();

        assert_eq!(tx1.id(), 1);
        assert_eq!(tx2.id(), 2);
        assert_eq!(tx1.isolation_level(), IsolationLevel::Serializable);
        assert_eq!(tx2.isolation_level(), IsolationLevel::Serializable);

        // Test custom isolation level
        let tx3 = manager.begin_with_isolation(IsolationLevel::ReadUncommitted);
        assert_eq!(tx3.id(), 3);
        assert_eq!(tx3.isolation_level(), IsolationLevel::ReadUncommitted);
    }
}