mnote/rust/crates/mnote-web/src/local_folder_watcher_registry.rs

use crate::document_buffer_store::BufferStore;
use crate::routes::{
    local_markdown_conflict_detection_key, local_workspace_id_from_root_uri,
    refresh_local_search_index_for_change_path_with_store,
    refresh_local_search_index_if_scheduled_due_with_store,
};
use control_plane::{ControlPlaneStore, SqliteControlPlaneStore};
use notify::event::ModifyKind;
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::{broadcast, mpsc, oneshot};
use tokio::time::MissedTickBehavior;

#[derive(Clone)]
pub struct LocalFolderWatcherRegistry {
    inner: Arc<LocalFolderWatcherRegistryInner>,
    buffer_store: BufferStore,
    control_plane: Arc<dyn ControlPlaneStore>,
}

impl std::fmt::Debug for LocalFolderWatcherRegistry {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("LocalFolderWatcherRegistry")
            .finish_non_exhaustive()
    }
}

impl LocalFolderWatcherRegistry {
    pub fn new(buffer_store: BufferStore, control_plane: Arc<SqliteControlPlaneStore>) -> Self {
        Self {
            inner: Arc::new(LocalFolderWatcherRegistryInner {
                entries: Mutex::new(HashMap::new()),
            }),
            buffer_store,
            control_plane,
        }
    }

    pub(crate) fn subscribe(
        &self,
        canonical_root: &Path,
    ) -> Result<LocalFolderWatcherSubscription, String> {
        let key = canonical_root_uri(canonical_root);
        let buffer_store = self.buffer_store.clone();
        let control_plane = self.control_plane.clone();
        let channel =
            self.inner
                .get_or_create_channel(&key, canonical_root, buffer_store, control_plane)?;
        channel.subscriber_count.fetch_add(1, Ordering::SeqCst);
        Ok(LocalFolderWatcherSubscription {
            receiver: channel.sender.subscribe(),
            guard: LocalFolderWatcherSubscriptionGuard {
                registry: Arc::downgrade(&self.inner),
                key,
                channel,
            },
        })
    }

    #[cfg(test)]
    pub fn active_watcher_count(&self) -> usize {
        self.inner.entries.lock().expect("registry lock").len()
    }
}

struct LocalFolderWatcherRegistryInner {
    entries: Mutex<HashMap<String, Arc<LocalFolderWatchChannel>>>,
}

impl LocalFolderWatcherRegistryInner {
    fn get_or_create_channel(
        &self,
        key: &str,
        canonical_root: &Path,
        buffer_store: BufferStore,
        control_plane: Arc<dyn ControlPlaneStore>,
    ) -> Result<Arc<LocalFolderWatchChannel>, String> {
        if let Some(existing) = self
            .entries
            .lock()
            .expect("registry lock")
            .get(key)
            .cloned()
        {
            return Ok(existing);
        }

        let channel = Arc::new(LocalFolderWatchChannel::new(
            key.to_string(),
            spawn_local_folder_watcher(
                key,
                canonical_root.to_path_buf(),
                buffer_store,
                control_plane,
            )?,
        ));

        let mut entries = self.entries.lock().expect("registry lock");
        if let Some(existing) = entries.get(key).cloned() {
            return Ok(existing);
        }
        entries.insert(key.to_string(), channel.clone());
        Ok(channel)
    }

    fn remove_if_idle(&self, key: &str, channel: &Arc<LocalFolderWatchChannel>) {
        let mut entries = self.entries.lock().expect("registry lock");
        let should_remove = entries
            .get(key)
            .map(|current| {
                Arc::ptr_eq(current, channel)
                    && channel.subscriber_count.load(Ordering::SeqCst) == 0
            })
            .unwrap_or(false);
        if should_remove {
            entries.remove(key);
            channel.shutdown();
        }
    }
}

struct LocalFolderWatchChannel {
    sender: broadcast::Sender<Value>,
    subscriber_count: AtomicUsize,
    shutdown_tx: Mutex<Option<oneshot::Sender<()>>>,
}

impl LocalFolderWatchChannel {
    fn new(_root_uri: String, parts: (broadcast::Sender<Value>, oneshot::Sender<()>)) -> Self {
        Self {
            sender: parts.0,
            subscriber_count: AtomicUsize::new(0),
            shutdown_tx: Mutex::new(Some(parts.1)),
        }
    }

    fn shutdown(&self) {
        if let Some(sender) = self.shutdown_tx.lock().expect("shutdown lock").take() {
            let _ = sender.send(());
        }
    }
}

pub(crate) struct LocalFolderWatcherSubscription {
    pub(crate) receiver: broadcast::Receiver<Value>,
    guard: LocalFolderWatcherSubscriptionGuard,
}

impl LocalFolderWatcherSubscription {
    pub(crate) fn root_uri(&self) -> &str {
        self.guard.key.as_str()
    }
}

struct LocalFolderWatcherSubscriptionGuard {
    registry: Weak<LocalFolderWatcherRegistryInner>,
    key: String,
    channel: Arc<LocalFolderWatchChannel>,
}

impl Drop for LocalFolderWatcherSubscriptionGuard {
    fn drop(&mut self) {
        let previous = self.channel.subscriber_count.fetch_sub(1, Ordering::SeqCst);
        if previous != 1 {
            return;
        }
        if let Some(registry) = self.registry.upgrade() {
            registry.remove_if_idle(&self.key, &self.channel);
        }
    }
}

fn spawn_local_folder_watcher(
    root_uri: &str,
    canonical_root: PathBuf,
    buffer_store: BufferStore,
    control_plane: Arc<dyn ControlPlaneStore>,
) -> Result<(broadcast::Sender<Value>, oneshot::Sender<()>), String> {
    let (event_sender, mut event_receiver) = mpsc::unbounded_channel::<notify::Result<Event>>();
    let mut watcher = RecommendedWatcher::new(
        move |result| {
            let _ = event_sender.send(result);
        },
        Config::default(),
    )
    .map_err(|error| format!("本地文件事件监听启动失败: {error}"))?;
    watcher
        .watch(&canonical_root, RecursiveMode::Recursive)
        .map_err(|error| format!("本地文件夹监听失败: {error}"))?;

    let (sender, _) = broadcast::channel::<Value>(256);
    let (shutdown_tx, mut shutdown_rx) = oneshot::channel::<()>();
    let sender_for_task = sender.clone();
    let root_uri_for_task = root_uri.to_string();
    let buffer_store_for_task = buffer_store.clone();
    let control_plane_for_task = control_plane.clone();
    tokio::spawn(async move {
        let _watcher = watcher;
        let mut index_schedule_tick = tokio::time::interval(Duration::from_secs(60));
        index_schedule_tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
        loop {
            tokio::select! {
                _ = &mut shutdown_rx => {
                    return;
                }
                _ = index_schedule_tick.tick() => {
                    refresh_local_search_index_for_schedule(
                        control_plane_for_task.as_ref(),
                        &canonical_root,
                        &root_uri_for_task,
                    );
                }
                maybe_result = event_receiver.recv() => {
                    let Some(result) = maybe_result else {
                        return;
                    };
                    let Ok(event) = result else {
                        continue;
                    };
                    if !should_emit_event_kind(&event.kind) {
                        continue;
                    }
                    for path in event.paths {
                        if !is_local_search_index_path(&path) {
                            continue;
                        }
                        let Some(relative_path) = relative_path_string(&canonical_root, &path) else {
                            continue;
                        };
                        refresh_local_search_index_for_event(
                            control_plane_for_task.as_ref(),
                            &canonical_root,
                            &root_uri_for_task,
                            &relative_path,
                        );
                        let mut observed_file_version: Option<String> = None;
                        let mut buffer_file_version: Option<String> = None;
                        let mut last_write_intent_id: Option<String> = None;
                        let mut last_save_operation_id: Option<String> = None;
                        let mut self_write_echo = false;
                        let event_kind = format!("{:?}", event.kind);
                        let document_id = if is_markdown_path(&path) {
                            let document_id =
                                format!("local-md:{}", encode_local_id_segment(&relative_path));
                            // 外部 Markdown 变更→更新 BufferStore
                            if let Ok(ws_id) =
                                local_workspace_id_from_root_uri(&root_uri_for_task)
                            {
                                let ws_path =
                                    crate::document_buffer_store::build_local_folder_workspace_path(
                                        &ws_id,
                                        &root_uri_for_task,
                                        &relative_path,
                                        &document_id,
                                    );
                                if event_kind.contains("Remove") {
                                    if let Some(buffer) =
                                        buffer_store_for_task.mark_local_folder_markdown_deleted(
                                            &ws_id,
                                            &root_uri_for_task,
                                            &relative_path,
                                            &document_id,
                                        )
                                    {
                                        buffer_file_version = buffer.file_version.clone();
                                        last_write_intent_id =
                                            buffer.last_write_intent_id.clone();
                                        last_save_operation_id =
                                            buffer.last_save_operation_id.clone();
                                    }
                                } else {
                                    observed_file_version =
                                        local_markdown_conflict_detection_key(&document_id, &path)
                                            .ok();
                                    if let Some(outcome) = buffer_store_for_task
                                        .mark_external_modified_if_version_changed_with_outcome(
                                            &ws_path,
                                            observed_file_version.clone(),
                                            Some("external-editor".into()),
                                        )
                                    {
                                        buffer_file_version = outcome.buffer.file_version.clone();
                                        last_write_intent_id =
                                            outcome.buffer.last_write_intent_id.clone();
                                        last_save_operation_id =
                                            outcome.buffer.last_save_operation_id.clone();
                                        self_write_echo = outcome.self_write_echo;
                                    }
                                }
                            }
                            document_id
                        } else {
                            String::new()
                        };

                        let payload = json!({
                            "sourceKind": "local_folder",
                            "rootUri": root_uri_for_task,
                            "relativePath": relative_path,
                            "documentId": document_id,
                            "eventKind": event_kind,
                            "revision": event_revision(&path),
                            "observedFileVersion": observed_file_version,
                            "bufferFileVersion": buffer_file_version,
                            "lastWriteIntentId": last_write_intent_id,
                            "lastSaveOperationId": last_save_operation_id,
                            "selfWriteEcho": self_write_echo,
                        });
                        let _ = sender_for_task.send(payload);
                    }
                }
            }
        }
    });

    Ok((sender, shutdown_tx))
}

fn refresh_local_search_index_for_event(
    control_plane: &dyn ControlPlaneStore,
    root: &Path,
    root_uri: &str,
    relative_path: &str,
) {
    let Ok(workspace_id) = local_workspace_id_from_root_uri(root_uri) else {
        return;
    };
    let _ = refresh_local_search_index_for_change_path_with_store(
        control_plane,
        root,
        root_uri,
        &workspace_id,
        relative_path,
    );
}

fn refresh_local_search_index_for_schedule(
    control_plane: &dyn ControlPlaneStore,
    root: &Path,
    root_uri: &str,
) {
    let Ok(workspace_id) = local_workspace_id_from_root_uri(root_uri) else {
        return;
    };
    let _ = refresh_local_search_index_if_scheduled_due_with_store(
        control_plane,
        root,
        root_uri,
        &workspace_id,
    );
}

fn canonical_root_uri(root: &Path) -> String {
    format!("file://{}", root.display())
}

fn relative_path_string(root: &Path, path: &Path) -> Option<String> {
    path.strip_prefix(root)
        .ok()
        .map(|relative| relative.to_string_lossy().replace('\\', "/"))
        .filter(|relative| !relative.is_empty())
}

fn is_markdown_path(path: &Path) -> bool {
    path.extension()
        .and_then(|extension| extension.to_str())
        .map(|extension| {
            extension.eq_ignore_ascii_case("md") || extension.eq_ignore_ascii_case("markdown")
        })
        .unwrap_or(false)
}

fn is_local_search_index_path(path: &Path) -> bool {
    if path.components().any(|component| {
        component
            .as_os_str()
            .to_str()
            .map(|value| value == ".mnote")
            .unwrap_or(false)
    }) {
        return false;
    }
    if is_markdown_path(path) {
        return true;
    }
    let file_name = path
        .file_name()
        .and_then(|value| value.to_str())
        .unwrap_or_default()
        .to_lowercase();
    if file_name.ends_with(".mindmap.json") {
        return true;
    }
    let extension = path
        .extension()
        .and_then(|extension| extension.to_str())
        .unwrap_or_default()
        .to_lowercase();
    matches!(
        extension.as_str(),
        "doc"
            | "docx"
            | "odt"
            | "ppt"
            | "pptx"
            | "odp"
            | "xls"
            | "xlsx"
            | "ods"
            | "pdf"
            | "png"
            | "jpg"
            | "jpeg"
            | "gif"
            | "webp"
            | "svg"
            | "txt"
            | "log"
            | "csv"
            | "json"
            | "zip"
    )
}

fn should_emit_event_kind(kind: &EventKind) -> bool {
    match kind {
        EventKind::Create(_) | EventKind::Remove(_) => true,
        EventKind::Modify(modify_kind) => matches!(
            modify_kind,
            ModifyKind::Any | ModifyKind::Data(_) | ModifyKind::Name(_) | ModifyKind::Metadata(_)
        ),
        _ => false,
    }
}

fn encode_local_id_segment(value: &str) -> String {
    let mut encoded = String::with_capacity(value.len());
    for byte in value.as_bytes() {
        let character = *byte as char;
        if character.is_ascii_alphanumeric() || matches!(character, '.' | '_' | '-') {
            encoded.push(character);
        } else {
            encoded.push('~');
            encoded.push_str(&format!("{byte:02X}"));
        }
    }
    encoded
}

fn event_revision(path: &Path) -> u128 {
    std::fs::metadata(path)
        .ok()
        .and_then(|metadata| metadata.modified().ok())
        .map(system_time_ms)
        .unwrap_or_else(|| system_time_ms(SystemTime::now()))
}

fn system_time_ms(time: SystemTime) -> u128 {
    time.duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_millis())
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::{
        is_local_search_index_path, refresh_local_search_index_for_event, should_emit_event_kind,
        LocalFolderWatcherRegistry,
    };
    use crate::document_buffer_store::BufferStore;
    use crate::routes::write_local_index_settings;
    use control_plane::SqliteControlPlaneStore;
    use notify::event::{AccessKind, CreateKind, DataChange, ModifyKind};
    use notify::EventKind;
    use std::sync::Arc;

    fn test_root(name: &str) -> std::path::PathBuf {
        let root = std::env::temp_dir().join(format!(
            "mnote-local-folder-watcher-{name}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|duration| duration.as_nanos())
                .unwrap_or(0)
        ));
        std::fs::create_dir_all(&root).expect("create temp root");
        root
    }

    #[tokio::test]
    async fn same_root_subscribers_share_single_watcher() {
        let control_plane =
            Arc::new(SqliteControlPlaneStore::in_memory().expect("init control plane"));
        let registry = LocalFolderWatcherRegistry::new(BufferStore::new(), control_plane);
        let root = test_root("shared");

        let first = registry.subscribe(&root).expect("first subscription");
        let second = registry.subscribe(&root).expect("second subscription");

        assert_eq!(registry.active_watcher_count(), 1);

        drop(first);
        tokio::task::yield_now().await;
        assert_eq!(registry.active_watcher_count(), 1);

        drop(second);
        tokio::task::yield_now().await;
        assert_eq!(registry.active_watcher_count(), 0);

        let _ = std::fs::remove_dir_all(root);
    }

    #[tokio::test]
    async fn different_roots_create_independent_watchers() {
        let control_plane =
            Arc::new(SqliteControlPlaneStore::in_memory().expect("init control plane"));
        let registry = LocalFolderWatcherRegistry::new(BufferStore::new(), control_plane);
        let first_root = test_root("first");
        let second_root = test_root("second");

        let first = registry
            .subscribe(&first_root)
            .expect("first root subscription");
        let second = registry
            .subscribe(&second_root)
            .expect("second root subscription");

        assert_eq!(registry.active_watcher_count(), 2);

        drop(first);
        drop(second);
        tokio::task::yield_now().await;
        assert_eq!(registry.active_watcher_count(), 0);

        let _ = std::fs::remove_dir_all(first_root);
        let _ = std::fs::remove_dir_all(second_root);
    }

    #[test]
    fn event_kind_filter_ignores_access_events() {
        assert!(should_emit_event_kind(&EventKind::Create(CreateKind::File)));
        assert!(should_emit_event_kind(&EventKind::Modify(
            ModifyKind::Data(DataChange::Content,)
        )));
        assert!(!should_emit_event_kind(&EventKind::Access(
            AccessKind::Read
        )));
    }

    #[test]
    fn watcher_index_path_filter_accepts_markdown_and_resources() {
        assert!(is_local_search_index_path(&std::path::Path::new(
            "docs/page.md"
        )));
        assert!(is_local_search_index_path(&std::path::Path::new(
            "maps/idea.mindmap.json"
        )));
        assert!(is_local_search_index_path(&std::path::Path::new(
            "office/report.xlsx"
        )));
        assert!(is_local_search_index_path(&std::path::Path::new(
            "attachments/report.pdf"
        )));
        assert!(is_local_search_index_path(&std::path::Path::new(
            "attachments/image.png"
        )));
        assert!(!is_local_search_index_path(&std::path::Path::new(
            ".mnote/page-options.json"
        )));
        assert!(!is_local_search_index_path(&std::path::Path::new(
            "docs/.DS_Store"
        )));
    }

    #[test]
    fn watcher_event_refreshes_local_search_index_path() {
        let root = test_root("search-index-event");
        std::fs::create_dir_all(root.join("docs")).expect("create docs");
        std::fs::write(root.join("README.md"), "# Home\n").expect("write home");
        std::fs::write(
            root.join("docs").join("watched.md"),
            "---\ntitle: Watched\n---\n# Watched\nWatcherToken\n",
        )
        .expect("write watched");

        let root_uri = format!("file://{}", root.display());
        let control_plane = SqliteControlPlaneStore::in_memory().expect("init control plane");
        write_local_index_settings(&root, &[String::from(".")], None, None, None, Some(true))
            .expect("enable run-on-change indexing");
        refresh_local_search_index_for_event(&control_plane, &root, &root_uri, "docs/watched.md");

        let index_path = root.join(".mnote").join("index").join("search-index.json");
        let index = std::fs::read_to_string(&index_path).expect("index exists");
        assert!(index.contains("docs/watched.md"));
        assert!(index.contains("WatcherToken"));

        let _ = std::fs::remove_dir_all(root);
    }
}