nftw 0.1.1

#![warn(missing_docs)]
#![forbid(unsafe_code)]

//! Nftw is a crate for walking a directory structure and asynchronously
//! working on the objects found.
use std::fmt;
use std::fs;
use std::io;
#[cfg(unix)]
use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};
use std::sync::mpsc;
use std::thread;

use rayon::prelude::*;
use rayon::ThreadPoolBuilder;

/// Error encountered during directory traversal or file processing.
///
/// This error type combines a file system path with the underlying I/O error
/// that occurred when accessing that path. It is returned when directory reading
/// or file operations fail during tree traversal.
///
/// # Fields
///
/// * `path` - The path where the error occurred
/// * `error` - The underlying I/O error with details about what went wrong
///
/// # Examples
///
/// ```
/// use nftw::{Nftw, NftwError};
/// use std::path::Path;
///
/// let walker = Nftw::new(Path::new("."));
/// let results = walker.map(|path| {
///     std::fs::read(path).map_err(|e| NftwError::from((path.to_path_buf(), e)))
/// });
///
/// for result in results {
///     if let Err(e) = result {
///         eprintln!("Failed to read {}: {}", e.path.display(), e.error);
///     }
/// }
/// ```
#[derive(Debug)]
pub struct NftwError {
    /// The path where the error occurred
    pub path: PathBuf,
    /// The underlying I/O error
    pub error: io::Error,
}

impl From<(PathBuf, io::Error)> for NftwError {
    /// Converts a tuple of path and I/O error into an `NftwError`.
    fn from((path, error): (PathBuf, io::Error)) -> Self {
        NftwError { path, error }
    }
}

impl fmt::Display for NftwError {
    /// Formats the error as a human-readable string combining the error and path.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Error {}: {}", self.error, self.path.display())
    }
}

/// Configuration for directory walking
pub struct Nftw {
    /// Root directory to start traversal from
    pub root: PathBuf,
    /// Number of threads to use for parallel processing
    pub nthreads: usize,
    /// Whether to ignore symbolic links
    pub ignore_symlinks: bool,
    /// Whether to ignore hidden files (names starting with .)
    pub ignore_hidden: bool,
    /// Whether to restrict traversal to the same device/filesystem
    pub same_device: bool,
    /// Device ID of the root directory (for same_device checks on Unix)
    pub device: Option<u64>,
    /// Whether to apply the function to only one file per leaf directory.
    /// A leaf directory is one that has no subdirectories.
    pub one_per_leaf: bool,
}

impl Nftw {
    /// Create a new Nftw walker with the given root directory and thread count
    pub fn new(root: &Path) -> Self {
        // Save one thread for the path walker and one for collector,
        // unless only have one core.
        Nftw {
            root: root.to_path_buf(),
            nthreads: num_cpus::get(),
            ignore_symlinks: false,
            ignore_hidden: false,
            same_device: false,
            #[cfg(unix)]
            device: Some(fs::metadata(root).unwrap().dev()),
            #[cfg(not(unix))]
            device: None,
            one_per_leaf: false,
        }
    }

    /// Configure to ignore symbolic links (builder pattern)
    pub fn ignore_symlinks(&mut self) -> &mut Self {
        self.ignore_symlinks = true;
        self
    }
    /// Configure to ignore hidden files (names starting with .)
    pub fn ignore_hidden(&mut self) -> &mut Self {
        self.ignore_hidden = true;
        self
    }
    /// Configure to not cross mount points (Unix only) (builder pattern)
    pub fn same_device(&mut self) -> &mut Self {
        self.same_device = true;
        self
    }
    /// Set the number of threads to use for parallel processing (builder pattern)
    pub fn threads(&mut self, n: usize) -> &mut Self {
        self.nthreads = n;
        self
    }
    /// Configure to apply the function to only one file per leaf directory (builder pattern).
    /// A leaf directory is one that contains no subdirectories.
    pub fn one_per_leaf(&mut self) -> &mut Self {
        self.one_per_leaf = true;
        self
    }

    /// Serially applies a mapping function to each path and collects results.
    ///
    /// This function walks the directory tree synchronously and applies the provided
    /// function to each file path. All results and errors are collected into separate
    /// vectors.
    ///
    /// # Arguments
    ///
    /// * `func` - A closure that takes a `&Path` and returns `Result<T, io::Error>`
    ///
    /// # Returns
    ///
    /// A vector of results from applying the function to each path.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let walker = Nftw::new(std::path::Path::new("."));
    /// let results = walker.map(|path| {
    ///     Ok(path.file_name().unwrap().to_string_lossy().to_string())
    /// });
    /// for result in results {
    ///     println!("Filename: {:?}", result);
    /// }
    /// ```
    pub fn map<T: 'static>(
        &self,
        func: impl Fn(&Path) -> Result<T, NftwError> + 'static,
    ) -> Vec<Result<T, NftwError>> {
        // Copy configuration flags to avoid borrowing self in the loop
        let ignore_symlinks = self.ignore_symlinks;
        let ignore_hidden = self.ignore_hidden;
        let one_per_leaf = self.one_per_leaf;

        // Use a vector as a stack for depth-first traversal of the directory tree.
        // This keeps memory usage bounded by the maximum depth of the tree.
        let mut dir_queue: Vec<PathBuf> = vec![self.root.to_path_buf()];
        let mut results: Vec<Result<T, NftwError>> = Vec::new();

        // Traverse the directory tree depth-first
        while let Some(dir) = dir_queue.pop() {
            // Read directory contents; errors are captured and included in results
            let entries = match fs::read_dir(&dir) {
                Ok(e) => e,
                Err(e) => {
                    // Record directory reading errors for the caller
                    results.push(Err((dir.clone(), e).into()));
                    continue;
                }
            };

            // Collect files and subdirectories separately so we can detect leaf directories
            let mut files: Vec<PathBuf> = Vec::new();
            let mut subdirs: Vec<PathBuf> = Vec::new();

            // Process each entry in the directory
            for entry_result in entries {
                let entry = match entry_result {
                    Ok(e) => e,
                    Err(e) => {
                        // Record entry reading errors
                        results.push(Err((dir.clone(), e).into()));
                        continue;
                    }
                };

                // Skip hidden files if configured to do so
                if ignore_hidden && entry.file_name().to_string_lossy().starts_with('.') {
                    continue;
                }

                let path = entry.path();
                match entry.file_type() {
                    Ok(ft) => {
                        if ft.is_dir() {
                            // Check if we should descend into this directory
                            // On Unix, respect the same_device restriction to prevent
                            // crossing filesystem boundaries
                            #[cfg(unix)]
                            if self.same_device {
                                let dev = match fs::metadata(&path) {
                                    Ok(m) => m.dev(),
                                    Err(e) => {
                                        results.push(Err((path.clone(), e).into()));
                                        continue;
                                    }
                                };
                                // Skip directories on different devices
                                if Some(dev) != self.device {
                                    continue;
                                }
                            }
                            subdirs.push(path);
                        } else if ft.is_file() {
                            // Skip symlinks if configured to do so
                            if ignore_symlinks && ft.is_symlink() {
                                continue;
                            }
                            files.push(path);
                        }
                    }
                    Err(e) => {
                        // Record file type checking errors
                        results.push(Err((entry.path(), e).into()));
                    }
                }
            }

            // Queue subdirectories for traversal
            dir_queue.extend(subdirs.iter().cloned());

            // Apply the function to files in this directory
            if one_per_leaf && subdirs.is_empty() {
                // In one_per_leaf mode, only process one file in each leaf directory
                if let Some(path) = files.first() {
                    results.push(func(path));
                }
            } else {
                for path in &files {
                    results.push(func(path));
                }
            }
        }

        results
    }

    /// Applies a mapping function to each path in parallel and collects results.
    ///
    /// This function walks the directory tree in a dedicated thread while a rayon
    /// thread pool processes paths in parallel. Results are collected in a separate
    /// thread. Use this when the cost of processing each file justifies the overhead
    /// of parallel processing.
    ///
    /// # Architecture
    ///
    /// - **Walker thread**: Traverses directories and sends file paths through a channel
    /// - **Worker threads**: Receive paths and apply the function in parallel
    ///
    /// # Arguments
    ///
    /// * `func` - A closure that takes a `&Path` and returns `Result<T, io::Error>`.
    ///   Must be `Send + Sync` for parallel execution.
    ///
    /// # Returns
    ///
    /// A vector of results from applying the function to each path.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let mut walker = Nftw::new(std::path::Path::new("."));
    /// walker.threads(4);
    /// let results = walker.par_map(|path| {
    ///     std::fs::read(path).map(|data| data.len())
    /// });
    /// for result in results {
    ///     println!("File size: {:?}", result);
    /// }
    /// ```
    pub fn par_map<T: Send + 'static>(
        &self,
        func: impl Fn(&Path) -> Result<T, NftwError> + Send + Sync + 'static,
    ) -> Vec<Result<T, NftwError>> {
        // Set up inter-thread communication channels:
        // path_tx/path_rx: Walker thread sends file paths; worker threads receive them
        let (path_tx, path_rx) = mpsc::channel::<Result<PathBuf, NftwError>>();
        // result_tx/result_rx: Worker threads send function results back to main thread
        let (result_tx, result_rx) = mpsc::channel();

        // Clone configuration for the walker thread
        let root_clone = self.root.to_path_buf();
        let ignore_symlinks = self.ignore_symlinks;
        let ignore_hidden = self.ignore_hidden;
        let same_device = self.same_device;
        let device = self.device;
        let one_per_leaf = self.one_per_leaf;

        // WALKER THREAD: Traverses the directory tree depth-first and sends file paths
        // through the channel. Runs independently from processing to decouple I/O from computation.
        // This allows processing to begin while the filesystem is still being scanned.
        let walker_thread = thread::spawn(move || {
            let mut dir_queue: Vec<PathBuf> = vec![root_clone];
            while let Some(dir) = dir_queue.pop() {
                // Read directory contents; send errors through to the worker threads
                let entries = match fs::read_dir(&dir) {
                    Ok(entries) => entries,
                    Err(e) => {
                        // Send read error; ignore send failure (workers may have finished)
                        let _ = path_tx.send(Err((dir.clone(), e).into()));
                        continue;
                    }
                };

                // Collect files and subdirectories separately so we can detect leaf directories
                let mut files: Vec<PathBuf> = Vec::new();
                let mut subdirs: Vec<PathBuf> = Vec::new();

                for entry_result in entries {
                    let entry = match entry_result {
                        Ok(entry) => entry,
                        Err(e) => {
                            let _ = path_tx.send(Err((dir.clone(), e).into()));
                            continue;
                        }
                    };

                    // Skip hidden files if configured
                    if ignore_hidden && entry.file_name().to_string_lossy().starts_with('.') {
                        continue;
                    }

                    let path = entry.path();
                    match entry.file_type() {
                        Ok(file_type) => {
                            if file_type.is_dir() {
                                // On Unix, skip directories on different filesystems if configured
                                #[cfg(unix)]
                                if same_device {
                                    let dev = match fs::metadata(&path) {
                                        Ok(m) => m.dev(),
                                        Err(e) => {
                                            let _ = path_tx.send(Err((path.clone(), e).into()));
                                            continue;
                                        }
                                    };
                                    if Some(dev) != device {
                                        continue;
                                    }
                                }
                                subdirs.push(path);
                            } else if file_type.is_file() {
                                // Skip symlinks if configured
                                if ignore_symlinks && file_type.is_symlink() {
                                    continue;
                                }
                                files.push(path);
                            }
                        }
                        Err(e) => {
                            // Send file type checking error
                            let _ = path_tx.send(Err((path, e).into()));
                        }
                    }
                }

                // Queue subdirectories for traversal
                for subdir in &subdirs {
                    dir_queue.push(subdir.clone());
                }

                // Send file paths to worker threads
                if one_per_leaf && subdirs.is_empty() {
                    // In one_per_leaf mode, one file per leaf directory is sent.
                    if let Some(path) = files.into_iter().next() {
                        let _ = path_tx.send(Ok(path));
                    }
                } else {
                    // Otherwise send all files.
                    for path in files {
                        let _ = path_tx.send(Ok(path));
                    }
                }
            }
            // Walker thread exits when directory queue is empty,
            // dropping path_tx causes channel to close and workers to finish
        });

        // WORKER THREADS: Create a Rayon thread pool to process paths in parallel.
        // The pool size is configurable via the `threads()` method.
        let pool = ThreadPoolBuilder::new()
            .num_threads(self.nthreads)
            .build()
            .unwrap();

        // Install the thread pool and begin parallel processing of paths.
        // `path_rx.into_iter()` yields each path as the walker sends them.
        // `par_bridge()` makes the iterator distribute work across threads.
        // `for_each_with(result_tx, ...)` applies the user function to each path
        // and sends results back through the channel.
        pool.install(|| {
            path_rx
                .into_iter()
                .par_bridge()
                .for_each_with(result_tx, |tx, path_result| {
                    // For each path received, apply the user function and send back the result.
                    // `path_result` is either `Ok(path)` or `Err(NftwError)` from the walker.
                    // We use and_then to short-circuit error handling:
                    // - If walker sent an error, pass it through
                    // - If walker sent a path, apply the user function
                    let _ = tx.send(path_result.and_then(|path| func(&path)));
                });
        });

        // Wait for the walker thread to finish traversing the directory tree
        let _ = walker_thread.join();

        // Collect all results from the worker threads.
        // This blocks until all workers have sent their results.
        result_rx.iter().collect()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs::{self, File};
    use tempfile::TempDir;

    /// Helper function to create a test directory structure
    fn create_test_dir_structure() -> TempDir {
        let temp_dir = TempDir::new().unwrap();
        let root = temp_dir.path();

        // Create some files and directories
        File::create(root.join("file1.txt")).unwrap();
        File::create(root.join("file2.txt")).unwrap();

        fs::create_dir(root.join("subdir1")).unwrap();
        File::create(root.join("subdir1/file3.txt")).unwrap();
        File::create(root.join("subdir1/file4.txt")).unwrap();

        fs::create_dir(root.join("subdir2")).unwrap();
        File::create(root.join("subdir2/file5.txt")).unwrap();

        fs::create_dir(root.join("subdir1/nested")).unwrap();
        File::create(root.join("subdir1/nested/file6.txt")).unwrap();

        // Create hidden files
        File::create(root.join(".hidden")).unwrap();
        File::create(root.join("subdir1/.hidden2")).unwrap();

        temp_dir
    }

    #[test]
    fn test_nftw_new() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let walker = Nftw::new(root);

        assert_eq!(walker.root, root);
        assert!(!walker.ignore_symlinks);
        assert!(!walker.ignore_hidden);
        assert!(!walker.same_device);
        assert!(walker.device.is_some());
    }

    #[test]
    fn test_builder_chaining() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let mut walker = Nftw::new(root);
        walker.ignore_symlinks().ignore_hidden().threads(2);

        assert!(walker.ignore_symlinks);
        assert!(walker.ignore_hidden);
        assert_eq!(walker.nthreads, 2);
    }

    #[test]
    fn test_map_basic() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let walker = Nftw::new(root);
        let results =
            walker.map(|path| Ok(path.file_name().unwrap().to_string_lossy().to_string()));

        let ok_results: Vec<_> = results.iter().filter_map(|r| r.as_ref().ok()).collect();

        // Should have 8 files (6 regular + 2 hidden)
        assert_eq!(ok_results.len(), 8);
    }

    #[test]
    fn test_map_with_error_handling() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let walker = Nftw::new(root);
        let results = walker.map(|path| {
            if path
                .file_name()
                .unwrap()
                .to_string_lossy()
                .contains("file1")
            {
                Err(NftwError {
                    path: "test/path.txt".into(),
                    error: io::ErrorKind::Other.into(),
                })
            } else {
                Ok(())
            }
        });

        // Check that we have both errors and successes
        let has_error = results.iter().any(|r| r.is_err());
        let has_ok = results.iter().any(|r| r.is_ok());

        assert!(has_error);
        assert!(has_ok);
    }

    #[test]
    fn test_map_ignore_hidden() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let mut walker = Nftw::new(root);
        walker.ignore_hidden();
        let results = walker.map(|_| Ok(()));

        // Should only have 6 regular files (excluding hidden)
        assert_eq!(results.len(), 6);
    }

    #[test]
    fn test_map_ignore_symlinks() {
        let temp_dir = TempDir::new().unwrap();
        let root = temp_dir.path();

        // Create a regular file and a symlink
        File::create(root.join("regular.txt")).unwrap();
        File::create(root.join("target.txt")).unwrap();
        #[cfg(unix)]
        std::os::unix::fs::symlink(root.join("target.txt"), root.join("link.txt")).unwrap();

        let mut walker = Nftw::new(root);
        walker.ignore_symlinks();
        let results = walker.map(|_| Ok(()));

        // On Unix, should only have 2 regular files (excluding symlink)
        // On other platforms, should have all files since symlinks may not exist
        #[cfg(unix)]
        assert_eq!(results.len(), 2);
        #[cfg(not(unix))]
        assert!(results.len() >= 2);
    }

    #[test]
    fn test_par_map_basic() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let walker = Nftw::new(root);
        let results =
            walker.par_map(|path| Ok(path.file_name().unwrap().to_string_lossy().to_string()));

        let ok_results: Vec<_> = results.iter().filter_map(|r| r.as_ref().ok()).collect();

        // Should have 8 files (6 regular + 2 hidden)
        assert_eq!(ok_results.len(), 8);
    }

    #[test]
    fn test_par_map_ignore_hidden() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let mut walker = Nftw::new(root);
        walker.ignore_hidden();
        let results = walker.par_map(|_| Ok(()));

        // Should only have 6 regular files (excluding hidden)
        assert_eq!(results.len(), 6);
    }

    #[test]
    fn test_par_map_ignore_symlinks() {
        let temp_dir = TempDir::new().unwrap();
        let root = temp_dir.path();

        // Create a regular file and a symlink
        File::create(root.join("regular.txt")).unwrap();
        File::create(root.join("target.txt")).unwrap();
        #[cfg(unix)]
        std::os::unix::fs::symlink(root.join("target.txt"), root.join("link.txt")).unwrap();

        let mut walker = Nftw::new(root);
        walker.ignore_symlinks();
        let results = walker.par_map(|_| Ok(()));

        // On Unix, should only have 2 regular files (excluding symlink)
        // On other platforms, should have all files since symlinks may not exist
        #[cfg(unix)]
        assert_eq!(results.len(), 2);
        #[cfg(not(unix))]
        assert!(results.len() >= 2);
    }

    #[test]
    fn test_nested_directories_with_map() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let walker = Nftw::new(root);
        let results = walker.map(|path| Ok(path.to_string_lossy().to_string()));

        let ok_results: Vec<_> = results.iter().filter_map(|r| r.as_ref().ok()).collect();

        // Verify nested file is processed
        assert!(ok_results
            .iter()
            .any(|p| p.contains("nested") && p.contains("file6")));
    }

    #[test]
    fn test_nested_directories_with_par_map() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let walker = Nftw::new(root);
        let results = walker.par_map(|path| Ok(path.to_string_lossy().to_string()));

        let ok_results: Vec<_> = results.iter().filter_map(|r| r.as_ref().ok()).collect();

        // Verify nested file is processed
        assert!(ok_results
            .iter()
            .any(|p| p.contains("nested") && p.contains("file6")));
    }

    #[test]
    fn test_map_one_per_leaf() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let mut walker = Nftw::new(root);
        walker.one_per_leaf();
        let results = walker.map(|path| Ok(path.to_string_lossy().to_string()));

        let ok_results: Vec<String> = results.into_iter().filter_map(|r| r.ok()).collect();

        // Non-leaf dirs process all files normally:
        //   root (non-leaf): file1.txt, file2.txt, .hidden (3)
        //   subdir1 (non-leaf): file3.txt, file4.txt, .hidden2 (3)
        // Leaf dirs process only one file:
        //   subdir1/nested (leaf): file6.txt (1)
        //   subdir2 (leaf): file5.txt (1)
        // Total: 8
        assert_eq!(
            ok_results.len(),
            8,
            "Expected 8 results (all non-leaf files + one per leaf dir), got: {:?}",
            ok_results
        );

        // Non-leaf directory files should all be present
        assert!(
            ok_results.iter().any(|p| p.contains("file1.txt")),
            "Expected file1.txt from root (non-leaf)"
        );
        assert!(
            ok_results.iter().any(|p| p.contains("file2.txt")),
            "Expected file2.txt from root (non-leaf)"
        );
        assert!(
            ok_results.iter().any(|p| p.contains("file3.txt")),
            "Expected file3.txt from subdir1 (non-leaf)"
        );
        // Leaf directories should contribute one file each
        assert!(
            ok_results.iter().any(|p| p.contains("nested")),
            "Expected a file from subdir1/nested (leaf)"
        );
        assert!(
            ok_results.iter().any(|p| p.contains("subdir2")),
            "Expected a file from subdir2 (leaf)"
        );
    }

    #[test]
    fn test_par_map_one_per_leaf() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let mut walker = Nftw::new(root);
        walker.one_per_leaf();
        let results = walker.par_map(|path| Ok(path.to_string_lossy().to_string()));

        let ok_results: Vec<String> = results.into_iter().filter_map(|r| r.ok()).collect();

        // Non-leaf dirs process all files; leaf dirs process only one file
        // root (non-leaf): 3 + subdir1 (non-leaf): 3 + nested (leaf): 1 + subdir2 (leaf): 1 = 8
        assert_eq!(
            ok_results.len(),
            8,
            "Expected 8 results (all non-leaf files + one per leaf dir), got: {:?}",
            ok_results
        );

        assert!(
            ok_results.iter().any(|p| p.contains("file1.txt")),
            "Expected file1.txt from root (non-leaf)"
        );
        assert!(
            ok_results.iter().any(|p| p.contains("file2.txt")),
            "Expected file2.txt from root (non-leaf)"
        );
        assert!(
            ok_results.iter().any(|p| p.contains("file3.txt")),
            "Expected file3.txt from subdir1 (non-leaf)"
        );
        assert!(
            ok_results.iter().any(|p| p.contains("nested")),
            "Expected a file from subdir1/nested (leaf)"
        );
        assert!(
            ok_results.iter().any(|p| p.contains("subdir2")),
            "Expected a file from subdir2 (leaf)"
        );
    }

    #[test]
    fn test_one_per_leaf_builder() {
        let temp_dir = create_test_dir_structure();
        let root = temp_dir.path();

        let mut walker = Nftw::new(root);
        assert!(!walker.one_per_leaf);

        walker.one_per_leaf();
        assert!(walker.one_per_leaf);
    }
}