stash/src/watcher.rs

use crate::{AppProject, error::StashError, scan_projects};
use notify::{Event, EventKind, RecursiveMode, Watcher};
use std::{path::PathBuf, time::Duration};
use tokio::sync::{broadcast, mpsc};
use tracing::warn;

const EVENT_DEBOUNCE: Duration = Duration::from_millis(150);

pub fn start(root: PathBuf, updates: broadcast::Sender<Vec<AppProject>>) -> Result<(), StashError> {
    let mut previous_projects = scan_projects(&root);
    let (event_tx, mut event_rx) = mpsc::unbounded_channel();
    let mut watcher = notify::recommended_watcher(move |result| {
        let _ = event_tx.send(result);
    })?;
    watcher.watch(&root, RecursiveMode::Recursive)?;

    tokio::spawn(async move {
        let _watcher = watcher;

        while let Some(event) = event_rx.recv().await {
            if !should_rescan(&event) {
                continue;
            }

            // Editors and build tools often emit a burst of file events for one change.
            tokio::time::sleep(EVENT_DEBOUNCE).await;
            while event_rx.try_recv().is_ok() {}

            let projects = scan_projects(&root);
            if projects == previous_projects {
                continue;
            }

            previous_projects = projects.clone();
            let _ = updates.send(projects);
        }
    });

    Ok(())
}

fn should_rescan(event: &notify::Result<Event>) -> bool {
    match event {
        Ok(event) => !matches!(event.kind, EventKind::Access(_)),
        Err(error) => {
            warn!("文件监听事件出错，将重新扫描项目: {error}");
            true
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use notify::{
        Event,
        event::{AccessKind, CreateKind, EventKind},
    };
    use std::{
        fs,
        path::{Path, PathBuf},
        sync::atomic::{AtomicU64, Ordering},
        time::Duration,
    };

    static NEXT_TEMP_ID: AtomicU64 = AtomicU64::new(0);

    struct TestDir {
        path: PathBuf,
    }

    impl TestDir {
        fn new() -> Self {
            let id = NEXT_TEMP_ID.fetch_add(1, Ordering::Relaxed);
            let path = std::env::temp_dir()
                .join(format!("stash-watcher-test-{}-{id}", std::process::id()));
            fs::create_dir_all(&path).expect("应能创建测试目录");
            Self { path }
        }

        fn path(&self) -> &Path {
            &self.path
        }
    }

    impl Drop for TestDir {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.path);
        }
    }

    #[test]
    fn access_events_do_not_trigger_project_scans() {
        let access = Event::new(EventKind::Access(AccessKind::Read));
        let create = Event::new(EventKind::Create(CreateKind::Folder));

        assert!(!should_rescan(&Ok(access)));
        assert!(should_rescan(&Ok(create)));
    }

    #[tokio::test]
    async fn broadcasts_when_a_project_is_added() {
        let root = TestDir::new();
        let (updates, _) = broadcast::channel(8);
        let mut receiver = updates.subscribe();

        start(root.path().to_path_buf(), updates).expect("应能启动文件监听");

        let project = root.path().join("new-project");
        fs::create_dir_all(&project).expect("应能创建项目目录");
        fs::write(project.join("index.html"), "<!doctype html>").expect("应能创建入口文件");

        let projects = tokio::time::timeout(Duration::from_secs(5), receiver.recv())
            .await
            .expect("应在超时前收到项目更新")
            .expect("项目更新通道不应关闭");

        assert_eq!(projects.len(), 1);
        assert_eq!(projects[0].name, "new-project");
        assert_eq!(projects[0].route_path, "/new-project/");
    }
}