prax_query/
transaction.rs

1#![allow(dead_code)]
2
3//! Transaction support with async closures and savepoints.
4//!
5//! Set `PRAX_DEBUG=true` to enable transaction debug logging.
6//!
7//! This module provides a type-safe transaction API that:
8//! - Automatically commits on success
9//! - Automatically rolls back on error or panic
10//! - Supports savepoints for nested transactions
11//! - Configurable isolation levels
12//!
13//! # Isolation Levels
14//!
15//! ```rust
16//! use prax_query::IsolationLevel;
17//!
18//! // Available isolation levels
19//! let level = IsolationLevel::ReadUncommitted;
20//! let level = IsolationLevel::ReadCommitted;  // Default
21//! let level = IsolationLevel::RepeatableRead;
22//! let level = IsolationLevel::Serializable;
23//!
24//! // Get SQL representation
25//! assert_eq!(IsolationLevel::Serializable.as_sql(), "SERIALIZABLE");
26//! assert_eq!(IsolationLevel::ReadCommitted.as_sql(), "READ COMMITTED");
27//! ```
28//!
29//! # Transaction Configuration
30//!
31//! ```rust
32//! use prax_query::{TransactionConfig, IsolationLevel};
33//!
34//! // Default configuration
35//! let config = TransactionConfig::new();
36//! assert_eq!(config.isolation, IsolationLevel::ReadCommitted);
37//!
38//! // Custom configuration
39//! let config = TransactionConfig::new()
40//!     .isolation(IsolationLevel::Serializable);
41//!
42//! // Access isolation as a public field
43//! assert_eq!(config.isolation, IsolationLevel::Serializable);
44//! ```
45//!
46//! # Transaction Usage (requires async runtime)
47//!
48//! ```rust,ignore
49//! // Basic transaction - commits on success, rolls back on error
50//! let result = client
51//!     .transaction(|tx| async move {
52//!         let user = tx.user().create(/* ... */).exec().await?;
53//!         tx.post().create(/* ... */).exec().await?;
54//!         Ok(user)
55//!     })
56//!     .await?;
57//!
58//! // With configuration
59//! let result = client
60//!     .transaction(|tx| async move {
61//!         // ... perform operations
62//!         Ok(())
63//!     })
64//!     .with_config(TransactionConfig::new()
65//!         .isolation(IsolationLevel::Serializable)
66//!         .timeout(Duration::from_secs(30)))
67//!     .await?;
68//!
69//! // With savepoints for partial rollback
70//! let result = client
71//!     .transaction(|tx| async move {
72//!         tx.user().create(/* ... */).exec().await?;
73//!
74//!         // This can be rolled back independently
75//!         let savepoint_result = tx.savepoint("sp1", |sp| async move {
76//!             sp.post().create(/* ... */).exec().await?;
77//!             Ok(())
78//!         }).await;
79//!
80//!         // Even if savepoint fails, outer transaction continues
81//!         if savepoint_result.is_err() {
82//!             // Handle partial failure
83//!         }
84//!
85//!         Ok(())
86//!     })
87//!     .await?;
88//! ```
89
90use std::future::Future;
91use std::time::Duration;
92use tracing::debug;
93
94use crate::error::QueryResult;
95
96/// Transaction isolation levels.
97#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
98pub enum IsolationLevel {
99    /// Read uncommitted - allows dirty reads.
100    ReadUncommitted,
101    /// Read committed - prevents dirty reads.
102    #[default]
103    ReadCommitted,
104    /// Repeatable read - prevents non-repeatable reads.
105    RepeatableRead,
106    /// Serializable - highest isolation level.
107    Serializable,
108}
109
110impl IsolationLevel {
111    /// Get the SQL clause for this isolation level.
112    pub fn as_sql(&self) -> &'static str {
113        match self {
114            Self::ReadUncommitted => "READ UNCOMMITTED",
115            Self::ReadCommitted => "READ COMMITTED",
116            Self::RepeatableRead => "REPEATABLE READ",
117            Self::Serializable => "SERIALIZABLE",
118        }
119    }
120}
121
122/// Access mode for transactions.
123#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
124pub enum AccessMode {
125    /// Read-write access (default).
126    #[default]
127    ReadWrite,
128    /// Read-only access.
129    ReadOnly,
130}
131
132impl AccessMode {
133    /// Get the SQL clause for this access mode.
134    pub fn as_sql(&self) -> &'static str {
135        match self {
136            Self::ReadWrite => "READ WRITE",
137            Self::ReadOnly => "READ ONLY",
138        }
139    }
140}
141
142/// Configuration for a transaction.
143#[derive(Debug, Clone, Default)]
144pub struct TransactionConfig {
145    /// Isolation level.
146    pub isolation: IsolationLevel,
147    /// Access mode.
148    pub access_mode: AccessMode,
149    /// Timeout for the transaction.
150    pub timeout: Option<Duration>,
151    /// Whether to defer constraint checking.
152    pub deferrable: bool,
153}
154
155impl TransactionConfig {
156    /// Create a new transaction config with defaults.
157    pub fn new() -> Self {
158        Self::default()
159    }
160
161    /// Set the isolation level.
162    pub fn isolation(mut self, level: IsolationLevel) -> Self {
163        self.isolation = level;
164        self
165    }
166
167    /// Set the access mode.
168    pub fn access_mode(mut self, mode: AccessMode) -> Self {
169        self.access_mode = mode;
170        self
171    }
172
173    /// Set the timeout.
174    pub fn timeout(mut self, timeout: Duration) -> Self {
175        self.timeout = Some(timeout);
176        self
177    }
178
179    /// Make the transaction read-only.
180    pub fn read_only(self) -> Self {
181        self.access_mode(AccessMode::ReadOnly)
182    }
183
184    /// Make the transaction deferrable.
185    pub fn deferrable(mut self) -> Self {
186        self.deferrable = true;
187        self
188    }
189
190    /// Generate the BEGIN TRANSACTION SQL.
191    pub fn to_begin_sql(&self) -> String {
192        let mut parts = vec!["BEGIN"];
193
194        // Isolation level
195        parts.push("ISOLATION LEVEL");
196        parts.push(self.isolation.as_sql());
197
198        // Access mode
199        parts.push(self.access_mode.as_sql());
200
201        // Deferrable (PostgreSQL specific, only valid for SERIALIZABLE READ ONLY)
202        if self.deferrable
203            && self.isolation == IsolationLevel::Serializable
204            && self.access_mode == AccessMode::ReadOnly
205        {
206            parts.push("DEFERRABLE");
207        }
208
209        let sql = parts.join(" ");
210        debug!(isolation = %self.isolation.as_sql(), access_mode = %self.access_mode.as_sql(), "Transaction BEGIN");
211        sql
212    }
213}
214
215/// A transaction handle that provides query operations.
216///
217/// The transaction will be committed when dropped if no error occurred,
218/// or rolled back if an error occurred or panic happened.
219pub struct Transaction<E> {
220    engine: E,
221    config: TransactionConfig,
222    committed: bool,
223    savepoint_count: u32,
224}
225
226impl<E> Transaction<E> {
227    /// Create a new transaction handle.
228    pub fn new(engine: E, config: TransactionConfig) -> Self {
229        Self {
230            engine,
231            config,
232            committed: false,
233            savepoint_count: 0,
234        }
235    }
236
237    /// Get the transaction configuration.
238    pub fn config(&self) -> &TransactionConfig {
239        &self.config
240    }
241
242    /// Get the underlying engine.
243    pub fn engine(&self) -> &E {
244        &self.engine
245    }
246
247    /// Create a savepoint.
248    pub fn savepoint_name(&mut self) -> String {
249        self.savepoint_count += 1;
250        format!("sp_{}", self.savepoint_count)
251    }
252
253    /// Mark the transaction as committed.
254    pub fn mark_committed(&mut self) {
255        self.committed = true;
256    }
257
258    /// Check if the transaction has been committed.
259    pub fn is_committed(&self) -> bool {
260        self.committed
261    }
262}
263
264/// Builder for executing a transaction with a closure.
265pub struct TransactionBuilder<E, F, Fut, T>
266where
267    F: FnOnce(Transaction<E>) -> Fut,
268    Fut: Future<Output = QueryResult<T>>,
269{
270    engine: E,
271    callback: F,
272    config: TransactionConfig,
273}
274
275impl<E, F, Fut, T> TransactionBuilder<E, F, Fut, T>
276where
277    F: FnOnce(Transaction<E>) -> Fut,
278    Fut: Future<Output = QueryResult<T>>,
279{
280    /// Create a new transaction builder.
281    pub fn new(engine: E, callback: F) -> Self {
282        Self {
283            engine,
284            callback,
285            config: TransactionConfig::default(),
286        }
287    }
288
289    /// Set the isolation level.
290    pub fn isolation(mut self, level: IsolationLevel) -> Self {
291        self.config.isolation = level;
292        self
293    }
294
295    /// Set read-only mode.
296    pub fn read_only(mut self) -> Self {
297        self.config.access_mode = AccessMode::ReadOnly;
298        self
299    }
300
301    /// Set the timeout.
302    pub fn timeout(mut self, timeout: Duration) -> Self {
303        self.config.timeout = Some(timeout);
304        self
305    }
306
307    /// Set deferrable mode.
308    pub fn deferrable(mut self) -> Self {
309        self.config.deferrable = true;
310        self
311    }
312}
313
314/// Interactive transaction for step-by-step operations.
315pub struct InteractiveTransaction<E> {
316    inner: Transaction<E>,
317    started: bool,
318}
319
320impl<E> InteractiveTransaction<E> {
321    /// Create a new interactive transaction.
322    pub fn new(engine: E) -> Self {
323        Self {
324            inner: Transaction::new(engine, TransactionConfig::default()),
325            started: false,
326        }
327    }
328
329    /// Create with configuration.
330    pub fn with_config(engine: E, config: TransactionConfig) -> Self {
331        Self {
332            inner: Transaction::new(engine, config),
333            started: false,
334        }
335    }
336
337    /// Get the engine.
338    pub fn engine(&self) -> &E {
339        &self.inner.engine
340    }
341
342    /// Check if the transaction has started.
343    pub fn is_started(&self) -> bool {
344        self.started
345    }
346
347    /// Get the BEGIN SQL.
348    pub fn begin_sql(&self) -> String {
349        self.inner.config.to_begin_sql()
350    }
351
352    /// Get the COMMIT SQL.
353    pub fn commit_sql(&self) -> &'static str {
354        "COMMIT"
355    }
356
357    /// Get the ROLLBACK SQL.
358    pub fn rollback_sql(&self) -> &'static str {
359        "ROLLBACK"
360    }
361
362    /// Get the SAVEPOINT SQL.
363    pub fn savepoint_sql(&mut self, name: Option<&str>) -> String {
364        let name = name
365            .map(|s| s.to_string())
366            .unwrap_or_else(|| self.inner.savepoint_name());
367        format!("SAVEPOINT {}", name)
368    }
369
370    /// Get the ROLLBACK TO SAVEPOINT SQL.
371    pub fn rollback_to_sql(&self, name: &str) -> String {
372        format!("ROLLBACK TO SAVEPOINT {}", name)
373    }
374
375    /// Get the RELEASE SAVEPOINT SQL.
376    pub fn release_savepoint_sql(&self, name: &str) -> String {
377        format!("RELEASE SAVEPOINT {}", name)
378    }
379
380    /// Mark as started.
381    pub fn mark_started(&mut self) {
382        self.started = true;
383    }
384}
385
386#[cfg(test)]
387mod tests {
388    use super::*;
389
390    #[test]
391    fn test_isolation_level() {
392        assert_eq!(IsolationLevel::ReadCommitted.as_sql(), "READ COMMITTED");
393        assert_eq!(IsolationLevel::Serializable.as_sql(), "SERIALIZABLE");
394    }
395
396    #[test]
397    fn test_access_mode() {
398        assert_eq!(AccessMode::ReadWrite.as_sql(), "READ WRITE");
399        assert_eq!(AccessMode::ReadOnly.as_sql(), "READ ONLY");
400    }
401
402    #[test]
403    fn test_transaction_config_default() {
404        let config = TransactionConfig::new();
405        assert_eq!(config.isolation, IsolationLevel::ReadCommitted);
406        assert_eq!(config.access_mode, AccessMode::ReadWrite);
407        assert!(config.timeout.is_none());
408        assert!(!config.deferrable);
409    }
410
411    #[test]
412    fn test_transaction_config_builder() {
413        let config = TransactionConfig::new()
414            .isolation(IsolationLevel::Serializable)
415            .read_only()
416            .deferrable()
417            .timeout(Duration::from_secs(30));
418
419        assert_eq!(config.isolation, IsolationLevel::Serializable);
420        assert_eq!(config.access_mode, AccessMode::ReadOnly);
421        assert!(config.deferrable);
422        assert_eq!(config.timeout, Some(Duration::from_secs(30)));
423    }
424
425    #[test]
426    fn test_begin_sql() {
427        let config = TransactionConfig::new();
428        let sql = config.to_begin_sql();
429        assert!(sql.contains("BEGIN"));
430        assert!(sql.contains("ISOLATION LEVEL READ COMMITTED"));
431        assert!(sql.contains("READ WRITE"));
432    }
433
434    #[test]
435    fn test_begin_sql_serializable_deferrable() {
436        let config = TransactionConfig::new()
437            .isolation(IsolationLevel::Serializable)
438            .read_only()
439            .deferrable();
440        let sql = config.to_begin_sql();
441        assert!(sql.contains("SERIALIZABLE"));
442        assert!(sql.contains("READ ONLY"));
443        assert!(sql.contains("DEFERRABLE"));
444    }
445
446    #[test]
447    fn test_interactive_transaction() {
448        #[derive(Clone)]
449        struct MockEngine;
450
451        let mut tx = InteractiveTransaction::new(MockEngine);
452        assert!(!tx.is_started());
453
454        let begin = tx.begin_sql();
455        assert!(begin.contains("BEGIN"));
456
457        let sp = tx.savepoint_sql(Some("test_sp"));
458        assert_eq!(sp, "SAVEPOINT test_sp");
459
460        let rollback_to = tx.rollback_to_sql("test_sp");
461        assert_eq!(rollback_to, "ROLLBACK TO SAVEPOINT test_sp");
462
463        let release = tx.release_savepoint_sql("test_sp");
464        assert_eq!(release, "RELEASE SAVEPOINT test_sp");
465    }
466}
prax_query/transaction.rs

prax_query/
transaction.rs
