lore_engine/engine/

watcher.rs

//! File watcher — monitors a vault directory for external `.md` changes and
//! syncs them into the DB and graph. Uses debounced `notify` events (500ms).

use super::{db, file_loader, vault_ops};
use super::error::LoreError;
use crate::state::AppState;
use notify::RecursiveMode;
use notify_debouncer_mini::{new_debouncer, DebouncedEvent, Debouncer};
use serde::Serialize;
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

/// A page that was created or modified by an external change.
#[derive(Debug, Clone, Serialize)]
pub struct ChangedPage {
    pub slug: String,
    pub title: String,
}

/// Payload emitted when the watcher detects vault changes.
#[derive(Debug, Clone, Serialize)]
pub struct VaultChangedPayload {
    pub changed: Vec<ChangedPage>,
    pub deleted: Vec<String>,
    pub page_count: usize,
    pub link_count: usize,
}

/// Callback type for vault change notifications.
/// The bridge layer sets this to emit a Qt signal.
pub type OnVaultChanged = Arc<dyn Fn(VaultChangedPayload) + Send + Sync>;

/// Start watching a vault directory for .md file changes.
/// Returns the debouncer handle — drop it to stop watching.
pub fn start_watcher(
    vault_path: &Path,
    state: Arc<AppState>,
    on_change: OnVaultChanged,
) -> Result<Debouncer<notify::RecommendedWatcher>, LoreError> {
    let vault = vault_path.to_path_buf();

    let mut debouncer = new_debouncer(
        Duration::from_millis(500),
        move |result: Result<Vec<DebouncedEvent>, notify::Error>| {
            match result {
                Ok(events) => {
                    if let Err(e) = handle_events(&state, &on_change, &vault, events) {
                        log::error!("Watcher event handler error: {e}");
                    }
                }
                Err(e) => log::error!("Watcher error: {e}"),
            }
        },
    )
    .map_err(|e| LoreError::WatcherError(e.to_string()))?;

    debouncer
        .watcher()
        .watch(vault_path, RecursiveMode::Recursive)
        .map_err(|e| LoreError::WatcherError(e.to_string()))?;

    log::info!("File watcher started for {}", vault_path.display());
    Ok(debouncer)
}

fn handle_events(
    state: &AppState,
    on_change: &OnVaultChanged,
    vault_path: &Path,
    events: Vec<DebouncedEvent>,
) -> Result<(), LoreError> {
    let mut paths: HashSet<PathBuf> = HashSet::new();
    for event in &events {
        // Only process paths within the vault directory
        if event.path.starts_with(vault_path) {
            paths.insert(event.path.clone());
        } else {
            log::debug!("Ignoring event outside vault: {}", event.path.display());
        }
    }

    if paths.is_empty() {
        return Ok(());
    }

    let payload = {
        let db_lock = state.db.lock()?;
        let conn = match db_lock.as_ref() {
            Some(c) => c,
            None => return Ok(()),
        };

        let existing_hashes = db::get_all_hashes(conn)?;
        let known_slugs = vault_ops::get_known_slugs_from(conn)?;
        let mut changed_pages: Vec<ChangedPage> = Vec::new();
        let mut deleted_slugs: Vec<String> = Vec::new();

        conn.execute_batch("BEGIN TRANSACTION")?;

        for path in &paths {
            if path.exists() {
                handle_file_change(conn, vault_path, path, &existing_hashes, &known_slugs, &mut changed_pages)?;
            } else {
                handle_file_delete(conn, vault_path, path, &mut deleted_slugs)?;
            }
        }

        conn.execute_batch("COMMIT")?;

        if changed_pages.is_empty() && deleted_slugs.is_empty() {
            return Ok(());
        }

        let mut graph = state.graph.lock()?;

        for cp in &changed_pages {
            graph.upsert_node(&cp.slug, &cp.title, false);

            let targets = db::get_outgoing_targets(conn, &cp.slug)?;
            graph.set_outgoing_edges(&cp.slug, &targets);

            for target in &targets {
                if graph.degree(target) == 0 {
                    if let Some(p) = db::get_page(conn, target)? {
                        graph.upsert_node(target, &p.title, p.is_placeholder);
                    }
                }
            }
        }

        for slug in &deleted_slugs {
            if let Some(p) = db::get_page(conn, slug)? {
                graph.upsert_node(slug, &p.title, true);
                graph.set_outgoing_edges(slug, &[]);
            } else {
                graph.remove_node(slug);
            }
        }

        let orphans = graph.cleanup_orphaned_placeholders();
        for slug in &orphans {
            if let Err(e) = db::delete_page(conn, slug) {
                log::warn!("Failed to clean up orphan placeholder '{slug}': {e}");
            }
        }

        let (page_count, link_count) = graph.stats();

        VaultChangedPayload {
            changed: changed_pages,
            deleted: deleted_slugs,
            page_count,
            link_count,
        }
    };

    log::info!(
        "Watcher: {} changed, {} deleted",
        payload.changed.len(),
        payload.deleted.len()
    );

    on_change(payload);

    Ok(())
}

fn handle_file_change(
    conn: &rusqlite::Connection,
    vault_path: &Path,
    path: &Path,
    existing_hashes: &std::collections::HashMap<String, String>,
    known_slugs: &HashSet<String>,
    changed: &mut Vec<ChangedPage>,
) -> Result<(), LoreError> {
    let scanned = match file_loader::scan_single_file(vault_path, path) {
        Some(f) => f,
        None => return Ok(()),
    };

    // Self-write detection: skip if hash matches AND page already exists in DB
    let page_exists = db::get_page(conn, &scanned.slug)?.is_some();
    if page_exists {
        if let Some(old_hash) = existing_hashes.get(&scanned.slug) {
            if old_hash == &scanned.content_hash {
                // Hash matches — DB is already up to date. Still report as changed
                // so in-memory state (graph, page list) gets refreshed.
                changed.push(ChangedPage {
                    slug: scanned.slug,
                    title: scanned.title,
                });
                return Ok(());
            }
        }
    }

    db::upsert_page(
        conn,
        &scanned.slug,
        &scanned.title,
        Some(&scanned.relative_path),
        Some(&scanned.content_hash),
        false,
    )?;

    // Use shared link-sync logic (with fuzzy resolution)
    vault_ops::sync_page_links(conn, &scanned.slug, &scanned.title, &scanned.content, known_slugs)?;

    changed.push(ChangedPage {
        slug: scanned.slug,
        title: scanned.title,
    });

    Ok(())
}

fn handle_file_delete(
    conn: &rusqlite::Connection,
    vault_path: &Path,
    path: &Path,
    deleted: &mut Vec<String>,
) -> Result<(), LoreError> {
    let relative = match path.strip_prefix(vault_path) {
        Ok(r) => r.to_string_lossy().replace('\\', "/"),
        Err(_) => return Ok(()),
    };
    let slug = file_loader::slug_from_path(&relative);

    let page = match db::get_page(conn, &slug)? {
        Some(p) => p,
        None => return Ok(()),
    };

    let backlinks = db::get_backlinks(conn, &slug)?;
    if backlinks.is_empty() {
        db::delete_page(conn, &slug)?;
    } else {
        db::upsert_page(conn, &slug, &page.title, None, None, true)?;
        db::delete_links_from(conn, &slug)?;
        db::delete_fts(conn, &slug)?;
    }

    deleted.push(slug);
    Ok(())
}