zng_ext_fs_watcher/

lib.rs

#![doc(html_favicon_url = "https://zng-ui.github.io/res/zng-logo-icon.png")]
#![doc(html_logo_url = "https://zng-ui.github.io/res/zng-logo.png")]
//!
//! File system events and service.
//!
//! # Events
//!
//! Events this extension provides.
//!
//! * [`FS_CHANGES_EVENT`]
//!
//! # Services
//!
//! Services this extension provides.
//!
//! * [`WATCHER`]
//!
//! # Crate
//!
#![doc = include_str!(concat!("../", std::env!("CARGO_PKG_README")))]
// suppress nag about very simple boxed closure signatures.
#![expect(clippy::type_complexity)]
#![warn(unused_extern_crates)]
#![warn(missing_docs)]

use std::{
    any::Any,
    fmt, fs,
    io::{self, Write as _},
    ops,
    path::{Path, PathBuf},
    sync::Arc,
    time::Duration,
};

use parking_lot::Mutex;
use path_absolutize::Absolutize;
use zng_app::{
    DInstant, INSTANT,
    event::event_args,
    handler::{Handler, HandlerExt as _},
};
use zng_clone_move::clmv;
use zng_handle::Handle;
use zng_txt::Txt;
use zng_unit::TimeUnits;
use zng_var::{AnyVar, BoxAnyVarValue, Var, VarHandle, VarValue, WeakAnyVar, var};

mod service;
use service::*;

mod lock;
use lock::*;

/// File system watcher service.
///
/// This is mostly a wrapper around the [`notify`](https://docs.rs/notify) crate, integrating it with events and variables.
pub struct WATCHER;
impl WATCHER {
    /// Gets a read-write variable that defines interval awaited between each [`FS_CHANGES_EVENT`]. If
    /// a watched path is constantly changing an event will be emitted every elapse of this interval,
    /// the event args will contain a list of all the changes observed during the interval.
    ///
    /// Note that the first event notifies immediately, only subsequent events within this interval are debounced.
    ///
    /// Is `100.ms()` by default.
    pub fn debounce(&self) -> Var<Duration> {
        WATCHER_SV.read().debounce.clone()
    }

    /// Gets a read-write variable that defines interval awaited between each [`sync`] write.
    ///
    /// Is `100.ms()` by default.
    ///
    /// Note that each [`sync`] is debounced in isolation and the first write happens immediately.
    ///
    /// [`sync`]: WATCHER::sync
    pub fn sync_debounce(&self) -> Var<Duration> {
        WATCHER_SV.read().sync_debounce.clone()
    }

    /// Gets a read-write variable that defines the fallback poll watcher interval.
    ///
    /// When an efficient watcher cannot be used a poll watcher fallback is used, the poll watcher reads
    /// the directory or path every elapse of this interval. The poll watcher is also used for paths that
    /// do not exist yet, that is also affected by this interval.
    ///
    /// Is `1.secs()` by default.
    pub fn poll_interval(&self) -> Var<Duration> {
        WATCHER_SV.read().poll_interval.clone()
    }

    /// Maximum time the service keeps the process alive to finish pending IO operations when the app shuts down.
    ///
    /// Is 1 minute by default.
    pub fn shutdown_timeout(&self) -> Var<Duration> {
        WATCHER_SV.read().shutdown_timeout.clone()
    }

    /// Enable [file change events] for the `file`.
    ///
    /// Returns a handle that will stop the file watch when dropped, if there is no other active handler for the same file.
    ///
    /// Note that this is implemented by actually watching the parent directory and filtering the events, this is done
    /// to ensure the watcher survives operations that remove the file and then move another file to the same path.
    ///
    /// See [`watch_dir`] for more details.
    ///
    /// [`watch_dir`]: WATCHER::watch_dir
    /// [file change events]: FS_CHANGES_EVENT
    pub fn watch(&self, file: impl Into<PathBuf>) -> WatcherHandle {
        WATCHER_SV.write().watch(file.into())
    }

    /// Enable [file change events] for files inside `dir`, also include inner directories if `recursive` is `true`.
    ///
    /// Returns a handle that will stop the dir watch when dropped, if there is no other active handler for the same directory.
    ///
    /// The directory will be watched using an OS specific efficient watcher provided by the [`notify`](https://docs.rs/notify) crate. If there is
    /// any error creating the watcher, such as if the directory does not exist yet a slower polling watcher will retry periodically    
    /// until the efficient watcher can be created or the handle is dropped.
    ///
    /// [file change events]: FS_CHANGES_EVENT
    pub fn watch_dir(&self, dir: impl Into<PathBuf>, recursive: bool) -> WatcherHandle {
        WATCHER_SV.write().watch_dir(dir.into(), recursive)
    }

    /// Read a file into a variable, the `init` value will start the variable and the `read` closure will be called
    /// once immediately and every time the file changes, if the closure returns `Some(O)` the variable updates with the new value.
    ///
    /// Dropping the variable drops the read watch. The `read` closure is non-blocking, it is called in a [`task::wait`]
    /// background thread.
    ///
    /// [`task::wait`]: zng_task::wait
    pub fn read<O: VarValue>(
        &self,
        file: impl Into<PathBuf>,
        init: O,
        mut read: impl FnMut(io::Result<WatchFile>) -> Option<O> + Send + 'static,
    ) -> Var<O> {
        let out = var(init);
        self.read_impl(
            absolutize_or_pass(file.into()),
            out.as_any().downgrade(),
            Box::new(move |r| read(r).map(BoxAnyVarValue::new)),
        );
        out.read_only()
    }
    fn read_impl(
        &self,
        file: PathBuf,
        out: WeakAnyVar,
        read: Box<dyn FnMut(io::Result<WatchFile>) -> Option<BoxAnyVarValue> + Send + 'static>,
    ) {
        let _handle = WATCHER.watch(file.clone());
        struct Data {
            file: PathBuf,
            read: Mutex<Box<dyn FnMut(io::Result<WatchFile>) -> Option<BoxAnyVarValue> + Send + 'static>>,
        }
        let data = Arc::new(Data {
            file,
            read: Mutex::new(read),
        });
        fn spawn_read(data: &Arc<Data>, out: AnyVar) {
            zng_task::spawn_wait(clmv!(data, || {
                if let Some(r) = data.read.lock()(WatchFile::open(&data.file)) {
                    // did read update, set output
                    out.modify(clmv!(data, |o| {
                        if o.set(r) {
                            // tag to avoid write back when `read_impl` is used in `write_impl`
                            o.push_tag(WatcherSyncWriteNote(data.file.clone()));
                        }
                    }));
                }
            }));
        }
        if data.file.exists() {
            // read initial value
            spawn_read(&data, out.upgrade().unwrap());
        }
        FS_CHANGES_EVENT
            .hook(move |args| {
                let _hold = &_handle;

                if let Some(out) = out.upgrade() {
                    if has_sync_changes(args, &data.file) {
                        // has changes, spawn read
                        spawn_read(&data, out);
                    }
                    true
                } else {
                    // out var dropped
                    false
                }
            })
            .perm();
    }

    /// Same operation as [`read`] but also tracks the operation status in a second var.
    ///
    /// The status variable is set to [`WatcherReadStatus::reading`] as soon as `read` starts and
    /// is set to [`WatcherReadStatus::idle`] when read returns. If read returns a value the status
    /// only updates to idle  when the new value is available on the var, or because read the same value.
    ///
    /// [`read`]: Self::read
    pub fn read_status<O, S, E>(
        &self,
        file: impl Into<PathBuf>,
        init: O,
        mut read: impl FnMut(io::Result<WatchFile>) -> Result<Option<O>, E> + Send + 'static,
    ) -> (Var<O>, Var<S>)
    where
        O: VarValue,
        S: WatcherReadStatus<E>,
        E: Any,
    {
        let out = var(init);
        let status = var(S::idle());
        self.read_status_impl(
            absolutize_or_pass(file.into()),
            out.as_any().downgrade(),
            status.as_any().clone(),
            Box::new(move |r| match read(r) {
                Ok(Some(r)) => Ok(Some(BoxAnyVarValue::new(r))),
                Ok(None) => Ok(None),
                Err(e) => Err(Box::new(e)),
            }),
            AnyWatcherReadStatus::new::<E, S>(),
        );
        (out.read_only(), status.read_only())
    }
    fn read_status_impl(
        &self,
        file: PathBuf,
        out: WeakAnyVar,
        status: AnyVar,
        read: Box<dyn FnMut(io::Result<WatchFile>) -> Result<Option<BoxAnyVarValue>, Box<dyn Any>> + Send + 'static>,
        status_mtds: AnyWatcherReadStatus,
    ) {
        let _handle = WATCHER.watch(file.clone());
        struct Data {
            file: PathBuf,
            read: Mutex<Box<dyn FnMut(io::Result<WatchFile>) -> Result<Option<BoxAnyVarValue>, Box<dyn Any>> + Send + 'static>>,
            status: AnyVar,
            status_mtds: AnyWatcherReadStatus,
        }
        let data = Arc::new(Data {
            file,
            read: Mutex::new(read),
            status,
            status_mtds,
        });
        fn spawn_read(data: &Arc<Data>, out: AnyVar) {
            data.status.set((data.status_mtds.reading)());
            zng_task::spawn_wait(clmv!(data, || {
                let mut read = match data.read.try_lock() {
                    Some(l) => l,
                    None => return, // another spawn either already has read lock on the file or will acquire it
                };
                match read(WatchFile::open(&data.file)) {
                    Ok(u) => {
                        if let Some(r) = u {
                            // did read update, set output
                            out.modify(clmv!(data, |o| {
                                if o.set(r) {
                                    // tag to avoid write back when `read_status_impl` is used in `sync_status_impl`
                                    o.push_tag(WatcherSyncWriteNote(data.file.clone()));
                                }
                            }));
                        }
                        data.status.set((data.status_mtds.idle)());
                    }
                    Err(e) => {
                        // read error
                        data.status.set((data.status_mtds.read_error)(Box::new(e)));
                    }
                }
            }));
        }
        if data.file.exists() {
            // read initial value
            spawn_read(&data, out.upgrade().unwrap());
        }
        FS_CHANGES_EVENT
            .hook(move |args| {
                let _hold = &_handle;
                if let Some(out) = out.upgrade() {
                    if has_sync_changes(args, &data.file) {
                        // has changes, spawn read
                        spawn_read(&data, out);
                    }
                    true
                } else {
                    // out dropped
                    false
                }
            })
            .perm();
    }

    /// Read a directory into a variable, the `init` value will start the variable and the `read` closure will be called
    /// once immediately and every time any changes happen inside the dir, if the closure returns `Some(O)` the variable updates with the new value.
    ///
    /// The `read` closure parameter is a directory walker from the [`walkdir`](https://docs.rs/walkdir) crate.
    ///
    /// The directory walker is pre-configured to skip the `dir` itself and to have a max-depth of 1 if not `recursive`, these configs can.
    ///
    /// Dropping the variable drops the read watch. The `read` closure is non-blocking, it is called in a [`task::wait`]
    /// background thread.
    ///
    /// [`task::wait`]: zng_task::wait
    pub fn read_dir<O: VarValue>(
        &self,
        dir: impl Into<PathBuf>,
        recursive: bool,
        init: O,
        mut read: impl FnMut(walkdir::WalkDir) -> Option<O> + Send + 'static,
    ) -> Var<O> {
        let out = var(init);
        self.read_dir_impl(
            absolutize_or_pass(dir.into()),
            out.as_any().downgrade(),
            recursive,
            Box::new(move |r| read(r).map(BoxAnyVarValue::new)),
        );
        out.read_only()
    }
    fn read_dir_impl(
        &self,
        dir: PathBuf,
        out: WeakAnyVar,
        recursive: bool,
        read: Box<dyn FnMut(walkdir::WalkDir) -> Option<BoxAnyVarValue> + Send + 'static>,
    ) {
        let _handle = WATCHER.watch_dir(dir.clone(), recursive);
        struct Data {
            dir: PathBuf,
            read: Mutex<Box<dyn FnMut(walkdir::WalkDir) -> Option<BoxAnyVarValue> + Send + 'static>>,
        }
        let data = Arc::new(Data {
            dir,
            read: Mutex::new(read),
        });
        fn spawn_read(data: &Arc<Data>, recursive: bool, out: AnyVar) {
            zng_task::spawn_wait(clmv!(data, || {
                let dir = if recursive {
                    walkdir::WalkDir::new(&data.dir).min_depth(1)
                } else {
                    walkdir::WalkDir::new(&data.dir).min_depth(1).max_depth(1)
                };
                if let Some(r) = data.read.lock()(dir) {
                    // did read update, set output
                    out.modify(clmv!(data, |o| {
                        if o.set(r) {
                            // tag just for parity with `read`
                            o.push_tag(WatcherSyncWriteNote(data.dir.clone()));
                        }
                    }));
                }
            }));
        }
        if data.dir.exists() {
            // read initial value
            spawn_read(&data, recursive, out.upgrade().unwrap());
        }
        FS_CHANGES_EVENT
            .hook(move |args| {
                let _hold = &_handle;
                if let Some(out) = out.upgrade() {
                    if has_sync_changes(args, &data.dir) {
                        // has changes, spawn read
                        spawn_read(&data, recursive, out);
                    }
                    true
                } else {
                    // out var dropped
                    false
                }
            })
            .perm();
    }

    /// Same operation as [`read_dir`] but also tracks the operation status in a second var.
    ///
    /// The status variable is set to [`WatcherReadStatus::reading`] as soon as `read` starts and
    /// is set to [`WatcherReadStatus::idle`] when read returns. If read returns a value the status
    /// only updates to idle when the new value is available on the var, or because read the same value.
    ///
    /// [`read_dir`]: Self::read_dir
    pub fn read_dir_status<O, S, E>(
        &self,
        dir: impl Into<PathBuf>,
        recursive: bool,
        init: O,
        mut read: impl FnMut(walkdir::WalkDir) -> Result<Option<O>, E> + Send + 'static,
    ) -> (Var<O>, Var<S>)
    where
        O: VarValue,
        S: WatcherReadStatus<E>,
        E: Any,
    {
        let out = var(init);
        let status = var(S::idle());

        self.read_dir_status_impl(
            absolutize_or_pass(dir.into()),
            out.as_any().downgrade(),
            status.as_any().clone(),
            recursive,
            Box::new(move |r| match read(r) {
                Ok(Some(r)) => Ok(Some(BoxAnyVarValue::new(r))),
                Ok(None) => Ok(None),
                Err(e) => Err(Box::new(e)),
            }),
            AnyWatcherReadStatus::new::<E, S>(),
        );
        (out.read_only(), status.read_only())
    }
    fn read_dir_status_impl(
        &self,
        dir: PathBuf,
        out: WeakAnyVar,
        status: AnyVar,
        recursive: bool,
        read: Box<dyn FnMut(walkdir::WalkDir) -> Result<Option<BoxAnyVarValue>, Box<dyn Any>> + Send + 'static>,
        status_mtds: AnyWatcherReadStatus,
    ) {
        let _handle = WATCHER.watch_dir(dir.clone(), recursive);
        struct Data {
            dir: PathBuf,
            read: Mutex<Box<dyn FnMut(walkdir::WalkDir) -> Result<Option<BoxAnyVarValue>, Box<dyn Any>> + Send + 'static>>,
            status: AnyVar,
            status_mtds: AnyWatcherReadStatus,
        }
        let data = Arc::new(Data {
            dir,
            read: Mutex::new(read),
            status,
            status_mtds,
        });
        fn spawn_read(data: &Arc<Data>, recursive: bool, out: AnyVar) {
            data.status.set((data.status_mtds.reading)());
            zng_task::spawn_wait(clmv!(data, || {
                let dir = if recursive {
                    walkdir::WalkDir::new(&data.dir).min_depth(1)
                } else {
                    walkdir::WalkDir::new(&data.dir).min_depth(1).max_depth(1)
                };
                match data.read.lock()(dir) {
                    Ok(u) => {
                        if let Some(r) = u {
                            // did read update, set output
                            out.modify(clmv!(data, |o| {
                                if o.set(r) {
                                    // tag just for parity with `read_status`
                                    o.push_tag(WatcherSyncWriteNote(data.dir.clone()));
                                }
                            }));
                        }
                        data.status.set((data.status_mtds.idle)());
                    }
                    Err(e) => {
                        data.status.set((data.status_mtds.read_error)(Box::new(e)));
                    }
                }
            }));
        }
        if data.dir.exists() {
            spawn_read(&data, recursive, out.upgrade().unwrap());
        }
        FS_CHANGES_EVENT
            .hook(move |args| {
                let _hold = &_handle;
                if let Some(out) = out.upgrade() {
                    if has_sync_changes(args, &data.dir) {
                        // has changes, spawn read
                        spawn_read(&data, recursive, out);
                    }
                    true
                } else {
                    // output var dropped
                    false
                }
            })
            .perm();
    }

    /// Bind a file with a variable, the `file` will be `read` when it changes and be `write` when the variable changes,
    /// writes are only applied on success and will not cause a `read` on the same sync task. The `init` value is used to
    /// create the variable, if the `file` exists it will be `read` once at the beginning.
    ///
    /// Dropping the variable drops the read watch. The `read` and `write` closures are non-blocking, they are called in a [`task::wait`]
    /// background thread.
    ///
    /// # Sync
    ///
    /// The file synchronization ensures that the file is only actually modified when write is finished by writing
    /// to a temporary file and committing a replace only if the write succeeded. The file is write-locked for the duration
    /// of `write` call, but the contents are not touched until commit. See [`WriteFile`] for more details.
    ///
    /// The service blocks on app exit until all writes commit or cancel. See [`WATCHER::shutdown_timeout`] for
    /// more details.
    ///
    /// ## Read Errors
    ///
    /// Not-found errors are handled by the watcher by calling `write` using the current variable value, other read errors
    /// are passed to `read`. If `read` returns a value for an error the `write` closure is called to override the file,
    /// otherwise only the variable is set and this variable update does not cause a `write`.
    ///
    /// ## Write Errors
    ///
    /// If `write` fails the file is not touched and the temporary file is removed, if the file path
    /// does not exit all missing parent folders and the file will be created automatically before the `write`
    /// call.
    ///
    /// Note that [`WriteFile::commit`] must be called to flush the temporary file and attempt to rename
    /// it, if the file is dropped without commit it will cancel and log an error, you must call [`WriteFile::cancel`]
    /// to correctly avoid writing.
    ///
    /// If the cleanup after commit fails the error is logged and ignored.
    ///
    /// If write fails to even create the file and/or acquire a write lock on it this error is the input for
    /// the `write` closure.
    ///
    /// ## Error Handling
    ///
    /// You can call services or set other variables from inside the `read` and `write` closures, this can be
    /// used to get a signal out to handle the error, perhaps drop the sync var (to stop watching), alert the user that the
    /// file is out of sync and initiate some sort of recovery routine.
    ///
    /// If the file synchronization is not important you can just ignore it, the watcher will try again
    /// on the next variable or file update.
    ///
    /// ## Status
    ///
    /// Note that `read` and `write` run in background task threads, so if you are tracking the operation
    /// status in a separate variable you may end-up with synchronization bugs between the status variable
    /// and the actual result variable, you can use [`sync_status`] to implement racing-free status tracking.
    ///
    /// [`sync_status`]: Self::sync_status
    /// [`task::wait`]: zng_task::wait
    pub fn sync<O: VarValue>(
        &self,
        file: impl Into<PathBuf>,
        init: O,
        mut read: impl FnMut(io::Result<WatchFile>) -> Option<O> + Send + 'static,
        mut write: impl FnMut(O, io::Result<WriteFile>) + Send + 'static,
    ) -> Var<O> {
        let out = var(init);
        self.sync_impl(
            absolutize_or_pass(file.into()),
            out.as_any().clone(),
            Box::new(move |r| read(r).map(BoxAnyVarValue::new)),
            Box::new(move |o, r| write(o.downcast().unwrap(), r)),
        );
        out
    }
    fn sync_impl(
        &self,
        file: PathBuf,
        out: AnyVar,
        read: Box<dyn FnMut(io::Result<WatchFile>) -> Option<BoxAnyVarValue> + Send + 'static>,
        write: Box<dyn FnMut(BoxAnyVarValue, io::Result<WriteFile>) + Send + 'static>,
    ) {
        self.read_impl(file.clone(), out.downgrade(), read);

        struct DataWrite {
            f: Box<dyn FnMut(BoxAnyVarValue, io::Result<WriteFile>) + Send + 'static>,
            last: DInstant,
        }
        struct Data {
            note: Arc<WatcherSyncWriteNote>,
            write: Mutex<DataWrite>,
            write_value: Mutex<Option<BoxAnyVarValue>>,
            debounce_flush: Arc<SyncFlushLock>,
        }
        let data = Arc::new(Data {
            note: Arc::new(WatcherSyncWriteNote(file)),
            write: Mutex::new(DataWrite {
                f: write,
                last: DInstant::EPOCH,
            }),
            write_value: Mutex::new(None),
            debounce_flush: Arc::new(SyncFlushLock::new()),
        });
        WATCHER_SV.write().push_sync_flush(&data.debounce_flush);

        out.hook(move |a| {
            // skip if value was set from `read_impl`
            for tag in a.downcast_tags::<WatcherSyncWriteNote>() {
                if *tag == *data.note {
                    return true;
                }
            }

            // spawn write
            let mut d = data.write_value.lock();
            let is_running = d.is_some() || data.write.try_lock().is_none();
            *d = Some(a.value().clone_boxed());
            if is_running {
                return true;
            }
            zng_task::spawn_wait(clmv!(data, || {
                // wait other write tasks
                let mut write = data.write.lock();
                // wait debounce of APP.on_deinit signal, holds _guard to block on_deinit so the process is not killed while writing
                let _guard = data
                    .debounce_flush
                    .begin_write(WATCHER_SV.read().sync_debounce.get().saturating_sub(write.last.elapsed()));

                // tag events during write so that `read_impl` can avoid reading this change back
                let _tag = WATCHER.annotate(data.note.clone());
                while let Some(value) = data.write_value.lock().take() {
                    (write.f)(value, WriteFile::open(data.note.0.clone()));
                }

                write.last = INSTANT.now();
            }));

            true
        })
        .perm();
    }

    /// Same operation as [`sync`] but also tracks the operation status in a second var.
    ///
    /// The status variable is set to [`WatcherReadStatus::reading`] as soon as `read` starts and
    /// is set to [`WatcherReadStatus::idle`] when read returns. If read returns a value the status
    /// only updates to idle when the new sync value is available, or because read the same value.
    ///
    /// The status variable is set to [`WatcherSyncStatus::writing`] as soon as it updates and
    /// is set to [`WatcherReadStatus::idle`] only when the new sync value is available, either
    /// by update or because read the same value.
    ///
    /// [`sync`]: Self::sync
    pub fn sync_status<O, S, ER, EW>(
        &self,
        file: impl Into<PathBuf>,
        init: O,
        mut read: impl FnMut(io::Result<WatchFile>) -> Result<Option<O>, ER> + Send + 'static,
        mut write: impl FnMut(O, io::Result<WriteFile>) -> Result<(), EW> + Send + 'static,
    ) -> (Var<O>, Var<S>)
    where
        O: VarValue,
        S: WatcherSyncStatus<ER, EW>,
        ER: Any,
        EW: Any,
    {
        let out = var(init);
        let status = var(S::idle());
        self.sync_status_impl(
            absolutize_or_pass(file.into()),
            out.as_any().clone(),
            status.as_any().clone(),
            Box::new(move |r| match read(r) {
                Ok(Some(r)) => Ok(Some(BoxAnyVarValue::new(r))),
                Ok(None) => Ok(None),
                Err(e) => Err(Box::new(e)),
            }),
            AnyWatcherReadStatus::new::<ER, S>(),
            Box::new(move |o, r| match write(o.downcast().unwrap(), r) {
                Ok(()) => Ok(()),
                Err(e) => Err(Box::new(e)),
            }),
            AnyWatcherWriteStatus::new::<ER, EW, S>(),
        );
        (out, status.read_only())
    }
    #[allow(clippy::too_many_arguments)]
    fn sync_status_impl(
        &self,
        file: PathBuf,
        out: AnyVar,
        status: AnyVar,
        read: Box<dyn FnMut(io::Result<WatchFile>) -> Result<Option<BoxAnyVarValue>, Box<dyn Any>> + Send + 'static>,
        status_mtds: AnyWatcherReadStatus,
        write: Box<dyn FnMut(BoxAnyVarValue, io::Result<WriteFile>) -> Result<(), Box<dyn Any>> + Send + 'static>,
        status_w_mtds: AnyWatcherWriteStatus,
    ) {
        self.read_status_impl(file.clone(), out.downgrade(), status.clone(), read, status_mtds);

        struct DataWrite {
            f: Box<dyn FnMut(BoxAnyVarValue, io::Result<WriteFile>) -> Result<(), Box<dyn Any>> + Send + 'static>,
            last: DInstant,
        }
        struct Data {
            note: Arc<WatcherSyncWriteNote>,
            write: Mutex<DataWrite>,
            write_value: Mutex<Option<BoxAnyVarValue>>,
            debounce_flush: Arc<SyncFlushLock>,
            status: AnyVar,
            status_w_mtds: AnyWatcherWriteStatus,
        }
        let data = Arc::new(Data {
            note: Arc::new(WatcherSyncWriteNote(file)),
            write: Mutex::new(DataWrite {
                f: write,
                last: DInstant::EPOCH,
            }),
            write_value: Mutex::new(None),
            debounce_flush: Arc::new(SyncFlushLock::new()),
            status,
            status_w_mtds,
        });
        WATCHER_SV.write().push_sync_flush(&data.debounce_flush);

        out.hook(move |a| {
            // skip if value was set from `read_impl`
            for tag in a.downcast_tags::<WatcherSyncWriteNote>() {
                if *tag == *data.note {
                    return true;
                }
            }

            // spawn write
            let mut d = data.write_value.lock();
            let is_running = d.is_some() || data.write.try_lock().is_none();
            *d = Some(a.value().clone_boxed());
            if is_running {
                return true;
            }
            data.status.set((data.status_w_mtds.writing)());
            zng_task::spawn_wait(clmv!(data, || {
                // wait other write tasks
                let mut write = data.write.lock();
                // wait debounce of APP.on_deinit signal, holds _guard to block on_deinit so the process is not killed while writing
                let debounce = WATCHER_SV.read().sync_debounce.get().saturating_sub(write.last.elapsed());
                let _guard = data.debounce_flush.begin_write(debounce);

                // tag events during write so that `read_impl` can avoid reading this change back
                let _tag = WATCHER.annotate(data.note.clone());

                while let Some(value) = data.write_value.lock().take() {
                    match (write.f)(value, WriteFile::open(data.note.0.clone())) {
                        Ok(()) => data.status.set((data.status_w_mtds.idle)()),
                        Err(e) => data.status.set((data.status_w_mtds.write_error)(e)),
                    }
                }

                write.last = INSTANT.now();
            }));

            true
        })
        .perm();
    }

    /// Watch `file` and calls `handler` every time it changes.
    ///
    /// Note that the `handler` is blocking, use [`async_hn!`] and [`task::wait`] to run IO without
    /// blocking the app.
    ///
    /// If `ignore_propagation` is `true` also calls `handler` when another handler marked as handled.
    ///
    /// [`async_hn!`]: macro@zng_app::handler::async_hn
    /// [`task::wait`]: zng_task::wait
    pub fn on_file_changed(&self, file: impl Into<PathBuf>, ignore_propagation: bool, handler: Handler<FsChangesArgs>) -> VarHandle {
        let file = absolutize_or_pass(file.into());
        let handle = self.watch(file.clone());
        FS_CHANGES_EVENT.on_event(
            ignore_propagation,
            handler.filtered(move |args| {
                let _handle = &handle;
                args.events_for_path(&file).next().is_some()
            }),
        )
    }

    /// Watch `dir` and calls `handler` every time something inside it changes.
    ///
    /// Note that the `handler` is blocking, use [`async_hn!`] and [`task::wait`] to run IO without
    /// blocking the app.
    ///
    /// If `ignore_propagation` is `true` also calls `handler` when another handler marked as handled.
    ///
    /// [`async_hn!`]: macro@zng_app::handler::async_hn
    /// [`task::wait`]: zng_task::wait
    pub fn on_dir_changed(
        &self,
        dir: impl Into<PathBuf>,
        recursive: bool,
        ignore_propagation: bool,
        handler: Handler<FsChangesArgs>,
    ) -> VarHandle {
        let dir = absolutize_or_pass(dir.into());
        let handle = self.watch_dir(dir.clone(), recursive);
        FS_CHANGES_EVENT.on_event(
            ignore_propagation,
            handler.filtered(move |args| {
                let _handle = &handle;
                args.events_for_path(&dir).next().is_some()
            }),
        )
    }

    /// Push a `note` that will be cloned on all subsequent change events until the returned handle is dropped.
    ///
    /// This can be used to tag all events that happened over a period of time, something you can't do just
    /// by receiving the events due to async delays caused by debounce.
    ///
    /// Note that the underlying system events the [`notify`](https://docs.rs/notify) crate uses are not guaranteed to be synchronous.
    pub fn annotate(&self, note: Arc<dyn FsChangeNote>) -> FsChangeNoteHandle {
        WATCHER_SV.write().annotate(note)
    }
}

fn has_sync_changes(args: &FsChangesArgs, path: &Path) -> bool {
    let mut any = false;
    for change in args.changes_for_path(path) {
        for note in change.notes::<WatcherSyncWriteNote>() {
            if note.as_path() == path {
                return false;
            }
        }
        any = true;
    }
    any
}

fn absolutize_or_pass(path: PathBuf) -> PathBuf {
    match path.absolutize() {
        Ok(p) if p != path => p.to_path_buf(),
        _ => path,
    }
}

/// Represents a status type for [`WATCHER.sync_status`].
///
/// [`WATCHER.sync_status`]: WATCHER::sync_status
pub trait WatcherSyncStatus<ER = io::Error, EW = io::Error>: WatcherReadStatus<ER> {
    /// New writing value.
    fn writing() -> Self;
    /// New write error value.
    fn write_error(e: EW) -> Self;
}

/// Represents a status type for [`WATCHER`] read-only operations.
pub trait WatcherReadStatus<ER = io::Error>: VarValue + PartialEq {
    /// New idle value.
    fn idle() -> Self;
    /// New reading value.
    fn reading() -> Self;
    /// New read error value.
    fn read_error(e: ER) -> Self;
}

struct AnyWatcherReadStatus {
    idle: fn() -> BoxAnyVarValue,
    reading: fn() -> BoxAnyVarValue,
    read_error: fn(Box<dyn Any>) -> BoxAnyVarValue,
}
impl AnyWatcherReadStatus {
    fn new<ER: Any, S: WatcherReadStatus<ER> + VarValue>() -> Self {
        fn idle_impl<ER, S: WatcherReadStatus<ER>>() -> BoxAnyVarValue {
            BoxAnyVarValue::new(S::idle())
        }
        fn reading_impl<ER, S: WatcherReadStatus<ER>>() -> BoxAnyVarValue {
            BoxAnyVarValue::new(S::reading())
        }
        fn read_error_impl<ER: Any, S: WatcherReadStatus<ER> + VarValue>(e: Box<dyn Any>) -> BoxAnyVarValue {
            BoxAnyVarValue::new(S::read_error(*e.downcast::<ER>().unwrap()))
        }
        Self {
            idle: idle_impl::<ER, S>,
            reading: reading_impl::<ER, S>,
            read_error: read_error_impl::<ER, S>,
        }
    }
}

struct AnyWatcherWriteStatus {
    idle: fn() -> BoxAnyVarValue,
    writing: fn() -> BoxAnyVarValue,
    write_error: fn(Box<dyn Any>) -> BoxAnyVarValue,
}
impl AnyWatcherWriteStatus {
    fn new<ER, EW: Any, S: WatcherSyncStatus<ER, EW> + VarValue>() -> Self {
        fn idle_impl<ER, S: WatcherReadStatus<ER>>() -> BoxAnyVarValue {
            BoxAnyVarValue::new(S::idle())
        }
        fn writing_impl<ER, EW, S: WatcherSyncStatus<ER, EW>>() -> BoxAnyVarValue {
            BoxAnyVarValue::new(S::writing())
        }
        fn write_error_impl<ER, EW: Any, S: WatcherSyncStatus<ER, EW> + VarValue>(e: Box<dyn Any>) -> BoxAnyVarValue {
            BoxAnyVarValue::new(S::write_error(*e.downcast::<EW>().unwrap()))
        }
        Self {
            idle: idle_impl::<ER, S>,
            writing: writing_impl::<ER, EW, S>,
            write_error: write_error_impl::<ER, EW, S>,
        }
    }
}

/// Represents an open read-only file provided by [`WATCHER.read`].
///
/// This type is a thin wrapper around the [`std::fs::File`] with some convenience parsing methods.
///
/// [`WATCHER.read`]: WATCHER::read
#[derive(Debug)]
pub struct WatchFile(fs::File);
impl WatchFile {
    /// Open read the file.
    pub fn open(file: impl AsRef<Path>) -> io::Result<Self> {
        Self::try_open_non_empty(file.as_ref(), true)
    }
    fn try_open_non_empty(path: &Path, retry: bool) -> io::Result<Self> {
        let file = fs::File::open(path)?;

        if retry && file.metadata()?.len() == 0 {
            // some apps create an empty file unlocked, then write.
            let _ = file;
            std::thread::sleep(5.ms());
            return Self::try_open_non_empty(path, false);
        }

        lock_shared(&file, Duration::from_secs(10))?;
        Ok(Self(file))
    }

    /// Read the file contents as a text string.
    pub fn text(&mut self) -> io::Result<Txt> {
        self.string().map(Txt::from)
    }

    /// Read the file contents as a string.
    pub fn string(&mut self) -> io::Result<String> {
        use std::io::Read;
        let mut s = String::new();
        self.0.read_to_string(&mut s)?;
        Ok(s)
    }

    /// Deserialize the file contents as JSON.
    #[cfg(feature = "json")]
    pub fn json<O>(&mut self) -> serde_json::Result<O>
    where
        O: serde::de::DeserializeOwned,
    {
        serde_json::from_reader(io::BufReader::new(&mut self.0))
    }

    /// Deserialize the file contents as TOML.
    #[cfg(feature = "toml")]
    pub fn toml<O>(&mut self) -> io::Result<O>
    where
        O: serde::de::DeserializeOwned,
    {
        use std::io::Read;
        let mut buf = io::BufReader::new(&mut self.0);

        let mut toml_str = String::new();
        buf.read_to_string(&mut toml_str)?;

        toml::de::from_str(&toml_str).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
    }

    /// Deserialize the file content as RON.
    #[cfg(feature = "ron")]
    pub fn ron<O>(&mut self) -> Result<O, ron::de::SpannedError>
    where
        O: serde::de::DeserializeOwned,
    {
        ron::de::from_reader(io::BufReader::new(&mut self.0))
    }

    /// Deserialize the file content as YAML.
    #[cfg(feature = "yaml")]
    pub fn yaml<O>(&mut self) -> serde_yaml::Result<O>
    where
        O: serde::de::DeserializeOwned,
    {
        serde_yaml::from_reader(io::BufReader::new(&mut self.0))
    }

    /// Read file and parse it.
    pub fn parse<O: std::str::FromStr>(&mut self) -> Result<O, WatchFileParseError<O::Err>> {
        use std::io::Read;
        let mut s = String::new();
        self.0.read_to_string(&mut s)?;
        O::from_str(&s).map_err(WatchFileParseError::Parse)
    }
}
impl ops::Deref for WatchFile {
    type Target = fs::File;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}
impl ops::DerefMut for WatchFile {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

const TRANSACTION_GUID: &str = "6eIw3bYMS0uKaQMkTIQacQ";
const TRANSACTION_LOCK_EXT: &str = "6eIw3bYMS0uKaQMkTIQacQ-lock.tmp";

/// Represents an open write file provided by [`WATCHER.sync`].
///
/// This struct writes to a temporary file and renames it over the actual file on commit only.
/// The dereferenced [`fs::File`] is the temporary file, not the actual one.
///
/// # Transaction
///
/// To minimize the risk of file corruption exclusive locks are used, both the target file and the temp file
/// are locked. An empty lock file is also used to cover the moment when both files are unlocked for the rename operation
/// and the moment the temp file is acquired.
///
/// The temp file is the actual file path with file extension replaced with `{path/.file-name.ext}.{GUID}-{n}.tmp`, the `n` is a
/// number from 0 to 999, if a temp file exists unlocked it will be reused.
///
/// The lock file is `{path/.file-name.ext}.{GUID}-lock.tmp`. Note that this
/// lock file only helps for apps that use [`WriteFile`], but even without it the risk is minimal as the slow
/// write operations are already flushed when it is time to commit.
///
/// [`WATCHER.sync`]: WATCHER::sync
pub struct WriteFile {
    temp_file: Option<fs::File>,
    actual_file: Option<fs::File>,
    transaction_lock: Option<fs::File>,

    actual_path: PathBuf,
    temp_path: PathBuf,
    transaction_path: PathBuf,

    cleaned: bool,
}
impl Drop for WriteFile {
    fn drop(&mut self) {
        if !self.cleaned {
            tracing::error!("dropped sync write file without commit or cancel");
            self.clean();
        }
    }
}
impl ops::Deref for WriteFile {
    type Target = fs::File;

    fn deref(&self) -> &Self::Target {
        self.temp_file.as_ref().unwrap()
    }
}
impl ops::DerefMut for WriteFile {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.temp_file.as_mut().unwrap()
    }
}
impl WriteFile {
    /// Open or create the file.
    pub fn open(path: PathBuf) -> io::Result<Self> {
        let actual_path = path.absolutize()?.into_owned();
        if !actual_path.exists()
            && let Some(parent) = actual_path.parent()
        {
            std::fs::create_dir_all(parent)?;
        }

        let hidden_name = match actual_path.file_name() {
            Some(n) => format!(".{}", n.to_string_lossy()),
            None => return Err(io::Error::new(io::ErrorKind::InvalidInput, "expected file name")),
        };

        let transaction_path = actual_path.with_file_name(format!("{hidden_name}.{TRANSACTION_LOCK_EXT}"));
        let transaction_lock = fs::OpenOptions::new()
            .create(true)
            .truncate(true)
            .write(true)
            .open(&transaction_path)?;

        const TIMEOUT: Duration = Duration::from_secs(10);

        lock_exclusive(&transaction_lock, TIMEOUT)?;

        let actual_file = fs::OpenOptions::new().write(true).create(true).truncate(false).open(&actual_path)?;
        lock_exclusive(&actual_file, TIMEOUT)?;

        let mut n = 0;
        let mut temp_path = actual_path.with_file_name(format!("{hidden_name}.{TRANSACTION_GUID}-{n}.tmp"));
        let temp_file = loop {
            if let Ok(f) = fs::OpenOptions::new().write(true).create(true).truncate(true).open(&temp_path)
                && f.try_lock().is_ok()
            {
                break f;
            }

            n += 1;
            temp_path = actual_path.with_file_name(format!("{hidden_name}.{TRANSACTION_GUID}-{n}.tmp"));
            n += 1;
            if n > 1000 {
                return Err(io::Error::new(io::ErrorKind::AlreadyExists, "cannot create temporary file"));
            }
        };

        Ok(Self {
            actual_file: Some(actual_file),
            temp_file: Some(temp_file),
            transaction_lock: Some(transaction_lock),
            actual_path,
            temp_path,
            transaction_path,
            cleaned: false,
        })
    }

    /// Write the text string.
    pub fn write_text(&mut self, txt: &str) -> io::Result<()> {
        self.write_all(txt.as_bytes())
    }

    /// Serialize and write.
    ///
    /// If `pretty` is `true` the JSON is formatted for human reading.
    #[cfg(feature = "json")]
    pub fn write_json<O: serde::Serialize>(&mut self, value: &O, pretty: bool) -> io::Result<()> {
        let mut buf = io::BufWriter::new(ops::DerefMut::deref_mut(self));
        if pretty {
            serde_json::to_writer_pretty(&mut buf, value)?;
        } else {
            serde_json::to_writer(&mut buf, value)?;
        }
        buf.flush()
    }

    /// Serialize and write.
    ///
    /// If `pretty` is `true` the TOML is formatted for human reading.
    #[cfg(feature = "toml")]
    pub fn write_toml<O: serde::Serialize>(&mut self, value: &O, pretty: bool) -> io::Result<()> {
        let toml = if pretty {
            toml::ser::to_string_pretty(value)
        } else {
            toml::ser::to_string(value)
        }
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;

        self.write_all(toml.as_bytes())
    }

    /// Serialize and write.
    ///
    /// If `pretty` is `true` the RON if formatted for human reading using the default pretty config.
    #[cfg(feature = "ron")]
    pub fn write_ron<O: serde::Serialize>(&mut self, value: &O, pretty: bool) -> io::Result<()> {
        let buf = io::BufWriter::new(ops::DerefMut::deref_mut(self));
        struct Ffs<'a> {
            w: io::BufWriter<&'a mut fs::File>,
        }
        impl fmt::Write for Ffs<'_> {
            fn write_str(&mut self, s: &str) -> fmt::Result {
                self.w.write_all(s.as_bytes()).map_err(|_| fmt::Error)
            }
        }
        let mut buf = Ffs { w: buf };
        if pretty {
            ron::ser::to_writer_pretty(&mut buf, value, Default::default()).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        } else {
            ron::ser::to_writer(&mut buf, value).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        }
        buf.w.flush()
    }

    /// Serialize and write.
    #[cfg(feature = "yaml")]
    pub fn write_yaml<O: serde::Serialize>(&mut self, value: &O) -> io::Result<()> {
        let mut buf = io::BufWriter::new(ops::DerefMut::deref_mut(self));
        serde_yaml::to_writer(&mut buf, value).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        buf.flush()
    }

    /// Commit write, flush and replace the actual file with the new one.
    pub fn commit(mut self) -> io::Result<()> {
        let r = self.replace_actual();
        self.clean();
        r
    }

    /// Cancel write, the file will not be updated.
    pub fn cancel(mut self) {
        self.clean();
    }

    fn replace_actual(&mut self) -> io::Result<()> {
        let mut temp_file = self.temp_file.take().unwrap();
        temp_file.flush()?;
        temp_file.sync_all()?;

        unlock_ok(&temp_file).unwrap();
        drop(temp_file);

        let actual_file = self.actual_file.take().unwrap();
        unlock_ok(&actual_file)?;
        drop(actual_file);

        let mut retries = 0;
        loop {
            // commit by replacing the actual_path with already on disk temp_path file.
            match fs::rename(&self.temp_path, &self.actual_path) {
                Ok(()) => {
                    break;
                }
                Err(e) => match e.kind() {
                    io::ErrorKind::PermissionDenied => {
                        if retries == 5 {
                            // Give-up, we managed to write lock both temp and actual just
                            // before this, but now we can't replace actual and remove temp.
                            // Hardware issue? Or another process holding a lock for 1s+50ms*5.
                            return Err(e);
                        } else if retries > 0 {
                            // Second+ retries:
                            //
                            // probably a system issue.
                            //
                            // Windows sporadically returns ACCESS_DENIED for kernel!SetRenameInformationFile in
                            // other apps that use the same save pattern (write-tmp -> close-tmp -> rename).
                            // see GIMP issue: https://gitlab.gnome.org/GNOME/gimp/-/issues/1370
                            //
                            // I used procmon to trace all file operations, there is no other app trying to use
                            // the temp and actual files when the ACCESS_DENIED occurs, both files are unlocked and
                            // closed before the rename calls start. This might be a Windows bug.
                            std::thread::sleep(30.ms());
                        } else {
                            // first retry:
                            //
                            // probably another process reading the `actual_path`.
                            //
                            // Reacquire a write lock and unlock, just to wait the external app.
                            match std::fs::File::options().write(true).open(&self.actual_path) {
                                Ok(f) => {
                                    if lock_exclusive(&f, 10.secs()).is_ok() {
                                        // acquired actual ok, retry
                                        let _ = unlock_ok(&f);
                                    }
                                }
                                Err(e) => match e.kind() {
                                    io::ErrorKind::NotFound => {
                                        // all good, rename will create actual
                                        continue;
                                    }
                                    _ => {
                                        // unknown error, let retry handle it
                                        std::thread::sleep(30.ms());
                                    }
                                },
                            }
                        }

                        retries += 1;
                    }
                    _ => return Err(e),
                },
            }
        }

        Ok(())
    }

    fn clean(&mut self) {
        self.cleaned = true;

        if let Some(tmp) = self.temp_file.take() {
            let _ = tmp.unlock();
        }
        if let Err(e) = fs::remove_file(&self.temp_path) {
            tracing::debug!("failed to cleanup temp file, {e}")
        }

        if let Some(file) = self.actual_file.take() {
            let _ = file.unlock();
        }

        let transaction = self.transaction_lock.take().unwrap();
        let _ = transaction.unlock();
        let _ = fs::remove_file(&self.transaction_path);
    }
}

/// Error for [`WatchFile::parse`].
#[derive(Debug)]
#[non_exhaustive]
pub enum WatchFileParseError<E> {
    /// Error reading the file.
    Io(io::Error),
    /// Error parsing the file.
    Parse(E),
}
impl<E> From<io::Error> for WatchFileParseError<E> {
    fn from(value: io::Error) -> Self {
        Self::Io(value)
    }
}
impl<E: fmt::Display> fmt::Display for WatchFileParseError<E> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            WatchFileParseError::Io(e) => write!(f, "read error, {e}"),
            WatchFileParseError::Parse(e) => write!(f, "parse error, {e}"),
        }
    }
}
impl<E: std::error::Error + 'static> std::error::Error for WatchFileParseError<E> {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            WatchFileParseError::Io(e) => Some(e),
            WatchFileParseError::Parse(e) => Some(e),
        }
    }
}

/// Represents a [`FsChange`] note.
///
/// This trait is already implemented for all types it applies.
#[diagnostic::on_unimplemented(note = "`FsChangeNote` is implemented for all `T: Debug + Any + Send + Sync`")]
pub trait FsChangeNote: fmt::Debug + std::any::Any + Send + Sync {
    /// Access any.
    fn as_any(&self) -> &dyn std::any::Any;
}
impl<T: fmt::Debug + std::any::Any + Send + Sync> FsChangeNote for T {
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

/// Handle that holds a [`WATCHER.annotate`] note.
///
/// [`WATCHER.annotate`]: WATCHER::annotate
#[derive(Clone)]
#[must_use = "the note is removed when the handle is dropped"]
pub struct FsChangeNoteHandle(#[expect(dead_code)] Arc<Arc<dyn FsChangeNote>>);

/// Annotation for file watcher events and var update tags.
///
/// Identifies the [`WATCHER.sync`] file that is currently being written to.
///
/// [`WATCHER.sync`]: WATCHER::sync
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct WatcherSyncWriteNote(PathBuf);
impl WatcherSyncWriteNote {
    /// Deref.
    pub fn as_path(&self) -> &Path {
        self
    }
}
impl ops::Deref for WatcherSyncWriteNote {
    type Target = Path;

    fn deref(&self) -> &Self::Target {
        self.0.as_path()
    }
}

/// File system change event types.
///
/// The event for each change is available in [`FsChange::event`].
///
/// This module re-exports types from the [`notify`](https://docs.rs/notify) crate.
pub mod fs_event {
    pub use notify::event::{
        AccessKind, AccessMode, CreateKind, DataChange, Event, EventKind, MetadataKind, ModifyKind, RemoveKind, RenameMode,
    };
    pub use notify::{Error, ErrorKind};
}

/// Represents a single file system change, annotated.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct FsChange {
    /// All [`WATCHER.annotate`] that where set when this event happened.
    ///
    /// [`WATCHER.annotate`]: WATCHER::annotate
    pub notes: Vec<Arc<dyn FsChangeNote>>,

    /// The actual notify event or error.
    pub event: Result<fs_event::Event, Arc<fs_event::Error>>,
}
impl PartialEq for FsChange {
    fn eq(&self, other: &Self) -> bool {
        self.notes.len() == other.notes.len()
            && self.notes.iter().zip(other.notes.iter()).all(|(a, b)| Arc::ptr_eq(a, b))
            && match (&self.event, &other.event) {
                (Ok(a), Ok(b)) => a == b,
                (Err(a), Err(b)) => Arc::ptr_eq(a, b),
                _ => false,
            }
    }
}
impl FsChange {
    /// If the change affects the `path`.
    pub fn is_for_path(&self, path: &Path) -> bool {
        if let Ok(ev) = &self.event {
            return ev.paths.iter().any(|p| p.starts_with(path));
        }
        false
    }

    /// If the change affects any path matched by the glob pattern.
    pub fn is_for_glob(&self, pattern: &glob::Pattern) -> bool {
        if let Ok(ev) = &self.event {
            return ev.paths.iter().any(|p| pattern.matches_path(p));
        }
        false
    }

    /// Iterate over all notes of the type `T`.
    pub fn notes<T: FsChangeNote>(&self) -> impl Iterator<Item = &T> {
        self.notes.iter().filter_map(|n| FsChangeNote::as_any(&**n).downcast_ref::<T>())
    }
}

event_args! {
    /// [`FS_CHANGES_EVENT`] arguments.
    pub struct FsChangesArgs {
        /// All notify changes since the last event.
        pub changes: Arc<Vec<FsChange>>,

        ..

        /// None, only app level handlers receive this event.
        fn is_in_target(&self, _id: WidgetId) -> bool {
            false
        }
    }
}
impl FsChangesArgs {
    /// Iterate over all change events.
    pub fn events(&self) -> impl Iterator<Item = &fs_event::Event> + '_ {
        self.changes.iter().filter_map(|r| r.event.as_ref().ok())
    }

    /// Iterate over all file watcher errors.
    pub fn errors(&self) -> impl Iterator<Item = &notify::Error> + '_ {
        self.changes.iter().filter_map(|r| r.event.as_ref().err().map(|e| &**e))
    }

    /// Returns `true` is some events where lost.
    ///
    /// This indicates either a lapse in the events or a change in the filesystem such that events
    /// received so far can no longer be relied on to represent the state of the filesystem now.
    ///
    /// An application that simply reacts to file changes may not care about this. An application
    /// that keeps an in-memory representation of the filesystem will need to care, and will need
    /// to refresh that representation directly from the filesystem.
    pub fn rescan(&self) -> bool {
        self.events().any(|e| e.need_rescan())
    }

    /// Iterate over all changes that affects paths selected by the `glob` pattern.
    pub fn changes_for(&self, glob: &str) -> Result<impl Iterator<Item = &FsChange> + '_, glob::PatternError> {
        let glob = glob::Pattern::new(glob)?;
        Ok(self.changes.iter().filter(move |c| c.is_for_glob(&glob)))
    }

    /// Iterate over all changes that affects paths that are equal to `path` or inside it.
    pub fn changes_for_path<'a>(&'a self, path: &'a Path) -> impl Iterator<Item = &'a FsChange> + 'a {
        self.changes.iter().filter(move |c| c.is_for_path(path))
    }

    /// Iterate over all change events that affects that are equal to `path` or inside it.
    pub fn events_for(&self, glob: &str) -> Result<impl Iterator<Item = &fs_event::Event> + '_, glob::PatternError> {
        let glob = glob::Pattern::new(glob)?;
        Ok(self.events().filter(move |ev| ev.paths.iter().any(|p| glob.matches_path(p))))
    }

    /// Iterate over all change events that affects paths that are equal to `path` or inside it.
    pub fn events_for_path<'a>(&'a self, path: &'a Path) -> impl Iterator<Item = &'a fs_event::Event> + 'a {
        self.events().filter(move |ev| ev.paths.iter().any(|p| p.starts_with(path)))
    }
}

zng_app::event::event! {
    /// Event sent by the [`WATCHER`] service on directories or files that are watched.
    pub static FS_CHANGES_EVENT: FsChangesArgs;
}

/// Represents an active file or directory watcher in [`WATCHER`].
#[derive(Clone)]
#[must_use = "the watcher is dropped if the handle is dropped"]
pub struct WatcherHandle(Handle<()>);

impl WatcherHandle {
    /// Handle to no watcher.
    pub fn dummy() -> Self {
        Self(Handle::dummy(()))
    }

    /// If [`perm`](Self::perm) was called in another clone of this handle.
    ///
    /// If `true` the resource will stay in memory for the duration of the app, unless [`force_drop`](Self::force_drop)
    /// is also called.
    pub fn is_permanent(&self) -> bool {
        self.0.is_permanent()
    }

    /// Force drops the watcher, meaning it will be dropped even if there are other handles active.
    pub fn force_drop(self) {
        self.0.force_drop()
    }

    /// If the watcher is dropped.
    pub fn is_dropped(&self) -> bool {
        self.0.is_dropped()
    }

    /// Drop the handle without dropping the watcher, the watcher will stay active for the
    /// duration of the app process.
    pub fn perm(self) {
        self.0.perm()
    }
}