graph_d 1.3.2

//! Multi-Version Concurrency Control (MVCC) implementation.
//!
//! Satisfies: RT-2 (Transactions MUST be ACID compliant)
//! Satisfies: B4 (ACID compliance for all transactions)
//! Satisfies: T6 (Thread-safe concurrent read operations)
//! Implements: TN4 resolution (Optimistic concurrency + configurable isolation)
//! Phase: C (Transaction Completion)
//!
//! # Overview
//!
//! MVCC allows multiple transactions to read and write data concurrently
//! without blocking each other. Each transaction sees a consistent snapshot
//! of the database as of its start time, and conflicts are detected at commit.
//!
//! # Architecture
//!
//! ```text
//! ┌────────────────────────────────────────────────────────────────────┐
//! │                    MVCC Transaction Flow                           │
//! │                                                                    │
//! │  Begin TX (t=100)                                                  │
//! │       │                                                            │
//! │       ▼                                                            │
//! │  ┌─────────────┐     ┌─────────────┐     ┌─────────────┐          │
//! │  │ Read Node 1 │ ──▶ │ Write Node 2│ ──▶ │   Commit    │          │
//! │  │ (sees v@t<100)│   │ (creates v@100)│  │(validate+write)│       │
//! │  └─────────────┘     └─────────────┘     └─────────────┘          │
//! │                                                                    │
//! │  Version Chain: Node 1                                             │
//! │  ┌────────┐    ┌────────┐    ┌────────┐                           │
//! │  │ v3@150 │ ◀─ │ v2@100 │ ◀─ │ v1@50  │                           │
//! │  └────────┘    └────────┘    └────────┘                           │
//! └────────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Isolation Levels
//!
//! - **Read Uncommitted**: See all writes (even uncommitted) - not recommended
//! - **Read Committed**: See only committed data at statement time
//! - **Repeatable Read**: See only committed data at transaction start
//! - **Serializable**: Full isolation with conflict detection
//!
//! # Example
//!
//! ```rust,ignore
//! use graph_d::transaction::mvcc::{MvccManager, MvccTransaction};
//! use graph_d::transaction::IsolationLevel;
//!
//! let mvcc = MvccManager::new();
//!
//! // Begin transaction
//! let mut tx = mvcc.begin(IsolationLevel::RepeatableRead);
//!
//! // Read sees consistent snapshot
//! let node = tx.read_node(1)?;
//!
//! // Write creates new version
//! tx.write_node(1, updated_node)?;
//!
//! // Commit validates and applies
//! tx.commit()?; // Fails if concurrent write to node 1
//! ```

use crate::error::{GraphError, Result};
use crate::graph::{Id, Node, Relationship};
use crate::transaction::{IsolationLevel, TransactionId, TransactionState};
use parking_lot::RwLock;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

/// Timestamp type for version ordering.
/// Satisfies: Monotonically increasing for correct ordering
pub type Timestamp = u64;

/// A versioned value with creation and deletion timestamps.
/// Satisfies: RT-2 (version history for isolation)
#[derive(Debug, Clone)]
pub struct VersionedValue<T> {
    /// The actual value
    pub value: T,

    /// Transaction that created this version
    pub created_by: TransactionId,

    /// Timestamp when this version was created
    pub created_at: Timestamp,

    /// Transaction that deleted this version (if any)
    pub deleted_by: Option<TransactionId>,

    /// Timestamp when this version was deleted (if any)
    pub deleted_at: Option<Timestamp>,
}

impl<T> VersionedValue<T> {
    /// Create a new version.
    pub fn new(value: T, tx_id: TransactionId, timestamp: Timestamp) -> Self {
        VersionedValue {
            value,
            created_by: tx_id,
            created_at: timestamp,
            deleted_by: None,
            deleted_at: None,
        }
    }

    /// Check if this version is visible to a transaction at the given timestamp.
    /// Satisfies: Isolation level semantics
    pub fn is_visible_at(
        &self,
        timestamp: Timestamp,
        committed_txs: &HashSet<TransactionId>,
    ) -> bool {
        // Version must be created by a committed transaction before our timestamp
        let created_visible =
            self.created_at <= timestamp && committed_txs.contains(&self.created_by);

        // Version must not be deleted, or deleted after our timestamp
        let not_deleted = match (self.deleted_at, self.deleted_by) {
            (Some(del_ts), Some(del_tx)) => del_ts > timestamp || !committed_txs.contains(&del_tx),
            _ => true,
        };

        created_visible && not_deleted
    }

    /// Mark this version as deleted.
    pub fn mark_deleted(&mut self, tx_id: TransactionId, timestamp: Timestamp) {
        self.deleted_by = Some(tx_id);
        self.deleted_at = Some(timestamp);
    }
}

/// Version chain for an entity (node or relationship).
/// Maintains multiple versions ordered by timestamp.
/// Satisfies: RT-2 (historical versions for repeatable reads)
#[derive(Debug)]
pub struct VersionChain<T> {
    /// Versions ordered by creation timestamp (newest first)
    versions: BTreeMap<Timestamp, VersionedValue<T>>,
}

impl<T: Clone> VersionChain<T> {
    /// Create a new empty version chain.
    pub fn new() -> Self {
        VersionChain {
            versions: BTreeMap::new(),
        }
    }

    /// Add a new version to the chain.
    pub fn add_version(&mut self, version: VersionedValue<T>) {
        self.versions.insert(version.created_at, version);
    }

    /// Find the visible version at the given timestamp.
    /// Satisfies: Correct version selection for isolation level
    pub fn find_visible_version(
        &self,
        timestamp: Timestamp,
        committed_txs: &HashSet<TransactionId>,
    ) -> Option<&VersionedValue<T>> {
        // Iterate from newest to oldest, find first visible version
        self.versions
            .iter()
            .rev()
            .map(|(_, version)| version)
            .find(|&version| version.is_visible_at(timestamp, committed_txs))
    }

    /// Get the latest version (regardless of visibility).
    pub fn latest_version(&self) -> Option<&VersionedValue<T>> {
        self.versions.values().next_back()
    }

    /// Find a version created by a specific transaction (for read-your-own-writes).
    pub fn find_version_by_tx(&self, tx_id: TransactionId) -> Option<&VersionedValue<T>> {
        self.versions
            .values()
            .rev()
            .find(|v| v.created_by == tx_id && v.deleted_at.is_none())
    }

    /// Mark the latest version as deleted.
    pub fn mark_latest_deleted(&mut self, tx_id: TransactionId, timestamp: Timestamp) -> bool {
        if let Some((_, version)) = self.versions.iter_mut().next_back() {
            version.mark_deleted(tx_id, timestamp);
            true
        } else {
            false
        }
    }

    /// Garbage collect old versions that are no longer needed.
    /// Satisfies: Memory efficiency (don't keep infinite history)
    pub fn gc(&mut self, oldest_active_timestamp: Timestamp) {
        // Keep versions that might still be visible to active transactions
        // Remove versions where both created_at and deleted_at are before oldest_active
        self.versions.retain(|_, version| {
            version.created_at >= oldest_active_timestamp
                || version
                    .deleted_at
                    .map(|t| t >= oldest_active_timestamp)
                    .unwrap_or(true)
        });
    }
}

impl<T: Clone> Default for VersionChain<T> {
    fn default() -> Self {
        Self::new()
    }
}

/// MVCC storage for nodes and relationships.
/// Satisfies: RT-2 (version storage for all data)
#[derive(Debug)]
pub struct MvccStore {
    /// Versioned nodes
    nodes: RwLock<HashMap<Id, VersionChain<Node>>>,

    /// Versioned relationships
    relationships: RwLock<HashMap<Id, VersionChain<Relationship>>>,

    /// Set of committed transaction IDs
    committed_txs: RwLock<HashSet<TransactionId>>,

    /// Oldest active transaction timestamp (for GC)
    oldest_active: AtomicU64,
}

impl MvccStore {
    /// Create a new MVCC store.
    pub fn new() -> Self {
        let mut committed = HashSet::new();
        committed.insert(0); // System "transaction" for initial data

        MvccStore {
            nodes: RwLock::new(HashMap::new()),
            relationships: RwLock::new(HashMap::new()),
            committed_txs: RwLock::new(committed),
            oldest_active: AtomicU64::new(0),
        }
    }

    /// Read a node at the given timestamp for a specific transaction.
    /// Satisfies: RT-2 (consistent read at transaction's snapshot)
    /// Also returns uncommitted writes from the current transaction (read-your-own-writes).
    pub fn read_node(
        &self,
        id: Id,
        timestamp: Timestamp,
        current_tx_id: Option<TransactionId>,
    ) -> Option<Node> {
        let nodes = self.nodes.read();
        let committed = self.committed_txs.read();

        nodes.get(&id).and_then(|chain| {
            // First check for uncommitted write by current transaction
            if let Some(tx_id) = current_tx_id {
                if let Some(version) = chain.find_version_by_tx(tx_id) {
                    return Some(version.value.clone());
                }
            }
            // Fall back to normal visibility check
            chain
                .find_visible_version(timestamp, &committed)
                .map(|v| v.value.clone())
        })
    }

    /// Read a relationship at the given timestamp for a specific transaction.
    /// Also returns uncommitted writes from the current transaction (read-your-own-writes).
    pub fn read_relationship(
        &self,
        id: Id,
        timestamp: Timestamp,
        current_tx_id: Option<TransactionId>,
    ) -> Option<Relationship> {
        let relationships = self.relationships.read();
        let committed = self.committed_txs.read();

        relationships.get(&id).and_then(|chain| {
            // First check for uncommitted write by current transaction
            if let Some(tx_id) = current_tx_id {
                if let Some(version) = chain.find_version_by_tx(tx_id) {
                    return Some(version.value.clone());
                }
            }
            // Fall back to normal visibility check
            chain
                .find_visible_version(timestamp, &committed)
                .map(|v| v.value.clone())
        })
    }

    /// Write a new node version (uncommitted).
    /// Satisfies: RT-2 (versioned writes)
    pub fn write_node(&self, node: Node, tx_id: TransactionId, timestamp: Timestamp) {
        let mut nodes = self.nodes.write();
        let id = node.id;

        let version = VersionedValue::new(node, tx_id, timestamp);

        nodes.entry(id).or_default().add_version(version);
    }

    /// Write a new relationship version (uncommitted).
    pub fn write_relationship(
        &self,
        rel: Relationship,
        tx_id: TransactionId,
        timestamp: Timestamp,
    ) {
        let mut relationships = self.relationships.write();
        let id = rel.id;

        let version = VersionedValue::new(rel, tx_id, timestamp);

        relationships.entry(id).or_default().add_version(version);
    }

    /// Delete a node (mark latest version as deleted).
    pub fn delete_node(&self, id: Id, tx_id: TransactionId, timestamp: Timestamp) -> bool {
        let mut nodes = self.nodes.write();
        if let Some(chain) = nodes.get_mut(&id) {
            chain.mark_latest_deleted(tx_id, timestamp)
        } else {
            false
        }
    }

    /// Delete a relationship.
    pub fn delete_relationship(&self, id: Id, tx_id: TransactionId, timestamp: Timestamp) -> bool {
        let mut relationships = self.relationships.write();
        if let Some(chain) = relationships.get_mut(&id) {
            chain.mark_latest_deleted(tx_id, timestamp)
        } else {
            false
        }
    }

    /// Mark a transaction as committed.
    /// Satisfies: RT-2 (commit makes writes visible)
    pub fn commit_transaction(&self, tx_id: TransactionId) {
        self.committed_txs.write().insert(tx_id);
    }

    /// Check if a transaction is committed.
    pub fn is_committed(&self, tx_id: TransactionId) -> bool {
        self.committed_txs.read().contains(&tx_id)
    }

    /// Update the oldest active transaction timestamp for GC.
    pub fn update_oldest_active(&self, timestamp: Timestamp) {
        self.oldest_active.store(timestamp, Ordering::SeqCst);
    }

    /// Run garbage collection on version chains.
    pub fn gc(&self) {
        let oldest = self.oldest_active.load(Ordering::SeqCst);

        {
            let mut nodes = self.nodes.write();
            for chain in nodes.values_mut() {
                chain.gc(oldest);
            }
        }

        {
            let mut relationships = self.relationships.write();
            for chain in relationships.values_mut() {
                chain.gc(oldest);
            }
        }
    }
}

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

/// Write set entry for conflict detection.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum WriteSetEntry {
    /// A node that was written by the transaction
    Node(Id),
    /// A relationship that was written by the transaction
    Relationship(Id),
}

/// Read set entry for serializable conflict detection.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum ReadSetEntry {
    /// A node that was read by the transaction
    Node(Id),
    /// A relationship that was read by the transaction
    Relationship(Id),
}

/// MVCC transaction for optimistic concurrency.
/// Satisfies: TN4 resolution (optimistic concurrency control)
#[derive(Debug)]
pub struct MvccTransaction {
    /// Transaction ID
    pub id: TransactionId,

    /// Isolation level for this transaction
    pub isolation_level: IsolationLevel,

    /// Transaction state
    pub state: TransactionState,

    /// Start timestamp (for snapshot isolation)
    pub start_timestamp: Timestamp,

    /// Commit timestamp (assigned at commit time)
    pub commit_timestamp: Option<Timestamp>,

    /// Write set for conflict detection
    write_set: HashSet<WriteSetEntry>,

    /// Read set for serializable conflict detection
    read_set: HashSet<ReadSetEntry>,
}

impl MvccTransaction {
    /// Create a new MVCC transaction.
    pub fn new(
        id: TransactionId,
        isolation_level: IsolationLevel,
        start_timestamp: Timestamp,
    ) -> Self {
        MvccTransaction {
            id,
            isolation_level,
            state: TransactionState::Active,
            start_timestamp,
            commit_timestamp: None,
            write_set: HashSet::new(),
            read_set: HashSet::new(),
        }
    }

    /// Record a node read (for serializable isolation).
    pub fn record_read_node(&mut self, id: Id) {
        if self.isolation_level == IsolationLevel::Serializable {
            self.read_set.insert(ReadSetEntry::Node(id));
        }
    }

    /// Record a relationship read.
    pub fn record_read_relationship(&mut self, id: Id) {
        if self.isolation_level == IsolationLevel::Serializable {
            self.read_set.insert(ReadSetEntry::Relationship(id));
        }
    }

    /// Record a node write.
    pub fn record_write_node(&mut self, id: Id) {
        self.write_set.insert(WriteSetEntry::Node(id));
    }

    /// Record a relationship write.
    pub fn record_write_relationship(&mut self, id: Id) {
        self.write_set.insert(WriteSetEntry::Relationship(id));
    }

    /// Get the write set.
    pub fn write_set(&self) -> &HashSet<WriteSetEntry> {
        &self.write_set
    }

    /// Get the read set.
    pub fn read_set(&self) -> &HashSet<ReadSetEntry> {
        &self.read_set
    }

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

/// MVCC Manager coordinates transactions and conflict detection.
/// Satisfies: RT-2 (transaction coordination for ACID)
pub struct MvccManager {
    /// MVCC store
    store: Arc<MvccStore>,

    /// Next transaction ID
    next_tx_id: AtomicU64,

    /// Global timestamp counter
    timestamp_counter: AtomicU64,

    /// Active transactions (for conflict detection and GC)
    active_txs: RwLock<HashMap<TransactionId, MvccTransaction>>,

    /// Recently committed transactions (for conflict detection window)
    recent_commits: RwLock<Vec<(TransactionId, Timestamp, HashSet<WriteSetEntry>)>>,

    /// How long to keep recent commits for conflict detection
    conflict_window: u64,
}

impl MvccManager {
    /// Create a new MVCC manager.
    pub fn new() -> Self {
        MvccManager {
            store: Arc::new(MvccStore::new()),
            next_tx_id: AtomicU64::new(1),
            timestamp_counter: AtomicU64::new(1),
            active_txs: RwLock::new(HashMap::new()),
            recent_commits: RwLock::new(Vec::new()),
            conflict_window: 10_000, // Keep last 10k timestamps
        }
    }

    /// Begin a new transaction.
    /// Satisfies: RT-2 (transaction isolation from start)
    pub fn begin(&self, isolation_level: IsolationLevel) -> TransactionId {
        let tx_id = self.next_tx_id.fetch_add(1, Ordering::SeqCst);
        let timestamp = self.timestamp_counter.fetch_add(1, Ordering::SeqCst);

        let tx = MvccTransaction::new(tx_id, isolation_level, timestamp);
        self.active_txs.write().insert(tx_id, tx);

        tx_id
    }

    /// Get the timestamp for a transaction to use for reads.
    pub fn get_read_timestamp(&self, tx_id: TransactionId) -> Option<Timestamp> {
        self.active_txs
            .read()
            .get(&tx_id)
            .map(|tx| tx.start_timestamp)
    }

    /// Read a node within a transaction.
    /// Supports read-your-own-writes: uncommitted writes by this transaction are visible.
    pub fn read_node(&self, tx_id: TransactionId, id: Id) -> Result<Option<Node>> {
        let mut active = self.active_txs.write();
        let tx = active
            .get_mut(&tx_id)
            .ok_or_else(|| GraphError::Transaction(format!("Transaction {tx_id} not found")))?;

        if tx.state != TransactionState::Active {
            return Err(GraphError::Transaction(
                "Transaction not active".to_string(),
            ));
        }

        tx.record_read_node(id);
        Ok(self.store.read_node(id, tx.start_timestamp, Some(tx_id)))
    }

    /// Read a relationship within a transaction.
    /// Supports read-your-own-writes: uncommitted writes by this transaction are visible.
    pub fn read_relationship(&self, tx_id: TransactionId, id: Id) -> Result<Option<Relationship>> {
        let mut active = self.active_txs.write();
        let tx = active
            .get_mut(&tx_id)
            .ok_or_else(|| GraphError::Transaction(format!("Transaction {tx_id} not found")))?;

        if tx.state != TransactionState::Active {
            return Err(GraphError::Transaction(
                "Transaction not active".to_string(),
            ));
        }

        tx.record_read_relationship(id);
        Ok(self
            .store
            .read_relationship(id, tx.start_timestamp, Some(tx_id)))
    }

    /// Write a node within a transaction.
    pub fn write_node(&self, tx_id: TransactionId, node: Node) -> Result<()> {
        let timestamp = self.timestamp_counter.fetch_add(1, Ordering::SeqCst);

        let mut active = self.active_txs.write();
        let tx = active
            .get_mut(&tx_id)
            .ok_or_else(|| GraphError::Transaction(format!("Transaction {tx_id} not found")))?;

        if tx.state != TransactionState::Active {
            return Err(GraphError::Transaction(
                "Transaction not active".to_string(),
            ));
        }

        let id = node.id;
        tx.record_write_node(id);
        self.store.write_node(node, tx_id, timestamp);

        Ok(())
    }

    /// Write a relationship within a transaction.
    pub fn write_relationship(&self, tx_id: TransactionId, rel: Relationship) -> Result<()> {
        let timestamp = self.timestamp_counter.fetch_add(1, Ordering::SeqCst);

        let mut active = self.active_txs.write();
        let tx = active
            .get_mut(&tx_id)
            .ok_or_else(|| GraphError::Transaction(format!("Transaction {tx_id} not found")))?;

        if tx.state != TransactionState::Active {
            return Err(GraphError::Transaction(
                "Transaction not active".to_string(),
            ));
        }

        let id = rel.id;
        tx.record_write_relationship(id);
        self.store.write_relationship(rel, tx_id, timestamp);

        Ok(())
    }

    /// Commit a transaction with conflict detection.
    /// Satisfies: TN4 resolution (detect conflicts at commit)
    pub fn commit(&self, tx_id: TransactionId) -> Result<()> {
        let commit_timestamp = self.timestamp_counter.fetch_add(1, Ordering::SeqCst);

        let mut active = self.active_txs.write();
        let tx = active
            .get_mut(&tx_id)
            .ok_or_else(|| GraphError::Transaction(format!("Transaction {tx_id} not found")))?;

        if tx.state != TransactionState::Active {
            return Err(GraphError::Transaction(
                "Transaction not active".to_string(),
            ));
        }

        // Conflict detection
        if tx.has_writes() {
            self.detect_conflicts(tx)?;
        }

        // Mark transaction as committed
        tx.state = TransactionState::Committed;
        tx.commit_timestamp = Some(commit_timestamp);

        // Make writes visible
        self.store.commit_transaction(tx_id);

        // Record for future conflict detection
        if tx.has_writes() {
            let write_set = tx.write_set().clone();
            self.recent_commits
                .write()
                .push((tx_id, commit_timestamp, write_set));
        }

        // Remove from active transactions
        let _tx = active.remove(&tx_id);
        drop(active);

        // Update oldest active for GC
        self.update_oldest_active();

        // Cleanup old commits
        self.cleanup_recent_commits(commit_timestamp);

        Ok(())
    }

    /// Rollback a transaction.
    /// Satisfies: RT-2 (atomicity - all or nothing)
    pub fn rollback(&self, tx_id: TransactionId) -> Result<()> {
        let mut active = self.active_txs.write();
        let tx = active
            .get_mut(&tx_id)
            .ok_or_else(|| GraphError::Transaction(format!("Transaction {tx_id} not found")))?;

        if tx.state != TransactionState::Active {
            return Err(GraphError::Transaction(
                "Transaction not active".to_string(),
            ));
        }

        tx.state = TransactionState::RolledBack;

        // Note: In a full implementation, we would also need to remove
        // the uncommitted versions from the store. For simplicity, we
        // rely on the visibility rules (uncommitted txs are not visible).

        active.remove(&tx_id);
        drop(active);

        self.update_oldest_active();

        Ok(())
    }

    /// Detect write-write conflicts.
    /// Satisfies: TN4 (optimistic concurrency - detect at commit)
    fn detect_conflicts(&self, tx: &MvccTransaction) -> Result<()> {
        let recent = self.recent_commits.read();

        for (other_tx_id, commit_ts, write_set) in recent.iter() {
            // Skip our own transaction
            if *other_tx_id == tx.id {
                continue;
            }

            // Check if this transaction committed after we started
            if *commit_ts > tx.start_timestamp {
                // Check for write-write conflict
                for entry in tx.write_set() {
                    if write_set.contains(entry) {
                        return Err(GraphError::Concurrency(format!(
                            "Write-write conflict on {entry:?} with transaction {other_tx_id}"
                        )));
                    }
                }

                // For serializable, also check read-write conflicts
                if tx.isolation_level == IsolationLevel::Serializable {
                    for entry in tx.read_set() {
                        let conflicting_write = match entry {
                            ReadSetEntry::Node(id) => write_set.contains(&WriteSetEntry::Node(*id)),
                            ReadSetEntry::Relationship(id) => {
                                write_set.contains(&WriteSetEntry::Relationship(*id))
                            }
                        };

                        if conflicting_write {
                            return Err(GraphError::Concurrency(format!(
                                "Read-write conflict on {entry:?} with transaction {other_tx_id}"
                            )));
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Update the oldest active transaction timestamp.
    fn update_oldest_active(&self) {
        let active = self.active_txs.read();
        let oldest = active
            .values()
            .map(|tx| tx.start_timestamp)
            .min()
            .unwrap_or(self.timestamp_counter.load(Ordering::SeqCst));

        self.store.update_oldest_active(oldest);
    }

    /// Cleanup old recent commits outside the conflict window.
    fn cleanup_recent_commits(&self, current_ts: Timestamp) {
        let mut recent = self.recent_commits.write();
        let cutoff = current_ts.saturating_sub(self.conflict_window);
        recent.retain(|(_, ts, _)| *ts >= cutoff);
    }

    /// Run garbage collection.
    pub fn gc(&self) {
        self.store.gc();
    }

    /// Get statistics about active transactions.
    pub fn get_stats(&self) -> MvccStats {
        let active = self.active_txs.read();
        let recent = self.recent_commits.read();

        MvccStats {
            active_transactions: active.len(),
            recent_commits: recent.len(),
            current_timestamp: self.timestamp_counter.load(Ordering::SeqCst),
        }
    }
}

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

/// MVCC statistics.
#[derive(Debug, Clone)]
pub struct MvccStats {
    /// Number of active transactions
    pub active_transactions: usize,

    /// Number of recent commits in conflict detection window
    pub recent_commits: usize,

    /// Current global timestamp
    pub current_timestamp: Timestamp,
}

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

    fn make_node(id: Id) -> Node {
        Node {
            id,
            properties: HashMap::new(),
        }
    }

    /// Test: Basic read-write transaction
    /// Validates: RT-2 (basic transaction flow)
    #[test]
    fn test_basic_transaction() {
        let mvcc = MvccManager::new();

        let tx_id = mvcc.begin(IsolationLevel::ReadCommitted);

        // Write a node
        mvcc.write_node(tx_id, make_node(1)).unwrap();

        // Read it back (should see our own write)
        let node = mvcc.read_node(tx_id, 1).unwrap();
        assert!(node.is_some());

        // Commit
        mvcc.commit(tx_id).unwrap();
    }

    /// Test: Isolation between transactions
    /// Validates: RT-2 (transaction isolation)
    #[test]
    fn test_isolation() {
        let mvcc = MvccManager::new();

        // TX1 writes node 1
        let tx1 = mvcc.begin(IsolationLevel::RepeatableRead);
        mvcc.write_node(tx1, make_node(1)).unwrap();
        mvcc.commit(tx1).unwrap();

        // TX2 starts after TX1 commits
        let tx2 = mvcc.begin(IsolationLevel::RepeatableRead);

        // TX3 starts, writes node 1 with new value
        let tx3 = mvcc.begin(IsolationLevel::RepeatableRead);
        let mut node = make_node(1);
        node.properties
            .insert("version".to_string(), serde_json::json!(2));
        mvcc.write_node(tx3, node).unwrap();
        mvcc.commit(tx3).unwrap();

        // TX2 should still see the old version (repeatable read)
        let node = mvcc.read_node(tx2, 1).unwrap().unwrap();
        assert!(!node.properties.contains_key("version"));

        mvcc.commit(tx2).unwrap();
    }

    /// Test: Write-write conflict detection
    /// Validates: TN4 (optimistic concurrency conflict detection)
    #[test]
    fn test_write_write_conflict() {
        let mvcc = MvccManager::new();

        // Setup: create node 1
        let setup_tx = mvcc.begin(IsolationLevel::ReadCommitted);
        mvcc.write_node(setup_tx, make_node(1)).unwrap();
        mvcc.commit(setup_tx).unwrap();

        // TX1 and TX2 both start
        let tx1 = mvcc.begin(IsolationLevel::RepeatableRead);
        let tx2 = mvcc.begin(IsolationLevel::RepeatableRead);

        // Both read node 1
        let _n1 = mvcc.read_node(tx1, 1).unwrap();
        let _n2 = mvcc.read_node(tx2, 1).unwrap();

        // TX1 writes and commits first
        let mut node1 = make_node(1);
        node1
            .properties
            .insert("writer".to_string(), serde_json::json!("tx1"));
        mvcc.write_node(tx1, node1).unwrap();
        mvcc.commit(tx1).unwrap();

        // TX2 tries to write the same node
        let mut node2 = make_node(1);
        node2
            .properties
            .insert("writer".to_string(), serde_json::json!("tx2"));
        mvcc.write_node(tx2, node2).unwrap();

        // TX2 commit should fail with conflict
        let result = mvcc.commit(tx2);
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("conflict"));
    }

    /// Test: Serializable read-write conflict
    /// Validates: Serializable isolation level
    #[test]
    fn test_serializable_read_write_conflict() {
        let mvcc = MvccManager::new();

        // Setup: create node 1
        let setup_tx = mvcc.begin(IsolationLevel::Serializable);
        mvcc.write_node(setup_tx, make_node(1)).unwrap();
        mvcc.commit(setup_tx).unwrap();

        // TX1 reads node 1
        let tx1 = mvcc.begin(IsolationLevel::Serializable);
        let _n = mvcc.read_node(tx1, 1).unwrap();

        // TX2 writes node 1 and commits
        let tx2 = mvcc.begin(IsolationLevel::Serializable);
        mvcc.write_node(tx2, make_node(1)).unwrap();
        mvcc.commit(tx2).unwrap();

        // TX1 tries to write something else and commit
        // Should fail because something it read was modified
        mvcc.write_node(tx1, make_node(2)).unwrap();
        let result = mvcc.commit(tx1);
        assert!(result.is_err());
    }

    /// Test: Rollback
    /// Validates: RT-2 (atomicity - rollback discards changes)
    #[test]
    fn test_rollback() {
        let mvcc = MvccManager::new();

        let tx1 = mvcc.begin(IsolationLevel::ReadCommitted);
        mvcc.write_node(tx1, make_node(1)).unwrap();
        mvcc.rollback(tx1).unwrap();

        // New transaction should not see the rolled back node
        let tx2 = mvcc.begin(IsolationLevel::ReadCommitted);
        let node = mvcc.read_node(tx2, 1).unwrap();
        assert!(node.is_none());
    }
}