oxirs-core 0.4.0

//! ACID transaction implementation with full isolation and durability guarantees

use super::{IsolationLevel, MvccSnapshot, WalEntry, WriteAheadLog};
use crate::model::Quad;
use crate::OxirsError;
use scirs2_core::metrics::{Counter, Timer};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::{Arc, RwLock};

/// Transaction ID (monotonically increasing)
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct TransactionId(pub u64);

impl TransactionId {
    /// Get the raw transaction ID
    pub fn raw(&self) -> u64 {
        self.0
    }
}

/// Transaction state
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransactionState {
    /// Transaction is active
    Active,
    /// Transaction is preparing to commit
    Preparing,
    /// Transaction committed successfully
    Committed,
    /// Transaction was aborted
    Aborted,
}

/// ACID transaction with full guarantees
pub struct AcidTransaction {
    /// Transaction ID
    id: TransactionId,
    /// Current state
    state: TransactionState,
    /// Isolation level
    isolation: IsolationLevel,
    /// MVCC snapshot (for snapshot isolation)
    #[allow(dead_code)]
    snapshot: Option<MvccSnapshot>,
    /// Pending quad insertions
    pending_inserts: Vec<Quad>,
    /// Pending quad deletions
    pending_deletes: Vec<Quad>,
    /// Read set (for conflict detection)
    read_set: HashMap<Quad, u64>,
    /// Write set (for validation)
    write_set: HashMap<Quad, QuadVersion>,
    /// Write-Ahead Log
    wal: Arc<RwLock<WriteAheadLog>>,
    /// Optional backing store to which committed inserts/deletes are applied
    /// during [`AcidTransaction::commit`]. When `None`, the transaction only
    /// records durability entries in the WAL (e.g. for tests or callers that
    /// apply the pending sets themselves via [`AcidTransaction::pending_inserts`]).
    store: Option<Arc<dyn crate::Store>>,
    /// Commit counter
    commit_counter: Arc<Counter>,
    /// Abort counter
    abort_counter: Arc<Counter>,
    /// Commit timer
    commit_timer: Arc<Timer>,
}

/// Versioned quad for MVCC
#[derive(Debug, Clone)]
struct QuadVersion {
    /// The quad
    #[allow(dead_code)]
    quad: Quad,
    /// Version number
    version: u64,
    /// Creating transaction ID
    #[allow(dead_code)]
    created_by: TransactionId,
    /// Deleting transaction ID (if deleted)
    deleted_by: Option<TransactionId>,
}

impl AcidTransaction {
    /// Create a new ACID transaction
    pub(super) fn new(
        id: TransactionId,
        isolation: IsolationLevel,
        snapshot: Option<MvccSnapshot>,
        wal: Arc<RwLock<WriteAheadLog>>,
        commit_counter: Arc<Counter>,
        abort_counter: Arc<Counter>,
        commit_timer: Arc<Timer>,
    ) -> Self {
        Self {
            id,
            state: TransactionState::Active,
            isolation,
            snapshot,
            pending_inserts: Vec::new(),
            pending_deletes: Vec::new(),
            read_set: HashMap::new(),
            write_set: HashMap::new(),
            wal,
            store: None,
            commit_counter,
            abort_counter,
            commit_timer,
        }
    }

    /// Attach a backing store so that [`AcidTransaction::commit`] applies the
    /// transaction's pending inserts and deletes to it atomically with the WAL
    /// commit record. Without a store the committed changes would only be
    /// recorded in the WAL and never become visible in the data set.
    pub fn with_store(mut self, store: Arc<dyn crate::Store>) -> Self {
        self.store = Some(store);
        self
    }

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

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

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

    /// Insert a quad into the transaction
    pub fn insert(&mut self, quad: Quad) -> Result<bool, OxirsError> {
        self.check_active()?;

        // Check if already in pending inserts
        if self.pending_inserts.contains(&quad) {
            return Ok(false);
        }

        // Check if in pending deletes (re-insertion)
        if let Some(pos) = self.pending_deletes.iter().position(|q| q == &quad) {
            self.pending_deletes.remove(pos);
            return Ok(true);
        }

        // Record in write set for validation
        self.write_set.insert(
            quad.clone(),
            QuadVersion {
                quad: quad.clone(),
                version: self.id.0,
                created_by: self.id,
                deleted_by: None,
            },
        );

        // Add to pending inserts
        self.pending_inserts.push(quad.clone());

        // Write to WAL for durability
        self.write_to_wal(WalEntry::Insert {
            tx_id: self.id.0,
            quad,
        })?;

        Ok(true)
    }

    /// Delete a quad from the transaction
    pub fn delete(&mut self, quad: Quad) -> Result<bool, OxirsError> {
        self.check_active()?;

        // Check if already in pending deletes
        if self.pending_deletes.contains(&quad) {
            return Ok(false);
        }

        // Check if in pending inserts (delete before commit)
        if let Some(pos) = self.pending_inserts.iter().position(|q| q == &quad) {
            self.pending_inserts.remove(pos);
            return Ok(true);
        }

        // Record in write set for validation
        if let Some(version) = self.write_set.get_mut(&quad) {
            version.deleted_by = Some(self.id);
        } else {
            self.write_set.insert(
                quad.clone(),
                QuadVersion {
                    quad: quad.clone(),
                    version: self.id.0,
                    created_by: self.id,
                    deleted_by: Some(self.id),
                },
            );
        }

        // Add to pending deletes
        self.pending_deletes.push(quad.clone());

        // Write to WAL for durability
        self.write_to_wal(WalEntry::Delete {
            tx_id: self.id.0,
            quad,
        })?;

        Ok(true)
    }

    /// Record a read for conflict detection
    pub fn record_read(&mut self, quad: &Quad) -> Result<(), OxirsError> {
        self.check_active()?;

        // Record in read set with current version
        self.read_set.insert(quad.clone(), self.id.0);

        Ok(())
    }

    /// Validate the transaction before commit
    fn validate(&self) -> Result<(), OxirsError> {
        // For serializable isolation, check for conflicts
        if self.isolation == IsolationLevel::Serializable {
            // Check read-write conflicts
            for (quad, version) in &self.read_set {
                if let Some(write_version) = self.write_set.get(quad) {
                    if write_version.version > *version {
                        return Err(OxirsError::ConcurrencyError(
                            "Read-write conflict detected".to_string(),
                        ));
                    }
                }
            }
        }

        // 1. Check for duplicate inserts within the same transaction
        let mut seen_inserts = std::collections::HashSet::new();
        for quad in &self.pending_inserts {
            if !seen_inserts.insert(quad) {
                return Err(OxirsError::Store(format!(
                    "Duplicate insert detected in transaction: {:?}",
                    quad
                )));
            }
        }

        // 2. Check for conflicting operations (insert and delete of the same quad)
        for insert_quad in &self.pending_inserts {
            for delete_quad in &self.pending_deletes {
                if insert_quad == delete_quad {
                    return Err(OxirsError::Store(format!(
                        "Conflicting insert/delete operations for quad: {:?}",
                        insert_quad
                    )));
                }
            }
        }

        // 3. Check for write-write conflicts in write set
        let mut write_conflicts = 0;
        for (quad, version) in &self.write_set {
            // If a quad has been both created and deleted in this transaction
            if version.deleted_by.is_some() && version.created_by == self.id {
                write_conflicts += 1;
            }

            // Check version consistency
            if version.version > self.id.0 {
                return Err(OxirsError::ConcurrencyError(format!(
                    "Version inconsistency detected for quad: {:?}",
                    quad
                )));
            }
        }

        // 4. Transaction size validation (prevent excessive operations)
        const MAX_PENDING_OPS: usize = 1_000_000; // 1 million operations per transaction
        let total_ops = self.pending_inserts.len() + self.pending_deletes.len();
        if total_ops > MAX_PENDING_OPS {
            return Err(OxirsError::Store(format!(
                "Transaction exceeds maximum operation limit: {} > {}",
                total_ops, MAX_PENDING_OPS
            )));
        }

        // 5. Validate quad components for inserts
        for quad in &self.pending_inserts {
            // Validate subject is not blank if strict validation needed
            // (This is a placeholder - actual validation depends on RDF semantics)

            // Validate predicate is a named node (RDF requirement)
            use crate::model::RdfTerm;
            if quad.predicate().as_str().is_empty() {
                return Err(OxirsError::Store(
                    "Invalid predicate in quad: predicate cannot be empty".to_string(),
                ));
            }

            // Validate object is not null/invalid
            // (Actual validation would check RDF term validity)
        }

        // 6. Check for snapshot isolation violations
        if self.isolation == IsolationLevel::Snapshot {
            // Ensure no reads have been invalidated by concurrent writes
            // This is a simplified check - full implementation would query the snapshot manager
            if !self.read_set.is_empty() && write_conflicts > self.pending_deletes.len() / 2 {
                return Err(OxirsError::ConcurrencyError(
                    "Snapshot isolation violation: too many write conflicts".to_string(),
                ));
            }
        }

        // 7. Validate referential integrity (graph-level constraints)
        // Check that deleted quads actually exist in pending operations or database
        // This is a simplified check - full implementation would query the store
        for delete_quad in &self.pending_deletes {
            // Ensure we're not deleting something we just inserted
            if self.pending_inserts.contains(delete_quad) {
                return Err(OxirsError::Store(format!(
                    "Cannot delete quad that was just inserted: {:?}",
                    delete_quad
                )));
            }
        }

        // 8. Check for potential deadlocks (simplified)
        // In a full implementation, this would use a deadlock detection algorithm
        if self.read_set.len() > 1000 && self.write_set.len() > 1000 {
            tracing::warn!(
                "Transaction {} has large read/write sets - potential deadlock risk",
                self.id.0
            );
        }

        tracing::debug!(
            "Transaction {} validated: {} inserts, {} deletes, {} reads, {} writes",
            self.id.0,
            self.pending_inserts.len(),
            self.pending_deletes.len(),
            self.read_set.len(),
            self.write_set.len()
        );

        Ok(())
    }

    /// Commit the transaction with full ACID guarantees
    pub fn commit(mut self) -> Result<(), OxirsError> {
        self.check_active()?;

        // Start timing the commit
        let _timer_guard = self.commit_timer.start();

        // Transition to preparing state
        self.state = TransactionState::Preparing;

        // Validate before commit
        self.validate()?;

        // Write commit record to WAL (durability)
        self.write_to_wal(WalEntry::Commit { tx_id: self.id.0 })?;

        // Force WAL to disk for durability guarantee
        self.flush_wal()?;

        // Apply the committed changes to the backing store (redo phase). The
        // commit record is already durable, so a crash between here and full
        // application is recoverable by replaying the WAL. Deletions are
        // applied before insertions to mirror the pending-operation ordering.
        if let Some(store) = &self.store {
            for quad in &self.pending_deletes {
                store.remove_quad(quad)?;
            }
            for quad in &self.pending_inserts {
                store.insert_quad(quad.clone())?;
            }
        }

        // Transition to committed state
        self.state = TransactionState::Committed;

        // Update metrics
        self.commit_counter.add(1);

        tracing::info!(
            "Transaction {} committed successfully with {} inserts and {} deletes",
            self.id.0,
            self.pending_inserts.len(),
            self.pending_deletes.len()
        );

        Ok(())
    }

    /// Abort the transaction
    pub fn abort(mut self) -> Result<(), OxirsError> {
        if self.state == TransactionState::Committed {
            return Err(OxirsError::Store(
                "Cannot abort committed transaction".to_string(),
            ));
        }

        // Write abort record to WAL
        self.write_to_wal(WalEntry::Abort { tx_id: self.id.0 })?;

        // Transition to aborted state
        self.state = TransactionState::Aborted;

        // Update metrics
        self.abort_counter.add(1);

        // Clear pending operations
        self.pending_inserts.clear();
        self.pending_deletes.clear();
        self.read_set.clear();
        self.write_set.clear();

        tracing::info!("Transaction {} aborted", self.id.0);

        Ok(())
    }

    /// Get pending insertions
    pub fn pending_inserts(&self) -> &[Quad] {
        &self.pending_inserts
    }

    /// Get pending deletions
    pub fn pending_deletes(&self) -> &[Quad] {
        &self.pending_deletes
    }

    /// Get the number of operations in this transaction
    pub fn operation_count(&self) -> usize {
        self.pending_inserts.len() + self.pending_deletes.len()
    }

    // Private helper methods

    /// Check if transaction is active
    fn check_active(&self) -> Result<(), OxirsError> {
        if self.state != TransactionState::Active {
            return Err(OxirsError::Store(format!(
                "Transaction is not active (state: {:?})",
                self.state
            )));
        }
        Ok(())
    }

    /// Write an entry to the WAL
    fn write_to_wal(&self, entry: WalEntry) -> Result<(), OxirsError> {
        let mut wal = self
            .wal
            .write()
            .map_err(|_| OxirsError::ConcurrencyError("WAL lock poisoned".to_string()))?;

        wal.append(entry)
    }

    /// Flush WAL to disk
    fn flush_wal(&self) -> Result<(), OxirsError> {
        let mut wal = self
            .wal
            .write()
            .map_err(|_| OxirsError::ConcurrencyError("WAL lock poisoned".to_string()))?;

        wal.flush()
    }
}

impl Drop for AcidTransaction {
    fn drop(&mut self) {
        // Auto-abort if not committed
        if self.state == TransactionState::Active {
            tracing::warn!(
                "Transaction {} dropped without commit or abort, auto-aborting",
                self.id.0
            );
            let _ = self.write_to_wal(WalEntry::Abort { tx_id: self.id.0 });
            self.abort_counter.add(1);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model::{GraphName, Literal, NamedNode, Object, Predicate, Subject};
    use tempfile::tempdir;

    fn create_test_quad(id: usize) -> Quad {
        Quad::new(
            Subject::NamedNode(
                NamedNode::new(format!("http://s{}", id)).expect("valid IRI from format"),
            ),
            Predicate::NamedNode(
                NamedNode::new(format!("http://p{}", id)).expect("valid IRI from format"),
            ),
            Object::Literal(Literal::new(format!("value{}", id))),
            GraphName::DefaultGraph,
        )
    }

    #[test]
    fn test_transaction_insert() -> Result<(), OxirsError> {
        let dir = tempdir().map_err(|e| OxirsError::Io(e.to_string()))?;
        let wal = Arc::new(RwLock::new(WriteAheadLog::new(dir.path())?));

        let mut tx = AcidTransaction::new(
            TransactionId(1),
            IsolationLevel::Snapshot,
            None,
            wal,
            Arc::new(Counter::new("test.commits".to_string())),
            Arc::new(Counter::new("test.aborts".to_string())),
            Arc::new(Timer::new("test.commit_time".to_string())),
        );

        let quad = create_test_quad(1);
        assert!(tx.insert(quad.clone())?);
        assert!(!tx.insert(quad)?); // Duplicate insert

        assert_eq!(tx.pending_inserts().len(), 1);

        Ok(())
    }

    #[test]
    fn test_transaction_delete() -> Result<(), OxirsError> {
        let dir = tempdir().map_err(|e| OxirsError::Io(e.to_string()))?;
        let wal = Arc::new(RwLock::new(WriteAheadLog::new(dir.path())?));

        let mut tx = AcidTransaction::new(
            TransactionId(1),
            IsolationLevel::Snapshot,
            None,
            wal,
            Arc::new(Counter::new("test.commits".to_string())),
            Arc::new(Counter::new("test.aborts".to_string())),
            Arc::new(Timer::new("test.commit_time".to_string())),
        );

        let quad = create_test_quad(1);
        assert!(tx.delete(quad.clone())?);
        assert!(!tx.delete(quad)?); // Duplicate delete

        assert_eq!(tx.pending_deletes().len(), 1);

        Ok(())
    }

    #[test]
    fn test_transaction_commit() -> Result<(), OxirsError> {
        let dir = tempdir().map_err(|e| OxirsError::Io(e.to_string()))?;
        let wal = Arc::new(RwLock::new(WriteAheadLog::new(dir.path())?));

        let mut tx = AcidTransaction::new(
            TransactionId(1),
            IsolationLevel::Snapshot,
            None,
            wal,
            Arc::new(Counter::new("test.commits".to_string())),
            Arc::new(Counter::new("test.aborts".to_string())),
            Arc::new(Timer::new("test.commit_time".to_string())),
        );

        let quad = create_test_quad(1);
        tx.insert(quad)?;

        assert_eq!(tx.state(), TransactionState::Active);

        tx.commit()?;
        // Note: Can't check state after commit since it consumes self

        Ok(())
    }

    #[test]
    fn test_transaction_abort() -> Result<(), OxirsError> {
        let dir = tempdir().map_err(|e| OxirsError::Io(e.to_string()))?;
        let wal = Arc::new(RwLock::new(WriteAheadLog::new(dir.path())?));

        let mut tx = AcidTransaction::new(
            TransactionId(1),
            IsolationLevel::Snapshot,
            None,
            wal,
            Arc::new(Counter::new("test.commits".to_string())),
            Arc::new(Counter::new("test.aborts".to_string())),
            Arc::new(Timer::new("test.commit_time".to_string())),
        );

        let quad = create_test_quad(1);
        tx.insert(quad)?;

        assert_eq!(tx.state(), TransactionState::Active);

        tx.abort()?;
        // Note: Can't check state after abort since it consumes self

        Ok(())
    }

    #[test]
    fn test_commit_applies_inserts_and_deletes_to_store() -> Result<(), OxirsError> {
        use crate::rdf_store::RdfStore;
        use crate::Store as StoreTrait;

        let dir = tempdir().map_err(|e| OxirsError::Io(e.to_string()))?;
        let wal = Arc::new(RwLock::new(WriteAheadLog::new(dir.path())?));
        let store: Arc<dyn crate::Store> = Arc::new(RdfStore::new()?);

        // Pre-seed a quad that the transaction will delete on commit.
        let existing = create_test_quad(99);
        StoreTrait::insert_quad(&*store, existing.clone())?;
        assert_eq!(StoreTrait::len(&*store)?, 1);

        let mut tx = AcidTransaction::new(
            TransactionId(1),
            IsolationLevel::Snapshot,
            None,
            wal,
            Arc::new(Counter::new("test.commits".to_string())),
            Arc::new(Counter::new("test.aborts".to_string())),
            Arc::new(Timer::new("test.commit_time".to_string())),
        )
        .with_store(store.clone());

        let new_quad = create_test_quad(1);
        tx.insert(new_quad.clone())?;
        tx.delete(existing.clone())?;

        // Before commit, the store is unchanged.
        assert_eq!(StoreTrait::len(&*store)?, 1);

        tx.commit()?;

        // After commit, the insert is visible and the delete has been applied.
        assert_eq!(StoreTrait::len(&*store)?, 1);
        let all = StoreTrait::quads(&*store)?;
        assert!(all.contains(&new_quad), "committed insert must be visible");
        assert!(!all.contains(&existing), "committed delete must be applied");

        Ok(())
    }
}