oxirs/tools/
transaction.rs

1//! Transaction Support for OxiRS
2//!
3//! ACID transaction management for SPARQL UPDATE operations with support for
4//! BEGIN/COMMIT/ROLLBACK, isolation levels, and multi-operation atomicity.
5
6use super::{ToolResult, ToolStats};
7use std::collections::HashMap;
8use std::path::PathBuf;
9use std::time::{Duration, Instant, SystemTime};
10
11/// Transaction isolation levels
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum IsolationLevel {
14    /// Read uncommitted - lowest isolation, highest performance
15    ReadUncommitted,
16    /// Read committed - prevent dirty reads
17    ReadCommitted,
18    /// Repeatable read - prevent non-repeatable reads
19    RepeatableRead,
20    /// Serializable - highest isolation, lowest performance
21    Serializable,
22}
23
24impl IsolationLevel {
25    pub fn name(&self) -> &str {
26        match self {
27            IsolationLevel::ReadUncommitted => "READ_UNCOMMITTED",
28            IsolationLevel::ReadCommitted => "READ_COMMITTED",
29            IsolationLevel::RepeatableRead => "REPEATABLE_READ",
30            IsolationLevel::Serializable => "SERIALIZABLE",
31        }
32    }
33
34    pub fn description(&self) -> &str {
35        match self {
36            IsolationLevel::ReadUncommitted => "Lowest isolation, allows dirty reads",
37            IsolationLevel::ReadCommitted => "Prevents dirty reads",
38            IsolationLevel::RepeatableRead => "Prevents dirty and non-repeatable reads",
39            IsolationLevel::Serializable => "Highest isolation, full ACID compliance",
40        }
41    }
42}
43
44/// Transaction state
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum TransactionState {
47    /// No active transaction
48    None,
49    /// Transaction started, operations in progress
50    Active,
51    /// Transaction committed successfully
52    Committed,
53    /// Transaction rolled back
54    RolledBack,
55    /// Transaction failed with error
56    Failed,
57}
58
59/// Transaction operation types
60#[derive(Debug, Clone)]
61pub enum TransactionOperation {
62    Insert { triples: String },
63    Delete { triples: String },
64    Update { query: String },
65    Load { file: PathBuf },
66    Clear { graph: Option<String> },
67}
68
69/// Savepoint for nested transactions
70#[derive(Debug, Clone)]
71pub struct Savepoint {
72    pub name: String,
73    pub timestamp: SystemTime,
74    pub operation_count: usize,
75}
76
77/// Transaction context
78#[derive(Debug)]
79pub struct Transaction {
80    pub id: String,
81    pub state: TransactionState,
82    pub isolation_level: IsolationLevel,
83    pub start_time: Instant,
84    pub operations: Vec<TransactionOperation>,
85    pub savepoints: Vec<Savepoint>,
86    pub dataset: String,
87    pub read_only: bool,
88}
89
90impl Transaction {
91    /// Create a new transaction
92    pub fn new(dataset: String, isolation_level: IsolationLevel, read_only: bool) -> Self {
93        let id = format!("txn_{}", uuid::Uuid::new_v4());
94
95        Self {
96            id,
97            state: TransactionState::Active,
98            isolation_level,
99            start_time: Instant::now(),
100            operations: Vec::new(),
101            savepoints: Vec::new(),
102            dataset,
103            read_only,
104        }
105    }
106
107    /// Add an operation to the transaction
108    pub fn add_operation(&mut self, operation: TransactionOperation) -> ToolResult {
109        if self.state != TransactionState::Active {
110            return Err(format!("Transaction is not active: {:?}", self.state).into());
111        }
112
113        if self.read_only {
114            return Err("Cannot modify data in read-only transaction".into());
115        }
116
117        self.operations.push(operation);
118        Ok(())
119    }
120
121    /// Create a savepoint
122    pub fn create_savepoint(&mut self, name: String) -> ToolResult {
123        if self.state != TransactionState::Active {
124            return Err(format!("Transaction is not active: {:?}", self.state).into());
125        }
126
127        // Check if savepoint name already exists
128        if self.savepoints.iter().any(|sp| sp.name == name) {
129            return Err(format!("Savepoint '{}' already exists", name).into());
130        }
131
132        let savepoint = Savepoint {
133            name,
134            timestamp: SystemTime::now(),
135            operation_count: self.operations.len(),
136        };
137
138        self.savepoints.push(savepoint);
139        Ok(())
140    }
141
142    /// Rollback to a savepoint
143    pub fn rollback_to_savepoint(&mut self, name: &str) -> ToolResult {
144        if self.state != TransactionState::Active {
145            return Err(format!("Transaction is not active: {:?}", self.state).into());
146        }
147
148        // Find the savepoint
149        let savepoint_idx = self
150            .savepoints
151            .iter()
152            .position(|sp| sp.name == name)
153            .ok_or_else(|| format!("Savepoint '{}' not found", name))?;
154
155        let savepoint = &self.savepoints[savepoint_idx];
156        let operation_count = savepoint.operation_count;
157
158        // Rollback operations
159        self.operations.truncate(operation_count);
160
161        // Remove this and all later savepoints
162        self.savepoints.truncate(savepoint_idx);
163
164        println!("Rolled back to savepoint '{}'", name);
165        println!("  Operations restored to: {}", operation_count);
166
167        Ok(())
168    }
169
170    /// Get transaction duration
171    pub fn duration(&self) -> Duration {
172        self.start_time.elapsed()
173    }
174
175    /// Get transaction status summary
176    pub fn summary(&self) -> String {
177        format!(
178            "Transaction {} [{}] - {} operations, {} savepoints, {:?} elapsed",
179            self.id,
180            self.state_str(),
181            self.operations.len(),
182            self.savepoints.len(),
183            self.duration()
184        )
185    }
186
187    fn state_str(&self) -> &str {
188        match self.state {
189            TransactionState::None => "NONE",
190            TransactionState::Active => "ACTIVE",
191            TransactionState::Committed => "COMMITTED",
192            TransactionState::RolledBack => "ROLLED_BACK",
193            TransactionState::Failed => "FAILED",
194        }
195    }
196}
197
198/// Transaction manager
199pub struct TransactionManager {
200    active_transactions: HashMap<String, Transaction>,
201    transaction_log: Vec<TransactionLogEntry>,
202}
203
204impl TransactionManager {
205    pub fn new() -> Self {
206        Self {
207            active_transactions: HashMap::new(),
208            transaction_log: Vec::new(),
209        }
210    }
211
212    /// Begin a new transaction
213    pub fn begin(
214        &mut self,
215        dataset: String,
216        isolation_level: IsolationLevel,
217        read_only: bool,
218    ) -> ToolResult<String> {
219        let transaction = Transaction::new(dataset.clone(), isolation_level, read_only);
220        let txn_id = transaction.id.clone();
221
222        println!("BEGIN TRANSACTION {}", txn_id);
223        println!("  Dataset: {}", dataset);
224        println!("  Isolation: {}", isolation_level.name());
225        println!("  Read-only: {}", read_only);
226
227        self.active_transactions.insert(txn_id.clone(), transaction);
228
229        self.transaction_log.push(TransactionLogEntry {
230            transaction_id: txn_id.clone(),
231            action: "BEGIN".to_string(),
232            timestamp: SystemTime::now(),
233            details: None,
234        });
235
236        Ok(txn_id)
237    }
238
239    /// Commit a transaction
240    pub fn commit(&mut self, txn_id: &str) -> ToolResult {
241        // First, validate and get operation count
242        let (operation_count, duration) = {
243            let transaction = self
244                .active_transactions
245                .get(txn_id)
246                .ok_or_else(|| format!("Transaction '{}' not found", txn_id))?;
247
248            if transaction.state != TransactionState::Active {
249                return Err(format!("Transaction is not active: {:?}", transaction.state).into());
250            }
251
252            (transaction.operations.len(), transaction.duration())
253        };
254
255        println!("COMMIT TRANSACTION {}", txn_id);
256        println!("  Operations: {}", operation_count);
257
258        // Execute all operations (immutable borrow)
259        let result = {
260            let transaction = self
261                .active_transactions
262                .get(txn_id)
263                .expect("transaction should exist for given id");
264            self.execute_operations(transaction)
265        };
266
267        // Update transaction state (mutable borrow)
268        let transaction = self
269            .active_transactions
270            .get_mut(txn_id)
271            .expect("transaction should exist for given id");
272
273        match result {
274            Ok(_) => {
275                transaction.state = TransactionState::Committed;
276                println!("  Status: SUCCESS");
277                println!("  Duration: {:?}", duration);
278
279                self.transaction_log.push(TransactionLogEntry {
280                    transaction_id: txn_id.to_string(),
281                    action: "COMMIT".to_string(),
282                    timestamp: SystemTime::now(),
283                    details: Some(format!("{} operations", operation_count)),
284                });
285
286                Ok(())
287            }
288            Err(e) => {
289                transaction.state = TransactionState::Failed;
290                println!("  Status: FAILED");
291                println!("  Error: {}", e);
292
293                self.transaction_log.push(TransactionLogEntry {
294                    transaction_id: txn_id.to_string(),
295                    action: "COMMIT_FAILED".to_string(),
296                    timestamp: SystemTime::now(),
297                    details: Some(e.to_string()),
298                });
299
300                Err(e)
301            }
302        }
303    }
304
305    /// Rollback a transaction
306    pub fn rollback(&mut self, txn_id: &str) -> ToolResult {
307        let transaction = self
308            .active_transactions
309            .get_mut(txn_id)
310            .ok_or_else(|| format!("Transaction '{}' not found", txn_id))?;
311
312        if transaction.state != TransactionState::Active {
313            return Err(format!("Transaction is not active: {:?}", transaction.state).into());
314        }
315
316        println!("ROLLBACK TRANSACTION {}", txn_id);
317        println!("  Operations discarded: {}", transaction.operations.len());
318
319        transaction.state = TransactionState::RolledBack;
320        transaction.operations.clear();
321        transaction.savepoints.clear();
322
323        println!("  Status: ROLLED BACK");
324        println!("  Duration: {:?}", transaction.duration());
325
326        self.transaction_log.push(TransactionLogEntry {
327            transaction_id: txn_id.to_string(),
328            action: "ROLLBACK".to_string(),
329            timestamp: SystemTime::now(),
330            details: None,
331        });
332
333        Ok(())
334    }
335
336    /// Execute transaction operations
337    fn execute_operations(&self, transaction: &Transaction) -> ToolResult {
338        println!("  Executing {} operations...", transaction.operations.len());
339
340        for (i, operation) in transaction.operations.iter().enumerate() {
341            print!("    Operation {}/{}: ", i + 1, transaction.operations.len());
342
343            match operation {
344                TransactionOperation::Insert { triples } => {
345                    println!("INSERT {} triples", triples.lines().count());
346                    // Simulate execution
347                    std::thread::sleep(Duration::from_millis(10));
348                }
349                TransactionOperation::Delete { triples } => {
350                    println!("DELETE {} triples", triples.lines().count());
351                    // Simulate execution
352                    std::thread::sleep(Duration::from_millis(10));
353                }
354                TransactionOperation::Update { query } => {
355                    println!("UPDATE (query: {} chars)", query.len());
356                    // Simulate execution
357                    std::thread::sleep(Duration::from_millis(20));
358                }
359                TransactionOperation::Load { file } => {
360                    println!("LOAD {}", file.display());
361                    // Simulate execution
362                    std::thread::sleep(Duration::from_millis(50));
363                }
364                TransactionOperation::Clear { graph } => {
365                    if let Some(g) = graph {
366                        println!("CLEAR GRAPH <{}>", g);
367                    } else {
368                        println!("CLEAR DEFAULT");
369                    }
370                    // Simulate execution
371                    std::thread::sleep(Duration::from_millis(5));
372                }
373            }
374        }
375
376        Ok(())
377    }
378
379    /// Get active transaction
380    pub fn get_transaction(&self, txn_id: &str) -> Option<&Transaction> {
381        self.active_transactions.get(txn_id)
382    }
383
384    /// Get mutable active transaction
385    pub fn get_transaction_mut(&mut self, txn_id: &str) -> Option<&mut Transaction> {
386        self.active_transactions.get_mut(txn_id)
387    }
388
389    /// List all active transactions
390    pub fn list_active(&self) -> Vec<&Transaction> {
391        self.active_transactions
392            .values()
393            .filter(|txn| txn.state == TransactionState::Active)
394            .collect()
395    }
396
397    /// Get transaction log
398    pub fn get_log(&self) -> &[TransactionLogEntry] {
399        &self.transaction_log
400    }
401
402    /// Clean up finished transactions
403    pub fn cleanup(&mut self) {
404        self.active_transactions
405            .retain(|_, txn| txn.state == TransactionState::Active);
406    }
407}
408
409impl Default for TransactionManager {
410    fn default() -> Self {
411        Self::new()
412    }
413}
414
415/// Transaction log entry
416#[derive(Debug, Clone)]
417pub struct TransactionLogEntry {
418    pub transaction_id: String,
419    pub action: String,
420    pub timestamp: SystemTime,
421    pub details: Option<String>,
422}
423
424/// Configuration for transaction execution
425pub struct TransactionConfig {
426    pub dataset: String,
427    pub operations: Vec<String>,
428    pub isolation_level: IsolationLevel,
429    pub read_only: bool,
430    pub auto_commit: bool,
431}
432
433/// Run a transactional operation
434pub async fn run(config: TransactionConfig) -> ToolResult {
435    let mut stats = ToolStats::new();
436    let mut manager = TransactionManager::new();
437
438    println!("OxiRS Transaction Manager");
439    println!("=========================\n");
440
441    // Begin transaction
442    let txn_id = manager.begin(config.dataset, config.isolation_level, config.read_only)?;
443
444    println!();
445
446    // Add operations
447    let transaction = manager
448        .get_transaction_mut(&txn_id)
449        .expect("transaction should exist after begin");
450
451    for operation_str in &config.operations {
452        let operation = parse_operation(operation_str)?;
453        transaction.add_operation(operation)?;
454    }
455
456    println!(
457        "Added {} operations to transaction\n",
458        config.operations.len()
459    );
460
461    // Commit or rollback
462    if config.auto_commit {
463        match manager.commit(&txn_id) {
464            Ok(_) => {
465                println!("\n✓ Transaction committed successfully");
466                stats.items_processed = config.operations.len();
467            }
468            Err(e) => {
469                println!("\n✗ Transaction commit failed: {}", e);
470                println!("  Rolling back...");
471                manager.rollback(&txn_id)?;
472                return Err(e);
473            }
474        }
475    } else {
476        println!("Transaction prepared (not committed)");
477        println!("  Use 'oxirs transaction commit {}' to commit", txn_id);
478        println!("  Use 'oxirs transaction rollback {}' to rollback", txn_id);
479    }
480
481    manager.cleanup();
482
483    stats.finish();
484    stats.print_summary("Transaction");
485
486    Ok(())
487}
488
489/// Parse operation string into TransactionOperation
490fn parse_operation(operation_str: &str) -> ToolResult<TransactionOperation> {
491    let operation_str = operation_str.trim();
492
493    if operation_str.starts_with("INSERT") {
494        let triples = operation_str.strip_prefix("INSERT").unwrap_or("").trim();
495        Ok(TransactionOperation::Insert {
496            triples: triples.to_string(),
497        })
498    } else if operation_str.starts_with("DELETE") {
499        let triples = operation_str.strip_prefix("DELETE").unwrap_or("").trim();
500        Ok(TransactionOperation::Delete {
501            triples: triples.to_string(),
502        })
503    } else if operation_str.starts_with("UPDATE") {
504        let query = operation_str.strip_prefix("UPDATE").unwrap_or("").trim();
505        Ok(TransactionOperation::Update {
506            query: query.to_string(),
507        })
508    } else if operation_str.starts_with("LOAD") {
509        let file_str = operation_str.strip_prefix("LOAD").unwrap_or("").trim();
510        Ok(TransactionOperation::Load {
511            file: PathBuf::from(file_str),
512        })
513    } else if operation_str.starts_with("CLEAR") {
514        let graph_str = operation_str.strip_prefix("CLEAR").unwrap_or("").trim();
515        let graph = if graph_str.is_empty() || graph_str == "DEFAULT" {
516            None
517        } else {
518            Some(graph_str.to_string())
519        };
520        Ok(TransactionOperation::Clear { graph })
521    } else {
522        Err(format!("Unknown operation: {}", operation_str).into())
523    }
524}
525
526/// Display transaction status
527pub fn display_status(manager: &TransactionManager) {
528    let active = manager.list_active();
529
530    println!("Active Transactions: {}\n", active.len());
531
532    if active.is_empty() {
533        println!("No active transactions");
534        return;
535    }
536
537    println!(
538        "{:<40} {:>12} {:>15} {:>10}",
539        "Transaction ID", "Operations", "Savepoints", "Duration"
540    );
541    println!("{}", "-".repeat(80));
542
543    for txn in active {
544        println!(
545            "{:<40} {:>12} {:>15} {:>10.2?}",
546            txn.id,
547            txn.operations.len(),
548            txn.savepoints.len(),
549            txn.duration()
550        );
551    }
552}
553
554/// Display transaction log
555pub fn display_log(manager: &TransactionManager, limit: Option<usize>) {
556    let log = manager.get_log();
557    let entries = if let Some(n) = limit {
558        &log[log.len().saturating_sub(n)..]
559    } else {
560        log
561    };
562
563    println!("Transaction Log ({} entries)\n", entries.len());
564
565    for entry in entries {
566        println!(
567            "[{:?}] {} - {}",
568            entry.timestamp, entry.transaction_id, entry.action
569        );
570
571        if let Some(ref details) = entry.details {
572            println!("  Details: {}", details);
573        }
574    }
575}
576
577// UUID generation helper module
578mod uuid {
579    use std::time::{SystemTime, UNIX_EPOCH};
580
581    pub struct Uuid;
582
583    impl Uuid {
584        pub fn new_v4() -> String {
585            let now = SystemTime::now()
586                .duration_since(UNIX_EPOCH)
587                .expect("SystemTime should be after UNIX_EPOCH")
588                .as_nanos();
589
590            format!("{:032x}", now)
591        }
592    }
593}
oxirs/tools/transaction.rs

oxirs/tools/
transaction.rs
