smartfs/rust/crates/smartfs-core/src/watch.rs

use crate::FsOps;
use notify::{Config, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use smartfs_protocol::{IpcEvent, WatchEvent};
use std::collections::HashMap;
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::SystemTime;

/// Manages file watchers, emitting events as IPC events to stdout.
pub struct WatchManager {
    watchers: Arc<Mutex<HashMap<String, RecommendedWatcher>>>,
}

impl WatchManager {
    pub fn new() -> Self {
        Self {
            watchers: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    pub fn add_watch(
        &self,
        id: String,
        path: &str,
        recursive: bool,
    ) -> Result<(), String> {
        let watch_id = id.clone();
        let mode = if recursive {
            RecursiveMode::Recursive
        } else {
            RecursiveMode::NonRecursive
        };

        let (tx, rx) = std::sync::mpsc::channel::<notify::Result<notify::Event>>();

        let mut watcher = RecommendedWatcher::new(tx, Config::default())
            .map_err(|e| format!("watch create: {}", e))?;

        watcher
            .watch(Path::new(path), mode)
            .map_err(|e| format!("watch path: {}", e))?;

        // Spawn a thread to read events and write IPC events to stdout
        let watch_id_clone = watch_id.clone();
        std::thread::spawn(move || {
            for event in rx {
                match event {
                    Ok(ev) => {
                        let event_type = match ev.kind {
                            EventKind::Create(_) => "add",
                            EventKind::Modify(_) => "change",
                            EventKind::Remove(_) => "delete",
                            _ => continue,
                        };

                        for ev_path in &ev.paths {
                            let stats = if event_type != "delete" {
                                FsOps::stat_path(ev_path).ok()
                            } else {
                                None
                            };

                            let watch_event = WatchEvent {
                                event_type: event_type.to_string(),
                                path: ev_path.to_string_lossy().into_owned(),
                                timestamp: {
                                    let d = SystemTime::now()
                                        .duration_since(SystemTime::UNIX_EPOCH)
                                        .unwrap_or_default();
                                    format!("{}.{:03}Z", d.as_secs(), d.subsec_millis())
                                },
                                stats,
                            };

                            let ipc_event = IpcEvent {
                                event: format!("watch:{}", watch_id_clone),
                                data: serde_json::to_value(&watch_event).unwrap_or_default(),
                            };

                            if let Ok(json) = serde_json::to_string(&ipc_event) {
                                // Write to stdout (IPC channel)
                                println!("{}", json);
                            }
                        }
                    }
                    Err(e) => {
                        eprintln!("watch error: {}", e);
                    }
                }
            }
        });

        // Store the watcher to keep it alive
        self.watchers
            .lock()
            .map_err(|e| format!("lock: {}", e))?
            .insert(id, watcher);

        Ok(())
    }

    pub fn remove_all(&self) -> Result<(), String> {
        self.watchers
            .lock()
            .map_err(|e| format!("lock: {}", e))?
            .clear();
        Ok(())
    }
}