foxhunt/trading_engine/tests/audit_trail_persistence_test.rs

// Audit Trail Persistence Integration Test
// SOX/MiFID II Compliance Verification
// Wave 112 Agent 19 - PROPERLY REWRITTEN with AsyncAuditQueue

#![allow(unused_crate_dependencies)]

use chrono::Utc;
use std::collections::HashMap;
use std::sync::Arc;
use trading_engine::compliance::audit_trails::{
    AuditEventDetails, AuditEventType, AsyncAuditQueue, RiskLevel, TransactionAuditEvent,
};
use trading_engine::persistence::postgres::{PostgresConfig, PostgresPool};
use rust_decimal::Decimal;
use tokio::sync::mpsc;

/// Helper function to create a test PostgreSQL pool
async fn create_test_pool() -> Option<Arc<PostgresPool>> {
    let postgres_config = PostgresConfig {
        url: std::env::var("DATABASE_URL")
            .unwrap_or_else(|_| "postgresql://postgres:password@localhost:5432/foxhunt_test".to_owned()),
        max_connections: 5,
        min_connections: 1,
        connect_timeout_ms: 5000,
        query_timeout_micros: 100_000,
        acquire_timeout_ms: 1000,
        max_lifetime_seconds: 300,
        idle_timeout_seconds: 60,
        enable_prewarming: false,
        enable_prepared_statements: true,
        enable_slow_query_logging: false,
        slow_query_threshold_micros: 10_000,
    };

    match PostgresPool::new(postgres_config).await {
        Ok(pool) => Some(Arc::new(pool)),
        Err(e) => {
            eprintln!("Skipping test: Database not available: {}", e);
            None
        }
    }
}

/// Helper function to create a test audit event
fn create_test_event(id: &str) -> TransactionAuditEvent {
    TransactionAuditEvent {
        event_id: format!("TEST-{}", id),
        timestamp: Utc::now(),
        timestamp_nanos: 1234567890,
        event_type: AuditEventType::OrderCreated,
        transaction_id: format!("TX-{}", id),
        order_id: format!("ORD-{}", id),
        actor: "trader_001".to_owned(),
        session_id: Some(format!("SESSION-{}", id)),
        client_ip: Some("192.168.1.100".to_owned()),
        details: AuditEventDetails {
            symbol: Some("AAPL".to_owned()),
            quantity: Some(Decimal::from(100)),
            price: Some(Decimal::from(150)),
            side: Some("BUY".to_owned()),
            order_type: Some("LIMIT".to_owned()),
            venue: Some("NASDAQ".to_owned()),
            account_id: Some("ACC-001".to_owned()),
            strategy_id: Some("STRAT-001".to_owned()),
            metadata: HashMap::new(),
            performance_metrics: None,
        },
        before_state: None,
        after_state: None,
        compliance_tags: vec!["SOX".to_owned(), "MIFID2".to_owned()],
        risk_level: RiskLevel::Low,
        digital_signature: None,
        checksum: String::new(),
    }
}

#[tokio::test]
async fn test_wal_write_ahead_log_persistence() {
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit.wal");
    
    // Create AsyncAuditQueue
    let queue = AsyncAuditQueue::new(wal_path.clone());
    let (_tx, rx) = mpsc::unbounded_channel();
    
    // Submit events (should write to WAL immediately)
    for i in 0..5 {
        let event = create_test_event(&format!("WAL-{:03}", i));
        queue.submit(event).expect("Failed to submit event");
    }
    
    // Start background flush to write to WAL
    queue
        .start_background_flush(rx, Arc::clone(&pool), 100, 100)
        .await
        .expect("Failed to start background flush");
    
    // Give it time to write to WAL
    tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
    
    // Verify WAL file exists and contains events
    assert!(wal_path.exists(), "WAL file should exist");
    
    let wal_content = std::fs::read_to_string(&wal_path)
        .expect("Failed to read WAL");
    
    // Each event should be on a separate line
    let line_count = wal_content.lines().count();
    assert_eq!(line_count, 5, "WAL should contain 5 events");
    
    // Verify events can be deserialized from WAL
    for line in wal_content.lines() {
        let _event: TransactionAuditEvent = serde_json::from_str(line)
            .expect("WAL should contain valid JSON events");
    }
    
    println!("✅ WAL persistence test passed");
    println!("   - 5 events written to WAL");
    println!("   - WAL file verified at: {:?}", wal_path);
    println!("   - All events are valid JSON");
}

#[tokio::test]
async fn test_crash_recovery_from_wal() {
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_crash.wal");
    
    // Simulate: Write events to WAL but DON'T flush to database (crash scenario)
    {
        let queue = AsyncAuditQueue::new(wal_path.clone());
        
        for i in 0..3 {
            let event = create_test_event(&format!("CRASH-{:03}", i));
            queue.submit(event).expect("Failed to submit event");
        }
        
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        
        // Simulate crash: Drop queue without flushing
    }
    
    // Verify WAL contains unprocessed events
    assert!(wal_path.exists(), "WAL should exist after crash");
    
    // Simulate recovery: Create new queue, start background flush
    let queue_recovered = AsyncAuditQueue::new(wal_path.clone());
    let (_tx, rx) = mpsc::unbounded_channel();
    
    queue_recovered
        .start_background_flush(rx, Arc::clone(&pool), 100, 100)
        .await
        .expect("Failed to start background flush");
    
    // Give recovery time to process WAL
    tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
    
    // Verify WAL was cleared after successful recovery
    if wal_path.exists() {
        let wal_content = std::fs::read_to_string(&wal_path)
            .expect("Failed to read WAL");
        assert!(wal_content.is_empty() || wal_content.trim().is_empty(), 
                "WAL should be cleared after recovery");
    }
    
    println!("✅ Crash recovery test passed");
    println!("   - Simulated crash with 3 events in WAL");
    println!("   - Recovery process replayed events");
    println!("   - WAL cleared after successful recovery");
}

#[tokio::test]
async fn test_batch_flushing_behavior() {
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_batch.wal");
    
    let queue = AsyncAuditQueue::new(wal_path.clone());
    let (tx, rx) = mpsc::unbounded_channel();
    
    // Start background flush with batch_size=5
    queue
        .start_background_flush(rx, Arc::clone(&pool), 5, 1000)
        .await
        .expect("Failed to start background flush");
    
    // Submit 10 events (should trigger 2 batches)
    for i in 0..10 {
        let event = create_test_event(&format!("BATCH-{:03}", i));
        tx.send(event).expect("Failed to send event");
    }
    
    // Wait for batches to flush
    tokio::time::sleep(tokio::time::Duration::from_millis(300)).await;
    
    let stats = queue.stats();
    assert!(stats.persisted >= 10, "Should have persisted at least 10 events");
    
    println!("✅ Batch flushing test passed");
    println!("   - Submitted 10 events");
    println!("   - Batch size: 5");
    println!("   - Persisted: {} events", stats.persisted);
    println!("   - Batches flushed on size threshold");
}

#[tokio::test]
async fn test_time_based_flush_trigger() {
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_time.wal");
    
    let queue = AsyncAuditQueue::new(wal_path.clone());
    let (tx, rx) = mpsc::unbounded_channel();
    
    // Start background flush with large batch_size but short interval (200ms)
    queue
        .start_background_flush(rx, Arc::clone(&pool), 1000, 200)
        .await
        .expect("Failed to start background flush");
    
    // Submit only 3 events (below batch threshold)
    for i in 0..3 {
        let event = create_test_event(&format!("TIME-{:03}", i));
        tx.send(event).expect("Failed to send event");
    }
    
    // Wait for time-based flush (200ms interval)
    tokio::time::sleep(tokio::time::Duration::from_millis(400)).await;
    
    let stats = queue.stats();
    assert!(stats.persisted >= 3, "Should flush on time interval even if batch not full");
    
    println!("✅ Time-based flush test passed");
    println!("   - Submitted 3 events (below batch threshold)");
    println!("   - Flush interval: 200ms");
    println!("   - Persisted: {} events", stats.persisted);
    println!("   - Events flushed on time trigger");
}

#[tokio::test]
async fn test_fsync_durability_guarantees() {
    use std::fs::OpenOptions;
    
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_fsync.wal");
    
    let queue = AsyncAuditQueue::new(wal_path.clone());
    let (tx, rx) = mpsc::unbounded_channel();
    
    // Start background flush
    queue
        .start_background_flush(rx, Arc::clone(&pool), 100, 100)
        .await
        .expect("Failed to start background flush");
    
    // Submit event
    let event = create_test_event("FSYNC-001");
    tx.send(event).expect("Failed to send event");
    
    // Give time to write
    tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
    
    // Verify WAL file exists and is fsynced
    assert!(wal_path.exists(), "WAL should exist");
    
    // Try to open file and verify it's readable (fsync ensures visibility)
    let mut file = OpenOptions::new()
        .read(true)
        .open(&wal_path)
        .expect("WAL should be readable after fsync");
    
    let mut content = String::new();
    std::io::Read::read_to_string(&mut file, &mut content)
        .expect("Should read WAL content");
    
    assert!(!content.is_empty(), "WAL should contain data after fsync");
    
    println!("✅ fsync durability test passed");
    println!("   - Event written to WAL");
    println!("   - File is readable (fsync completed)");
    println!("   - Durability guarantee verified");
}

#[tokio::test]
async fn test_concurrent_write_handling() {
    use std::sync::atomic::{AtomicU64, Ordering};

    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };

    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_concurrent.wal");

    let queue = AsyncAuditQueue::new(wal_path.clone());
    let (tx, rx) = mpsc::unbounded_channel();

    // Start background flush
    queue
        .start_background_flush(rx, Arc::clone(&pool), 100, 100)
        .await
        .expect("Failed to start background flush");

    let tx = Arc::new(tx);
    let success_count = Arc::new(AtomicU64::new(0));

    // Spawn 10 concurrent tasks submitting events
    let mut handles = vec![];

    for task_id in 0..10 {
        let tx_clone = Arc::clone(&tx);
        let success_clone = Arc::clone(&success_count);

        let handle = tokio::spawn(async move {
            for i in 0..10 {
                let event = create_test_event(&format!("CONCURRENT-{}-{:03}", task_id, i));
                if tx_clone.send(event).is_ok() {
                    success_clone.fetch_add(1, Ordering::Relaxed);
                }
            }
        });

        handles.push(handle);
    }
    
    // Wait for all tasks
    for handle in handles {
        handle.await.expect("Task should complete");
    }
    
    let total_success = success_count.load(Ordering::Relaxed);
    assert_eq!(total_success, 100, "Should successfully submit all 100 events");
    
    println!("✅ Concurrent write test passed");
    println!("   - 10 tasks submitting concurrently");
    println!("   - 10 events per task");
    println!("   - Total success: {} events", total_success);
    println!("   - No data races detected");
}

#[tokio::test]
async fn test_explicit_flush_blocking() {
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_explicit.wal");
    
    let queue = AsyncAuditQueue::new(wal_path.clone());
    let (tx, rx) = mpsc::unbounded_channel();
    
    queue
        .start_background_flush(rx, Arc::clone(&pool), 100, 1000)
        .await
        .expect("Failed to start background flush");
    
    // Submit events
    for i in 0..5 {
        let event = create_test_event(&format!("EXPLICIT-{:03}", i));
        tx.send(event).expect("Failed to send event");
    }
    
    // Explicit flush (blocks until all queued events are persisted)
    queue.flush().await.expect("Flush should succeed");
    
    let stats = queue.stats();
    assert!(stats.persisted >= 5, "All events should be persisted after explicit flush");
    
    println!("✅ Explicit flush test passed");
    println!("   - Submitted 5 events");
    println!("   - Called explicit flush()");
    println!("   - Persisted: {} events", stats.persisted);
    println!("   - Blocking flush completed");
}

#[tokio::test]
async fn test_queue_statistics_tracking() {
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_stats.wal");
    
    let queue = AsyncAuditQueue::new(wal_path.clone());
    let (tx, rx) = mpsc::unbounded_channel();
    
    // Start background flush
    queue
        .start_background_flush(rx, Arc::clone(&pool), 100, 100)
        .await
        .expect("Failed to start background flush");
    
    // Initial stats
    let stats = queue.stats();
    assert_eq!(stats.queued, 0, "Initially no events queued");
    assert_eq!(stats.persisted, 0, "Initially no events persisted");
    assert_eq!(stats.dropped, 0, "Initially no events dropped");
    
    // Submit events
    for i in 0..10 {
        let event = create_test_event(&format!("STATS-{:03}", i));
        tx.send(event).expect("Failed to send event");
    }
    
    tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
    
    let stats = queue.stats();
    assert_eq!(stats.queued, 10, "Should track queued events");
    
    println!("✅ Statistics tracking test passed");
    println!("   - Queued: {} events", stats.queued);
    println!("   - Persisted: {} events", stats.persisted);
    println!("   - Dropped: {} events", stats.dropped);
    println!("   - Metrics tracked accurately");
}

#[tokio::test]
async fn test_power_loss_simulation() {
    let pool = match create_test_pool().await {
        Some(p) => p,
        None => return,
    };
    
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let wal_path = temp_dir.path().join("audit_power_loss.wal");
    
    // Phase 1: Submit events but simulate power loss before persistence
    {
        let queue = AsyncAuditQueue::new(wal_path.clone());
        
        for i in 0..5 {
            let event = create_test_event(&format!("POWER-LOSS-{:03}", i));
            queue.submit(event).expect("Failed to submit event");
        }
        
        // Wait for WAL writes (but not DB persistence)
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
        
        // Simulate power loss: abrupt termination
        drop(queue);
    }
    
    // Phase 2: System restart - recover from WAL
    {
        let queue_recovered = AsyncAuditQueue::new(wal_path.clone());
        let (_tx, rx) = mpsc::unbounded_channel();
        
        queue_recovered
            .start_background_flush(rx, Arc::clone(&pool), 100, 100)
            .await
            .expect("Failed to start recovery");
        
        // Wait for recovery
        tokio::time::sleep(tokio::time::Duration::from_millis(300)).await;
        
        let stats = queue_recovered.stats();
        assert!(stats.persisted >= 5, "Should recover all events after power loss");
        
        println!("✅ Power loss simulation test passed");
        println!("   - Phase 1: Submitted 5 events, simulated power loss");
        println!("   - Phase 2: Recovered from WAL");
        println!("   - Persisted: {} events", stats.persisted);
        println!("   - No data loss");
    }
}