ObsiGate/backend/watcher.py

import asyncio
import logging
import os
import time
from collections.abc import Callable, Iterable
from pathlib import Path

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
from watchdog.observers.polling import PollingObserver

from backend.indexer import IGNORED_DIRS, SUPPORTED_EXTENSIONS

logger = logging.getLogger("obsigate.watcher")

# Filesystem types where inotify does NOT fire for changes made by other
# clients (the edit happens on another machine — e.g. Obsidian on Windows
# writing to an NFS/SMB export). For those mounts watchdog's native
# Observer silently misses everything and polling is the only option.
NETWORK_FSTYPES = {
    "nfs", "nfs4", "cifs", "smbfs", "smb2", "smb3",
    "fuse", "fuse.sshfs", "glusterfs", "9p", "virtiofs", "lustre",
}


def find_mount_fstype(resolved_path: str, mount_lines: Iterable[str]) -> str:
    """Return the fstype of the longest /proc/mounts entry covering the path.

    Pure function (mount_lines are text lines) so it can be unit-tested
    without /proc. Returns "" when no mount matches.
    """
    best_point = ""
    best_fstype = ""
    for line in mount_lines:
        parts = line.split()
        if len(parts) < 3:
            continue
        mount_point, fstype = parts[1], parts[2]
        # /proc/mounts escapes special chars in octal
        mount_point = mount_point.replace("\\040", " ").replace("\\011", "\t")
        root = mount_point.rstrip("/") or "/"
        if resolved_path == root or resolved_path.startswith(root + "/"):
            if len(mount_point) > len(best_point):
                best_point = mount_point
                best_fstype = fstype
    return best_fstype


def is_network_mount(path: str) -> bool:
    """True if *path* sits on a network/fuse filesystem (Linux only).

    inotify events are local to the client that made the write, so vaults
    shared over NFS/SMB never see Obsidian's edits. On those mounts we
    must fall back to PollingObserver, which detects changes by stat().
    """
    if os.name != "posix":
        return False
    try:
        with open("/proc/mounts", "r", encoding="utf-8") as fh:
            fstype = find_mount_fstype(os.path.realpath(path), fh)
    except OSError:
        return False
    return fstype in NETWORK_FSTYPES


class VaultEventHandler(FileSystemEventHandler):
    """Gestionnaire d'événements filesystem pour une vault Obsidian.

    Collecte les événements dans une queue asyncio pour traitement
    différé (debounce) par le VaultWatcher.
    """

    def __init__(
        self,
        vault_name: str,
        event_queue: asyncio.Queue,
        loop: asyncio.AbstractEventLoop,
    ):
        self.vault_name = vault_name
        self.event_queue = event_queue
        self.loop = loop

    def _is_relevant(self, path: str) -> bool:
        """Ignorer les fichiers non-markdown et les dossiers système."""
        p = Path(path)
        if any(part in IGNORED_DIRS for part in p.parts):
            return False
        suffix = p.suffix.lower()
        basename_lower = p.name.lower()
        return suffix in SUPPORTED_EXTENSIONS or basename_lower in ("dockerfile", "makefile", "cmakelists.txt")

    def _enqueue(self, event_type: str, src: str, dest: str | None = None):
        """Thread-safe : envoyer l'événement vers l'event loop asyncio."""
        event = {
            'type': event_type,
            'vault': self.vault_name,
            'src': src,
            'dest': dest,
            'timestamp': time.time(),
        }
        self.loop.call_soon_threadsafe(self.event_queue.put_nowait, event)

    def on_created(self, event):
        if not event.is_directory and self._is_relevant(event.src_path):
            self._enqueue('created', event.src_path)

    def on_modified(self, event):
        if not event.is_directory and self._is_relevant(event.src_path):
            self._enqueue('modified', event.src_path)

    def on_deleted(self, event):
        if not event.is_directory and self._is_relevant(event.src_path):
            self._enqueue('deleted', event.src_path)

    def on_moved(self, event):
        if not event.is_directory:
            src_ok = self._is_relevant(event.src_path)
            dest_ok = self._is_relevant(event.dest_path)
            if src_ok or dest_ok:
                self._enqueue('moved', event.src_path, event.dest_path)


class VaultWatcher:
    """Gestionnaire principal de surveillance de toutes les vaults.

    Démarre un Observer watchdog par vault et traite les événements
    avec debounce pour éviter les réindexations en rafale.
    """

    def __init__(
        self,
        on_file_change: Callable,
        debounce_seconds: float = 2.0,
        use_polling: bool = False,
        polling_interval: float = 5.0,
    ):
        self.on_file_change = on_file_change
        self.debounce_seconds = debounce_seconds
        self.use_polling = use_polling
        self.polling_interval = polling_interval
        self.observers: dict[str, object] = {}
        self.event_queue: asyncio.Queue = asyncio.Queue()
        self._processor_task: asyncio.Task | None = None
        self._running = False

    async def start(self, vaults: dict[str, str]):
        """Démarrer la surveillance de toutes les vaults.

        Args:
            vaults: dict { vault_name: vault_path }
        """
        self._running = True
        loop = asyncio.get_event_loop()

        for vault_name, vault_path in vaults.items():
            await self._watch_vault(vault_name, vault_path, loop)

        self._processor_task = asyncio.create_task(self._process_events())
        logger.info(f"VaultWatcher started on {len(vaults)} vault(s)")

    async def _watch_vault(
        self,
        vault_name: str,
        vault_path: str,
        loop: asyncio.AbstractEventLoop,
    ):
        """Créer et démarrer un observer pour une vault.

        Le mode (natif inotify vs polling) est choisi par vault :
        les mounts réseau (NFS/SMB/fuse) sont surveillés en polling car
        inotify ne voit pas les écritures faites depuis d'autres clients.
        """
        path = Path(vault_path)
        if not path.exists():
            logger.warning(f"Vault '{vault_name}' path not found: {vault_path}")
            return

        handler = VaultEventHandler(vault_name, self.event_queue, loop)

        polling = self.use_polling or is_network_mount(vault_path)
        if polling and not self.use_polling:
            logger.info(
                f"Vault '{vault_name}' is on a network mount — "
                "using polling watcher (inotify cannot see remote edits)"
            )

        ObserverClass = PollingObserver if polling else Observer
        try:
            observer = ObserverClass(
                timeout=self.polling_interval if polling else 1
            )
            observer.schedule(handler, str(path), recursive=True)
            observer.daemon = True
            observer.start()
            self.observers[vault_name] = observer
            mode = "polling" if polling else "native"
            logger.info(f"Watching ({mode}): {vault_name} -> {vault_path}")
        except Exception as e:
            logger.error(f"Failed to start watcher for '{vault_name}': {e}")
            # Fallback to polling if native fails
            if not self.use_polling:
                logger.info(f"Falling back to polling for '{vault_name}'")
                try:
                    observer = PollingObserver(timeout=self.polling_interval)
                    observer.schedule(handler, str(path), recursive=True)
                    observer.daemon = True
                    observer.start()
                    self.observers[vault_name] = observer
                    logger.info(f"Watching (polling fallback): {vault_name} -> {vault_path}")
                except Exception as e2:
                    logger.error(f"Polling fallback also failed for '{vault_name}': {e2}")

    async def add_vault(self, vault_name: str, vault_path: str):
        """Ajouter une nouvelle vault à surveiller sans redémarrage."""
        loop = asyncio.get_event_loop()
        await self._watch_vault(vault_name, vault_path, loop)

    async def remove_vault(self, vault_name: str):
        """Arrêter la surveillance d'une vault."""
        if vault_name in self.observers:
            observer = self.observers[vault_name]
            try:
                observer.stop()   # type: ignore[attr-defined]
                observer.join(timeout=5)  # type: ignore[attr-defined]
            except Exception as e:
                logger.warning(f"Error stopping observer for '{vault_name}': {e}")
            del self.observers[vault_name]
            logger.info(f"Stopped watching: {vault_name}")

    async def _process_events(self):
        """Traiter les événements filesystem avec debounce.

        Accumule les événements pendant debounce_seconds puis appelle
        le callback une seule fois avec la liste consolidée.
        """
        pending: dict[str, dict] = {}

        while self._running:
            try:
                event = await asyncio.wait_for(
                    self.event_queue.get(),
                    timeout=self.debounce_seconds,
                )
                # Dédupliquer : garder seulement le dernier événement par fichier
                key = event.get('dest') or event['src']
                pending[key] = event

            except asyncio.TimeoutError:
                if pending:
                    events_to_process = list(pending.values())
                    pending.clear()
                    await self._dispatch(events_to_process)

            except asyncio.CancelledError:
                break

            except Exception as e:
                logger.error(f"Event processor error: {e}")

    async def _dispatch(self, events: list[dict]):
        """Appeler le callback avec les événements consolidés."""
        try:
            await self.on_file_change(events)
        except Exception as e:
            logger.error(f"Change callback error: {e}")

    async def stop(self):
        """Arrêter proprement tous les observers."""
        self._running = False
        if self._processor_task:
            self._processor_task.cancel()
            try:
                await self._processor_task
            except asyncio.CancelledError:
                pass
        for name, observer in self.observers.items():
            try:
                observer.stop()
            except Exception as e:
                logger.warning(f"Error stopping observer '{name}': {e}")
        for observer in self.observers.values():
            try:
                observer.join(timeout=5)
            except Exception:  # nosec B110 — best-effort shutdown, ignore failures
                pass
        self.observers.clear()
        logger.info("VaultWatcher stopped")