moirai-runtime 0.7.0

//! Financial Transaction Processing - Real-World Concurrency Example
//!
//! This example demonstrates:
//! - Race condition handling in financial transfers
//! - Transaction isolation and consistency
//! - Deadlock prevention in multi-account operations
//! - Audit trail maintenance under high concurrency
//! - Error handling and recovery in financial systems

#![expect(
    clippy::unwrap_used,
    reason = "test scope: failed precondition = test failure"
)]
#![expect(
    dead_code,
    reason = "This example defines audit and transaction domain variants beyond the short executable scenario"
)]

use moirai::{Moirai, Priority};
use std::collections::HashMap;
use std::fmt;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Instant, SystemTime, UNIX_EPOCH};

/// Represents an account in the financial system
#[derive(Debug, Clone)]
struct Account {
    id: u64,
    balance: Arc<AtomicU64>, // Using atomic for lock-free balance operations
    version: Arc<AtomicU64>, // Optimistic locking version
}

impl Account {
    fn new(id: u64, initial_balance: u64) -> Self {
        Self {
            id,
            balance: Arc::new(AtomicU64::new(initial_balance)),
            version: Arc::new(AtomicU64::new(1)),
        }
    }

    fn balance(&self) -> u64 {
        self.balance.load(Ordering::Acquire)
    }

    fn version(&self) -> u64 {
        self.version.load(Ordering::Acquire)
    }
}

/// Different types of financial transactions
#[derive(Debug, Clone, PartialEq)]
enum TransactionType {
    Transfer,
    Deposit,
    Withdrawal,
    Fee,
}

/// Represents a financial transaction
#[derive(Debug, Clone)]
struct Transaction {
    id: u64,
    from_account: Option<u64>,
    to_account: Option<u64>,
    amount: u64,
    transaction_type: TransactionType,
    timestamp: u64,
    status: TransactionStatus,
}

#[derive(Debug, Clone, PartialEq)]
enum TransactionStatus {
    Pending,
    Completed,
    Failed,
    Cancelled,
}

impl fmt::Display for TransactionStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            TransactionStatus::Pending => write!(f, "PENDING"),
            TransactionStatus::Completed => write!(f, "COMPLETED"),
            TransactionStatus::Failed => write!(f, "FAILED"),
            TransactionStatus::Cancelled => write!(f, "CANCELLED"),
        }
    }
}

/// Audit trail for financial transactions
#[derive(Debug)]
struct AuditTrail {
    entries: Arc<Mutex<Vec<AuditEntry>>>,
    entry_count: Arc<AtomicUsize>,
}

#[derive(Debug, Clone)]
struct AuditEntry {
    transaction_id: u64,
    account_id: u64,
    balance_before: u64,
    balance_after: u64,
    timestamp: u64,
    operation: String,
}

impl AuditTrail {
    fn new() -> Self {
        Self {
            entries: Arc::new(Mutex::new(Vec::new())),
            entry_count: Arc::new(AtomicUsize::new(0)),
        }
    }

    fn log_entry(&self, entry: AuditEntry) -> Result<(), String> {
        let mut entries = self
            .entries
            .lock()
            .map_err(|_| "Failed to acquire audit lock")?;
        entries.push(entry);
        self.entry_count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    fn entry_count(&self) -> usize {
        self.entry_count.load(Ordering::Relaxed)
    }

    fn get_entries_for_account(&self, account_id: u64) -> Result<Vec<AuditEntry>, String> {
        let entries = self
            .entries
            .lock()
            .map_err(|_| "Failed to acquire audit lock")?;
        Ok(entries
            .iter()
            .filter(|entry| entry.account_id == account_id)
            .cloned()
            .collect())
    }
}

/// Financial transaction processing engine
struct TransactionEngine {
    accounts: Arc<RwLock<HashMap<u64, Account>>>,
    audit_trail: Arc<AuditTrail>,
    transaction_counter: Arc<AtomicU64>,
    successful_transactions: Arc<AtomicUsize>,
    failed_transactions: Arc<AtomicUsize>,
    runtime: Moirai,
}

impl TransactionEngine {
    fn new() -> Result<Self, String> {
        let runtime = Moirai::new().map_err(|_| "Failed to create Moirai runtime")?;

        Ok(Self {
            accounts: Arc::new(RwLock::new(HashMap::new())),
            audit_trail: Arc::new(AuditTrail::new()),
            transaction_counter: Arc::new(AtomicU64::new(1)),
            successful_transactions: Arc::new(AtomicUsize::new(0)),
            failed_transactions: Arc::new(AtomicUsize::new(0)),
            runtime,
        })
    }

    fn create_account(&self, account_id: u64, initial_balance: u64) -> Result<(), String> {
        let mut accounts = self
            .accounts
            .write()
            .map_err(|_| "Failed to acquire accounts write lock")?;

        if accounts.contains_key(&account_id) {
            return Err(format!("Account {} already exists", account_id));
        }

        accounts.insert(account_id, Account::new(account_id, initial_balance));
        Ok(())
    }

    fn get_account_balance(&self, account_id: u64) -> Result<u64, String> {
        let accounts = self
            .accounts
            .read()
            .map_err(|_| "Failed to acquire accounts read lock")?;

        accounts
            .get(&account_id)
            .map(|account| account.balance())
            .ok_or_else(|| format!("Account {} not found", account_id))
    }

    /// Process a transfer with comprehensive error handling and audit trailing
    fn process_transfer(
        &self,
        from_account_id: u64,
        to_account_id: u64,
        amount: u64,
        priority: Priority,
    ) -> Result<u64, String> {
        if from_account_id == to_account_id {
            return Err("Cannot transfer to the same account".to_string());
        }

        if amount == 0 {
            return Err("Transfer amount must be greater than zero".to_string());
        }

        let transaction_id = self.transaction_counter.fetch_add(1, Ordering::Relaxed);
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();

        let accounts = self.accounts.clone();
        let audit_trail = self.audit_trail.clone();
        let successful_counter = self.successful_transactions.clone();
        let failed_counter = self.failed_transactions.clone();

        // Use async execution for the transaction to demonstrate real-world async processing
        let handle = self.runtime.spawn_fn_with_priority(
            move || {
                // Always acquire locks in consistent order (by account ID) to prevent deadlocks
                let (_first_id, _second_id) = if from_account_id < to_account_id {
                    (from_account_id, to_account_id)
                } else {
                    (to_account_id, from_account_id)
                };

                // Get account references
                let accounts_read = accounts.read().map_err(|_| {
                    format!(
                        "Failed to acquire accounts lock for transaction {}",
                        transaction_id
                    )
                })?;

                let from_account = accounts_read
                    .get(&from_account_id)
                    .ok_or_else(|| format!("Source account {} not found", from_account_id))?
                    .clone();

                let to_account = accounts_read
                    .get(&to_account_id)
                    .ok_or_else(|| format!("Destination account {} not found", to_account_id))?
                    .clone();

                // Release read lock before processing
                drop(accounts_read);

                // Check balance with optimistic locking
                let from_balance_before = from_account.balance();
                let _from_version_before = from_account.version();

                if from_balance_before < amount {
                    failed_counter.fetch_add(1, Ordering::Relaxed);
                    return Err(format!(
                        "Insufficient funds in account {}: {} < {}",
                        from_account_id, from_balance_before, amount
                    ));
                }

                // Perform atomic balance updates
                let to_balance_before = to_account.balance();

                // Debit from source account
                let from_balance_after = from_account.balance.fetch_sub(amount, Ordering::AcqRel);
                if from_balance_after < amount {
                    // Race condition detected - restore balance and fail
                    from_account.balance.fetch_add(amount, Ordering::AcqRel);
                    failed_counter.fetch_add(1, Ordering::Relaxed);
                    return Err(format!(
                        "Race condition detected in account {}",
                        from_account_id
                    ));
                }
                let from_balance_after = from_balance_after - amount;

                // Credit to destination account
                let to_balance_after =
                    to_account.balance.fetch_add(amount, Ordering::AcqRel) + amount;

                // Update versions for optimistic locking
                from_account.version.fetch_add(1, Ordering::AcqRel);
                to_account.version.fetch_add(1, Ordering::AcqRel);

                // Log audit entries
                let from_audit = AuditEntry {
                    transaction_id,
                    account_id: from_account_id,
                    balance_before: from_balance_before,
                    balance_after: from_balance_after,
                    timestamp,
                    operation: format!("DEBIT {}", amount),
                };

                let to_audit = AuditEntry {
                    transaction_id,
                    account_id: to_account_id,
                    balance_before: to_balance_before,
                    balance_after: to_balance_after,
                    timestamp,
                    operation: format!("CREDIT {}", amount),
                };

                audit_trail
                    .log_entry(from_audit)
                    .map_err(|e| format!("Failed to log debit audit: {}", e))?;
                audit_trail
                    .log_entry(to_audit)
                    .map_err(|e| format!("Failed to log credit audit: {}", e))?;

                successful_counter.fetch_add(1, Ordering::Relaxed);
                Ok(transaction_id)
            },
            priority,
        );

        // Return transaction ID immediately for async processing
        match handle.join() {
            Some(Ok(result)) => result,
            Some(Err(_)) | None => {
                self.failed_transactions.fetch_add(1, Ordering::Relaxed);
                Err("Transaction execution failed".to_string())
            }
        }
    }

    /// Process multiple transactions concurrently to test edge cases
    fn process_batch_transfers(
        &self,
        transfers: Vec<(u64, u64, u64)>,
    ) -> Result<Vec<Result<u64, String>>, String> {
        let mut handles = Vec::new();

        for (from, to, amount) in transfers {
            let engine = self;
            let priority = if amount > 1000 {
                Priority::High
            } else {
                Priority::Normal
            };

            // Process each transfer with appropriate priority
            match engine.process_transfer(from, to, amount, priority) {
                Ok(tx_id) => handles.push(Ok(tx_id)),
                Err(e) => handles.push(Err(e)),
            }
        }

        Ok(handles)
    }

    fn get_statistics(&self) -> (usize, usize, usize) {
        (
            self.successful_transactions.load(Ordering::Relaxed),
            self.failed_transactions.load(Ordering::Relaxed),
            self.audit_trail.entry_count(),
        )
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("Financial Transaction Processing - Real-World Concurrency");
    println!("=========================================================");

    let engine = TransactionEngine::new()?;

    // Create test accounts
    println!("\n1. Setting up test accounts...");
    engine.create_account(1001, 10000)?; // Alice
    engine.create_account(1002, 5000)?; // Bob
    engine.create_account(1003, 7500)?; // Carol
    engine.create_account(1004, 2000)?; // Dave
    engine.create_account(1005, 15000)?; // Eve

    println!("  Created 5 accounts with initial balances");

    // Edge Case 1: High-frequency concurrent transfers
    println!("\n2. High-frequency concurrent transfers...");
    let start_time = Instant::now();

    let concurrent_transfers = vec![
        (1001, 1002, 100), // Alice -> Bob
        (1002, 1003, 50),  // Bob -> Carol
        (1003, 1004, 75),  // Carol -> Dave
        (1004, 1005, 25),  // Dave -> Eve
        (1005, 1001, 200), // Eve -> Alice
        (1001, 1003, 150), // Alice -> Carol
        (1002, 1004, 80),  // Bob -> Dave
        (1003, 1005, 120), // Carol -> Eve
        (1004, 1001, 90),  // Dave -> Alice
        (1005, 1002, 110), // Eve -> Bob
    ];

    let results = engine.process_batch_transfers(concurrent_transfers)?;
    let processing_time = start_time.elapsed();

    println!(
        "  Processed {} transfers in {:?}",
        results.len(),
        processing_time
    );

    let successful_count = results.iter().filter(|r| r.is_ok()).count();
    let failed_count = results.len() - successful_count;
    println!(
        "  Results: {} successful, {} failed",
        successful_count, failed_count
    );

    // Edge Case 2: Insufficient funds scenario
    println!("\n3. Testing insufficient funds edge case...");
    match engine.process_transfer(1004, 1005, 5000, Priority::High) {
        Ok(_) => println!("  ERROR: Transfer should have failed!"),
        Err(e) => println!("  Expected failure: {}", e),
    }

    // Edge Case 3: Same account transfer
    println!("\n4. Testing same account transfer edge case...");
    match engine.process_transfer(1001, 1001, 100, Priority::Normal) {
        Ok(_) => println!("  ERROR: Same account transfer should have failed!"),
        Err(e) => println!("  Expected failure: {}", e),
    }

    // Edge Case 4: Zero amount transfer
    println!("\n5. Testing zero amount transfer edge case...");
    match engine.process_transfer(1001, 1002, 0, Priority::Normal) {
        Ok(_) => println!("  ERROR: Zero amount transfer should have failed!"),
        Err(e) => println!("  Expected failure: {}", e),
    }

    // Edge Case 5: Race condition simulation
    println!("\n6. Simulating race conditions with rapid transfers...");
    let race_start = Instant::now();

    // Create multiple rapid transfers from the same account
    let rapid_transfers = vec![
        (1001, 1002, 1000),
        (1001, 1003, 1000),
        (1001, 1004, 1000),
        (1001, 1005, 1000),
        (1001, 1002, 1000),
        (1001, 1003, 1000),
        (1001, 1004, 1000),
        (1001, 1005, 1000),
    ];

    let race_results = engine.process_batch_transfers(rapid_transfers)?;
    let race_time = race_start.elapsed();

    let race_successful = race_results.iter().filter(|r| r.is_ok()).count();
    let race_failed = race_results.len() - race_successful;

    println!(
        "  Rapid transfers: {} successful, {} failed in {:?}",
        race_successful, race_failed, race_time
    );

    // Display final account balances
    println!("\n7. Final account balances:");
    for account_id in 1001..=1005 {
        match engine.get_account_balance(account_id) {
            Ok(balance) => println!("  Account {}: ${}", account_id, balance),
            Err(e) => println!("  Account {}: Error - {}", account_id, e),
        }
    }

    // Display audit statistics
    let (successful, failed, audit_entries) = engine.get_statistics();
    println!("\n8. Transaction Statistics:");
    println!("  Successful transactions: {}", successful);
    println!("  Failed transactions: {}", failed);
    println!("  Audit entries: {}", audit_entries);
    println!(
        "  Success rate: {:.2}%",
        (successful as f64 / (successful + failed) as f64) * 100.0
    );

    // Edge Case 6: Audit trail verification
    println!("\n9. Audit trail verification for Account 1001:");
    match engine.audit_trail.get_entries_for_account(1001) {
        Ok(entries) => {
            println!("  Found {} audit entries:", entries.len());
            for (i, entry) in entries.iter().take(5).enumerate() {
                println!(
                    "    {}. TX{}: {} (${} -> ${})",
                    i + 1,
                    entry.transaction_id,
                    entry.operation,
                    entry.balance_before,
                    entry.balance_after
                );
            }
            if entries.len() > 5 {
                println!("    ... and {} more entries", entries.len() - 5);
            }
        }
        Err(e) => println!("  Failed to get audit entries: {}", e),
    }

    println!("\nFinancial transaction processing completed successfully!");
    println!("All edge cases handled appropriately with proper error recovery.");

    Ok(())
}