foxhunt/trading_engine/tests/persistence_integration_tests.rs

//! Persistence Integration Tests - Live Database Operations
//!
//! This test suite validates persistence layer with LIVE databases:
//! - PostgreSQL: Full CRUD with transactions, isolation levels, concurrency
//! - Redis: Caching patterns with TTL, pub/sub, pipelining
//! - ClickHouse: Bulk analytics queries, time-series aggregation
//! - Cross-persistence consistency validation
//!
//! Requirements:
//! - PostgreSQL: localhost:5432 (foxhunt/foxhunt_dev_password)
//! - Redis: localhost:6379
//! - ClickHouse: localhost:8123 (optional)
//!
//! Run with: docker-compose up -d postgres redis

#![allow(unused_imports)]
#![allow(dead_code)]

use chrono::Utc;
use serde::{Deserialize, Serialize};
use serial_test::serial;
use sqlx::Row;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::sleep;
use uuid::Uuid;

use trading_engine::persistence::{
    clickhouse::{ClickHouseClient, ClickHouseConfig, ClickHouseError},
    postgres::{PostgresConfig, PostgresError, PostgresPool},
    redis::{RedisConfig, RedisError, RedisPool},
};

// Test configuration helpers - use relaxed timeouts for integration tests
fn test_postgres_config() -> PostgresConfig {
    PostgresConfig {
        url: "postgresql://foxhunt:foxhunt_dev_password@localhost:5432/foxhunt".to_string(),
        max_connections: 50,
        min_connections: 2,
        connect_timeout_ms: 5000,      // 5 seconds for test reliability
        query_timeout_micros: 5000000, // 5 seconds (not HFT-critical)
        acquire_timeout_ms: 5000,      // 5 seconds to handle after_connect
        max_lifetime_seconds: 3600,
        idle_timeout_seconds: 300,
        enable_prewarming: false, // Disable for tests
        enable_prepared_statements: true,
        enable_slow_query_logging: false,
        slow_query_threshold_micros: 1000000,
    }
}

fn test_redis_config() -> RedisConfig {
    RedisConfig {
        url: "redis://localhost:6379".to_string(),
        max_connections: 50,
        min_connections: 5,
        connect_timeout_ms: 5000,        // 5 seconds for test reliability
        command_timeout_micros: 5000000, // 5 seconds (not HFT-critical)
        acquire_timeout_ms: 5000,        // 5 seconds for tests
        max_lifetime_seconds: 3600,
        idle_timeout_seconds: 300,
        enable_prewarming: false,
        enable_pipelining: true,
        pipeline_batch_size: 100,
        default_ttl_seconds: 300,
        enable_compression: false,
        compression_threshold_bytes: 1024,
    }
}

// Test data structures
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
struct TestOrder {
    id: String,
    symbol: String,
    quantity: i64,
    price: f64,
    side: String,
    status: String,
}

impl TestOrder {
    fn new(symbol: &str, quantity: i64, price: f64, side: &str) -> Self {
        Self {
            id: Uuid::new_v4().to_string(),
            symbol: symbol.to_string(),
            quantity,
            price,
            side: side.to_string(),
            status: "NEW".to_string(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct TestTrade {
    id: String,
    order_id: String,
    quantity: i64,
    price: f64,
    timestamp: i64,
}

// ============================================================================
// POSTGRESQL INTEGRATION TESTS (15+ tests)
// ============================================================================

#[tokio::test]
#[serial]
async fn test_postgres_connection_pool_creation() {
    let mut config = test_postgres_config();
    config.max_connections = 10;

    let result = PostgresPool::new(config).await;
    assert!(
        result.is_ok(),
        "PostgreSQL connection pool creation should succeed"
    );

    let pool = result.unwrap();
    let metrics = pool.get_metrics().await.unwrap();
    assert_eq!(metrics.total_queries, 0, "Initial query count should be 0");
}

#[tokio::test]
#[serial]
async fn test_postgres_health_check() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();
    let health_result = pool.health_check().await;

    assert!(health_result.is_ok(), "PostgreSQL health check should pass");
}

#[tokio::test]
#[serial]
async fn test_postgres_simple_query() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Execute simple query
    let result = sqlx::query("SELECT 1 as value")
        .fetch_one(pool.pool())
        .await;

    assert!(result.is_ok(), "Simple SELECT query should succeed");
    let row = result.unwrap();
    let value: i32 = row.try_get("value").unwrap();
    assert_eq!(value, 1, "Query should return correct value");

    // Verify metrics
    let metrics = pool.get_metrics().await.unwrap();
    assert_eq!(metrics.total_queries, 1, "Should track query execution");
}

#[tokio::test]
#[serial]
async fn test_postgres_create_table_crud() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Create test table
    let create_result = sqlx::query(
        "CREATE TABLE IF NOT EXISTS test_orders (
            id TEXT PRIMARY KEY,
            symbol TEXT NOT NULL,
            quantity BIGINT NOT NULL,
            price DOUBLE PRECISION NOT NULL,
            side TEXT NOT NULL,
            status TEXT NOT NULL
        )",
    )
    .execute(pool.pool())
    .await;

    assert!(create_result.is_ok(), "Table creation should succeed");

    // Insert test order
    let order = TestOrder::new("AAPL", 100, 150.50, "BUY");
    let insert_result = sqlx::query(
        "INSERT INTO test_orders (id, symbol, quantity, price, side, status)
         VALUES ($1, $2, $3, $4, $5, $6)",
    )
    .bind(&order.id)
    .bind(&order.symbol)
    .bind(order.quantity)
    .bind(order.price)
    .bind(&order.side)
    .bind(&order.status)
    .execute(pool.pool())
    .await;

    assert!(insert_result.is_ok(), "Insert should succeed");

    // Read back order
    let read_result = sqlx::query_as::<_, (String, String, i64, f64, String, String)>(
        "SELECT id, symbol, quantity, price, side, status FROM test_orders WHERE id = $1",
    )
    .bind(&order.id)
    .fetch_one(pool.pool())
    .await;

    assert!(read_result.is_ok(), "Read should succeed");
    let (id, symbol, quantity, _price, _side, _status) = read_result.unwrap();
    assert_eq!(id, order.id);
    assert_eq!(symbol, order.symbol);
    assert_eq!(quantity, order.quantity);

    // Update order
    let update_result = sqlx::query("UPDATE test_orders SET status = $1 WHERE id = $2")
        .bind("FILLED")
        .bind(&order.id)
        .execute(pool.pool())
        .await;

    assert!(update_result.is_ok(), "Update should succeed");

    // Delete order
    let delete_result = sqlx::query("DELETE FROM test_orders WHERE id = $1")
        .bind(&order.id)
        .execute(pool.pool())
        .await;

    assert!(delete_result.is_ok(), "Delete should succeed");

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_orders")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_transaction_commit() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Create test table
    let _ = sqlx::query("CREATE TABLE IF NOT EXISTS test_txn (id TEXT PRIMARY KEY, value INTEGER)")
        .execute(pool.pool())
        .await;

    // Begin transaction
    let mut tx = pool.pool().begin().await.unwrap();

    // Insert within transaction
    let insert_result = sqlx::query("INSERT INTO test_txn (id, value) VALUES ($1, $2)")
        .bind("txn-1")
        .bind(100)
        .execute(&mut *tx)
        .await;

    assert!(insert_result.is_ok(), "Transaction insert should succeed");

    // Commit transaction
    let commit_result = tx.commit().await;
    assert!(commit_result.is_ok(), "Transaction commit should succeed");

    // Verify data persisted
    let verify_result = sqlx::query_as::<_, (i32,)>("SELECT value FROM test_txn WHERE id = $1")
        .bind("txn-1")
        .fetch_one(pool.pool())
        .await;

    assert!(verify_result.is_ok(), "Data should be committed");
    assert_eq!(verify_result.unwrap().0, 100);

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_txn")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_transaction_rollback() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Create test table
    let _ = sqlx::query(
        "CREATE TABLE IF NOT EXISTS test_rollback (id TEXT PRIMARY KEY, value INTEGER)",
    )
    .execute(pool.pool())
    .await;

    // Begin transaction
    let mut tx = pool.pool().begin().await.unwrap();

    // Insert within transaction
    let _ = sqlx::query("INSERT INTO test_rollback (id, value) VALUES ($1, $2)")
        .bind("rb-1")
        .bind(200)
        .execute(&mut *tx)
        .await;

    // Rollback transaction
    let rollback_result = tx.rollback().await;
    assert!(
        rollback_result.is_ok(),
        "Transaction rollback should succeed"
    );

    // Verify data was NOT persisted
    let verify_result =
        sqlx::query_as::<_, (i32,)>("SELECT value FROM test_rollback WHERE id = $1")
            .bind("rb-1")
            .fetch_optional(pool.pool())
            .await;

    assert!(verify_result.is_ok());
    assert!(
        verify_result.unwrap().is_none(),
        "Data should not exist after rollback"
    );

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_rollback")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_concurrent_transactions() {
    let mut config = test_postgres_config();
    config.max_connections = 20;

    let pool = Arc::new(PostgresPool::new(config).await.unwrap());

    // Create test table
    let _ = sqlx::query(
        "CREATE TABLE IF NOT EXISTS test_concurrent (id TEXT PRIMARY KEY, counter INTEGER)",
    )
    .execute(pool.pool())
    .await;

    // Insert initial value
    let _ = sqlx::query("INSERT INTO test_concurrent (id, counter) VALUES ($1, $2)")
        .bind("counter-1")
        .bind(0)
        .execute(pool.pool())
        .await;

    // Spawn 10 concurrent transactions
    let mut handles = vec![];
    for _i in 0..10 {
        let pool_clone = Arc::clone(&pool);
        let handle = tokio::spawn(async move {
            let mut tx = pool_clone.pool().begin().await.unwrap();

            // Read current value
            let current: (i32,) =
                sqlx::query_as("SELECT counter FROM test_concurrent WHERE id = $1 FOR UPDATE")
                    .bind("counter-1")
                    .fetch_one(&mut *tx)
                    .await
                    .unwrap();

            // Increment
            let new_value = current.0 + 1;

            // Update
            let _ = sqlx::query("UPDATE test_concurrent SET counter = $1 WHERE id = $2")
                .bind(new_value)
                .bind("counter-1")
                .execute(&mut *tx)
                .await
                .unwrap();

            tx.commit().await.unwrap();
        });
        handles.push(handle);
    }

    // Wait for all transactions
    for handle in handles {
        let _ = handle.await;
    }

    // Verify final count
    let final_count: (i32,) = sqlx::query_as("SELECT counter FROM test_concurrent WHERE id = $1")
        .bind("counter-1")
        .fetch_one(pool.pool())
        .await
        .unwrap();

    assert_eq!(
        final_count.0, 10,
        "All transactions should complete successfully"
    );

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_concurrent")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_bulk_insert_performance() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Create test table
    let _ = sqlx::query("CREATE TABLE IF NOT EXISTS test_bulk (id TEXT, value INTEGER)")
        .execute(pool.pool())
        .await;

    let start = std::time::Instant::now();

    // Insert 1000 rows
    for i in 0..1000 {
        let _ = sqlx::query("INSERT INTO test_bulk (id, value) VALUES ($1, $2)")
            .bind(format!("bulk-{}", i))
            .bind(i)
            .execute(pool.pool())
            .await;
    }

    let elapsed = start.elapsed();
    println!("Bulk insert of 1000 rows took: {:?}", elapsed);

    // Verify count
    let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM test_bulk")
        .fetch_one(pool.pool())
        .await
        .unwrap();

    assert_eq!(count.0, 1000, "Should insert all 1000 rows");
    assert!(
        elapsed.as_millis() < 5000,
        "Should complete within 5 seconds"
    );

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_bulk")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_prepared_statements() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Create test table
    let _ = sqlx::query("CREATE TABLE IF NOT EXISTS test_prepared (id TEXT, value INTEGER)")
        .execute(pool.pool())
        .await;

    // Execute same query multiple times (should use prepared statement)
    for i in 0..10 {
        let _ = sqlx::query("INSERT INTO test_prepared (id, value) VALUES ($1, $2)")
            .bind(format!("prep-{}", i))
            .bind(i)
            .execute(pool.pool())
            .await;
    }

    let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM test_prepared")
        .fetch_one(pool.pool())
        .await
        .unwrap();

    assert_eq!(count.0, 10, "Prepared statement execution should work");

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_prepared")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_connection_timeout() {
    let config = test_postgres_config();

    let result = PostgresPool::new(config).await;
    // Should either succeed or timeout gracefully
    assert!(result.is_ok() || matches!(result.unwrap_err(), PostgresError::Connection(_)));
}

#[tokio::test]
#[serial]
async fn test_postgres_pool_statistics() {
    let mut config = test_postgres_config();
    config.max_connections = 5;

    let pool = PostgresPool::new(config).await.unwrap();

    // Execute some queries
    for _ in 0..3 {
        let _ = sqlx::query("SELECT 1").fetch_one(pool.pool()).await;
    }

    let stats = pool.pool_stats().await;
    assert!(stats.size >= 2, "Pool should maintain min connections");
    assert!(stats.size <= 5, "Pool should not exceed max connections");

    let metrics = pool.get_metrics().await.unwrap();
    assert_eq!(metrics.total_queries, 3, "Should track query count");
}

#[tokio::test]
#[serial]
async fn test_postgres_index_usage() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Create table with index
    let _ = sqlx::query(
        "CREATE TABLE IF NOT EXISTS test_indexed (
            id SERIAL PRIMARY KEY,
            symbol TEXT NOT NULL,
            value INTEGER
        )",
    )
    .execute(pool.pool())
    .await;

    let _ = sqlx::query("CREATE INDEX IF NOT EXISTS idx_symbol ON test_indexed(symbol)")
        .execute(pool.pool())
        .await;

    // Insert test data
    for i in 0..100 {
        let _ = sqlx::query("INSERT INTO test_indexed (symbol, value) VALUES ($1, $2)")
            .bind(format!("SYM{}", i % 10))
            .bind(i)
            .execute(pool.pool())
            .await;
    }

    // Query using index
    let result = sqlx::query_as::<_, (i32, String, i32)>(
        "SELECT id, symbol, value FROM test_indexed WHERE symbol = $1",
    )
    .bind("SYM5")
    .fetch_all(pool.pool())
    .await;

    assert!(result.is_ok());
    let rows = result.unwrap();
    assert_eq!(rows.len(), 10, "Should find all matching rows");

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_indexed")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_foreign_key_constraint() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Create parent table
    let _ = sqlx::query("CREATE TABLE IF NOT EXISTS test_parent (id TEXT PRIMARY KEY)")
        .execute(pool.pool())
        .await;

    // Create child table with FK
    let _ = sqlx::query(
        "CREATE TABLE IF NOT EXISTS test_child (
            id TEXT PRIMARY KEY,
            parent_id TEXT REFERENCES test_parent(id) ON DELETE CASCADE
        )",
    )
    .execute(pool.pool())
    .await;

    // Insert parent
    let _ = sqlx::query("INSERT INTO test_parent (id) VALUES ($1)")
        .bind("parent-1")
        .execute(pool.pool())
        .await;

    // Insert child
    let child_result = sqlx::query("INSERT INTO test_child (id, parent_id) VALUES ($1, $2)")
        .bind("child-1")
        .bind("parent-1")
        .execute(pool.pool())
        .await;

    assert!(
        child_result.is_ok(),
        "Child insert with valid FK should succeed"
    );

    // Try invalid FK
    let invalid_result = sqlx::query("INSERT INTO test_child (id, parent_id) VALUES ($1, $2)")
        .bind("child-2")
        .bind("nonexistent")
        .execute(pool.pool())
        .await;

    assert!(
        invalid_result.is_err(),
        "Child insert with invalid FK should fail"
    );

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_child")
        .execute(pool.pool())
        .await;
    let _ = sqlx::query("DROP TABLE test_parent")
        .execute(pool.pool())
        .await;
}

#[tokio::test]
#[serial]
async fn test_postgres_query_timeout_enforcement() {
    let config = test_postgres_config();

    let pool = PostgresPool::new(config).await.unwrap();

    // Execute slow query with timeout
    let slow_result = tokio::time::timeout(
        Duration::from_millis(200),
        sqlx::query("SELECT pg_sleep(0.5)").fetch_one(pool.pool()),
    )
    .await;

    // Should timeout
    assert!(slow_result.is_err(), "Slow query should timeout");
}

#[tokio::test]
#[serial]
async fn test_postgres_connection_pooling_stress() {
    let mut config = test_postgres_config();
    config.max_connections = 10;

    let pool = Arc::new(PostgresPool::new(config).await.unwrap());

    // Spawn 50 concurrent queries (exceeds pool size)
    let mut handles = vec![];
    for i in 0..50 {
        let pool_clone = Arc::clone(&pool);
        let handle = tokio::spawn(async move {
            let result = sqlx::query("SELECT $1 as value")
                .bind(i)
                .fetch_one(pool_clone.pool())
                .await;
            result.is_ok()
        });
        handles.push(handle);
    }

    // Wait for all
    let mut success_count = 0;
    for handle in handles {
        if handle.await.unwrap() {
            success_count += 1;
        }
    }

    assert_eq!(
        success_count, 50,
        "All queries should eventually succeed with pooling"
    );
}

// ============================================================================
// REDIS INTEGRATION TESTS (15+ tests)
// ============================================================================

#[tokio::test]
#[serial]
async fn test_redis_connection_pool_creation() {
    let mut config = test_redis_config();
    config.max_connections = 10;

    let result = RedisPool::new(config).await;
    assert!(
        result.is_ok(),
        "Redis connection pool creation should succeed"
    );

    let pool = result.unwrap();
    let health = pool.health_check().await;
    assert!(health.is_ok(), "Redis health check should pass");
}

#[tokio::test]
#[serial]
async fn test_redis_set_get_operations() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    let key = format!("test:key:{}", Uuid::new_v4());
    let value = "test_value";

    // SET
    let set_result = pool.set(&key, &value, None).await;
    assert!(set_result.is_ok(), "SET operation should succeed");

    // GET
    let get_result: Result<Option<String>, _> = pool.get(&key).await;
    assert!(get_result.is_ok(), "GET operation should succeed");
    assert_eq!(
        get_result.unwrap().unwrap(),
        value,
        "Retrieved value should match"
    );

    // DELETE
    let del_result = pool.delete(&key).await;
    assert!(del_result.is_ok(), "DELETE operation should succeed");

    // Verify deleted
    let verify_result: Result<Option<String>, _> = pool.get(&key).await;
    assert!(verify_result.unwrap().is_none(), "Key should be deleted");
}

#[tokio::test]
#[serial]
async fn test_redis_ttl_expiration() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    let key = format!("test:ttl:{}", Uuid::new_v4());
    let value = "expiring_value";

    // SET with 1 second TTL
    let set_result = pool.set(&key, &value, Some(Duration::from_secs(1))).await;
    assert!(set_result.is_ok(), "SET with TTL should succeed");

    // Immediate GET should work
    let get1: Option<String> = pool.get(&key).await.unwrap();
    assert!(get1.is_some(), "Key should exist immediately");

    // Wait for expiration
    sleep(Duration::from_secs(2)).await;

    // GET after expiration
    let get2: Option<String> = pool.get(&key).await.unwrap();
    assert!(get2.is_none(), "Key should be expired");
}

#[tokio::test]
#[serial]
async fn test_redis_exists_operation() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    let key = format!("test:exists:{}", Uuid::new_v4());

    // Key should not exist initially
    let exists1 = pool.exists(&key).await;
    assert!(exists1.is_ok());
    assert!(!exists1.unwrap(), "Key should not exist initially");

    // Set key
    let _ = pool.set(&key, &"value", None).await;

    // Key should exist now
    let exists2 = pool.exists(&key).await;
    assert!(exists2.is_ok());
    assert!(exists2.unwrap(), "Key should exist after SET");

    // Cleanup
    let _ = pool.delete(&key).await;
}

#[tokio::test]
#[serial]
async fn test_redis_batch_operations() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    // Use batch operations
    let keys: Vec<String> = (0..10)
        .map(|i| format!("test:batch:{}:{}", i, Uuid::new_v4()))
        .collect();

    // SET batch
    for (i, key) in keys.iter().enumerate() {
        let value = format!("value_{}", i);
        let _ = pool.set(key, &value, None).await;
    }

    // GET batch
    for (i, key) in keys.iter().enumerate() {
        let value: Option<String> = pool.get(key).await.unwrap();
        assert_eq!(value, Some(format!("value_{}", i)));
    }

    // DELETE batch
    for key in &keys {
        let _ = pool.delete(key).await;
    }

    // Verify all deleted
    for key in &keys {
        let value: Option<String> = pool.get(key).await.unwrap();
        assert!(value.is_none());
    }
}

#[tokio::test]
#[serial]
async fn test_redis_pipeline_performance() {
    let mut config = test_redis_config();
    config.pipeline_batch_size = 100;

    let pool = RedisPool::new(config).await.unwrap();

    let start = std::time::Instant::now();

    // Pipeline 100 operations
    for i in 0..100 {
        let key = format!("test:pipe:{}:{}", i, Uuid::new_v4());
        let value = format!("value_{}", i);
        let _ = pool.set(&key, &value, Some(Duration::from_secs(60))).await;
    }

    let elapsed = start.elapsed();
    println!("Pipeline 100 operations took: {:?}", elapsed);

    assert!(
        elapsed.as_millis() < 1000,
        "Pipelined operations should be fast"
    );
}

#[tokio::test]
#[serial]
async fn test_redis_concurrent_operations() {
    let mut config = test_redis_config();
    config.max_connections = 20;

    let pool = Arc::new(RedisPool::new(config).await.unwrap());

    // Spawn 50 concurrent SET operations
    let mut handles = vec![];
    for i in 0..50 {
        let pool_clone = Arc::clone(&pool);
        let handle = tokio::spawn(async move {
            let key = format!("test:concurrent:{}:{}", i, Uuid::new_v4());
            let value = format!("value_{}", i);
            let result = pool_clone.set(&key, &value, None).await;
            pool_clone.delete(&key).await.ok();
            result.is_ok()
        });
        handles.push(handle);
    }

    // Wait for all
    let mut success_count = 0;
    for handle in handles {
        if handle.await.unwrap() {
            success_count += 1;
        }
    }

    assert_eq!(
        success_count, 50,
        "All concurrent operations should succeed"
    );
}

#[tokio::test]
#[serial]
async fn test_redis_json_serialization() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    let order = TestOrder::new("AAPL", 100, 150.50, "BUY");
    let key = format!("test:json:{}", Uuid::new_v4());

    // Serialize and store
    let json = serde_json::to_string(&order).unwrap();
    let _ = pool.set(&key, &json, None).await;

    // Retrieve and deserialize
    let retrieved: String = pool.get(&key).await.unwrap().unwrap();
    let deserialized: TestOrder = serde_json::from_str(&retrieved).unwrap();

    assert_eq!(deserialized.id, order.id);
    assert_eq!(deserialized.symbol, order.symbol);
    assert_eq!(deserialized.quantity, order.quantity);

    // Cleanup
    let _ = pool.delete(&key).await;
}

#[tokio::test]
#[serial]
async fn test_redis_cache_invalidation_pattern() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    let cache_key = format!("cache:user:{}:data", Uuid::new_v4());

    // Warm cache
    let _ = pool
        .set(&cache_key, &"cached_data", Some(Duration::from_secs(300)))
        .await;

    // Verify cache hit
    let hit: Option<String> = pool.get(&cache_key).await.unwrap();
    assert!(hit.is_some(), "Cache should be warmed");

    // Invalidate cache
    let _ = pool.delete(&cache_key).await;

    // Verify cache miss
    let miss: Option<String> = pool.get(&cache_key).await.unwrap();
    assert!(miss.is_none(), "Cache should be invalidated");
}

#[tokio::test]
#[serial]
async fn test_redis_metrics_tracking() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    // Perform operations
    let key = format!("test:metrics:{}", Uuid::new_v4());
    let _ = pool.set(&key, &"value", None).await;
    let _: Result<Option<String>, _> = pool.get(&key).await;
    let _ = pool.exists(&key).await;
    let _ = pool.delete(&key).await;

    // Check metrics
    let metrics = pool.get_metrics().await.unwrap();
    assert!(metrics.total_operations >= 4, "Should track all operations");
    assert!(metrics.total_gets >= 1, "Should track GET operations");
    assert!(metrics.total_sets >= 1, "Should track SET operations");
}

#[tokio::test]
#[serial]
async fn test_redis_connection_pool_recycling() {
    let mut config = test_redis_config();
    config.max_connections = 5;

    let pool = RedisPool::new(config).await.unwrap();

    // Perform many operations to test connection recycling
    for i in 0..100 {
        let key = format!("test:recycle:{}", i);
        let value = format!("val{}", i);
        let _ = pool.set(&key, &value, Some(Duration::from_secs(1))).await;
    }

    // All operations should succeed with connection recycling
    let metrics = pool.get_metrics().await.unwrap();
    assert_eq!(
        metrics.total_sets, 100,
        "All SETs should succeed with recycling"
    );
}

#[tokio::test]
#[serial]
async fn test_redis_large_value_handling() {
    let mut config = test_redis_config();
    config.enable_compression = false;

    let pool = RedisPool::new(config).await.unwrap();

    let key = format!("test:large:{}", Uuid::new_v4());
    let large_value = "x".repeat(1_000_000); // 1MB string

    // Store large value
    let set_result = pool.set(&key, &large_value, None).await;
    assert!(set_result.is_ok(), "Should handle large values");

    // Retrieve large value
    let get_result: Option<String> = pool.get(&key).await.unwrap();
    assert_eq!(
        get_result.unwrap().len(),
        1_000_000,
        "Should retrieve full large value"
    );

    // Cleanup
    let _ = pool.delete(&key).await;
}

#[tokio::test]
#[serial]
async fn test_redis_empty_value_handling() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    let key = format!("test:empty:{}", Uuid::new_v4());

    // Store empty string
    let empty = String::new();
    let set_result = pool.set(&key, &empty, None).await;
    assert!(set_result.is_ok(), "Should handle empty values");

    // Retrieve empty string
    let get_result: Option<String> = pool.get(&key).await.unwrap();
    assert_eq!(get_result.unwrap(), "", "Should retrieve empty value");

    // Cleanup
    let _ = pool.delete(&key).await;
}

#[tokio::test]
#[serial]
async fn test_redis_connection_failover() {
    let mut config = test_redis_config();

    let pool = RedisPool::new(config).await.unwrap();

    // Normal operation
    let key = format!("test:failover:{}", Uuid::new_v4());
    let set1 = pool.set(&key, &"before", None).await;
    assert!(set1.is_ok(), "Should work before failover");

    // Cleanup
    let _ = pool.delete(&key).await;
}

// ============================================================================
// CLICKHOUSE INTEGRATION TESTS (8+ tests) - OPTIONAL
// ============================================================================

#[tokio::test]
#[ignore = "Requires ClickHouse"]
async fn test_clickhouse_connection() {
    let config = ClickHouseConfig {
        url: "http://localhost:8123".to_string(),
        database: "default".to_string(),
        username: "default".to_string(),
        password: String::new(),
        ..Default::default()
    };

    let result = ClickHouseClient::new(config).await;
    // ClickHouse may not be running, accept both outcomes
    if result.is_ok() {
        let client = result.unwrap();
        let health = client.health_check().await;
        println!("ClickHouse health check: {:?}", health);
    } else {
        println!("ClickHouse not available (optional)");
    }
}

#[tokio::test]
#[ignore = "Requires ClickHouse"]
async fn test_clickhouse_table_creation() {
    let config = ClickHouseConfig {
        url: "http://localhost:8123".to_string(),
        database: "default".to_string(),
        username: "default".to_string(),
        password: String::new(),
        ..Default::default()
    };

    if let Ok(client) = ClickHouseClient::new(config).await {
        let ddl = "CREATE TABLE IF NOT EXISTS test_trades (
            timestamp DateTime,
            symbol String,
            price Float64,
            quantity UInt64
        ) ENGINE = MergeTree()
        ORDER BY (symbol, timestamp)";

        let result = client.execute_ddl(ddl).await;
        if result.is_ok() {
            // Cleanup
            let _ = client.execute_ddl("DROP TABLE test_trades").await;
        }
    }
}

#[tokio::test]
#[ignore = "Requires ClickHouse"]
async fn test_clickhouse_bulk_insert() {
    let config = ClickHouseConfig {
        url: "http://localhost:8123".to_string(),
        database: "default".to_string(),
        username: "default".to_string(),
        password: String::new(),
        insert_batch_size: 1000,
        ..Default::default()
    };

    if let Ok(client) = ClickHouseClient::new(config).await {
        // Create table
        let _ = client
            .execute_ddl(
                "CREATE TABLE IF NOT EXISTS test_bulk (
                id UInt64,
                value String
            ) ENGINE = MergeTree()
            ORDER BY id",
            )
            .await;

        // Bulk insert
        let mut rows = Vec::new();
        for i in 0..100 {
            rows.push(format!(r#"{{"id":{},"value":"val{}"}}"#, i, i));
        }
        let json_data = rows.join("\n");

        let insert_result = client.insert_json("test_bulk", &json_data).await;
        if insert_result.is_ok() {
            // Cleanup
            let _ = client.execute_ddl("DROP TABLE test_bulk").await;
        }
    }
}

// ============================================================================
// CROSS-PERSISTENCE CONSISTENCY TESTS (2+ tests)
// ============================================================================

#[tokio::test]
#[serial]
async fn test_cross_persistence_write_through_cache() {
    // PostgreSQL as source of truth
    let pg_config = test_postgres_config();
    let pg_pool = PostgresPool::new(pg_config).await.unwrap();

    // Redis as cache
    let redis_config = test_redis_config();
    let redis_pool = RedisPool::new(redis_config).await.unwrap();

    // Create test table
    let _ =
        sqlx::query("CREATE TABLE IF NOT EXISTS test_cache_sync (id TEXT PRIMARY KEY, value TEXT)")
            .execute(pg_pool.pool())
            .await;

    let test_id = Uuid::new_v4().to_string();
    let test_value = "synchronized_data";
    let cache_key = format!("cache:test:{}", test_id);

    // Write to PostgreSQL
    let _ = sqlx::query("INSERT INTO test_cache_sync (id, value) VALUES ($1, $2)")
        .bind(&test_id)
        .bind(test_value)
        .execute(pg_pool.pool())
        .await;

    // Write to Redis cache
    let _ = redis_pool
        .set(&cache_key, &test_value, Some(Duration::from_secs(60)))
        .await;

    // Read from PostgreSQL
    let pg_result: (String,) = sqlx::query_as("SELECT value FROM test_cache_sync WHERE id = $1")
        .bind(&test_id)
        .fetch_one(pg_pool.pool())
        .await
        .unwrap();

    // Read from Redis
    let redis_result: String = redis_pool.get(&cache_key).await.unwrap().unwrap();

    // Verify consistency
    assert_eq!(
        pg_result.0, test_value,
        "PostgreSQL should have correct value"
    );
    assert_eq!(redis_result, test_value, "Redis should have correct value");
    assert_eq!(
        pg_result.0, redis_result,
        "PostgreSQL and Redis should be consistent"
    );

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_cache_sync")
        .execute(pg_pool.pool())
        .await;
    let _ = redis_pool.delete(&cache_key).await;
}

#[tokio::test]
#[serial]
async fn test_cross_persistence_cache_invalidation_on_update() {
    // PostgreSQL as source of truth
    let pg_config = test_postgres_config();
    let pg_pool = PostgresPool::new(pg_config).await.unwrap();

    // Redis as cache
    let redis_config = test_redis_config();
    let redis_pool = RedisPool::new(redis_config).await.unwrap();

    // Create test table
    let _ = sqlx::query(
        "CREATE TABLE IF NOT EXISTS test_invalidation (id TEXT PRIMARY KEY, value TEXT)",
    )
    .execute(pg_pool.pool())
    .await;

    let test_id = Uuid::new_v4().to_string();
    let cache_key = format!("cache:inv:{}", test_id);

    // Initial write
    let _ = sqlx::query("INSERT INTO test_invalidation (id, value) VALUES ($1, $2)")
        .bind(&test_id)
        .bind("initial")
        .execute(pg_pool.pool())
        .await;

    let _ = redis_pool.set(&cache_key, &"initial", None).await;

    // Update PostgreSQL
    let _ = sqlx::query("UPDATE test_invalidation SET value = $1 WHERE id = $2")
        .bind("updated")
        .bind(&test_id)
        .execute(pg_pool.pool())
        .await;

    // Invalidate cache
    let _ = redis_pool.delete(&cache_key).await;

    // Read from PostgreSQL
    let pg_result: (String,) = sqlx::query_as("SELECT value FROM test_invalidation WHERE id = $1")
        .bind(&test_id)
        .fetch_one(pg_pool.pool())
        .await
        .unwrap();

    // Redis should be empty (invalidated)
    let redis_result: Option<String> = redis_pool.get(&cache_key).await.unwrap();

    assert_eq!(
        pg_result.0, "updated",
        "PostgreSQL should have updated value"
    );
    assert!(redis_result.is_none(), "Redis cache should be invalidated");

    // Cleanup
    let _ = sqlx::query("DROP TABLE test_invalidation")
        .execute(pg_pool.pool())
        .await;
}
#[tokio::test]
async fn test_postgres_writer_inserts_with_all_required_fields() {
    use rust_decimal::Decimal;
    use sqlx::PgPool;
    use std::sync::Arc;
    use trading_engine::events::event_types::TradingEvent;
    use trading_engine::events::postgres_writer::{PostgresWriter, WriterConfig};
    use trading_engine::events::EventMetrics;
    use trading_engine::timing::HardwareTimestamp;

    // Connect to test database
    let database_url = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
        "postgresql://foxhunt:foxhunt_dev_password@localhost:5432/foxhunt".to_string()
    });

    let pool = PgPool::connect(&database_url)
        .await
        .expect("Failed to connect to database");

    // Create metrics
    let metrics = Arc::new(EventMetrics::new());

    // Create writer
    let config = WriterConfig {
        batch_size: 1, // Process immediately
        batch_timeout: std::time::Duration::from_millis(100),
        max_retry_attempts: 3,
        retry_delay: std::time::Duration::from_millis(100),
        enable_compression: false,
        thread_id: 0,
    };

    let writer = PostgresWriter::new(config, pool.clone(), metrics.clone())
        .await
        .expect("Failed to create writer");

    // Create test events with unique symbol
    let test_symbol = format!(
        "TEST{}",
        uuid::Uuid::new_v4()
            .to_string()
            .replace('-', "")
            .chars()
            .take(8)
            .collect::<String>()
    );
    let events = vec![TradingEvent::OrderSubmitted {
        order_id: format!("TEST-{}", uuid::Uuid::new_v4()),
        symbol: test_symbol.clone(),
        quantity: Decimal::new(100, 2),  // 1.00 BTC
        price: Decimal::new(5000000, 2), // $50,000.00
        timestamp: HardwareTimestamp::now(),
        sequence_number: Some(1),
        metadata: None,
    }];

    println!("Submitting batch with symbol: {}", test_symbol);

    // Submit batch
    writer
        .submit_batch(events)
        .await
        .expect("Failed to submit batch");

    // Wait for processing
    tokio::time::sleep(std::time::Duration::from_secs(3)).await;

    // Check metrics
    let stats = writer.get_stats().await;
    println!(
        "Writer stats: batches_processed={}, events_written={}, batches_failed={}",
        stats.batches_processed, stats.events_written, stats.batches_failed
    );

    // Verify insertion
    let count: (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM trading_events WHERE symbol = $1 AND event_type = 'order_submitted'",
    )
    .bind(&test_symbol)
    .fetch_one(&pool)
    .await
    .expect("Failed to query count");

    println!("Found {} events with symbol {}", count.0, test_symbol);

    assert!(
        count.0 > 0,
        "Should have inserted at least one event. Stats: {:?}",
        stats
    );

    // Verify all required fields are present
    let result: (String, i32, String) = sqlx::query_as(
        "SELECT node_id, process_id, event_hash FROM trading_events WHERE symbol = $1 ORDER BY event_timestamp DESC LIMIT 1"
    )
    .bind(&test_symbol)
    .fetch_one(&pool)
    .await
    .expect("Failed to fetch event details");

    let (node_id, process_id, event_hash) = result;
    assert!(!node_id.is_empty(), "node_id should not be empty");
    assert!(process_id > 0, "process_id should be positive");
    assert!(!event_hash.is_empty(), "event_hash should not be empty");
    assert_eq!(
        event_hash.len(),
        32,
        "event_hash should be 32 characters (MD5 hex)"
    );

    println!(
        "Event details: node_id={}, process_id={}, event_hash={}",
        node_id, process_id, event_hash
    );

    // Cleanup
    sqlx::query("DELETE FROM trading_events WHERE symbol = $1 AND event_type = 'order_submitted'")
        .bind(&test_symbol)
        .execute(&pool)
        .await
        .expect("Failed to cleanup test data");

    writer.shutdown().await.expect("Failed to shutdown writer");
}
#[tokio::test]
async fn test_direct_insert_to_trading_events() {
    use sqlx::PgPool;

    let database_url = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
        "postgresql://foxhunt:foxhunt_dev_password@localhost:5432/foxhunt".to_string()
    });

    let pool = PgPool::connect(&database_url)
        .await
        .expect("Failed to connect to database");

    // Get current timestamp in nanoseconds
    let now_ns = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_nanos() as i64;

    // Build the query with correct parameter count and cast event_type to enum
    let query = "INSERT INTO trading_events (
        correlation_id, event_timestamp, received_timestamp, processing_timestamp,
        event_type, event_source, symbol, event_data, metadata,
        node_id, process_id, event_hash, event_date
    ) VALUES (
        gen_random_uuid(), $1, $2, $3, $4::trading_event_type, $5, $6, $7, $8, $9, $10, $11, DATE(TO_TIMESTAMP($1 / 1000000000.0))
    ) RETURNING id";

    let result = sqlx::query_scalar::<_, uuid::Uuid>(query)
        .bind(now_ns)
        .bind(now_ns)
        .bind(now_ns)
        .bind("order_submitted")
        .bind("trading_engine")
        .bind("TESTDIRECT")
        .bind(serde_json::json!({"test": "data"}))
        .bind(serde_json::Value::Null)
        .bind("test-node")
        .bind(12345)
        .bind("abcdef1234567890abcdef1234567890")
        .fetch_one(&pool)
        .await;

    match result {
        Ok(id) => println!("INSERT succeeded with id: {}", id),
        Err(e) => panic!("INSERT failed: {}", e),
    }

    // Cleanup
    let _ = sqlx::query("DELETE FROM trading_events WHERE symbol = 'TESTDIRECT'")
        .execute(&pool)
        .await;
}