//! Integration tests for Index.db size=0 handling with real Cassandra 5.0 data
//!
//! Tests validate Issue #149 sequential scan fallback when Index.db reports size=0.
//! NOTE: Tests currently validate fallback triggers correctly. After issues #150
//! (chunked decompression) and #151 (schema loading) are complete, update tests
//! to verify correct data is returned.

use cqlite_core::{
    platform::Platform,
    storage::sstable::{index_reader::IndexReader, SSTableReader},
    types::{RowKey, TableId},
    Config,
};
use std::sync::Arc;
use tokio::fs;

mod common;
use common::sstable_test_utils::TestContext;

/// Helper function to find a file with a specific pattern in a directory
/// Returns None if file not found (for refs-only datasets in CI)
async fn find_file_with_pattern(
    table_path: &std::path::Path,
    pattern: &str,
) -> Option<std::path::PathBuf> {
    let read_dir = match fs::read_dir(table_path).await {
        Ok(dir) => dir,
        Err(_) => return None,
    };
    let mut read_dir = read_dir;

    while let Some(entry) = read_dir.next_entry().await.ok()? {
        let path = entry.path();
        if let Some(name) = path.file_name().and_then(|n| n.to_str()) {
            // Skip .jsonl reference files when looking for actual SSTable components
            if name.contains(pattern) && (pattern.contains(".jsonl") || !name.contains(".jsonl")) {
                return Some(path);
            }
        }
    }

    None
}

/// Test 1: Verify that get() with size=0 fallback completes without panic
///
/// Current expectation: Returns None (until #150/#151 complete)
/// Future behavior: Should return correct data after chunked decompression is implemented
#[tokio::test]
async fn test_get_with_size_zero_fallback() {
    eprintln!("=== Test 1: get() with size=0 fallback ===");
    eprintln!(
        "CQLITE_DATASETS_ROOT = {:?}",
        std::env::var("CQLITE_DATASETS_ROOT")
    );

    let mut context = TestContext::new("test_basic").await.unwrap();
    eprintln!("TestContext dataset_path = {:?}", context.dataset_path);

    let table_path = context.prepare_sstable("simple_table").await.unwrap();
    eprintln!("Prepared SSTable at: {}", table_path.display());

    let config = Config::default();
    let platform = Arc::new(Platform::new(&config).await.unwrap());

    // Find the actual Data.db file - must be present for this test
    let data_file = find_file_with_pattern(&table_path, "-Data.db")
        .await
        .expect("Data.db file must exist for this test - requires full SSTable binary files, not just reference data");

    // Open SSTableReader
    let reader = match SSTableReader::open(&data_file, &config, platform.clone()).await {
        Ok(reader) => reader,
        Err(e) => {
            println!(
                "⚠️  SSTable loading failed: {}. This might indicate file format incompatibility.",
                e
            );
            return;
        }
    };

    // Verify Index.db has entries with size=0
    let index_file = find_file_with_pattern(&table_path, "-Index.db")
        .await
        .expect("Index.db file must exist for this test");

    let index_reader = match IndexReader::open(&index_file, platform).await {
        Ok(reader) => reader,
        Err(e) => {
            println!(
                "⚠️  Index loading failed: {}. This might indicate file format incompatibility.",
                e
            );
            return;
        }
    };

    let partition_entries = index_reader.get_partition_entries();
    eprintln!(
        "Found {} partition entries in Index.db",
        partition_entries.len()
    );

    // Check for size=0 entries
    let zero_size_count = partition_entries
        .iter()
        .filter(|e| e.data_size == 0)
        .count();
    eprintln!(
        "Partition entries with size=0: {} out of {}",
        zero_size_count,
        partition_entries.len()
    );

    assert!(
        zero_size_count > 0,
        "Index.db must contain size=0 entries for this test (requires Cassandra 5.0 format). Found 0 out of {} entries",
        partition_entries.len()
    );

    // Test get() operation with a synthetic key
    // This should trigger the size=0 fallback path in data_access.rs lines 27-32
    let test_key = RowKey::from(b"test_key_1".to_vec());
    let table_id = TableId::from("test");

    eprintln!(
        "Calling reader.get() with table_id='{}', key={:?}",
        table_id, test_key
    );
    let result = reader.get(&table_id, &test_key).await;

    // Validate operation completes without panic
    assert!(
        result.is_ok(),
        "get() should complete without error, got: {:?}",
        result.err()
    );

    // Current expectation: Returns None (blocked by #150/#151)
    // TODO(Issue #150, #151): After chunked decompression and schema loading are complete,
    // verify that correct data is returned instead of None
    let value = result.unwrap();
    if value.is_none() {
        eprintln!("✓ get() returned None as expected (sequential scan not yet fully implemented)");
        eprintln!("  NOTE: After #150/#151, this test should verify correct data is returned");
    } else {
        eprintln!("✓ get() returned data: {:?}", value);
    }

    context.record_bytes_read(0);
    let _metrics = context.cleanup().unwrap();
    println!("✅ Test 1 passed: get() with size=0 fallback completes without panic");
}

/// Test 2: Verify scan() detects mixed size entries and triggers sequential fallback
///
/// Current expectation: Returns empty Vec (until #150/#151 complete)
/// Future behavior: Should return correct data after implementation completes
#[tokio::test]
async fn test_scan_with_mixed_sizes() {
    eprintln!("=== Test 2: scan() with mixed sizes fallback ===");

    let mut context = TestContext::new("test_basic").await.unwrap();
    let table_path = context.prepare_sstable("simple_table").await.unwrap();

    let config = Config::default();
    let platform = Arc::new(Platform::new(&config).await.unwrap());

    // Find the actual Data.db file
    let data_file = find_file_with_pattern(&table_path, "-Data.db")
        .await
        .expect("Data.db file must exist for this test");

    let reader = match SSTableReader::open(&data_file, &config, platform.clone()).await {
        Ok(reader) => reader,
        Err(e) => {
            println!(
                "⚠️  SSTable loading failed: {}. This might indicate file format incompatibility.",
                e
            );
            return;
        }
    };

    // Verify Index.db has size=0 entries
    let index_file = find_file_with_pattern(&table_path, "-Index.db")
        .await
        .expect("Index.db file must exist for this test");

    let index_reader = IndexReader::open(&index_file, platform)
        .await
        .expect("Index.db file must be loadable for this test");

    let partition_entries = index_reader.get_partition_entries();
    let zero_size_count = partition_entries
        .iter()
        .filter(|e| e.data_size == 0)
        .count();

    assert!(
        zero_size_count > 0,
        "Index.db must contain size=0 entries for this test (requires Cassandra 5.0 format). Found 0 out of {} entries",
        partition_entries.len()
    );

    eprintln!(
        "Index has {} entries with size=0 - scan should trigger sequential fallback",
        zero_size_count
    );

    // Call scan() which should detect size=0 entries and use sequential fallback
    // This triggers the fallback path in data_access.rs lines 86-92
    let table_id = TableId::from("test");
    let result = reader.scan(&table_id, None, None, Some(10), None).await;

    assert!(
        result.is_ok(),
        "scan() should complete without error, got: {:?}",
        result.err()
    );

    let entries = result.unwrap();

    // Current expectation: Returns empty Vec (blocked by #150/#151)
    // The sequential scan path is triggered but data parsing is not yet complete
    if entries.is_empty() {
        eprintln!(
            "✓ scan() returned empty results as expected (sequential scan blocked by #150/#151)"
        );
        eprintln!("  NOTE: After chunked decompression and schema loading are complete,");
        eprintln!("  this test should verify that correct data is returned");
    } else {
        eprintln!(
            "✓ scan() returned {} entries (sequential scan may be working)",
            entries.len()
        );
    }

    context.record_bytes_read(0);
    let _metrics = context.cleanup().unwrap();
    println!("✅ Test 2 passed: scan() triggers sequential fallback correctly");
}

/// Test 3: Verify sequential scan performance is reasonable (smoke test)
///
/// Validates that sequential scan completes within reasonable time bounds
/// to catch any O(n²) or exponential behavior bugs
#[tokio::test]
async fn test_sequential_scan_performance() {
    eprintln!("=== Test 3: Sequential scan performance smoke test ===");

    let mut context = TestContext::new("test_basic").await.unwrap();
    let table_path = context.prepare_sstable("simple_table").await.unwrap();

    let config = Config::default();
    let platform = Arc::new(Platform::new(&config).await.unwrap());

    let data_file = find_file_with_pattern(&table_path, "-Data.db")
        .await
        .expect("Data.db file must exist for this test");

    let reader = SSTableReader::open(&data_file, &config, platform)
        .await
        .expect("SSTable must be loadable for this test");

    // Time the scan operation
    let table_id = TableId::from("test");
    let start = std::time::Instant::now();
    let result = reader.scan(&table_id, None, None, Some(100), None).await;
    let elapsed = start.elapsed();

    assert!(
        result.is_ok(),
        "scan() should complete without error, got: {:?}",
        result.err()
    );

    eprintln!("Scan completed in {:?}", elapsed);

    // Smoke test: Should complete within 5 seconds (not a benchmark, just sanity check)
    assert!(
        elapsed.as_secs() < 5,
        "Scan took too long: {:?} (should be < 5s). Possible performance regression.",
        elapsed
    );

    eprintln!(
        "✓ Sequential scan performance is acceptable ({:?})",
        elapsed
    );

    context.record_bytes_read(0);
    let _metrics = context.cleanup().unwrap();
    println!("✅ Test 3 passed: Sequential scan completes within acceptable time");
}

/// Test 4: Verify proper error handling when encountering corrupt data with size=0
///
/// Ensures that the system returns proper errors instead of panicking when
/// corrupt or malformed blocks are encountered during sequential scan fallback
#[tokio::test]
async fn test_size_zero_with_corrupt_data() {
    eprintln!("=== Test 4: Error handling for corrupt data with size=0 ===");

    let mut context = TestContext::new("test_basic").await.unwrap();
    let table_path = context.prepare_sstable("simple_table").await.unwrap();

    let config = Config::default();
    let platform = Arc::new(Platform::new(&config).await.unwrap());

    let data_file = find_file_with_pattern(&table_path, "-Data.db")
        .await
        .expect("Data.db file must exist for this test");

    let reader = SSTableReader::open(&data_file, &config, platform)
        .await
        .expect("SSTable must be loadable for this test");

    // Attempt lookup that might hit corrupt/malformed blocks
    // Using a variety of synthetic keys to exercise different code paths
    let test_keys = vec![
        RowKey::from(b"".to_vec()),                    // Empty key
        RowKey::from(vec![0xFF; 1024]),                // Large invalid key
        RowKey::from(b"nonexistent_key_999".to_vec()), // Likely non-existent
    ];

    let table_id = TableId::from("test");
    let completed_without_panic = true;

    for test_key in test_keys {
        eprintln!("Testing with key of length {}", test_key.as_bytes().len());
        let result = reader.get(&table_id, &test_key).await;

        // Validate that operation either succeeds or returns proper error
        // The key requirement is: NO PANICS
        match result {
            Ok(Some(_)) => {
                eprintln!("  ✓ Returned data successfully");
            }
            Ok(None) => {
                eprintln!("  ✓ Returned None (key not found or filtered)");
            }
            Err(e) => {
                eprintln!("  ✓ Returned proper error: {:?}", e);
                // Verify error type is reasonable (not a panic)
                assert!(
                    !format!("{:?}", e).contains("panic"),
                    "Error should not contain panic: {:?}",
                    e
                );
            }
        }
    }

    if completed_without_panic {
        eprintln!("✓ All lookups completed without panic");
    }

    context.record_bytes_read(0);
    let _metrics = context.cleanup().unwrap();
    println!("✅ Test 4 passed: Proper error handling, no panics");
}
