lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0

//! Historical tree walker for traversing filesystem state at a specific TXG.
//!
//! This module provides the ability to walk the DMU tree at any historical
//! transaction group, enabling point-in-time queries of filesystem state.

use alloc::boxed::Box;
use alloc::string::{String, ToString};
use alloc::vec;
use alloc::vec::Vec;

use super::types::{ChangeType, FileType, Filter, QueryRow, TimeError, Value};

// ═══════════════════════════════════════════════════════════════════════════════
// HISTORICAL ENTRY
// ═══════════════════════════════════════════════════════════════════════════════

/// A directory entry at a specific point in time.
#[derive(Debug, Clone)]
pub struct HistoricalEntry {
    /// Entry name.
    pub name: String,
    /// Full path.
    pub path: String,
    /// Object ID (dnode number).
    pub object_id: u64,
    /// Parent object ID.
    pub parent_id: u64,
    /// File type.
    pub file_type: FileType,
    /// File size in bytes.
    pub size: u64,
    /// Mode/permissions.
    pub mode: u32,
    /// Owner UID.
    pub uid: u32,
    /// Owner GID.
    pub gid: u32,
    /// Access time.
    pub atime: u64,
    /// Modification time.
    pub mtime: u64,
    /// Creation time.
    pub ctime: u64,
    /// Transaction group when created/modified.
    pub txg: u64,
    /// Data checksum.
    pub checksum: [u64; 4],
    /// Number of links (for directories: child count).
    pub nlinks: u64,
    /// Block count.
    pub blocks: u64,
    /// Generation number.
    pub generation: u64,
}

impl HistoricalEntry {
    /// Check if this entry is a directory.
    pub fn is_dir(&self) -> bool {
        matches!(self.file_type, FileType::Directory)
    }

    /// Check if this entry is a regular file.
    pub fn is_file(&self) -> bool {
        matches!(self.file_type, FileType::Regular)
    }

    /// Check if this entry is a symlink.
    pub fn is_symlink(&self) -> bool {
        matches!(self.file_type, FileType::Symlink)
    }

    /// Convert to a QueryRow.
    pub fn to_query_row(&self) -> QueryRow {
        QueryRow {
            path: self.path.clone(),
            object_id: self.object_id,
            size: self.size,
            mtime: self.mtime,
            ctime: self.ctime,
            atime: self.atime,
            mode: self.mode,
            uid: self.uid,
            gid: self.gid,
            txg: self.txg,
            file_type: self.file_type,
            checksum: self.checksum,
        }
    }

    /// Get file extension (if any).
    pub fn extension(&self) -> Option<&str> {
        if let Some(pos) = self.name.rfind('.') {
            if pos > 0 && pos < self.name.len() - 1 {
                return Some(&self.name[pos + 1..]);
            }
        }
        None
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// HISTORICAL TREE PROVIDER
// ═══════════════════════════════════════════════════════════════════════════════

/// Trait for accessing historical filesystem tree state.
///
/// This trait must be implemented by the DMU/ZAP layer to provide
/// access to historical tree state at specific TXGs.
pub trait HistoricalTreeProvider {
    /// Get the root directory entry at the given TXG.
    fn root_at_txg(&self, txg: u64) -> Result<HistoricalEntry, TimeError>;

    /// Get a specific entry by path at the given TXG.
    fn lookup_at_txg(&self, path: &str, txg: u64) -> Result<HistoricalEntry, TimeError>;

    /// List directory contents at the given TXG.
    fn readdir_at_txg(&self, path: &str, txg: u64) -> Result<Vec<HistoricalEntry>, TimeError>;

    /// Get an entry by object ID at the given TXG.
    fn lookup_by_id_at_txg(&self, object_id: u64, txg: u64) -> Result<HistoricalEntry, TimeError>;

    /// Check if a path existed at the given TXG.
    fn exists_at_txg(&self, path: &str, txg: u64) -> bool;

    /// Read symlink target at the given TXG.
    fn readlink_at_txg(&self, path: &str, txg: u64) -> Result<String, TimeError>;
}

// ═══════════════════════════════════════════════════════════════════════════════
// TREE WALKER
// ═══════════════════════════════════════════════════════════════════════════════

/// Options for tree walking.
#[derive(Debug, Clone)]
pub struct WalkOptions {
    /// Maximum depth to recurse (-1 for unlimited).
    pub max_depth: i32,
    /// Follow symlinks.
    pub follow_symlinks: bool,
    /// Include hidden files (starting with .).
    pub include_hidden: bool,
    /// Only include directories.
    pub dirs_only: bool,
    /// Only include files.
    pub files_only: bool,
    /// Filter to apply.
    pub filter: Option<Filter>,
    /// Maximum results to return.
    pub limit: Option<usize>,
}

impl Default for WalkOptions {
    fn default() -> Self {
        Self {
            max_depth: -1,
            follow_symlinks: false,
            include_hidden: true,
            dirs_only: false,
            files_only: false,
            filter: None,
            limit: None,
        }
    }
}

/// Historical tree walker.
pub struct HistoricalTreeWalker<'a, P: HistoricalTreeProvider> {
    provider: &'a P,
    txg: u64,
}

impl<'a, P: HistoricalTreeProvider> HistoricalTreeWalker<'a, P> {
    /// Create a new walker for the given TXG.
    pub fn new(provider: &'a P, txg: u64) -> Self {
        Self { provider, txg }
    }

    /// Get the TXG this walker is operating at.
    pub fn txg(&self) -> u64 {
        self.txg
    }

    /// Look up a path at the walker's TXG.
    pub fn lookup(&self, path: &str) -> Result<HistoricalEntry, TimeError> {
        self.provider.lookup_at_txg(path, self.txg)
    }

    /// Check if a path exists at the walker's TXG.
    pub fn exists(&self, path: &str) -> bool {
        self.provider.exists_at_txg(path, self.txg)
    }

    /// List directory contents at the walker's TXG.
    pub fn readdir(&self, path: &str) -> Result<Vec<HistoricalEntry>, TimeError> {
        self.provider.readdir_at_txg(path, self.txg)
    }

    /// Walk a directory tree with options.
    pub fn walk(
        &self,
        path: &str,
        options: &WalkOptions,
    ) -> Result<Vec<HistoricalEntry>, TimeError> {
        let mut results = Vec::new();
        let mut count = 0;

        self.walk_recursive(path, 0, options, &mut results, &mut count)?;

        Ok(results)
    }

    /// Walk a directory tree with default options.
    pub fn walk_default(&self, path: &str) -> Result<Vec<HistoricalEntry>, TimeError> {
        self.walk(path, &WalkOptions::default())
    }

    /// Recursive walk implementation.
    fn walk_recursive(
        &self,
        path: &str,
        depth: i32,
        options: &WalkOptions,
        results: &mut Vec<HistoricalEntry>,
        count: &mut usize,
    ) -> Result<(), TimeError> {
        // Check depth limit
        if options.max_depth >= 0 && depth > options.max_depth {
            return Ok(());
        }

        // Check result limit
        if let Some(limit) = options.limit {
            if *count >= limit {
                return Ok(());
            }
        }

        // Read directory contents
        let entries = self.provider.readdir_at_txg(path, self.txg)?;

        for entry in entries {
            // Skip hidden files if requested
            if !options.include_hidden && entry.name.starts_with('.') {
                continue;
            }

            // Apply type filters
            if options.dirs_only && !entry.is_dir() {
                continue;
            }
            if options.files_only && !entry.is_file() {
                continue;
            }

            // Apply custom filter
            if let Some(ref filter) = options.filter {
                if !self.apply_filter(&entry, filter) {
                    continue;
                }
            }

            // Check limit
            if let Some(limit) = options.limit {
                if *count >= limit {
                    return Ok(());
                }
            }

            results.push(entry.clone());
            *count += 1;

            // Recurse into directories
            if entry.is_dir() {
                self.walk_recursive(&entry.path, depth + 1, options, results, count)?;
            } else if entry.is_symlink() && options.follow_symlinks {
                // Follow symlink if requested
                if let Ok(target) = self.provider.readlink_at_txg(&entry.path, self.txg) {
                    // Check if target is a directory
                    if let Ok(target_entry) = self.provider.lookup_at_txg(&target, self.txg) {
                        if target_entry.is_dir() {
                            self.walk_recursive(&target, depth + 1, options, results, count)?;
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Apply a filter to an entry.
    fn apply_filter(&self, entry: &HistoricalEntry, filter: &Filter) -> bool {
        match filter {
            Filter::Eq { column, value } => {
                let entry_value = self.get_entry_value(entry, column);
                self.values_equal(&entry_value, value)
            }
            Filter::Ne { column, value } => {
                let entry_value = self.get_entry_value(entry, column);
                !self.values_equal(&entry_value, value)
            }
            Filter::Lt { column, value } => {
                let entry_value = self.get_entry_value(entry, column);
                self.compare_values(&entry_value, value) == Some(core::cmp::Ordering::Less)
            }
            Filter::Le { column, value } => {
                let entry_value = self.get_entry_value(entry, column);
                matches!(
                    self.compare_values(&entry_value, value),
                    Some(core::cmp::Ordering::Less) | Some(core::cmp::Ordering::Equal)
                )
            }
            Filter::Gt { column, value } => {
                let entry_value = self.get_entry_value(entry, column);
                self.compare_values(&entry_value, value) == Some(core::cmp::Ordering::Greater)
            }
            Filter::Ge { column, value } => {
                let entry_value = self.get_entry_value(entry, column);
                matches!(
                    self.compare_values(&entry_value, value),
                    Some(core::cmp::Ordering::Greater) | Some(core::cmp::Ordering::Equal)
                )
            }
            Filter::Like { column, pattern } => {
                let entry_value = self.get_entry_value(entry, column);
                if let Value::String(s) = entry_value {
                    self.match_like(&s, pattern)
                } else {
                    false
                }
            }
            Filter::In { column, values } => {
                let entry_value = self.get_entry_value(entry, column);
                values.iter().any(|v| self.values_equal(&entry_value, v))
            }
            Filter::Between { column, low, high } => {
                let entry_value = self.get_entry_value(entry, column);
                let ge_low = matches!(
                    self.compare_values(&entry_value, low),
                    Some(core::cmp::Ordering::Greater) | Some(core::cmp::Ordering::Equal)
                );
                let le_high = matches!(
                    self.compare_values(&entry_value, high),
                    Some(core::cmp::Ordering::Less) | Some(core::cmp::Ordering::Equal)
                );
                ge_low && le_high
            }
            Filter::IsNull { column } => {
                let entry_value = self.get_entry_value(entry, column);
                matches!(entry_value, Value::Null)
            }
            Filter::IsNotNull { column } => {
                let entry_value = self.get_entry_value(entry, column);
                !matches!(entry_value, Value::Null)
            }
            Filter::And(left, right) => {
                self.apply_filter(entry, left) && self.apply_filter(entry, right)
            }
            Filter::Or(left, right) => {
                self.apply_filter(entry, left) || self.apply_filter(entry, right)
            }
            Filter::Not(inner) => !self.apply_filter(entry, inner),
        }
    }

    /// Get a value from an entry by column name.
    fn get_entry_value(&self, entry: &HistoricalEntry, column: &str) -> Value {
        match column.to_lowercase().as_str() {
            "name" => Value::String(entry.name.clone()),
            "path" => Value::String(entry.path.clone()),
            "size" => Value::Unsigned(entry.size),
            "mode" => Value::Unsigned(entry.mode as u64),
            "uid" => Value::Unsigned(entry.uid as u64),
            "gid" => Value::Unsigned(entry.gid as u64),
            "atime" => Value::Unsigned(entry.atime),
            "mtime" => Value::Unsigned(entry.mtime),
            "ctime" => Value::Unsigned(entry.ctime),
            "txg" => Value::Unsigned(entry.txg),
            "object_id" | "oid" => Value::Unsigned(entry.object_id),
            "type" | "file_type" => Value::String(entry.file_type.name().into()),
            "extension" | "ext" => {
                if let Some(ext) = entry.extension() {
                    Value::String(ext.into())
                } else {
                    Value::Null
                }
            }
            "nlinks" => Value::Unsigned(entry.nlinks),
            "blocks" => Value::Unsigned(entry.blocks),
            "generation" | "gen" => Value::Unsigned(entry.generation),
            _ => Value::Null,
        }
    }

    /// Check if two values are equal.
    fn values_equal(&self, a: &Value, b: &Value) -> bool {
        match (a, b) {
            (Value::String(s1), Value::String(s2)) => s1 == s2,
            (Value::Integer(n1), Value::Integer(n2)) => n1 == n2,
            (Value::Unsigned(n1), Value::Unsigned(n2)) => n1 == n2,
            (Value::Integer(n1), Value::Unsigned(n2)) => *n1 >= 0 && *n1 as u64 == *n2,
            (Value::Unsigned(n1), Value::Integer(n2)) => *n2 >= 0 && *n1 == *n2 as u64,
            (Value::Bool(b1), Value::Bool(b2)) => b1 == b2,
            (Value::Null, Value::Null) => true,
            _ => false,
        }
    }

    /// Compare two values.
    fn compare_values(&self, a: &Value, b: &Value) -> Option<core::cmp::Ordering> {
        match (a, b) {
            (Value::String(s1), Value::String(s2)) => Some(s1.cmp(s2)),
            (Value::Integer(n1), Value::Integer(n2)) => Some(n1.cmp(n2)),
            (Value::Unsigned(n1), Value::Unsigned(n2)) => Some(n1.cmp(n2)),
            (Value::Integer(n1), Value::Unsigned(n2)) => {
                if *n1 < 0 {
                    Some(core::cmp::Ordering::Less)
                } else {
                    Some((*n1 as u64).cmp(n2))
                }
            }
            (Value::Unsigned(n1), Value::Integer(n2)) => {
                if *n2 < 0 {
                    Some(core::cmp::Ordering::Greater)
                } else {
                    Some(n1.cmp(&(*n2 as u64)))
                }
            }
            _ => None,
        }
    }

    /// Match a value against a SQL LIKE pattern.
    fn match_like(&self, value: &str, pattern: &str) -> bool {
        // Convert SQL LIKE pattern to simple matching
        // % = any sequence, _ = single character
        let mut pattern_chars = pattern.chars().peekable();
        let mut value_chars = value.chars().peekable();

        self.match_like_recursive(&mut pattern_chars, &mut value_chars)
    }

    /// Recursive LIKE pattern matching.
    fn match_like_recursive(
        &self,
        pattern: &mut core::iter::Peekable<core::str::Chars>,
        value: &mut core::iter::Peekable<core::str::Chars>,
    ) -> bool {
        loop {
            match pattern.peek() {
                None => return value.peek().is_none(),
                Some('%') => {
                    pattern.next();
                    // % matches zero or more characters
                    if pattern.peek().is_none() {
                        return true;
                    }
                    // Try matching at each position
                    loop {
                        let mut p_clone = pattern.clone();
                        let mut v_clone = value.clone();
                        if self.match_like_recursive(&mut p_clone, &mut v_clone) {
                            return true;
                        }
                        if value.next().is_none() {
                            return false;
                        }
                    }
                }
                Some('_') => {
                    pattern.next();
                    if value.next().is_none() {
                        return false;
                    }
                }
                Some(&pc) => {
                    if let Some(&vc) = value.peek() {
                        if pc.to_lowercase().next() == vc.to_lowercase().next() {
                            pattern.next();
                            value.next();
                        } else {
                            return false;
                        }
                    } else {
                        return false;
                    }
                }
            }
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// IN-MEMORY TREE PROVIDER (FOR TESTING)
// ═══════════════════════════════════════════════════════════════════════════════

/// In-memory implementation of HistoricalTreeProvider for testing.
#[derive(Debug, Default)]
pub struct InMemoryTreeProvider {
    /// Entries organized by (path, txg).
    pub entries: Vec<(u64, HistoricalEntry)>,
}

impl InMemoryTreeProvider {
    /// Create a new empty provider.
    pub fn new() -> Self {
        Self::default()
    }

    /// Add an entry at a specific TXG.
    pub fn add_entry(&mut self, txg: u64, entry: HistoricalEntry) {
        self.entries.push((txg, entry));
    }

    /// Find entries for a path that existed at or before the given TXG.
    fn find_at_txg(&self, path: &str, target_txg: u64) -> Option<&HistoricalEntry> {
        self.entries
            .iter()
            .filter(|(txg, e)| e.path == path && *txg <= target_txg)
            .max_by_key(|(txg, _)| *txg)
            .map(|(_, e)| e)
    }
}

impl HistoricalTreeProvider for InMemoryTreeProvider {
    fn root_at_txg(&self, txg: u64) -> Result<HistoricalEntry, TimeError> {
        self.find_at_txg("/", txg)
            .cloned()
            .ok_or(TimeError::PathNotFound("/".into()))
    }

    fn lookup_at_txg(&self, path: &str, txg: u64) -> Result<HistoricalEntry, TimeError> {
        self.find_at_txg(path, txg)
            .cloned()
            .ok_or_else(|| TimeError::PathNotFound(path.into()))
    }

    fn readdir_at_txg(&self, path: &str, txg: u64) -> Result<Vec<HistoricalEntry>, TimeError> {
        // Normalize path
        let parent = if path == "/" {
            "/".to_string()
        } else {
            path.trim_end_matches('/').to_string()
        };

        // Find all entries whose parent is this path and existed at this TXG
        let mut children = Vec::new();
        let mut seen_names = Vec::new();

        // Sort by TXG descending to get most recent first
        let mut sorted: Vec<_> = self.entries.iter().filter(|(t, _)| *t <= txg).collect();
        sorted.sort_by(|a, b| b.0.cmp(&a.0));

        for (_, entry) in sorted {
            // Check if this entry's parent matches
            let entry_parent = if entry.path == "/" {
                "".to_string()
            } else if let Some(pos) = entry.path.rfind('/') {
                if pos == 0 {
                    "/".to_string()
                } else {
                    entry.path[..pos].to_string()
                }
            } else {
                "/".to_string()
            };

            if entry_parent == parent && !seen_names.contains(&entry.name) {
                seen_names.push(entry.name.clone());
                children.push(entry.clone());
            }
        }

        Ok(children)
    }

    fn lookup_by_id_at_txg(&self, object_id: u64, txg: u64) -> Result<HistoricalEntry, TimeError> {
        self.entries
            .iter()
            .filter(|(t, e)| e.object_id == object_id && *t <= txg)
            .max_by_key(|(t, _)| *t)
            .map(|(_, e)| e.clone())
            .ok_or_else(|| TimeError::PathNotFound(alloc::format!("oid:{}", object_id)))
    }

    fn exists_at_txg(&self, path: &str, txg: u64) -> bool {
        self.find_at_txg(path, txg).is_some()
    }

    fn readlink_at_txg(&self, path: &str, txg: u64) -> Result<String, TimeError> {
        let entry = self.lookup_at_txg(path, txg)?;
        if entry.is_symlink() {
            // For testing, store symlink target in name field prefixed with "-> "
            if entry.name.starts_with("-> ") {
                Ok(entry.name[3..].to_string())
            } else {
                Err(TimeError::NotSupported("symlink target not stored".into()))
            }
        } else {
            Err(TimeError::NotSupported("not a symlink".into()))
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// TESTS
// ═══════════════════════════════════════════════════════════════════════════════

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

    fn create_entry(
        path: &str,
        name: &str,
        file_type: FileType,
        size: u64,
        txg: u64,
    ) -> HistoricalEntry {
        HistoricalEntry {
            name: name.into(),
            path: path.into(),
            object_id: path.len() as u64,
            parent_id: 1,
            file_type,
            size,
            mode: 0o644,
            uid: 1000,
            gid: 1000,
            atime: txg * 1000,
            mtime: txg * 1000,
            ctime: txg * 1000,
            txg,
            checksum: [0; 4],
            nlinks: 1,
            blocks: size.div_ceil(512),
            generation: txg,
        }
    }

    fn create_test_provider() -> InMemoryTreeProvider {
        let mut provider = InMemoryTreeProvider::new();

        // Add root
        provider.add_entry(100, create_entry("/", "", FileType::Directory, 0, 100));

        // Add /data directory
        provider.add_entry(
            100,
            create_entry("/data", "data", FileType::Directory, 0, 100),
        );

        // Add files at TXG 100
        provider.add_entry(
            100,
            create_entry("/data/file1.txt", "file1.txt", FileType::Regular, 100, 100),
        );
        provider.add_entry(
            100,
            create_entry("/data/file2.pdf", "file2.pdf", FileType::Regular, 2000, 100),
        );

        // Add file at TXG 200
        provider.add_entry(
            200,
            create_entry("/data/file3.doc", "file3.doc", FileType::Regular, 500, 200),
        );

        // Modify file1.txt at TXG 200
        provider.add_entry(
            200,
            create_entry("/data/file1.txt", "file1.txt", FileType::Regular, 150, 200),
        );

        provider
    }

    #[test]
    fn test_lookup_at_txg() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 100);

        let entry = walker.lookup("/data/file1.txt").unwrap();
        assert_eq!(entry.size, 100);

        // At TXG 200, file1.txt should be larger
        let walker200 = HistoricalTreeWalker::new(&provider, 200);
        let entry = walker200.lookup("/data/file1.txt").unwrap();
        assert_eq!(entry.size, 150);
    }

    #[test]
    fn test_readdir() {
        let provider = create_test_provider();

        // At TXG 100, should have 2 files
        let walker = HistoricalTreeWalker::new(&provider, 100);
        let entries = walker.readdir("/data").unwrap();
        assert_eq!(entries.len(), 2);

        // At TXG 200, should have 3 files
        let walker200 = HistoricalTreeWalker::new(&provider, 200);
        let entries = walker200.readdir("/data").unwrap();
        assert_eq!(entries.len(), 3);
    }

    #[test]
    fn test_exists() {
        let provider = create_test_provider();

        let walker100 = HistoricalTreeWalker::new(&provider, 100);
        assert!(walker100.exists("/data/file1.txt"));
        assert!(!walker100.exists("/data/file3.doc")); // Not created yet

        let walker200 = HistoricalTreeWalker::new(&provider, 200);
        assert!(walker200.exists("/data/file3.doc"));
    }

    #[test]
    fn test_walk_default() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        let results = walker.walk_default("/data").unwrap();
        assert_eq!(results.len(), 3);
    }

    #[test]
    fn test_walk_with_limit() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        let options = WalkOptions {
            limit: Some(2),
            ..Default::default()
        };
        let results = walker.walk("/data", &options).unwrap();
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_walk_files_only() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        let options = WalkOptions {
            files_only: true,
            ..Default::default()
        };
        let results = walker.walk("/data", &options).unwrap();
        assert!(results.iter().all(|e| e.is_file()));
    }

    #[test]
    fn test_filter_size() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        // Find files larger than 100 bytes
        let options = WalkOptions {
            filter: Some(Filter::Gt {
                column: "size".into(),
                value: Value::Unsigned(100),
            }),
            ..Default::default()
        };
        let results = walker.walk("/data", &options).unwrap();
        assert!(results.iter().all(|e| e.size > 100));
    }

    #[test]
    fn test_filter_extension() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        let options = WalkOptions {
            filter: Some(Filter::Eq {
                column: "extension".into(),
                value: Value::String("txt".into()),
            }),
            ..Default::default()
        };
        let results = walker.walk("/data", &options).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].name, "file1.txt");
    }

    #[test]
    fn test_filter_like() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        let options = WalkOptions {
            filter: Some(Filter::Like {
                column: "name".into(),
                pattern: "file%.txt".into(),
            }),
            ..Default::default()
        };
        let results = walker.walk("/data", &options).unwrap();
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn test_filter_and() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        // size > 100 AND size < 1000
        let options = WalkOptions {
            filter: Some(Filter::And(
                Box::new(Filter::Gt {
                    column: "size".into(),
                    value: Value::Unsigned(100),
                }),
                Box::new(Filter::Lt {
                    column: "size".into(),
                    value: Value::Unsigned(1000),
                }),
            )),
            ..Default::default()
        };
        let results = walker.walk("/data", &options).unwrap();
        assert!(results.iter().all(|e| e.size > 100 && e.size < 1000));
    }

    #[test]
    fn test_filter_in() {
        let provider = create_test_provider();
        let walker = HistoricalTreeWalker::new(&provider, 200);

        let options = WalkOptions {
            filter: Some(Filter::In {
                column: "extension".into(),
                values: vec![Value::String("txt".into()), Value::String("pdf".into())],
            }),
            ..Default::default()
        };
        let results = walker.walk("/data", &options).unwrap();
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_match_like_patterns() {
        let provider = InMemoryTreeProvider::new();
        let walker = HistoricalTreeWalker::new(&provider, 100);

        // Test various patterns
        assert!(walker.match_like("hello.txt", "%.txt"));
        assert!(walker.match_like("hello.txt", "hello%"));
        assert!(walker.match_like("hello.txt", "%llo%"));
        assert!(walker.match_like("hello.txt", "h_llo.txt"));
        assert!(!walker.match_like("hello.txt", "%.pdf"));
        assert!(walker.match_like("file123.txt", "file___.txt"));
    }

    #[test]
    fn test_entry_extension() {
        let entry = create_entry("/test/file.txt", "file.txt", FileType::Regular, 100, 100);
        assert_eq!(entry.extension(), Some("txt"));

        let entry = create_entry("/test/noext", "noext", FileType::Regular, 100, 100);
        assert_eq!(entry.extension(), None);

        // Hidden files with dot prefix have no extension
        let entry = create_entry("/test/.hidden", ".hidden", FileType::Regular, 100, 100);
        assert_eq!(entry.extension(), None);

        // Hidden file with extension
        let entry = create_entry(
            "/test/.hidden.txt",
            ".hidden.txt",
            FileType::Regular,
            100,
            100,
        );
        assert_eq!(entry.extension(), Some("txt"));
    }

    #[test]
    fn test_to_query_row() {
        let entry = create_entry("/data/file.txt", "file.txt", FileType::Regular, 1024, 100);
        let row = entry.to_query_row();

        assert_eq!(row.path, "/data/file.txt");
        assert_eq!(row.size, 1024);
        assert_eq!(row.txg, 100);
    }
}