//! Integration tests for concurrent database operations.
//!
//! These tests verify that the storage works correctly with concurrent access.

use dig_l2_storage::{BlockHeader, BlockStorage, DataLocation, Emission, StoredBlock};
use std::sync::Arc;
use std::thread;
use tempfile::TempDir;
/// Helper to create a test block.
fn create_test_block(epoch: u64, height: u64, data: &[u8]) -> StoredBlock {
    StoredBlock {
        header: BlockHeader {
            epoch,
            height,
            prev_block_root: [0u8; 32],
            timestamp: 1700000000 + height,
            proposer_pubkey: data.to_vec(),
            extra_data: [0u8; 32],
        },
        data_hashes: vec![[1u8; 32], [2u8; 32]],
        emissions: vec![Emission {
            pubkey: data.to_vec(),
            weight: 100,
        }],
        data_root: [3u8; 32],
        emissions_root: [4u8; 32],
    }
}

fn make_root(epoch: u64, height: u64) -> [u8; 32] {
    let mut root = [0u8; 32];
    root[0..8].copy_from_slice(&epoch.to_le_bytes());
    root[8..16].copy_from_slice(&height.to_le_bytes());
    root
}

/// Test concurrent reads from multiple threads.
#[test]
fn test_concurrent_reads() {
    let tmp = TempDir::new().unwrap();
    let storage = BlockStorage::open(tmp.path()).unwrap();

    // Populate with data
    for epoch in 0..10u64 {
        for height in 0..10u64 {
            let root = make_root(epoch, height);
            let block = create_test_block(epoch, height, &[epoch as u8, height as u8]);
            storage.put_block(epoch, &root, &block).unwrap();
        }
    }
    storage.flush().unwrap();

    // Clone storage for concurrent access (BlockStorage is Clone)
    let storage = Arc::new(storage);

    // Spawn multiple reader threads
    let mut handles = vec![];
    for thread_id in 0..8 {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            for _ in 0..100 {
                // Read random blocks
                let epoch = (thread_id * 13 + 7) % 10;
                let height = (thread_id * 17 + 3) % 10;
                let root = make_root(epoch, height);

                let result = storage_clone.get_block(epoch, &root);
                assert!(result.is_ok(), "Read should succeed");
                assert!(result.unwrap().is_some(), "Block should exist");
            }
        });
        handles.push(handle);
    }

    // Wait for all threads
    for handle in handles {
        handle.join().expect("Thread should complete without panic");
    }
}

/// Test concurrent writes to different keys.
#[test]
fn test_concurrent_writes_different_keys() {
    let tmp = TempDir::new().unwrap();
    let storage = Arc::new(BlockStorage::open(tmp.path()).unwrap());

    let threads = 8usize;
    let writes_per_thread = 50usize;

    // Each thread writes to different epochs
    let mut handles = vec![];
    for thread_id in 0..threads {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            let epoch = thread_id as u64;
            for height in 0..writes_per_thread {
                let root = make_root(epoch, height as u64);
                let block = create_test_block(epoch, height as u64, &[thread_id as u8, height as u8]);
                storage_clone.put_block(epoch, &root, &block).unwrap();
            }
        });
        handles.push(handle);
    }

    // Wait for all threads
    for handle in handles {
        handle.join().expect("Thread should complete without panic");
    }

    // Verify all writes
    for thread_id in 0..threads {
        let epoch = thread_id as u64;
        let roots = storage.get_epoch_blocks(epoch).unwrap();
        assert_eq!(
            roots.len(),
            writes_per_thread,
            "Thread {} should have written {} blocks",
            thread_id,
            writes_per_thread
        );
    }

    let total = storage.block_count().unwrap();
    assert_eq!(total, threads * writes_per_thread);
}

/// Test mixed concurrent reads and writes.
#[test]
fn test_concurrent_read_write() {
    let tmp = TempDir::new().unwrap();
    let storage = Arc::new(BlockStorage::open(tmp.path()).unwrap());

    // Pre-populate some data
    for epoch in 0..5u64 {
        for height in 0..10u64 {
            let root = make_root(epoch, height);
            let block = create_test_block(epoch, height, &[epoch as u8]);
            storage.put_block(epoch, &root, &block).unwrap();
        }
    }
    storage.flush().unwrap();

    let mut handles = vec![];

    // Writer threads (write to new epochs)
    for thread_id in 0..4 {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            let epoch = (10 + thread_id) as u64;
            for height in 0..20u64 {
                let root = make_root(epoch, height);
                let block = create_test_block(epoch, height, &[thread_id as u8]);
                storage_clone.put_block(epoch, &root, &block).unwrap();
            }
        });
        handles.push(handle);
    }

    // Reader threads (read from pre-existing data)
    for thread_id in 0..4 {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            for _ in 0..50 {
                let epoch = (thread_id % 5) as u64;
                let height = (thread_id * 2) as u64 % 10;
                let root = make_root(epoch, height);

                let result = storage_clone.get_block(epoch, &root);
                assert!(result.is_ok(), "Concurrent read should succeed");
            }
        });
        handles.push(handle);
    }

    // Wait for all threads
    for handle in handles {
        handle.join().expect("Thread should complete");
    }
}

/// Test concurrent state operations.
#[test]
fn test_concurrent_state_operations() {
    let tmp = TempDir::new().unwrap();
    let storage = Arc::new(BlockStorage::open(tmp.path()).unwrap());

    let mut handles = vec![];

    // Multiple threads writing different state keys
    for thread_id in 0..8 {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            for i in 0..100 {
                let key = format!("thread_{}_key_{}", thread_id, i);
                let value = format!("value_{}_{}", thread_id, i);
                storage_clone.put_state(&key, value.as_bytes()).unwrap();
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("Thread should complete");
    }

    // Verify all state keys
    for thread_id in 0..8 {
        for i in 0..100 {
            let key = format!("thread_{}_key_{}", thread_id, i);
            let expected = format!("value_{}_{}", thread_id, i);
            let actual = storage.get_state(&key).unwrap().expect("State should exist");
            assert_eq!(String::from_utf8(actual).unwrap(), expected);
        }
    }
}

/// Test concurrent data index operations.
#[test]
fn test_concurrent_data_indexing() {
    let tmp = TempDir::new().unwrap();
    let storage = Arc::new(BlockStorage::open(tmp.path()).unwrap());

    let mut handles = vec![];

    // Multiple threads indexing data
    for thread_id in 0..8u64 {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            for i in 0..50u64 {
                let mut hash = [0u8; 32];
                hash[0..8].copy_from_slice(&thread_id.to_le_bytes());
                hash[8..16].copy_from_slice(&i.to_le_bytes());

                let location = DataLocation {
                    epoch: thread_id,
                    block_root: make_root(thread_id, i),
                    index: i as usize,
                };

                storage_clone.index_data(&hash, &location).unwrap();
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("Thread should complete");
    }

    // Verify all indexed data
    for thread_id in 0..8u64 {
        for i in 0..50u64 {
            let mut hash = [0u8; 32];
            hash[0..8].copy_from_slice(&thread_id.to_le_bytes());
            hash[8..16].copy_from_slice(&i.to_le_bytes());

            let location = storage.find_data(&hash).unwrap().expect("Data should be indexed");
            assert_eq!(location.epoch, thread_id);
            assert_eq!(location.index, i as usize);
        }
    }
}

/// Test concurrent canonical chain updates.
#[test]
fn test_concurrent_canonical_updates() {
    let tmp = TempDir::new().unwrap();
    let storage = Arc::new(BlockStorage::open(tmp.path()).unwrap());

    let mut handles = vec![];

    // Each thread manages canonical for different epochs
    for thread_id in 0..8 {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            let epoch = thread_id as u64;
            for height in 0..100u64 {
                let root = make_root(epoch, height);
                storage_clone.set_canonical(epoch, &root).unwrap();
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("Thread should complete");
    }

    // Verify final canonical state (should be last write for each epoch)
    for thread_id in 0..8 {
        let epoch = thread_id as u64;
        let canonical = storage.get_canonical(epoch).unwrap().expect("Canonical should exist");
        // Last write should be height 99
        let expected = make_root(epoch, 99);
        assert_eq!(canonical, expected);
    }
}

/// Test storage clone behavior.
#[test]
fn test_storage_clone_shares_db() {
    let tmp = TempDir::new().unwrap();
    let storage1 = BlockStorage::open(tmp.path()).unwrap();
    let storage2 = storage1.clone();

    // Write through first handle
    let root = make_root(1, 1);
    let block = create_test_block(1, 1, &[1, 2, 3]);
    storage1.put_block(1, &root, &block).unwrap();

    // Should be visible through second handle
    assert!(storage2.has_block(1, &root).unwrap());

    // Write through second handle
    let root2 = make_root(2, 2);
    let block2 = create_test_block(2, 2, &[4, 5, 6]);
    storage2.put_block(2, &root2, &block2).unwrap();

    // Should be visible through first handle
    assert!(storage1.has_block(2, &root2).unwrap());
}

/// Test heavy concurrent load.
#[test]
fn test_heavy_concurrent_load() {
    let tmp = TempDir::new().unwrap();
    let storage = Arc::new(BlockStorage::open(tmp.path()).unwrap());

    let num_threads = 16usize;
    let operations_per_thread = 100usize;

    let mut handles = vec![];

    for thread_id in 0..num_threads {
        let storage_clone = Arc::clone(&storage);
        let handle = thread::spawn(move || {
            for op in 0..operations_per_thread {
                let epoch = (thread_id as u64 * 1000) + op as u64;
                let root = make_root(epoch, op as u64);

                // Write block
                let block = create_test_block(epoch, op as u64, &[thread_id as u8, op as u8]);
                storage_clone.put_block(epoch, &root, &block).unwrap();

                // Set canonical
                storage_clone.set_canonical(epoch, &root).unwrap();

                // Write state
                let key = format!("key_{}_{}", thread_id, op);
                storage_clone.put_state(&key, b"value").unwrap();

                // Index data
                let hash = root;
                let location = DataLocation {
                    epoch,
                    block_root: root,
                    index: op,
                };
                storage_clone.index_data(&hash, &location).unwrap();
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("Thread should complete");
    }

    // Verify final count
    let expected_blocks = num_threads * operations_per_thread;
    let actual_blocks = storage.block_count().unwrap();
    assert_eq!(actual_blocks, expected_blocks, "Should have all blocks");
}
