foxhunt/testing/service-load/tests/database_stress_test.rs

//! Database stress testing for PostgreSQL performance validation
//!
//! This test suite validates PostgreSQL can handle production load:
//! - 10,000 inserts/sec sustained for 60 seconds
//! - Concurrent writes (10, 100 connections)
//! - Connection pool behavior under stress
//! - Query performance degradation under load
//! - Transaction rollback performance
//!
//! Run with: cargo test -p load_tests --test database_stress_test -- --ignored --nocapture

use anyhow::Result;
use chrono::Utc;
use sqlx::postgres::{PgPool, PgPoolOptions};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::task::JoinSet;
use uuid::Uuid;

const DATABASE_URL: &str = "postgresql://foxhunt:foxhunt_dev_password@localhost:5432/foxhunt";
const TEST_SYMBOL: &str = "STRESS_TEST";
const TEST_ACCOUNT: &str = "stress_test_account";

/// Metrics for database operations
#[derive(Debug)]
struct DbMetrics {
    inserts: AtomicU64,
    selects: AtomicU64,
    updates: AtomicU64,
    errors: AtomicU64,
    deadlocks: AtomicU64,
    timeouts: AtomicU64,
}

impl DbMetrics {
    fn new() -> Self {
        Self {
            inserts: AtomicU64::new(0),
            selects: AtomicU64::new(0),
            updates: AtomicU64::new(0),
            errors: AtomicU64::new(0),
            deadlocks: AtomicU64::new(0),
            timeouts: AtomicU64::new(0),
        }
    }

    fn print_summary(&self, duration: Duration, test_name: &str) {
        let inserts = self.inserts.load(Ordering::Relaxed);
        let selects = self.selects.load(Ordering::Relaxed);
        let updates = self.updates.load(Ordering::Relaxed);
        let errors = self.errors.load(Ordering::Relaxed);
        let deadlocks = self.deadlocks.load(Ordering::Relaxed);
        let timeouts = self.timeouts.load(Ordering::Relaxed);

        let secs = duration.as_secs_f64();
        let insert_rate = inserts as f64 / secs;
        let select_rate = selects as f64 / secs;
        let total_ops = inserts + selects + updates;
        let error_rate = if total_ops > 0 {
            (errors as f64 / total_ops as f64) * 100.0
        } else {
            0.0
        };

        println!("\n{}", "=".repeat(80));
        println!("Database Stress Test: {}", test_name);
        println!("{}", "=".repeat(80));
        println!("Duration: {:.2}s", secs);
        println!("Operations:");
        println!("  Inserts: {} ({:.2}/sec)", inserts, insert_rate);
        println!("  Selects: {} ({:.2}/sec)", selects, select_rate);
        println!("  Updates: {}", updates);
        println!("Errors:");
        println!("  Total: {} ({:.2}%)", errors, error_rate);
        println!("  Deadlocks: {}", deadlocks);
        println!("  Timeouts: {}", timeouts);
        println!("{}\n", "=".repeat(80));
    }
}

/// Test 1: Baseline insert performance (single connection)
async fn test_baseline_insert_performance() -> Result<()> {
    println!("\n🚀 Test 1: Baseline Insert Performance (single connection)");

    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect(DATABASE_URL)
        .await?;

    let metrics = Arc::new(DbMetrics::new());
    let start = Instant::now();
    let test_duration = Duration::from_secs(10);

    while start.elapsed() < test_duration {
        let order_id = Uuid::new_v4();
        let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);

        match sqlx::query(
            r#"
            INSERT INTO orders (
                id, symbol, side, order_type, time_in_force, quantity,
                filled_quantity, remaining_quantity, status, created_at,
                updated_at, account_id, venue
            ) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
            "#,
        )
        .bind(order_id)
        .bind(TEST_SYMBOL)
        .bind(created_at)
        .bind(TEST_ACCOUNT)
        .execute(&pool)
        .await
        {
            Ok(_) => metrics.inserts.fetch_add(1, Ordering::Relaxed),
            Err(e) => {
                eprintln!("Insert error: {:?}", e);
                metrics.errors.fetch_add(1, Ordering::Relaxed)
            },
        };
    }

    let duration = start.elapsed();
    metrics.print_summary(duration, "Baseline Insert Performance");

    let insert_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
    println!("✅ Baseline: {:.2} inserts/sec", insert_rate);

    // Cleanup
    cleanup_test_data(&pool).await?;

    Ok(())
}

/// Test 2: Concurrent writes (10 connections)
async fn test_concurrent_writes_10_connections() -> Result<()> {
    println!("\n🚀 Test 2: Concurrent Writes (10 connections)");

    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(DATABASE_URL)
        .await?;

    let metrics = Arc::new(DbMetrics::new());
    let mut join_set = JoinSet::new();
    let test_duration = Duration::from_secs(30);

    for worker_id in 0..10 {
        let pool = pool.clone();
        let metrics = Arc::clone(&metrics);

        join_set.spawn(async move {
            let start = Instant::now();
            while start.elapsed() < test_duration {
                let order_id = Uuid::new_v4();
                let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
                let symbol = format!("{}_W{}", TEST_SYMBOL, worker_id);

                match sqlx::query(r#"
                    INSERT INTO orders (
                        id, symbol, side, order_type, time_in_force, quantity,
                        filled_quantity, remaining_quantity, status, created_at,
                        updated_at, account_id, venue
                    ) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
                    "#)
                    .bind(order_id)
                    .bind(symbol)
                    .bind(created_at)
                    .bind(TEST_ACCOUNT)
                .execute(&pool)
                .await
                {
                    Ok(_) => metrics.inserts.fetch_add(1, Ordering::Relaxed),
                    Err(e) => {
                        if e.to_string().contains("deadlock") {
                            metrics.deadlocks.fetch_add(1, Ordering::Relaxed);
                        }
                        metrics.errors.fetch_add(1, Ordering::Relaxed)
                    }
                };

                tokio::time::sleep(Duration::from_micros(1000)).await;
            }
        });
    }

    let start = Instant::now();
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            eprintln!("Worker error: {:?}", e);
        }
    }
    let duration = start.elapsed();

    metrics.print_summary(duration, "Concurrent Writes (10 connections)");

    let insert_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
    println!("✅ Throughput: {:.2} inserts/sec", insert_rate);
    println!(
        "✅ Deadlocks: {}",
        metrics.deadlocks.load(Ordering::Relaxed)
    );

    // Cleanup
    cleanup_test_data(&pool).await?;

    Ok(())
}

/// Test 3: High throughput (100 connections, target 10K inserts/sec)
async fn test_high_throughput_100_connections() -> Result<()> {
    println!("\n🚀 Test 3: High Throughput (100 connections, 60s sustained)");

    let pool = PgPoolOptions::new()
        .max_connections(100)
        .acquire_timeout(Duration::from_secs(5))
        .connect(DATABASE_URL)
        .await?;

    let metrics = Arc::new(DbMetrics::new());
    let mut join_set = JoinSet::new();
    let test_duration = Duration::from_secs(60);

    for worker_id in 0..100 {
        let pool = pool.clone();
        let metrics = Arc::clone(&metrics);

        join_set.spawn(async move {
            let start = Instant::now();
            while start.elapsed() < test_duration {
                let order_id = Uuid::new_v4();
                let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
                let symbol = format!("{}_W{}", TEST_SYMBOL, worker_id % 20);

                match sqlx::query(r#"
                    INSERT INTO orders (
                        id, symbol, side, order_type, time_in_force, quantity,
                        filled_quantity, remaining_quantity, status, created_at,
                        updated_at, account_id, venue
                    ) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
                    "#)
                    .bind(order_id)
                    .bind(symbol)
                    .bind(created_at)
                    .bind(TEST_ACCOUNT)
                .execute(&pool)
                .await
                {
                    Ok(_) => metrics.inserts.fetch_add(1, Ordering::Relaxed),
                    Err(e) => {
                        let err_str = e.to_string();
                        if err_str.contains("deadlock") {
                            metrics.deadlocks.fetch_add(1, Ordering::Relaxed);
                        } else if err_str.contains("timeout") || err_str.contains("timed out") {
                            metrics.timeouts.fetch_add(1, Ordering::Relaxed);
                        }
                        metrics.errors.fetch_add(1, Ordering::Relaxed)
                    }
                };

                // Target: 100 inserts/sec per worker = 10K total
                tokio::time::sleep(Duration::from_micros(10000)).await;
            }
        });
    }

    let start = Instant::now();
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            eprintln!("Worker error: {:?}", e);
        }
    }
    let duration = start.elapsed();

    metrics.print_summary(duration, "High Throughput (100 connections)");

    let insert_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
    let inserts = metrics.inserts.load(Ordering::Relaxed);
    let errors = metrics.errors.load(Ordering::Relaxed);

    println!("🎯 Target: 10,000 inserts/sec");
    println!("✅ Achieved: {:.2} inserts/sec", insert_rate);
    println!(
        "✅ Success rate: {:.2}%",
        (inserts as f64 / (inserts + errors) as f64) * 100.0
    );

    assert!(
        insert_rate >= 9000.0,
        "Insert rate too low: {:.2} (expected >= 9000)",
        insert_rate
    );

    // Cleanup
    cleanup_test_data(&pool).await?;

    Ok(())
}

/// Test 4: Connection pool stress (exceed pool limits)
async fn test_connection_pool_stress() -> Result<()> {
    println!("\n🚀 Test 4: Connection Pool Stress (150 tasks, 100 max connections)");

    let pool = PgPoolOptions::new()
        .max_connections(100)
        .acquire_timeout(Duration::from_secs(10))
        .connect(DATABASE_URL)
        .await?;

    let metrics = Arc::new(DbMetrics::new());
    let mut join_set = JoinSet::new();
    let test_duration = Duration::from_secs(20);

    // Spawn 150 tasks (more than pool size)
    for worker_id in 0..150 {
        let pool = pool.clone();
        let metrics = Arc::clone(&metrics);

        join_set.spawn(async move {
            let start = Instant::now();
            let mut wait_times = Vec::new();

            while start.elapsed() < test_duration {
                let acquire_start = Instant::now();
                let order_id = Uuid::new_v4();
                let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);

                match sqlx::query(r#"
                    INSERT INTO orders (
                        id, symbol, side, order_type, time_in_force, quantity,
                        filled_quantity, remaining_quantity, status, created_at,
                        updated_at, account_id, venue
                    ) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
                    "#)
                    .bind(order_id)
                    .bind(format!("{}_P{}", TEST_SYMBOL, worker_id % 30))
                    .bind(created_at)
                    .bind(TEST_ACCOUNT)
                .execute(&pool)
                .await
                {
                    Ok(_) => {
                        metrics.inserts.fetch_add(1, Ordering::Relaxed);
                        wait_times.push(acquire_start.elapsed().as_millis());
                    }
                    Err(e) => {
                        if e.to_string().contains("timeout") {
                            metrics.timeouts.fetch_add(1, Ordering::Relaxed);
                        }
                        metrics.errors.fetch_add(1, Ordering::Relaxed);
                    }
                };

                tokio::time::sleep(Duration::from_millis(20)).await;
            }

            if !wait_times.is_empty() {
                let avg_wait = wait_times.iter().sum::<u128>() / wait_times.len() as u128;
                let max_wait = wait_times.iter().max().unwrap_or(&0);
                println!(
                    "Worker {}: avg wait {}ms, max wait {}ms",
                    worker_id, avg_wait, max_wait
                );
            }
        });
    }

    let start = Instant::now();
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            eprintln!("Worker error: {:?}", e);
        }
    }
    let duration = start.elapsed();

    metrics.print_summary(duration, "Connection Pool Stress");

    let timeouts = metrics.timeouts.load(Ordering::Relaxed);
    let errors = metrics.errors.load(Ordering::Relaxed);

    println!("✅ Connection timeouts: {}", timeouts);
    println!("✅ Total errors: {}", errors);

    assert!(
        errors < 100,
        "Too many errors under pool stress: {} (expected < 100)",
        errors
    );

    // Cleanup
    cleanup_test_data(&pool).await?;

    Ok(())
}

/// Test 5: Query performance under write load
async fn test_query_performance_under_load() -> Result<()> {
    println!("\n🚀 Test 5: Query Performance Under Write Load");

    let pool = PgPoolOptions::new()
        .max_connections(50)
        .connect(DATABASE_URL)
        .await?;

    let metrics = Arc::new(DbMetrics::new());
    let mut join_set = JoinSet::new();
    let test_duration = Duration::from_secs(30);

    // Spawn 30 writers
    for writer_id in 0..30 {
        let pool = pool.clone();
        let metrics = Arc::clone(&metrics);

        join_set.spawn(async move {
            let start = Instant::now();
            while start.elapsed() < test_duration {
                let order_id = Uuid::new_v4();
                let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);

                if sqlx::query(r#"
                    INSERT INTO orders (
                        id, symbol, side, order_type, time_in_force, quantity,
                        filled_quantity, remaining_quantity, status, created_at,
                        updated_at, account_id, venue
                    ) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
                    "#)
                    .bind(order_id)
                    .bind(format!("{}_Q{}", TEST_SYMBOL, writer_id))
                    .bind(created_at)
                    .bind(TEST_ACCOUNT)
                .execute(&pool)
                .await
                .is_ok()
                {
                    metrics.inserts.fetch_add(1, Ordering::Relaxed);
                }

                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        });
    }

    // Spawn 20 readers
    for reader_id in 0..20 {
        let pool = pool.clone();
        let metrics = Arc::clone(&metrics);

        join_set.spawn(async move {
            let start = Instant::now();
            let mut query_times = Vec::new();

            while start.elapsed() < test_duration {
                let query_start = Instant::now();

                match sqlx::query(
                    r#"
                    SELECT id, symbol, status, quantity, filled_quantity
                    FROM orders
                    WHERE account_id = $1 AND status = 'pending'
                    ORDER BY created_at DESC
                    LIMIT 100
                    "#,
                )
                .bind(TEST_ACCOUNT)
                .fetch_all(&pool)
                .await
                {
                    Ok(_) => {
                        metrics.selects.fetch_add(1, Ordering::Relaxed);
                        query_times.push(query_start.elapsed().as_micros());
                    },
                    Err(_) => {
                        metrics.errors.fetch_add(1, Ordering::Relaxed);
                    },
                }

                tokio::time::sleep(Duration::from_millis(50)).await;
            }

            if !query_times.is_empty() {
                let avg_time = query_times.iter().sum::<u128>() / query_times.len() as u128;
                let p95_idx = (query_times.len() as f64 * 0.95) as usize;
                let mut sorted = query_times.clone();
                sorted.sort_unstable();
                let p95_time = sorted.get(p95_idx).unwrap_or(&0);

                println!(
                    "Reader {}: avg {}μs, p95 {}μs",
                    reader_id, avg_time, p95_time
                );
            }
        });
    }

    let start = Instant::now();
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            eprintln!("Worker error: {:?}", e);
        }
    }
    let duration = start.elapsed();

    metrics.print_summary(duration, "Query Performance Under Load");

    let selects = metrics.selects.load(Ordering::Relaxed);
    let select_rate = selects as f64 / duration.as_secs_f64();

    println!("✅ Read throughput: {:.2} queries/sec", select_rate);

    // Cleanup
    cleanup_test_data(&pool).await?;

    Ok(())
}

/// Test 6: Transaction stress (with rollbacks)
async fn test_transaction_stress() -> Result<()> {
    println!("\n🚀 Test 6: Transaction Stress (commits and rollbacks)");

    let pool = PgPoolOptions::new()
        .max_connections(20)
        .connect(DATABASE_URL)
        .await?;

    let metrics = Arc::new(DbMetrics::new());
    let mut join_set = JoinSet::new();
    let test_duration = Duration::from_secs(20);

    for worker_id in 0..20 {
        let pool = pool.clone();
        let metrics = Arc::clone(&metrics);

        join_set.spawn(async move {
            let start = Instant::now();
            let mut commits = 0u64;
            let mut rollbacks = 0u64;

            while start.elapsed() < test_duration {
                let mut tx = match pool.begin().await {
                    Ok(tx) => tx,
                    Err(_) => {
                        metrics.errors.fetch_add(1, Ordering::Relaxed);
                        continue;
                    }
                };

                let order_id = Uuid::new_v4();
                let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);

                if sqlx::query(r#"
                    INSERT INTO orders (
                        id, symbol, side, order_type, time_in_force, quantity,
                        filled_quantity, remaining_quantity, status, created_at,
                        updated_at, account_id, venue
                    ) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
                    "#)
                    .bind(order_id)
                    .bind(format!("{}_T{}", TEST_SYMBOL, worker_id))
                    .bind(created_at)
                    .bind(TEST_ACCOUNT)
                .execute(&mut *tx)
                .await
                .is_ok()
                {
                    metrics.inserts.fetch_add(1, Ordering::Relaxed);

                    // Randomly commit or rollback (70% commit, 30% rollback)
                    if worker_id % 10 < 7 {
                        if tx.commit().await.is_ok() {
                            commits += 1;
                        }
                    } else if tx.rollback().await.is_ok() {
                        rollbacks += 1;
                    }
                }

                tokio::time::sleep(Duration::from_millis(10)).await;
            }

            println!(
                "Worker {}: commits={}, rollbacks={}",
                worker_id, commits, rollbacks
            );
        });
    }

    let start = Instant::now();
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            eprintln!("Worker error: {:?}", e);
        }
    }
    let duration = start.elapsed();

    metrics.print_summary(duration, "Transaction Stress");

    let tx_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
    println!("✅ Transaction rate: {:.2} tx/sec", tx_rate);

    // Cleanup
    cleanup_test_data(&pool).await?;

    Ok(())
}

/// Cleanup test data
async fn cleanup_test_data(pool: &PgPool) -> Result<()> {
    println!("🧹 Cleaning up test data...");

    let result = sqlx::query(r#"DELETE FROM orders WHERE symbol LIKE $1"#)
        .bind(format!("{}%", TEST_SYMBOL))
        .execute(pool)
        .await?;

    println!("🧹 Deleted {} test orders", result.rows_affected());

    Ok(())
}

/// Integration test: Run all database stress tests
#[tokio::test]
#[ignore = "Requires PostgreSQL"]
async fn test_comprehensive_database_stress() -> Result<()> {
    println!("\n{}", "=".repeat(80));
    println!("🎯 Comprehensive Database Stress Test Suite");
    println!("{}\n", "=".repeat(80));

    test_baseline_insert_performance().await?;
    tokio::time::sleep(Duration::from_secs(2)).await;

    test_concurrent_writes_10_connections().await?;
    tokio::time::sleep(Duration::from_secs(2)).await;

    test_high_throughput_100_connections().await?;
    tokio::time::sleep(Duration::from_secs(2)).await;

    test_connection_pool_stress().await?;
    tokio::time::sleep(Duration::from_secs(2)).await;

    test_query_performance_under_load().await?;
    tokio::time::sleep(Duration::from_secs(2)).await;

    test_transaction_stress().await?;

    println!("\n{}", "=".repeat(80));
    println!("✅ All database stress tests completed successfully!");
    println!("{}\n", "=".repeat(80));

    Ok(())
}