rtb-config 0.6.5

//! File-system hot reload for [`crate::Config`].
//!
//! Opt-in via the `hot-reload` Cargo feature. Contract: spec
//! [0014, rtb-config hot reload](https://gitlab.com/phpboyscout/rust-tool-base/-/wikis/specs/0014-rtb-config-hot-reload).

#![cfg(feature = "hot-reload")]

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::Duration;

use notify::{EventKind, RecursiveMode};
use notify_debouncer_full::{new_debouncer, DebouncedEvent};
use serde::de::DeserializeOwned;

use crate::config::Config;
use crate::error::ConfigError;

/// Debounce window for coalescing file-system events. 250 ms is slow
/// enough for editors that save via rename-and-replace (vim, VS Code)
/// and fast enough that users perceive reload as instantaneous.
const DEBOUNCE_MS: u64 = 250;

/// Owns the background watcher for a [`Config`]. Dropping the handle
/// stops the watcher; the thread exits on its next tick.
#[must_use = "dropping `WatchHandle` immediately stops the file watcher"]
pub struct WatchHandle {
    stop: Arc<AtomicBool>,
    thread: Option<JoinHandle<()>>,
}

impl std::fmt::Debug for WatchHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("WatchHandle")
            .field("stopped", &self.stop.load(Ordering::Relaxed))
            .finish_non_exhaustive()
    }
}

impl Drop for WatchHandle {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
        if let Some(h) = self.thread.take() {
            // Best-effort join. Worker thread exits within one
            // debouncer tick (~100ms) once `stop` is set.
            let _ = h.join();
        }
    }
}

impl<C> Config<C>
where
    C: DeserializeOwned + Send + Sync + 'static,
{
    /// Start a background watcher for every path registered via
    /// [`crate::ConfigBuilder::user_file`]. Debounced change events
    /// call [`Config::reload`] — successful reloads wake every
    /// [`Config::subscribe`] receiver.
    ///
    /// The watcher follows the file's directory rather than the file,
    /// so in-place writes, rename-replace saves, a file created after
    /// the watch started, and a symlink swap (as Kubernetes does for a
    /// mounted `ConfigMap`) are all seen.
    ///
    /// # Errors
    ///
    /// - [`ConfigError::Watch`] when no user-file paths were
    ///   registered, when a path's directory does not exist, or when
    ///   the underlying `notify` backend fails to create a watcher.
    pub fn watch_files(&self) -> Result<WatchHandle, ConfigError> {
        if self.sources.files.is_empty() {
            return Err(ConfigError::Watch("no user files registered".into()));
        }
        let mut targets = self
            .sources
            .files
            .iter()
            .map(|p| WatchedFile::new(p))
            .collect::<Result<Vec<_>, _>>()?;

        let (tx, rx) = mpsc::channel();
        let mut debouncer = new_debouncer(Duration::from_millis(DEBOUNCE_MS), None, tx)
            .map_err(|e| ConfigError::Watch(format!("debouncer: {e}")))?;
        let mut watched_dirs = BTreeSet::new();
        for dir in targets.iter().flat_map(WatchedFile::dirs) {
            if watched_dirs.insert(dir.clone()) {
                debouncer
                    .watch(&dir, RecursiveMode::NonRecursive)
                    .map_err(|e| ConfigError::Watch(format!("watch {}: {e}", dir.display())))?;
            }
        }

        let this = self.clone();
        let stop = Arc::new(AtomicBool::new(false));
        let stop_thread = Arc::clone(&stop);
        let thread = std::thread::Builder::new()
            .name("rtb-config-watcher".into())
            .spawn(move || {
                while !stop_thread.load(Ordering::Relaxed) {
                    match rx.recv_timeout(Duration::from_millis(100)) {
                        Ok(Ok(events)) => {
                            if !should_reload(&events, &mut targets) {
                                continue;
                            }
                            // A symlink swap can move the real file into a
                            // directory nobody is watching yet.
                            for dir in targets.iter().flat_map(WatchedFile::dirs) {
                                if watched_dirs.insert(dir.clone()) {
                                    let _ = debouncer.watch(&dir, RecursiveMode::NonRecursive);
                                }
                            }
                            // A bad write shouldn't kill the watcher; the
                            // previous snapshot stays live.
                            let _ = this.reload();
                        }
                        Ok(Err(_notify_errs)) => { /* transient notify errors — keep watching */ }
                        Err(mpsc::RecvTimeoutError::Timeout) => {}
                        Err(mpsc::RecvTimeoutError::Disconnected) => break,
                    }
                }
            })
            .map_err(|e| ConfigError::Watch(format!("spawn watcher thread: {e}")))?;

        Ok(WatchHandle { stop, thread: Some(thread) })
    }
}

/// A registered user file, addressed through its canonical directory so
/// event paths compare equal on every backend.
struct WatchedFile {
    path: PathBuf,
    resolved: Option<PathBuf>,
}

impl WatchedFile {
    fn new(registered: &Path) -> Result<Self, ConfigError> {
        let name = registered.file_name().ok_or_else(|| {
            ConfigError::Watch(format!("watch {}: not a file path", registered.display()))
        })?;
        let dir = match registered.parent() {
            Some(d) if !d.as_os_str().is_empty() => d,
            _ => Path::new("."),
        };
        let dir = dir
            .canonicalize()
            .map_err(|e| ConfigError::Watch(format!("watch {}: {e}", dir.display())))?;
        let path = dir.join(name);
        let resolved = path.canonicalize().ok();
        Ok(Self { path, resolved })
    }

    fn dirs(&self) -> impl Iterator<Item = PathBuf> + '_ {
        [Some(&self.path), self.resolved.as_ref()]
            .into_iter()
            .flatten()
            .filter_map(|p| p.parent().map(Path::to_path_buf))
    }

    fn is_named_by(&self, event_path: &Path) -> bool {
        event_path == self.path || self.resolved.as_deref() == Some(event_path)
    }

    /// Re-resolves symlinks; true when the file now lives somewhere else.
    fn refresh_resolved(&mut self) -> bool {
        let now = self.path.canonicalize().ok();
        let moved = now != self.resolved;
        self.resolved = now;
        moved
    }
}

/// Access events are ignored because `reload` itself opens every file,
/// and on inotify that open would otherwise trigger the next reload.
fn should_reload(events: &[DebouncedEvent], targets: &mut [WatchedFile]) -> bool {
    let changes: Vec<&DebouncedEvent> =
        events.iter().filter(|e| !matches!(e.kind, EventKind::Access(_))).collect();
    if changes.is_empty() {
        return false;
    }
    let named = changes
        .iter()
        .flat_map(|e| e.paths.iter())
        .any(|p| targets.iter().any(|t| t.is_named_by(p)));
    let mut moved = false;
    for t in targets.iter_mut() {
        moved |= t.refresh_resolved();
    }
    named || moved
}