oxirs-tdb 0.3.1

//! LSM-tree (Log-Structured Merge-tree) storage engine
//!
//! Provides write-optimized storage using LSM-tree architecture.
//! Optimized for high write throughput with efficient range scans.
//!
//! ## Architecture
//!
//! - **MemTable** - In-memory write buffer (red-black tree)
//! - **Immutable MemTable** - Frozen write buffer being flushed
//! - **SSTable** - Sorted String Table on disk
//! - **Compaction** - Background merging of SSTables
//! - **Bloom Filters** - Fast existence checks per SSTable
//! - **Write-Ahead Log** - Durability guarantee
//!
//! ## Performance Characteristics
//!
//! - **Writes**: O(log n) in-memory, amortized O(1) to disk
//! - **Reads**: O(log n) for MemTable + O(k) SSTable checks
//! - **Range Scans**: O(k + log n) where k is result size
//! - **Space**: Amplification factor ~2-3x due to multiple levels
//!
//! ## Example
//!
//! ```rust,ignore
//! use oxirs_tdb::storage::lsm_tree::{LsmTree, LsmConfig};
//!
//! let config = LsmConfig::default();
//! let mut lsm = LsmTree::new(config)?;
//!
//! // Write key-value pairs
//! lsm.put(b"key1", b"value1")?;
//! lsm.put(b"key2", b"value2")?;
//!
//! // Read value
//! let value = lsm.get(b"key1")?;
//! assert_eq!(value, Some(b"value1".to_vec()));
//!
//! // Range scan
//! let range = lsm.scan(b"key1"..b"key3")?;
//! ```

use crate::compression::BloomFilter;
use crate::error::{Result, TdbError};
use oxicode::Decode;
use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write};
use std::ops::Bound;
use std::path::{Path, PathBuf};
use std::sync::Arc;

/// LSM-tree configuration
#[derive(Debug, Clone)]
pub struct LsmConfig {
    /// Directory for SSTable files
    pub data_dir: PathBuf,
    /// MemTable size threshold for flushing (bytes)
    pub memtable_size_threshold: usize,
    /// Number of SSTable levels (typically 4-7)
    pub num_levels: usize,
    /// Size multiplier between levels (typically 10)
    pub level_size_multiplier: usize,
    /// Enable bloom filters per SSTable
    pub enable_bloom_filters: bool,
    /// Bloom filter false positive rate
    pub bloom_filter_fpr: f64,
    /// Compaction strategy
    pub compaction_strategy: CompactionStrategy,
}

impl Default for LsmConfig {
    fn default() -> Self {
        Self {
            data_dir: PathBuf::from("lsm_data"),
            memtable_size_threshold: 4 * 1024 * 1024, // 4 MB
            num_levels: 5,
            level_size_multiplier: 10,
            enable_bloom_filters: true,
            bloom_filter_fpr: 0.01,
            compaction_strategy: CompactionStrategy::Leveled,
        }
    }
}

/// Compaction strategy
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompactionStrategy {
    /// Size-tiered compaction (better for write-heavy)
    SizeTiered,
    /// Leveled compaction (better for read-heavy)
    Leveled,
    /// Universal compaction (balanced)
    Universal,
}

/// Entry in the LSM-tree (key-value pair with metadata)
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Entry {
    /// Key
    key: Vec<u8>,
    /// Value (None for deletion tombstone)
    value: Option<Vec<u8>>,
    /// Sequence number for MVCC
    sequence: u64,
}

/// Type alias for scan result entry
type ScanEntry = (Vec<u8>, Option<Vec<u8>>, u64);

/// MemTable (in-memory write buffer)
struct MemTable {
    /// Entries stored in sorted order (BTreeMap)
    entries: BTreeMap<Vec<u8>, (Option<Vec<u8>>, u64)>,
    /// Approximate size in bytes
    size_bytes: usize,
}

impl MemTable {
    fn new() -> Self {
        Self {
            entries: BTreeMap::new(),
            size_bytes: 0,
        }
    }

    fn put(&mut self, key: Vec<u8>, value: Vec<u8>, sequence: u64) {
        let entry_size = key.len() + value.len() + 16; // 16 for metadata overhead
        self.entries.insert(key, (Some(value), sequence));
        self.size_bytes += entry_size;
    }

    fn delete(&mut self, key: Vec<u8>, sequence: u64) {
        let entry_size = key.len() + 16;
        self.entries.insert(key, (None, sequence));
        self.size_bytes += entry_size;
    }

    fn get(&self, key: &[u8]) -> Option<(Option<Vec<u8>>, u64)> {
        self.entries.get(key).cloned()
    }

    fn size(&self) -> usize {
        self.size_bytes
    }

    fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    fn iter(&self) -> impl Iterator<Item = (&Vec<u8>, &(Option<Vec<u8>>, u64))> {
        self.entries.iter()
    }
}

/// SSTable (Sorted String Table on disk)
#[derive(Debug, Clone)]
struct SsTable {
    /// File path
    path: PathBuf,
    /// Level number
    level: usize,
    /// Smallest key in this SSTable
    min_key: Vec<u8>,
    /// Largest key in this SSTable
    max_key: Vec<u8>,
    /// Number of entries
    num_entries: usize,
    /// File size in bytes
    file_size: usize,
    /// Bloom filter for existence checks
    bloom_filter: Option<BloomFilter>,
}

impl SsTable {
    /// Create a new SSTable by flushing MemTable to disk
    fn flush_from_memtable(
        memtable: &MemTable,
        path: PathBuf,
        level: usize,
        enable_bloom: bool,
        bloom_fpr: f64,
    ) -> Result<Self> {
        let file = OpenOptions::new()
            .create(true)
            .write(true)
            .truncate(true)
            .open(&path)
            .map_err(TdbError::Io)?;

        let mut writer = BufWriter::new(file);
        let mut min_key = None;
        let mut max_key = None;
        let mut num_entries = 0;

        // Initialize bloom filter
        let mut bloom_filter = if enable_bloom {
            Some(BloomFilter::new(memtable.entries.len(), bloom_fpr))
        } else {
            None
        };

        // Write entries in sorted order
        for (key, (value, sequence)) in memtable.iter() {
            // Track min/max keys
            if min_key.is_none() {
                min_key = Some(key.clone());
            }
            max_key = Some(key.clone());

            // Add to bloom filter
            if let Some(ref mut bloom) = bloom_filter {
                bloom.insert_bytes(key);
            }

            // Serialize entry
            let entry = Entry {
                key: key.clone(),
                value: value.clone(),
                sequence: *sequence,
            };
            let encoded = oxicode::serde::encode_to_vec(&entry, oxicode::config::standard())
                .map_err(|e| TdbError::Other(format!("Serialization error: {}", e)))?;

            // Write length prefix + entry
            let len = encoded.len() as u32;
            writer.write_all(&len.to_le_bytes()).map_err(TdbError::Io)?;
            writer.write_all(&encoded).map_err(TdbError::Io)?;

            num_entries += 1;
        }

        writer.flush().map_err(TdbError::Io)?;

        let file_size = std::fs::metadata(&path).map_err(TdbError::Io)?.len() as usize;

        Ok(Self {
            path,
            level,
            min_key: min_key.unwrap_or_default(),
            max_key: max_key.unwrap_or_default(),
            num_entries,
            file_size,
            bloom_filter,
        })
    }

    /// Check if key might exist (using bloom filter)
    fn might_contain(&self, key: &[u8]) -> bool {
        if let Some(ref bloom) = self.bloom_filter {
            bloom.contains_bytes(key)
        } else {
            true // No bloom filter, assume it might exist
        }
    }

    /// Read an entry from SSTable
    fn get(&self, key: &[u8]) -> Result<Option<(Option<Vec<u8>>, u64)>> {
        // Quick bloom filter check
        if !self.might_contain(key) {
            return Ok(None);
        }

        // Check key range
        if key < self.min_key.as_slice() || key > self.max_key.as_slice() {
            return Ok(None);
        }

        // Sequential scan through SSTable (can be optimized with index blocks)
        let file = File::open(&self.path).map_err(TdbError::Io)?;
        let mut reader = BufReader::new(file);

        loop {
            // Read length prefix
            let mut len_buf = [0u8; 4];
            match reader.read_exact(&mut len_buf) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
                Err(e) => return Err(TdbError::Io(e)),
            }

            let len = u32::from_le_bytes(len_buf) as usize;

            // Read entry
            let mut entry_buf = vec![0u8; len];
            reader.read_exact(&mut entry_buf).map_err(TdbError::Io)?;

            let entry: Entry =
                oxicode::serde::decode_from_slice(&entry_buf, oxicode::config::standard())
                    .map_err(|e| TdbError::Other(format!("Deserialization error: {}", e)))?
                    .0;

            if entry.key == key {
                return Ok(Some((entry.value, entry.sequence)));
            }
        }

        Ok(None)
    }

    /// Scan a range of keys
    fn scan(&self, start: &[u8], end: &[u8]) -> Result<Vec<ScanEntry>> {
        let file = File::open(&self.path).map_err(TdbError::Io)?;
        let mut reader = BufReader::new(file);
        let mut results = Vec::new();

        loop {
            // Read length prefix
            let mut len_buf = [0u8; 4];
            match reader.read_exact(&mut len_buf) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
                Err(e) => return Err(TdbError::Io(e)),
            }

            let len = u32::from_le_bytes(len_buf) as usize;

            // Read entry
            let mut entry_buf = vec![0u8; len];
            reader.read_exact(&mut entry_buf).map_err(TdbError::Io)?;

            let entry: Entry =
                oxicode::serde::decode_from_slice(&entry_buf, oxicode::config::standard())
                    .map_err(|e| TdbError::Other(format!("Deserialization error: {}", e)))?
                    .0;

            if entry.key.as_slice() >= start && entry.key.as_slice() < end {
                results.push((entry.key, entry.value, entry.sequence));
            }
        }

        Ok(results)
    }
}

/// LSM-tree storage engine
pub struct LsmTree {
    /// Configuration
    config: LsmConfig,
    /// Active MemTable (receiving writes)
    active_memtable: Arc<RwLock<MemTable>>,
    /// Immutable MemTables being flushed
    immutable_memtables: Arc<RwLock<Vec<MemTable>>>,
    /// SSTables organized by level
    sstables: Arc<RwLock<Vec<Vec<SsTable>>>>,
    /// Current sequence number for MVCC
    sequence: Arc<RwLock<u64>>,
    /// SSTable counter for unique file names
    sstable_counter: Arc<RwLock<usize>>,
}

impl LsmTree {
    /// Create a new LSM-tree
    pub fn new(config: LsmConfig) -> Result<Self> {
        // Create data directory
        std::fs::create_dir_all(&config.data_dir).map_err(TdbError::Io)?;

        // Initialize empty levels
        let mut levels = Vec::new();
        for _ in 0..config.num_levels {
            levels.push(Vec::new());
        }

        Ok(Self {
            config,
            active_memtable: Arc::new(RwLock::new(MemTable::new())),
            immutable_memtables: Arc::new(RwLock::new(Vec::new())),
            sstables: Arc::new(RwLock::new(levels)),
            sequence: Arc::new(RwLock::new(0)),
            sstable_counter: Arc::new(RwLock::new(0)),
        })
    }

    /// Insert a key-value pair
    pub fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
        let sequence = {
            let mut seq = self.sequence.write();
            *seq += 1;
            *seq
        };

        // Write to active MemTable
        {
            let mut memtable = self.active_memtable.write();
            memtable.put(key.to_vec(), value.to_vec(), sequence);

            // Check if MemTable is full
            if memtable.size() >= self.config.memtable_size_threshold {
                // Move active to immutable
                let old_memtable = std::mem::replace(&mut *memtable, MemTable::new());
                self.immutable_memtables.write().push(old_memtable);

                // Trigger flush (in production, this would be async)
                drop(memtable); // Release lock before flushing
                self.flush_immutable_memtables()?;
            }
        }

        Ok(())
    }

    /// Delete a key
    pub fn delete(&self, key: &[u8]) -> Result<()> {
        let sequence = {
            let mut seq = self.sequence.write();
            *seq += 1;
            *seq
        };

        // Write tombstone to active MemTable
        {
            let mut memtable = self.active_memtable.write();
            memtable.delete(key.to_vec(), sequence);

            // Check if MemTable is full
            if memtable.size() >= self.config.memtable_size_threshold {
                let old_memtable = std::mem::replace(&mut *memtable, MemTable::new());
                self.immutable_memtables.write().push(old_memtable);
                drop(memtable);
                self.flush_immutable_memtables()?;
            }
        }

        Ok(())
    }

    /// Get a value by key
    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        // Check active MemTable first
        {
            let memtable = self.active_memtable.read();
            if let Some((value, _)) = memtable.get(key) {
                return Ok(value);
            }
        }

        // Check immutable MemTables
        {
            let immutable = self.immutable_memtables.read();
            for memtable in immutable.iter().rev() {
                if let Some((value, _)) = memtable.get(key) {
                    return Ok(value);
                }
            }
        }

        // Check SSTables from level 0 downwards
        {
            let sstables = self.sstables.read();
            for level in sstables.iter() {
                for sstable in level.iter().rev() {
                    if let Some((value, _)) = sstable.get(key)? {
                        return Ok(value);
                    }
                }
            }
        }

        Ok(None)
    }

    /// Scan a range of keys
    pub fn scan(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        let mut results = BTreeMap::new();

        // Scan active MemTable
        {
            let memtable = self.active_memtable.read();
            for (key, (value, sequence)) in memtable.iter() {
                if key.as_slice() >= start && key.as_slice() < end {
                    results.insert(key.clone(), (value.clone(), *sequence));
                }
            }
        }

        // Scan immutable MemTables
        {
            let immutable = self.immutable_memtables.read();
            for memtable in immutable.iter() {
                for (key, (value, sequence)) in memtable.iter() {
                    if key.as_slice() >= start && key.as_slice() < end {
                        // Keep newest version
                        results
                            .entry(key.clone())
                            .and_modify(|e| {
                                if *sequence > e.1 {
                                    *e = (value.clone(), *sequence);
                                }
                            })
                            .or_insert((value.clone(), *sequence));
                    }
                }
            }
        }

        // Scan SSTables
        {
            let sstables = self.sstables.read();
            for level in sstables.iter() {
                for sstable in level.iter() {
                    let entries = sstable.scan(start, end)?;
                    for (key, value, sequence) in entries {
                        results
                            .entry(key)
                            .and_modify(|e| {
                                if sequence > e.1 {
                                    *e = (value.clone(), sequence);
                                }
                            })
                            .or_insert((value, sequence));
                    }
                }
            }
        }

        // Filter out tombstones and return results
        Ok(results
            .into_iter()
            .filter_map(|(k, (v, _))| v.map(|val| (k, val)))
            .collect())
    }

    /// Flush immutable MemTables to level 0 SSTables
    fn flush_immutable_memtables(&self) -> Result<()> {
        let memtables = {
            let mut immutable = self.immutable_memtables.write();
            std::mem::take(&mut *immutable)
        };

        for memtable in memtables {
            let sstable_id = {
                let mut counter = self.sstable_counter.write();
                *counter += 1;
                *counter
            };

            let path = self
                .config
                .data_dir
                .join(format!("level0_{:08}.sst", sstable_id));

            let sstable = SsTable::flush_from_memtable(
                &memtable,
                path,
                0,
                self.config.enable_bloom_filters,
                self.config.bloom_filter_fpr,
            )?;

            self.sstables.write()[0].push(sstable);
        }

        // Trigger compaction if needed
        self.maybe_compact()?;

        Ok(())
    }

    /// Check if compaction is needed and trigger it
    fn maybe_compact(&self) -> Result<()> {
        // Simple compaction trigger: if level has > 4 SSTables, compact
        let sstables = self.sstables.read();
        for (level_idx, level) in sstables.iter().enumerate() {
            if level.len() > 4 && level_idx + 1 < self.config.num_levels {
                drop(sstables);
                self.compact_level(level_idx)?;
                break;
            }
        }
        Ok(())
    }

    /// Compact a level into the next level
    fn compact_level(&self, level: usize) -> Result<()> {
        // This is a simplified compaction - production would be more sophisticated
        // For now, we just merge all SSTables in the level

        let source_tables = {
            let mut sstables = self.sstables.write();
            std::mem::take(&mut sstables[level])
        };

        if source_tables.is_empty() {
            return Ok(());
        }

        // Merge all entries from source tables
        let mut merged = BTreeMap::new();
        for sstable in &source_tables {
            let all_entries = sstable.scan(&[], &[0xff; 256])?;
            for (key, value, sequence) in all_entries {
                merged
                    .entry(key)
                    .and_modify(|e: &mut (Option<Vec<u8>>, u64)| {
                        if sequence > e.1 {
                            *e = (value.clone(), sequence);
                        }
                    })
                    .or_insert((value, sequence));
            }
        }

        // Create new SSTable in next level
        let mut temp_memtable = MemTable::new();
        for (key, (value, sequence)) in merged {
            if let Some(val) = value {
                temp_memtable.put(key, val, sequence);
            }
        }

        let sstable_id = {
            let mut counter = self.sstable_counter.write();
            *counter += 1;
            *counter
        };

        let path = self
            .config
            .data_dir
            .join(format!("level{}_{:08}.sst", level + 1, sstable_id));

        let new_sstable = SsTable::flush_from_memtable(
            &temp_memtable,
            path,
            level + 1,
            self.config.enable_bloom_filters,
            self.config.bloom_filter_fpr,
        )?;

        // Add to next level
        self.sstables.write()[level + 1].push(new_sstable);

        // Delete old SSTable files
        for sstable in source_tables {
            std::fs::remove_file(&sstable.path).ok();
        }

        Ok(())
    }

    /// Get statistics about the LSM-tree
    pub fn stats(&self) -> LsmStats {
        let active_size = self.active_memtable.read().size();
        let immutable_count = self.immutable_memtables.read().len();

        let sstables = self.sstables.read();
        let level_stats: Vec<LevelStats> = sstables
            .iter()
            .enumerate()
            .map(|(level, tables)| LevelStats {
                level,
                num_sstables: tables.len(),
                total_size_bytes: tables.iter().map(|t| t.file_size).sum(),
                num_entries: tables.iter().map(|t| t.num_entries).sum(),
            })
            .collect();

        LsmStats {
            active_memtable_size: active_size,
            immutable_memtable_count: immutable_count,
            level_stats,
            total_sstables: sstables.iter().map(|l| l.len()).sum(),
            total_size_bytes: sstables
                .iter()
                .flat_map(|l| l.iter())
                .map(|t| t.file_size)
                .sum(),
        }
    }

    /// Flush all MemTables and wait for completion
    pub fn flush(&self) -> Result<()> {
        // Move active to immutable
        {
            let mut active = self.active_memtable.write();
            if !active.is_empty() {
                let old_memtable = std::mem::replace(&mut *active, MemTable::new());
                self.immutable_memtables.write().push(old_memtable);
            }
        }

        // Flush all immutable
        self.flush_immutable_memtables()
    }
}

/// LSM-tree statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LsmStats {
    /// Active MemTable size in bytes
    pub active_memtable_size: usize,
    /// Number of immutable MemTables
    pub immutable_memtable_count: usize,
    /// Statistics per level
    pub level_stats: Vec<LevelStats>,
    /// Total number of SSTables
    pub total_sstables: usize,
    /// Total size in bytes
    pub total_size_bytes: usize,
}

/// Statistics for a single level
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LevelStats {
    /// Level number
    pub level: usize,
    /// Number of SSTables in this level
    pub num_sstables: usize,
    /// Total size in bytes
    pub total_size_bytes: usize,
    /// Total number of entries
    pub num_entries: usize,
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::env;

    #[test]
    fn test_lsm_tree_creation() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_creation");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();
        let stats = lsm.stats();

        assert_eq!(stats.active_memtable_size, 0);
        assert_eq!(stats.immutable_memtable_count, 0);
        assert_eq!(stats.total_sstables, 0);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_lsm_put_get() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_put_get");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();

        lsm.put(b"key1", b"value1").unwrap();
        lsm.put(b"key2", b"value2").unwrap();

        assert_eq!(lsm.get(b"key1").unwrap(), Some(b"value1".to_vec()));
        assert_eq!(lsm.get(b"key2").unwrap(), Some(b"value2".to_vec()));
        assert_eq!(lsm.get(b"key3").unwrap(), None);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_lsm_delete() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_delete");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();

        lsm.put(b"key1", b"value1").unwrap();
        assert_eq!(lsm.get(b"key1").unwrap(), Some(b"value1".to_vec()));

        lsm.delete(b"key1").unwrap();
        assert_eq!(lsm.get(b"key1").unwrap(), None);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_lsm_scan() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_scan");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();

        lsm.put(b"key1", b"value1").unwrap();
        lsm.put(b"key2", b"value2").unwrap();
        lsm.put(b"key3", b"value3").unwrap();
        lsm.put(b"key5", b"value5").unwrap();

        let results = lsm.scan(b"key1", b"key4").unwrap();
        assert_eq!(results.len(), 3);
        assert_eq!(results[0], (b"key1".to_vec(), b"value1".to_vec()));
        assert_eq!(results[1], (b"key2".to_vec(), b"value2".to_vec()));
        assert_eq!(results[2], (b"key3".to_vec(), b"value3".to_vec()));

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_lsm_memtable_flush() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_flush");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            memtable_size_threshold: 100, // Very small to trigger flush
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();

        // Write enough data to trigger flush
        for i in 0..10 {
            let key = format!("key{:03}", i);
            let value = format!("value{:03}", i);
            lsm.put(key.as_bytes(), value.as_bytes()).unwrap();
        }

        let stats = lsm.stats();
        assert!(stats.total_sstables > 0);

        // Verify data is still readable
        assert_eq!(lsm.get(b"key001").unwrap(), Some(b"value001".to_vec()));

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_lsm_overwrite() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_overwrite");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();

        lsm.put(b"key1", b"value1").unwrap();
        assert_eq!(lsm.get(b"key1").unwrap(), Some(b"value1".to_vec()));

        lsm.put(b"key1", b"value2").unwrap();
        assert_eq!(lsm.get(b"key1").unwrap(), Some(b"value2".to_vec()));

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_lsm_stats() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_stats");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();

        lsm.put(b"key1", b"value1").unwrap();
        lsm.put(b"key2", b"value2").unwrap();

        let stats = lsm.stats();
        assert!(stats.active_memtable_size > 0);
        assert_eq!(stats.level_stats.len(), 5); // Default 5 levels

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_lsm_flush_manually() {
        let temp_dir = env::temp_dir().join("oxirs_lsm_manual_flush");
        std::fs::remove_dir_all(&temp_dir).ok();

        let config = LsmConfig {
            data_dir: temp_dir.clone(),
            ..Default::default()
        };

        let lsm = LsmTree::new(config).unwrap();

        lsm.put(b"key1", b"value1").unwrap();
        lsm.flush().unwrap();

        let stats = lsm.stats();
        assert_eq!(stats.active_memtable_size, 0);
        assert!(stats.total_sstables > 0);

        // Verify data is still readable
        assert_eq!(lsm.get(b"key1").unwrap(), Some(b"value1".to_vec()));

        std::fs::remove_dir_all(&temp_dir).ok();
    }
}