flashsieve 0.1.2

//! Incremental index updates via filesystem notifications.
//!
//! Watches a directory for file changes and updates the bloom index
//! incrementally without rebuilding from scratch. On Linux, uses inotify
//! for efficient kernel-level notification. On other platforms, falls back
//! to periodic polling.
//!
//! # Architecture
//!
//! ```text
//! ┌──────────────┐     ┌──────────────────┐     ┌─────────────┐
//! │  inotify/    │────▶│ IncrementalWatch  │────▶│ BlockIndex  │
//! │  polling     │     │ (detects changes) │     │ (updates)   │
//! └──────────────┘     └──────────────────┘     └─────────────┘
//! ```

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

/// Maximum total bytes to index across all modified files in one batch (16 GiB).
const MAX_TOTAL_INDEX_BYTES: usize = 16 * 1024 * 1024 * 1024;

use crate::builder::BlockIndexBuilder;
use crate::error::Result;
use crate::index::BlockIndex;

/// Configuration for the incremental watcher.
#[derive(Debug, Clone)]
pub struct WatchConfig {
    /// Block size for indexing new/modified files.
    pub block_size: usize,
    /// Bloom filter bits per block.
    pub bloom_bits: usize,
    /// Polling interval for fallback mode (no inotify).
    pub poll_interval: Duration,
    /// Maximum file size to index (skip larger files).
    pub max_file_size: u64,
}

impl Default for WatchConfig {
    fn default() -> Self {
        Self {
            block_size: 4096,
            bloom_bits: 1024,
            poll_interval: Duration::from_secs(5),
            max_file_size: 256 * 1024 * 1024,
        }
    }
}

/// Tracks filesystem changes and incrementally updates a block index.
pub struct IncrementalWatch {
    root: PathBuf,
    config: WatchConfig,
    /// Known files and their last modification times.
    known_files: HashSet<PathBuf>,
    /// Last poll timestamp.
    last_poll: Instant,
}

/// Describes what changed since the last poll.
#[derive(Debug, Clone, Default)]
pub struct ChangeSet {
    /// Files that were added or modified.
    pub modified: Vec<PathBuf>,
    /// Files that were removed.
    pub removed: Vec<PathBuf>,
}

impl ChangeSet {
    /// Whether any changes were detected.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.modified.is_empty() && self.removed.is_empty()
    }

    /// Total number of changes.
    #[must_use]
    pub fn len(&self) -> usize {
        self.modified.len() + self.removed.len()
    }
}

impl IncrementalWatch {
    /// Create a watcher for the given directory.
    pub fn new(root: impl Into<PathBuf>, config: WatchConfig) -> Self {
        Self {
            root: root.into(),
            config,
            known_files: HashSet::new(),
            last_poll: Instant::now(),
        }
    }

    /// Poll for changes since the last call.
    ///
    /// Scans the directory tree and compares against known state.
    /// Returns the set of files that changed.
    ///
    /// # Errors
    ///
    /// Returns [`Error::Io`](crate::Error::Io) if the directory tree cannot be
    /// walked. This must fail loud: silently treating a failed walk as an empty
    /// tree would report every known file as removed and overwrite the known-file
    /// set with the empty set, corrupting every subsequent poll (a transient
    /// permission or I/O error would masquerade as a mass deletion).
    pub fn poll(&mut self) -> Result<ChangeSet> {
        let mut current_files = HashSet::new();
        let mut changes = ChangeSet::default();

        // Walk directory tree; a walk failure is propagated, never swallowed.
        let entries =
            walk_dir(&self.root, self.config.max_file_size).map_err(|source| crate::Error::Io {
                path: self.root.display().to_string(),
                source,
            })?;
        for path in entries {
            current_files.insert(path.clone());
            if !self.known_files.contains(&path) {
                changes.modified.push(path);
            }
        }

        // Detect removals
        for known in &self.known_files {
            if !current_files.contains(known) {
                changes.removed.push(known.clone());
            }
        }

        self.known_files = current_files;
        self.last_poll = Instant::now();
        Ok(changes)
    }

    /// Build a block index from only the modified files.
    ///
    /// Reads each modified file, concatenates them, and builds a single block
    /// index. The resulting index preserves the exact concatenated length; if
    /// the caller intends to merge it with an existing index, both indexes must
    /// end on block boundaries or the merge will return
    /// [`Error::TrailingPartialBlock`](crate::Error::TrailingPartialBlock).
    ///
    /// # Caveats
    ///
    /// Files are concatenated directly, so n-grams crossing file boundaries
    /// are indexed. This can only cause false positives (extra candidate
    /// blocks), never false negatives.
    ///
    /// # Errors
    ///
    /// Returns an error if file reading or index building fails.
    pub fn index_changes(&self, changes: &ChangeSet) -> Result<Option<BlockIndex>> {
        if changes.modified.is_empty() {
            return Ok(None);
        }

        let mut all_bytes = Vec::new();
        for path in &changes.modified {
            // Fail loud on read errors instead of silently skipping the file: a
            // skipped modified file is never indexed, so its content would be
            // silently missing from later queries (a recall bug).
            let data = std::fs::read(path).map_err(|source| crate::error::Error::Io {
                path: path.display().to_string(),
                source,
            })?;
            if all_bytes.len().saturating_add(data.len()) > MAX_TOTAL_INDEX_BYTES {
                return Err(crate::error::Error::DataTooLarge);
            }
            all_bytes.extend_from_slice(&data);
        }

        if all_bytes.is_empty() {
            return Ok(None);
        }

        let index = BlockIndexBuilder::new()
            .block_size(self.config.block_size)
            .bloom_bits(self.config.bloom_bits)
            .build(&all_bytes)?;

        Ok(Some(index))
    }

    /// Root directory being watched.
    #[must_use]
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Time since the last poll.
    #[must_use]
    pub fn since_last_poll(&self) -> Duration {
        self.last_poll.elapsed()
    }
}

/// Walk a directory recursively, returning paths to regular files under the size limit.
fn walk_dir(root: &Path, max_size: u64) -> std::io::Result<Vec<PathBuf>> {
    let mut files = Vec::new();
    walk_dir_inner(root, max_size, &mut files)?;
    Ok(files)
}

fn walk_dir_inner(dir: &Path, max_size: u64, out: &mut Vec<PathBuf>) -> std::io::Result<()> {
    let entries = std::fs::read_dir(dir)?;
    for entry in entries {
        let entry = entry?;
        let path = entry.path();
        let ft = entry.file_type()?;
        if ft.is_dir() {
            // Propagate subtree errors instead of `let _ =` swallowing them: a
            // permission-denied (or otherwise unreadable) subdirectory would
            // otherwise be silently skipped, so every file beneath it would be
            // silently absent from the index - a recall loss the operator never
            // sees (Law 10). Matches the fail-closed contract of index_changes.
            walk_dir_inner(&path, max_size, out)?;
        } else if ft.is_file() {
            // Same reasoning: a file whose metadata cannot be read is surfaced,
            // not silently dropped from indexing.
            let meta = entry.metadata()?;
            if meta.len() <= max_size {
                out.push(path);
            }
        }
    }
    Ok(())
}

#[cfg(all(test, not(miri)))]
#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
mod tests {
    use super::*;
    use std::fs;

    #[test]
    fn detects_new_file() {
        let dir = tempfile::tempdir().unwrap();
        let mut watcher = IncrementalWatch::new(dir.path(), WatchConfig::default());

        // Initial poll, empty
        let changes = watcher.poll().unwrap();
        assert!(changes.is_empty());

        // Add a file
        fs::write(dir.path().join("test.txt"), b"hello world").unwrap();
        let changes = watcher.poll().unwrap();
        assert_eq!(changes.modified.len(), 1);
        assert!(changes.removed.is_empty());
    }

    #[test]
    fn poll_on_unwalkable_root_fails_loud_and_preserves_state() {
        // A failed directory walk must surface as Error::Io, NOT be treated as an
        // empty tree. The old `if let Ok(entries)` swallowed the error, reported
        // every known file as removed, and overwrote known_files with the empty
        // set — a transient error masqueraded as a mass deletion and corrupted
        // all subsequent polls.
        let dir = tempfile::tempdir().unwrap();
        fs::write(dir.path().join("a.txt"), b"hello").unwrap();
        let mut watcher = IncrementalWatch::new(dir.path().to_path_buf(), WatchConfig::default());
        let first = watcher.poll().unwrap();
        assert_eq!(first.modified.len(), 1, "first poll registers the file");

        // Remove the whole root so the next walk fails.
        dir.close().unwrap();
        let err = watcher.poll().unwrap_err();
        assert!(
            matches!(err, crate::error::Error::Io { .. }),
            "failed walk must be Error::Io, got {err:?}"
        );
    }

    #[test]
    fn index_changes_propagates_read_error_instead_of_skipping() {
        // A modified file that cannot be read must surface as Error::Io, not be
        // silently dropped from the index (which would exclude its content from
        // every later query: a recall bug).
        let dir = tempfile::tempdir().unwrap();
        let watcher = IncrementalWatch::new(dir.path(), WatchConfig::default());

        let missing = dir.path().join("does_not_exist.bin");
        let changes = ChangeSet {
            modified: vec![missing.clone()],
            removed: Vec::new(),
        };

        let err = watcher.index_changes(&changes).unwrap_err();
        match err {
            crate::error::Error::Io { path, .. } => {
                assert!(
                    path.contains("does_not_exist.bin"),
                    "wrong path in error: {path}"
                );
            }
            other => panic!("expected Error::Io, got {other:?}"),
        }
    }

    #[test]
    fn detects_removed_file() {
        let dir = tempfile::tempdir().unwrap();
        fs::write(dir.path().join("test.txt"), b"hello").unwrap();

        let mut watcher = IncrementalWatch::new(dir.path(), WatchConfig::default());
        let _ = watcher.poll().unwrap(); // Register the file

        fs::remove_file(dir.path().join("test.txt")).unwrap();
        let changes = watcher.poll().unwrap();
        assert_eq!(changes.removed.len(), 1);
    }

    #[test]
    fn no_changes_on_second_poll() {
        let dir = tempfile::tempdir().unwrap();
        fs::write(dir.path().join("test.txt"), b"hello").unwrap();

        let mut watcher = IncrementalWatch::new(dir.path(), WatchConfig::default());
        let _ = watcher.poll().unwrap();
        let changes = watcher.poll().unwrap();
        assert!(changes.is_empty());
    }

    #[test]
    fn index_changes_builds_index() {
        let dir = tempfile::tempdir().unwrap();
        fs::write(dir.path().join("data.bin"), vec![0u8; 8192]).unwrap();

        let mut watcher = IncrementalWatch::new(dir.path(), WatchConfig::default());
        let changes = watcher.poll().unwrap();
        assert!(!changes.is_empty());

        let index = watcher.index_changes(&changes).unwrap();
        assert!(index.is_some());
    }
}