oj_notify 0.2.48

//! Watcher implementation for Darwin's FSEvents API
//!
//! The FSEvents API provides a mechanism to notify clients about directories they ought to re-scan
//! in order to keep their internal data structures up-to-date with respect to the true state of
//! the file system. (For example, when files or directories are created, modified, or removed.) It
//! sends these notifications "in bulk", possibly notifying the client of changes to several
//! directories in a single callback.
//!
//! For more information see the [FSEvents API reference][ref].
//!
//! TODO: document event translation
//!
//! [ref]: https://developer.apple.com/library/mac/documentation/Darwin/Reference/FSEvents_Ref/

#![allow(non_upper_case_globals, dead_code)]

use crate::paths::{absolute_path, reported_path};
use crate::{
    Config, Error, ErrorKind, EventHandler, EventKindMask, RecursiveMode, Result, Sender, Watcher,
    unbounded,
};
use crate::{PathOp, event::*};
use objc2_core_foundation as cf;
use objc2_core_services as fs;
use std::collections::HashMap;
use std::ffi::{CStr, OsStr};
use std::fmt;
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};
use std::ptr::{self, NonNull};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;

bitflags::bitflags! {
  #[repr(C)]
  #[derive(Debug)]
  struct StreamFlags: u32 {
    const NONE = fs::kFSEventStreamEventFlagNone;
    const MUST_SCAN_SUBDIRS = fs::kFSEventStreamEventFlagMustScanSubDirs;
    const USER_DROPPED = fs::kFSEventStreamEventFlagUserDropped;
    const KERNEL_DROPPED = fs::kFSEventStreamEventFlagKernelDropped;
    const IDS_WRAPPED = fs::kFSEventStreamEventFlagEventIdsWrapped;
    const HISTORY_DONE = fs::kFSEventStreamEventFlagHistoryDone;
    const ROOT_CHANGED = fs::kFSEventStreamEventFlagRootChanged;
    const MOUNT = fs::kFSEventStreamEventFlagMount;
    const UNMOUNT = fs::kFSEventStreamEventFlagUnmount;
    const ITEM_CREATED = fs::kFSEventStreamEventFlagItemCreated;
    const ITEM_REMOVED = fs::kFSEventStreamEventFlagItemRemoved;
    const INODE_META_MOD = fs::kFSEventStreamEventFlagItemInodeMetaMod;
    const ITEM_RENAMED = fs::kFSEventStreamEventFlagItemRenamed;
    const ITEM_MODIFIED = fs::kFSEventStreamEventFlagItemModified;
    const FINDER_INFO_MOD = fs::kFSEventStreamEventFlagItemFinderInfoMod;
    const ITEM_CHANGE_OWNER = fs::kFSEventStreamEventFlagItemChangeOwner;
    const ITEM_XATTR_MOD = fs::kFSEventStreamEventFlagItemXattrMod;
    const IS_FILE = fs::kFSEventStreamEventFlagItemIsFile;
    const IS_DIR = fs::kFSEventStreamEventFlagItemIsDir;
    const IS_SYMLINK = fs::kFSEventStreamEventFlagItemIsSymlink;
    const OWN_EVENT = fs::kFSEventStreamEventFlagOwnEvent;
    const IS_HARDLINK = fs::kFSEventStreamEventFlagItemIsHardlink;
    const IS_LAST_HARDLINK = fs::kFSEventStreamEventFlagItemIsLastHardlink;
    const ITEM_CLONED = fs::kFSEventStreamEventFlagItemCloned;
  }
}

const RUNLOOP_STOP_RECHECK_SECONDS: cf::CFTimeInterval = 0.5;

/// FSEvents-based `Watcher` implementation
pub struct FsEventWatcher {
    since_when: Option<fs::FSEventStreamEventId>,
    latency: cf::CFTimeInterval,
    flags: fs::FSEventStreamCreateFlags,
    event_handler: Arc<Mutex<dyn EventHandler>>,
    runloop: Option<RunLoopHandle>,
    watches: HashMap<PathBuf, WatchEntry>,
    event_kinds: EventKindMask,
}

// `cf_path` is kept out of `WatchInfo` because `WatchInfo` is shared with the stream
// context, which must stay `Send + Sync`.
#[derive(Debug)]
struct WatchEntry {
    info: WatchInfo,
    cf_path: cf::CFRetained<cf::CFString>,
    device: u64,
    inode: u64,
}

#[derive(Clone, Debug)]
struct WatchInfo {
    is_recursive: bool,
    reported_path: PathBuf,
}

// FSEvents applies the path limit across live streams, so all watcher instances
// in this process must share the same count.
static ACTIVE_FSEVENTS_PATHS: AtomicUsize = AtomicUsize::new(0);

#[derive(Debug)]
struct FseventsPathReservation {
    active_paths: &'static AtomicUsize,
    path_count: usize,
}

impl FseventsPathReservation {
    fn acquire(
        active_paths: &'static AtomicUsize,
        path_count: usize,
        budget: usize,
    ) -> std::result::Result<Self, usize> {
        active_paths
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |active_path_count| {
                active_path_count
                    .checked_add(path_count)
                    .filter(|&combined_path_count| combined_path_count <= budget)
            })
            .map(|_| Self {
                active_paths,
                path_count,
            })
    }
}

impl Drop for FseventsPathReservation {
    fn drop(&mut self) {
        let previous = self
            .active_paths
            .fetch_sub(self.path_count, Ordering::Relaxed);
        debug_assert!(previous >= self.path_count);
    }
}

#[derive(Debug)]
struct RunLoopHandle {
    runloop: cf::CFRetained<cf::CFRunLoop>,
    stop_flag: Arc<AtomicBool>,
    thread_handle: thread::JoinHandle<fs::FSEventStreamEventId>,
    recursive_info: Arc<Mutex<HashMap<PathBuf, WatchInfo>>>,
}

impl fmt::Debug for FsEventWatcher {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.debug_struct("FsEventWatcher")
            .field("since_when", &self.since_when)
            .field("latency", &self.latency)
            .field("flags", &self.flags)
            .field("event_handler", &Arc::as_ptr(&self.event_handler))
            .field("runloop", &self.runloop)
            .field("watches", &self.watches)
            .finish()
    }
}

// FsEventWatcher is not Send/Sync automatically.
// It's Send because the pointer is not used in other threads.
unsafe impl Send for FsEventWatcher {}

// It's Sync because all methods that change the mutable state use `&mut self`.
unsafe impl Sync for FsEventWatcher {}

fn translate_flags(flags: StreamFlags, precise: bool) -> Vec<Event> {
    let mut evs = Vec::new();
    translate_flags_with(flags, precise, |ev| evs.push(ev));
    evs
}

// Keep this in sync with `translate_flags_with`; the callback uses it to avoid path clones.
fn translated_event_count(flags: &StreamFlags, precise: bool) -> usize {
    if flags.contains(StreamFlags::HISTORY_DONE) {
        return 0;
    }

    let mut count = usize::from(flags.contains(StreamFlags::MUST_SCAN_SUBDIRS));
    if !precise {
        return count + 1;
    }

    let root_changed = flags.contains(StreamFlags::ROOT_CHANGED);
    count += usize::from(root_changed);
    count += usize::from(flags.contains(StreamFlags::MOUNT));
    count += usize::from(flags.contains(StreamFlags::UNMOUNT));
    count += usize::from(flags.contains(StreamFlags::ITEM_CREATED));
    count += usize::from(flags.contains(StreamFlags::ITEM_RENAMED) && !root_changed);
    count += usize::from(flags.contains(StreamFlags::INODE_META_MOD));
    count += usize::from(flags.contains(StreamFlags::FINDER_INFO_MOD));
    count += usize::from(flags.contains(StreamFlags::ITEM_CHANGE_OWNER));
    count += usize::from(flags.contains(StreamFlags::ITEM_XATTR_MOD));
    count += usize::from(flags.contains(StreamFlags::ITEM_MODIFIED));
    count += usize::from(flags.contains(StreamFlags::ITEM_REMOVED) && !root_changed);
    count
}

fn translate_flags_with(flags: StreamFlags, precise: bool, mut emit: impl FnMut(Event)) {
    // «Denotes a sentinel event sent to mark the end of the "historical" events
    // sent as a result of specifying a `sinceWhen` value in the FSEvents.Create
    // call that created this event stream. After invoking the client's callback
    // with all the "historical" events that occurred before now, the client's
    // callback will be invoked with an event where the HistoryDone flag is set.
    // The client should ignore the path supplied in this callback.»
    // — https://www.mbsplugins.eu/FSEventsNextEvent.shtml
    //
    // As a result, we just stop processing here and return an empty vec, which
    // will ignore this completely and not emit any Events whatsoever.
    if flags.contains(StreamFlags::HISTORY_DONE) {
        return;
    }

    // `ITEM_CLONED` can be present alongside other flags (including create/modify/remove).
    // Preserve any existing `info` (like "root changed"), but annotate otherwise so downstream
    // can detect and filter clone-related events. See https://github.com/notify-rs/notify/issues/465.
    let clone_related = precise && flags.contains(StreamFlags::ITEM_CLONED);
    let own_process_id = if precise && flags.contains(StreamFlags::OWN_EVENT) {
        Some(std::process::id())
    } else {
        None
    };

    let mut emit_event = |mut ev: Event| {
        if clone_related && ev.info().is_none() {
            ev.attrs.set_info("is: clone");
        }
        if let Some(process_id) = own_process_id {
            ev.attrs.set_process_id(process_id);
        }
        emit(ev);
    };

    // FSEvents provides two possible hints as to why events were dropped,
    // however documentation on what those mean is scant, so we just pass them
    // through in the info attr field. The intent is clear enough, and the
    // additional information is provided if the user wants it.
    if flags.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
        let e = Event::new(EventKind::Other).set_flag(Flag::Rescan);
        emit_event(if flags.contains(StreamFlags::USER_DROPPED) {
            e.set_info("rescan: user dropped")
        } else if flags.contains(StreamFlags::KERNEL_DROPPED) {
            e.set_info("rescan: kernel dropped")
        } else {
            e
        });
    }

    // In imprecise mode, let's not even bother parsing the kind of the event
    // except for the above very special events.
    if !precise {
        emit(Event::new(EventKind::Any));
        return;
    }

    // A watched root changed (renamed or removed). If the flags provide a hint,
    // prefer that over guessing. Otherwise, treat it as a removal to avoid
    // misclassifying a delete as a rename.
    let root_changed = flags.contains(StreamFlags::ROOT_CHANGED);
    if root_changed {
        let kind = if flags.contains(StreamFlags::ITEM_REMOVED) {
            if flags.contains(StreamFlags::IS_DIR) {
                EventKind::Remove(RemoveKind::Folder)
            } else if flags.contains(StreamFlags::IS_FILE) {
                EventKind::Remove(RemoveKind::File)
            } else {
                EventKind::Remove(RemoveKind::Any)
            }
        } else if flags.contains(StreamFlags::ITEM_RENAMED) {
            EventKind::Modify(ModifyKind::Name(RenameMode::From))
        } else {
            EventKind::Remove(RemoveKind::Any)
        };

        emit_event(Event::new(kind).set_info("root changed"));
    }

    // A path was mounted at the event path; we treat that as a create.
    if flags.contains(StreamFlags::MOUNT) {
        emit_event(Event::new(EventKind::Create(CreateKind::Other)).set_info("mount"));
    }

    // A path was unmounted at the event path; we treat that as a remove.
    if flags.contains(StreamFlags::UNMOUNT) {
        emit_event(Event::new(EventKind::Remove(RemoveKind::Other)).set_info("mount"));
    }

    if flags.contains(StreamFlags::ITEM_CREATED) {
        emit_event(if flags.contains(StreamFlags::IS_DIR) {
            Event::new(EventKind::Create(CreateKind::Folder))
        } else if flags.contains(StreamFlags::IS_FILE) {
            Event::new(EventKind::Create(CreateKind::File))
        } else {
            let e = Event::new(EventKind::Create(CreateKind::Other));
            if flags.contains(StreamFlags::IS_SYMLINK) {
                e.set_info("is: symlink")
            } else if flags.contains(StreamFlags::IS_HARDLINK) {
                e.set_info("is: hardlink")
            } else if flags.contains(StreamFlags::ITEM_CLONED) {
                e.set_info("is: clone")
            } else {
                Event::new(EventKind::Create(CreateKind::Any))
            }
        });
    }

    // FSEvents provides no mechanism to associate the old and new sides of a
    // rename event.
    // Avoid emitting duplicate events around a root change by checking `root_changed`.
    if flags.contains(StreamFlags::ITEM_RENAMED) && !root_changed {
        emit_event(Event::new(EventKind::Modify(ModifyKind::Name(
            RenameMode::Any,
        ))));
    }

    // This is only described as "metadata changed", but it may be that it's
    // only emitted for some more precise subset of events... if so, will need
    // amending, but for now we have an Any-shaped bucket to put it in.
    if flags.contains(StreamFlags::INODE_META_MOD) {
        emit_event(Event::new(EventKind::Modify(ModifyKind::Metadata(
            MetadataKind::Any,
        ))));
    }

    if flags.contains(StreamFlags::FINDER_INFO_MOD) {
        emit_event(
            Event::new(EventKind::Modify(ModifyKind::Metadata(MetadataKind::Other)))
                .set_info("meta: finder info"),
        );
    }

    if flags.contains(StreamFlags::ITEM_CHANGE_OWNER) {
        emit_event(Event::new(EventKind::Modify(ModifyKind::Metadata(
            MetadataKind::Ownership,
        ))));
    }

    if flags.contains(StreamFlags::ITEM_XATTR_MOD) {
        emit_event(Event::new(EventKind::Modify(ModifyKind::Metadata(
            MetadataKind::Extended,
        ))));
    }

    // This is specifically described as a data change, which we take to mean
    // is a content change.
    if flags.contains(StreamFlags::ITEM_MODIFIED) {
        emit_event(Event::new(EventKind::Modify(ModifyKind::Data(
            DataChange::Content,
        ))));
    }

    // Avoid emitting duplicate events around a root change by checking `root_changed`.
    if flags.contains(StreamFlags::ITEM_REMOVED) && !root_changed {
        emit_event(if flags.contains(StreamFlags::IS_DIR) {
            Event::new(EventKind::Remove(RemoveKind::Folder))
        } else if flags.contains(StreamFlags::IS_FILE) {
            Event::new(EventKind::Remove(RemoveKind::File))
        } else {
            let e = Event::new(EventKind::Remove(RemoveKind::Other));
            if flags.contains(StreamFlags::IS_SYMLINK) {
                e.set_info("is: symlink")
            } else if flags.contains(StreamFlags::IS_HARDLINK) {
                e.set_info("is: hardlink")
            } else if flags.contains(StreamFlags::ITEM_CLONED) {
                e.set_info("is: clone")
            } else {
                Event::new(EventKind::Remove(RemoveKind::Any))
            }
        });
    }
}

struct StreamContextInfo {
    event_handler: Arc<Mutex<dyn EventHandler>>,
    recursive_info: Arc<Mutex<HashMap<PathBuf, WatchInfo>>>,
    event_kinds: EventKindMask,
}

// Free the context when the stream created by `FSEventStreamCreate` is released.
unsafe extern "C-unwind" fn release_context(info: *const libc::c_void) {
    // Safety:
    // - The [documentation] for `FSEventStreamContext` states that `release` is only
    //   called when the stream is deallocated, so it is safe to convert `info` back into a
    //   box and drop it.
    //
    // [docs]: https://developer.apple.com/documentation/coreservices/fseventstreamcontext?language=objc
    unsafe {
        drop(Box::from_raw(
            info as *const StreamContextInfo as *mut StreamContextInfo,
        ));
    }
}

impl FsEventWatcher {
    fn from_event_handler(
        event_handler: Arc<Mutex<dyn EventHandler>>,
        event_kinds: EventKindMask,
        latency: cf::CFTimeInterval,
    ) -> Result<Self> {
        Ok(FsEventWatcher {
            since_when: None,
            latency,
            flags: fs::kFSEventStreamCreateFlagFileEvents
                | fs::kFSEventStreamCreateFlagNoDefer
                | fs::kFSEventStreamCreateFlagWatchRoot,
            event_handler,
            runloop: None,
            watches: HashMap::new(),
            event_kinds,
        })
    }

    fn watch_inner(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()> {
        let (path, previous) = self.append_path(path, recursive_mode)?;
        let entry = &self.watches[&path];
        let unchanged = previous.is_some_and(|previous| {
            previous.device == entry.device
                && previous.inode == entry.inode
                && previous.info.is_recursive == entry.info.is_recursive
        });
        if (unchanged || self.is_covered(&path, entry.device))
            && let Some(runloop) = &self.runloop
        {
            // The stream already delivers these events. Update the callback's path
            // representation and recursive mode without rebuilding the stream.
            runloop
                .recursive_info
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .insert(path, entry.info.clone());
            return Ok(());
        }

        self.stop();
        self.run()
    }

    fn unwatch_inner(&mut self, path: &Path) -> Result<()> {
        let (path, entry) = self.remove_path(path)?;
        if self.is_covered(&path, entry.device)
            && let Some(runloop) = &self.runloop
        {
            runloop
                .recursive_info
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .remove(&path);
            return Ok(());
        }

        self.stop();
        self.run()
    }

    fn update_paths_inner(
        &mut self,
        ops: Vec<crate::PathOp>,
    ) -> crate::StdResult<(), crate::UpdatePathsError> {
        if ops.is_empty() {
            return Ok(());
        }
        let previous_roots: HashMap<_, _> = self
            .stream_roots()
            .map(|(path, entry)| (path.clone(), (entry.device, entry.inode)))
            .collect();

        let result = crate::update_paths(ops, |op| match op {
            crate::PathOp::Watch(path, config) => self
                .append_path(&path, config.recursive_mode())
                .map(|_| ())
                .map_err(|e| (PathOp::Watch(path, config), e)),
            crate::PathOp::Unwatch(path) => self
                .remove_path(&path)
                .map(|_| ())
                .map_err(|e| (PathOp::Unwatch(path), e)),
        });

        let roots_unchanged = self.stream_roots().count() == previous_roots.len()
            && self.stream_roots().all(|(path, entry)| {
                previous_roots.get(path) == Some(&(entry.device, entry.inode))
            });
        if roots_unchanged && let Some(runloop) = &self.runloop {
            *runloop
                .recursive_info
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner()) = self
                .watches
                .iter()
                .map(|(path, entry)| (path.clone(), entry.info.clone()))
                .collect();
            return result;
        }

        self.stop();
        match self.run() {
            Err(run_error) => match result {
                Ok(()) => Err(crate::UpdatePathsError {
                    source: run_error,
                    origin: None,
                    remaining: Default::default(),
                }),
                Err(path_op_error) => {
                    log::error!(
                        "Unable to run fsevents watcher after updating paths error: {run_error:?}"
                    );
                    Err(path_op_error)
                }
            },
            Ok(()) => result,
        }
    }

    #[inline]
    fn is_running(&self) -> bool {
        self.runloop.is_some()
    }

    fn stop(&mut self) {
        if !self.is_running() {
            return;
        }

        if let Some(RunLoopHandle {
            runloop,
            stop_flag,
            thread_handle,
            ..
        }) = self.runloop.take()
        {
            // Don't wait for the runloop to become "waiting" before stopping; if the
            // stream is under heavy load that can delay shutdown indefinitely.
            stop_flag.store(true, Ordering::Release);
            runloop.stop();
            runloop.wake_up();
            // Wait for the thread to shut down.
            self.since_when = Some(thread_handle.join().expect("thread to shut down"));
        }
    }

    fn remove_path(&mut self, path: &Path) -> Result<(PathBuf, WatchEntry)> {
        let p = path
            .canonicalize()
            .ok()
            .or_else(|| {
                self.watches
                    .iter()
                    .find(|(_, entry)| entry.info.reported_path == path)
                    .map(|(path, _)| path.clone())
            })
            .or_else(|| absolute_path(path).ok())
            .unwrap_or_else(|| path.to_owned());

        match self.watches.remove(&p) {
            Some(entry) => Ok((p, entry)),
            None => Err(Error::watch_not_found()),
        }
    }

    // https://github.com/thibaudgg/rb-fsevent/blob/master/ext/fsevent_watch/main.c
    fn append_path(
        &mut self,
        path: &Path,
        recursive_mode: RecursiveMode,
    ) -> Result<(PathBuf, Option<WatchEntry>)> {
        if !path.exists() {
            return Err(Error::path_not_found().add_path(path.into()));
        }
        let canonical_path = path.to_path_buf().canonicalize()?;
        let mut err: *mut cf::CFError = ptr::null_mut();
        let Some(cf_path) = (unsafe { path_to_cfstring_ref(&canonical_path, &mut err) }) else {
            if let Some(err) = NonNull::new(err) {
                let _ = unsafe { cf::CFRetained::from_raw(err) };
            }
            // Most likely the directory was deleted, or permissions changed,
            // while the above code was running.
            return Err(Error::path_not_found().add_path(path.into()));
        };

        let metadata = std::fs::metadata(&canonical_path)?;

        let previous = self.watches.insert(
            canonical_path.clone(),
            WatchEntry {
                info: WatchInfo {
                    is_recursive: recursive_mode.is_recursive(),
                    reported_path: path.to_path_buf(),
                },
                cf_path,
                device: metadata.dev(),
                inode: metadata.ino(),
            },
        );
        Ok((canonical_path, previous))
    }

    // A recursive watch covers nested watches on the same volume. Non-recursive
    // ancestors may filter out deeper events, and FSEvents may not cross mounts.
    fn is_covered(&self, path: &Path, device: u64) -> bool {
        path.ancestors().skip(1).any(|ancestor| {
            self.watches
                .get(ancestor)
                .is_some_and(|covering| covering.info.is_recursive && covering.device == device)
        })
    }

    fn stream_roots(&self) -> impl Iterator<Item = (&PathBuf, &WatchEntry)> {
        self.watches
            .iter()
            .filter(|(path, entry)| !self.is_covered(path, entry.device))
    }

    fn stream_paths(&self) -> cf::CFRetained<cf::CFMutableArray<cf::CFString>> {
        let paths: cf::CFRetained<cf::CFMutableArray<cf::CFString>> = cf::CFMutableArray::empty();
        for (_, entry) in self.stream_roots() {
            paths.append(&entry.cf_path);
        }
        paths
    }

    fn run(&mut self) -> Result<()> {
        let stream_paths = self.stream_paths();
        if stream_paths.is_empty() {
            self.since_when = None;
            return Ok(());
        }

        // Over roughly RLIMIT_NOFILE/10 paths across all live streams, FSEvents
        // closes fd 0, which this process owns. The corruption then surfaces as
        // EBADF on unrelated files.
        let path_count = stream_paths.iter().count();
        let budget = fsevents_path_budget().unwrap_or(usize::MAX);
        let path_reservation =
            match FseventsPathReservation::acquire(&ACTIVE_FSEVENTS_PATHS, path_count, budget) {
                Ok(reservation) => reservation,
                Err(active_path_count) => {
                    let combined_path_count = active_path_count.saturating_add(path_count);
                    log::error!(
                        "refusing FSEvents stream: {combined_path_count} active paths exceed the \
                         safe limit of {budget}. Raise RLIMIT_NOFILE, watch fewer paths, or use \
                         macos_kqueue."
                    );
                    return Err(Error::new(ErrorKind::MaxFilesWatch));
                }
            };

        // We need to associate the stream context with our callback in order to propagate events
        // to the rest of the system. This will be owned by the stream, and will be freed when the
        // stream is closed. This means we will leak the context if we panic before reaching
        // `FSEventStreamRelease`.
        let recursive_info = Arc::new(Mutex::new(
            self.watches
                .iter()
                .map(|(path, entry)| (path.clone(), entry.info.clone()))
                .collect(),
        ));
        let context = Box::into_raw(Box::new(StreamContextInfo {
            event_handler: self.event_handler.clone(),
            recursive_info: recursive_info.clone(),
            event_kinds: self.event_kinds,
        }));

        let stream_context = fs::FSEventStreamContext {
            version: 0,
            info: context as *mut libc::c_void,
            retain: None,
            release: Some(release_context),
            copyDescription: None,
        };

        // Use a concrete starting ID even before the first callback. Otherwise a
        // restart before any events have been delivered would still use SinceNow
        // and discard the old stream's pending events.
        let since_when = self
            .since_when
            .unwrap_or_else(|| unsafe { fs::FSEventsGetCurrentEventId() });
        let stream = unsafe {
            fs::FSEventStreamCreate(
                cf::kCFAllocatorDefault,
                Some(callback),
                &stream_context as *const _ as *mut _,
                stream_paths.as_opaque(),
                since_when,
                self.latency,
                self.flags,
            )
        };

        // Wrapper to help send CFRunLoop types across threads.
        struct CFRunLoopSendWrapper(cf::CFRetained<cf::CFRunLoop>);

        // Safety:
        // - According to the Apple documentation, it's safe to move `CFRunLoop`s across threads.
        //   https://developer.apple.com/library/archive/documentation/Cocoa/Conceptual/Multithreading/ThreadSafetySummary/ThreadSafetySummary.html
        unsafe impl Send for CFRunLoopSendWrapper {}

        // Wrapper to help send FSEventStreamRef types across threads.
        struct FSEventStreamSendWrapper(fs::FSEventStreamRef);

        // TODO: Write docs for the safety of this impl.
        // SAFETY: Unclear?
        unsafe impl Send for FSEventStreamSendWrapper {}

        // move into thread
        let stream = FSEventStreamSendWrapper(stream);

        // channel to pass runloop around
        let (rl_tx, rl_rx) = unbounded();

        // Used to stop the runloop thread without relying on privileged APIs or
        // on `CFRunLoopIsWaiting()` becoming true under heavy event load.
        let stop_flag = Arc::new(AtomicBool::new(false));
        let stop_flag_thread = Arc::clone(&stop_flag);

        let thread_handle = thread::Builder::new()
            .name("notify-rs fsevents loop".to_string())
            .spawn(move || {
                // Keep the shared path count reserved until this stream is released.
                let _path_reservation = path_reservation;
                let _ = &stream;
                let stream = stream.0;

                unsafe {
                    // Safety:
                    // This may panic if OOM occurs.
                    // Related: https://github.com/madsmtm/objc2/issues/797
                    let cur_runloop =
                        cf::CFRunLoop::current().expect("Failed to get current runloop");

                    #[allow(deprecated)]
                    fs::FSEventStreamScheduleWithRunLoop(
                        stream,
                        &cur_runloop,
                        cf::kCFRunLoopDefaultMode.expect("Failed to get default runloop mode"),
                    );
                    if !fs::FSEventStreamStart(stream) {
                        fs::FSEventStreamInvalidate(stream);
                        fs::FSEventStreamRelease(stream);
                        rl_tx
                            .send(Err(Error::generic("unable to start FSEvent stream")))
                            .expect("Unable to send error for FSEventStreamStart");
                        return since_when;
                    }

                    // `stop()` will call `CFRunLoopStop` + `CFRunLoopWakeUp` and then join this
                    // thread.
                    rl_tx
                        .send(Ok(CFRunLoopSendWrapper(cur_runloop)))
                        .expect("Unable to send runloop to watcher");

                    // `CFRunLoopStop` is discarded if `stop()` calls it before this thread
                    // enters the runloop, so re-check the stop flag after each bounded run.
                    while !stop_flag_thread.load(Ordering::Acquire) {
                        let result = cf::CFRunLoop::run_in_mode(
                            cf::kCFRunLoopDefaultMode,
                            RUNLOOP_STOP_RECHECK_SECONDS,
                            false,
                        );
                        if result == cf::CFRunLoopRunResult::Finished {
                            break;
                        }
                    }
                    fs::FSEventStreamStop(stream);
                    // The runloop has finished all callbacks. Preserve the last
                    // delivered ID before releasing the stream so its replacement
                    // also receives events queued before or during the restart.
                    // RootChanged events have ID zero; never rewind past the
                    // point where this stream started.
                    let since_when = fs::FSEventStreamGetLatestEventId(stream).max(since_when);
                    fs::FSEventStreamInvalidate(stream);
                    fs::FSEventStreamRelease(stream);
                    since_when
                }
            })?;
        // block until runloop has been sent
        let runloop_wrapper = match rl_rx.recv() {
            Ok(Ok(runloop_wrapper)) => runloop_wrapper,
            Ok(Err(err)) => {
                thread_handle
                    .join()
                    .expect("thread to shut down after FSEvent stream startup failure");
                return Err(err);
            }
            Err(_) => {
                thread_handle
                    .join()
                    .expect("thread to shut down after FSEvent stream startup channel close");
                return Err(Error::generic(
                    "unable to receive FSEvent stream startup result",
                ));
            }
        };
        self.runloop = Some(RunLoopHandle {
            runloop: runloop_wrapper.0,
            stop_flag,
            thread_handle,
            recursive_info,
        });

        Ok(())
    }

    fn configure_raw_mode(&mut self, _config: Config, tx: Sender<Result<bool>>) {
        tx.send(Ok(false))
            .expect("configuration channel disconnect");
    }
}

// A twelfth rather than a tenth: the edge also shifts with how many descriptors
// the process already holds.
fn fsevents_path_budget() -> Option<usize> {
    let mut limit = unsafe { std::mem::zeroed::<libc::rlimit>() };
    if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit) } != 0 {
        return None;
    }
    let soft = usize::try_from(limit.rlim_cur).ok()?;
    Some(soft / 12)
}

unsafe extern "C-unwind" fn callback(
    stream_ref: fs::ConstFSEventStreamRef,
    info: *mut libc::c_void,
    num_events: libc::size_t,                          // size_t numEvents
    event_paths: NonNull<libc::c_void>,                // void *eventPaths
    event_flags: NonNull<fs::FSEventStreamEventFlags>, // const FSEventStreamEventFlags eventFlags[]
    event_ids: NonNull<fs::FSEventStreamEventId>,      // const FSEventStreamEventId eventIds[]
) {
    // Never unwind into CoreServices; if something goes wrong, drop the events and log.
    // This also protects against panics from user-provided `EventHandler` implementations.
    let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| unsafe {
        callback_impl(
            stream_ref,
            info,
            num_events,
            event_paths,
            event_flags,
            event_ids,
        )
    }))
    .map_err(|_| {
        log::error!("panic in FSEvents callback; dropping pending events");
    });
}

/// The watch an event at `path` belongs to, if any: the most specific of the
/// path itself, its parent, or an ancestor watched recursively.
///
/// The path and its ancestors are looked up by hash, nearest first, rather
/// than scanning every watched path, so deciding an event costs one lookup
/// per component of its path, however many paths are watched. The callback
/// runs this for every event FSEvents delivers, and a scan of every watch
/// slows delivery in proportion to how many there are.
fn watch_for<'a>(
    recursive_info: &'a HashMap<PathBuf, WatchInfo>,
    path: &Path,
) -> Option<(&'a PathBuf, &'a WatchInfo)> {
    if let Some(watch) = recursive_info.get_key_value(path) {
        return Some(watch);
    }
    path.ancestors()
        .skip(1)
        .enumerate()
        .find_map(|(depth, ancestor)| {
            let (watch_path, watch_info) = recursive_info.get_key_value(ancestor)?;
            (watch_info.is_recursive || depth == 0).then_some((watch_path, watch_info))
        })
}

unsafe fn callback_impl(
    _stream_ref: fs::ConstFSEventStreamRef,
    info: *mut libc::c_void,
    num_events: libc::size_t,                          // size_t numEvents
    event_paths: NonNull<libc::c_void>,                // void *eventPaths
    event_flags: NonNull<fs::FSEventStreamEventFlags>, // const FSEventStreamEventFlags eventFlags[]
    _event_ids: NonNull<fs::FSEventStreamEventId>,     // const FSEventStreamEventId eventIds[]
) {
    let event_paths = event_paths.as_ptr() as *const *const libc::c_char;
    let info = info as *const StreamContextInfo;
    let event_handler_mutex = &(*info).event_handler;
    let event_kinds = (*info).event_kinds;
    let mut event_handler_guard = None;

    for p in 0..num_events {
        // Paths are not guaranteed to be valid UTF-8 (e.g. NFS); keep them as raw bytes.
        let path = CStr::from_ptr(*event_paths.add(p));
        let path = Path::new(OsStr::from_bytes(path.to_bytes()));

        let raw_flag = *event_flags.as_ptr().add(p) as u32;
        let flag = StreamFlags::from_bits_truncate(raw_flag);
        let unknown_bits = raw_flag & !StreamFlags::all().bits();
        if unknown_bits != 0 {
            // `FSEventStreamEventFlags` is an extensible bitfield; tolerate future flags.
            log::trace!("unknown FSEventStreamEventFlags bits: 0x{unknown_bits:08x}");
        }

        let recursive_info = (*info)
            .recursive_info
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let Some((watch_path, watch_info)) = watch_for(&recursive_info, path) else {
            // FSEvents reports MustScanSubDirs at an ancestor of the
            // registered watches (often `/` or the common parent). Those
            // paths fail membership, but the recovery notice must still
            // reach the handler.
            if flag.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
                drop(recursive_info);
                let mut ev = Event::new(EventKind::Other).set_flag(Flag::Rescan);
                ev = if flag.contains(StreamFlags::USER_DROPPED) {
                    ev.set_info("rescan: user dropped")
                } else if flag.contains(StreamFlags::KERNEL_DROPPED) {
                    ev.set_info("rescan: kernel dropped")
                } else {
                    ev
                };
                ev.paths.push(path.to_path_buf());
                if event_kinds.matches(&ev.kind) {
                    let event_handler = event_handler_guard.get_or_insert_with(|| {
                        match event_handler_mutex.lock() {
                            Ok(guard) => guard,
                            Err(poisoned) => poisoned.into_inner(),
                        }
                    });
                    let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                        event_handler.handle_event(Ok(ev));
                    }))
                    .map_err(|_| {
                        log::error!("panic in FSEvents event handler; dropping event");
                    });
                }
            }
            continue;
        };
        let translated_count = translated_event_count(&flag, true);
        if translated_count == 0 {
            continue;
        }
        // Most FSEvents flags produce one Event; move the reported path in that case.
        let mut event_path = Some(reported_path(watch_path, &watch_info.reported_path, path));
        // Do not hold the watch information lock while invoking user code.
        drop(recursive_info);
        let single_translated_event = translated_count == 1;

        log::trace!("FSEvent: path = `{}`, flag = {:?}", path.display(), flag);

        translate_flags_with(flag, true, |mut ev| {
            // Filter events based on EventKindMask before adding the path.
            if !event_kinds.matches(&ev.kind) {
                return;
            }
            if single_translated_event {
                ev.paths
                    .push(event_path.take().expect("single translated event path"));
            } else {
                ev.paths
                    .push(event_path.as_ref().expect("translated event path").clone());
            }

            let event_handler =
                event_handler_guard.get_or_insert_with(|| match event_handler_mutex.lock() {
                    Ok(guard) => guard,
                    Err(poisoned) => poisoned.into_inner(),
                });
            // Protect against panicking event handlers, which would otherwise unwind into
            // the CoreServices callback.
            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                event_handler.handle_event(Ok(ev));
            }))
            .map_err(|_| {
                log::error!("panic in FSEvents event handler; dropping event");
            });
        });
    }
}

impl Watcher for FsEventWatcher {
    /// Create a new watcher.
    fn new<F: EventHandler>(event_handler: F, config: Config) -> Result<Self> {
        Self::from_event_handler(
            Arc::new(Mutex::new(event_handler)),
            config.event_kinds(),
            config.fsevent_latency().as_secs_f64(),
        )
    }

    fn watch(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()> {
        self.watch_inner(path, recursive_mode)
    }

    fn unwatch(&mut self, path: &Path) -> Result<()> {
        self.unwatch_inner(path)
    }

    fn update_paths(&mut self, ops: Vec<PathOp>) -> crate::StdResult<(), crate::UpdatePathsError> {
        self.update_paths_inner(ops)
    }

    fn configure(&mut self, config: Config) -> Result<bool> {
        let (tx, rx) = unbounded();
        self.configure_raw_mode(config, tx);
        rx.recv()?
    }

    fn watched_paths(&self) -> Result<Vec<(PathBuf, RecursiveMode)>> {
        // Unlike the channel-based backends, FSEvents keeps watch state on the watcher itself.
        // The runloop callback reads a separate map of watch information, so it does not
        // mutate or read this map concurrently.
        Ok(self
            .watches
            .iter()
            .map(|(_path, WatchEntry { info, .. })| {
                (
                    info.reported_path.clone(),
                    if info.is_recursive {
                        RecursiveMode::Recursive
                    } else {
                        RecursiveMode::NonRecursive
                    },
                )
            })
            .collect())
    }

    fn kind() -> crate::WatcherKind {
        crate::WatcherKind::Fsevent
    }
}

impl Drop for FsEventWatcher {
    fn drop(&mut self) {
        self.stop();
    }
}

/// Grabbed from <https://docs.rs/fsevent-sys/4.1.0/src/fsevent_sys/core_foundation.rs.html#149-230>.
///
/// TODO: Could we simplify this?
unsafe fn path_to_cfstring_ref(
    source: &Path,
    err: &mut *mut cf::CFError,
) -> Option<cf::CFRetained<cf::CFString>> {
    let url = cf::CFURL::from_file_path(source)?;

    let mut placeholder = url.absolute_url()?;

    let imaginary = cf::CFMutableArray::empty();

    while !unsafe { placeholder.resource_is_reachable(err) } {
        if let Some(child) = placeholder.last_path_component() {
            imaginary.insert(0, &*child);
        }

        placeholder = cf::CFURL::new_copy_deleting_last_path_component(None, Some(&placeholder))?;
    }

    let url = unsafe { cf::CFURL::new_file_reference_url(None, Some(&placeholder), err) }?;

    let mut placeholder = unsafe { cf::CFURL::new_file_path_url(None, Some(&url), err) }?;

    for component in imaginary {
        placeholder = cf::CFURL::new_copy_appending_path_component(
            None,
            Some(&placeholder),
            Some(&component),
            false,
        )?;
    }

    placeholder.file_system_path(cf::CFURLPathStyle::CFURLPOSIXPathStyle)
}

#[cfg(test)]
mod tests {
    use std::sync::mpsc;
    use std::time::{Duration, Instant};

    use crate::{ErrorKind, WatchPathConfig};

    use super::*;
    use crate::test::*;

    fn watcher() -> (TestWatcher<FsEventWatcher>, Receiver) {
        channel()
    }

    fn watch_info(is_recursive: bool) -> WatchInfo {
        WatchInfo {
            is_recursive,
            reported_path: PathBuf::from("/reported"),
        }
    }

    #[test]
    fn watch_for_matches_each_kind_of_watch() {
        let mut recursive_info = HashMap::new();
        recursive_info.insert(PathBuf::from("/repo"), watch_info(false));
        recursive_info.insert(PathBuf::from("/repo/src"), watch_info(false));
        recursive_info.insert(PathBuf::from("/deep"), watch_info(true));
        recursive_info.insert(PathBuf::from("/deep/inner"), watch_info(false));
        recursive_info.insert(PathBuf::from("/single/file.txt"), watch_info(false));

        let cases = [
            // A watched path itself, recursive or not.
            ("/repo", Some("/repo")),
            ("/deep", Some("/deep")),
            ("/single/file.txt", Some("/single/file.txt")),
            // A direct child of a non-recursive watch.
            ("/repo/readme.md", Some("/repo")),
            ("/repo/src/main.rs", Some("/repo/src")),
            // A grandchild of a non-recursive watch whose own parent is unwatched.
            ("/repo/target/debug", None),
            // Any descendant of a recursive watch.
            ("/deep/a", Some("/deep")),
            ("/deep/a/b/c", Some("/deep")),
            // The more specific of a recursive watch and a nested watch.
            ("/deep/inner/x", Some("/deep/inner")),
            ("/deep/inner/x/y", Some("/deep")),
            // A sibling sharing a prefix of characters, not of components.
            ("/repository/readme.md", None),
            ("/deeper/a", None),
            // The parent of a watched file, and its other children.
            ("/single", None),
            ("/single/other.txt", None),
            // Nothing watched above it at all.
            ("/elsewhere/x", None),
            ("/", None),
        ];
        for (path, expected) in cases {
            let matched = watch_for(&recursive_info, Path::new(path))
                .map(|(watch_path, _)| watch_path.as_path());
            assert_eq!(matched, expected.map(Path::new), "{path}");
        }
    }

    #[test]
    fn watch_for_agrees_with_a_scan_of_every_watch() {
        // The decision a scan of every watched path makes, keeping the
        // longest match, which `watch_for` replaces: kept here as the
        // reference it must agree with.
        fn scan<'a>(
            recursive_info: &'a HashMap<PathBuf, WatchInfo>,
            path: &Path,
        ) -> Option<&'a PathBuf> {
            recursive_info
                .iter()
                .filter(|(watch_path, watch_info)| {
                    path.starts_with(watch_path)
                        && (watch_info.is_recursive
                            || path == watch_path.as_path()
                            || path.parent() == Some(watch_path.as_path()))
                })
                .map(|(watch_path, _)| watch_path)
                .max_by_key(|watch_path| watch_path.as_os_str().as_bytes().len())
        }

        let watches: [(&str, bool); 7] = [
            ("/a", false),
            ("/a/b", true),
            ("/a/b/c", false),
            ("/a/b/c/d", true),
            ("/x/y", false),
            ("/x/y/z.txt", false),
            ("/r", true),
        ];
        let paths = [
            "/",
            "/a",
            "/a/b",
            "/a/b/c",
            "/a/b/c/d",
            "/a/b/c/d/e",
            "/a/b/c/d/e/f",
            "/a/b/q/r",
            "/a/z",
            "/a/z/q",
            "/ab",
            "/x",
            "/x/y",
            "/x/y/z.txt",
            "/x/y/w",
            "/x/y/w/v",
            "/r",
            "/r/s/t",
            "/rs",
        ];
        // Every subset of the watches, so each path is judged against watches
        // that are recursive, non-recursive, nested in each other and absent.
        for mask in 0u32..(1 << watches.len()) {
            let recursive_info: HashMap<PathBuf, WatchInfo> = watches
                .iter()
                .enumerate()
                .filter(|(index, _)| mask & (1 << index) != 0)
                .map(|(_, &(watch_path, is_recursive))| {
                    (PathBuf::from(watch_path), watch_info(is_recursive))
                })
                .collect();
            for path in paths {
                assert_eq!(
                    watch_for(&recursive_info, Path::new(path)).map(|(watch_path, _)| watch_path),
                    scan(&recursive_info, Path::new(path)),
                    "{path} against {recursive_info:?}"
                );
            }
        }
    }

    fn runloop_thread_id(watcher: &FsEventWatcher) -> thread::ThreadId {
        watcher
            .runloop
            .as_ref()
            .expect("watcher to be running")
            .thread_handle
            .thread()
            .id()
    }

    fn wait_for_path(rx: &mpsc::Receiver<Result<Event>>, path: &Path) {
        let deadline = Instant::now() + Duration::from_secs(10);
        loop {
            let event = rx
                .recv_timeout(deadline.saturating_duration_since(Instant::now()))
                .unwrap_or_else(|err| panic!("missing event for {}: {err}", path.display()))
                .expect("watcher event");
            if event.paths.iter().any(|event_path| event_path == path) {
                return;
            }
        }
    }

    // Block the native runloop, rather than racing a sleep against event delivery.
    // The source owns its context and is released along with the runloop.
    fn pause_runloop(watcher: &FsEventWatcher) -> mpsc::Sender<()> {
        type Pause = (mpsc::Sender<()>, mpsc::Receiver<()>);

        unsafe extern "C-unwind" fn perform(info: *mut libc::c_void) {
            let (entered, resume) = unsafe { &*info.cast::<Pause>() };
            let _ = entered.send(());
            let _ = resume.recv_timeout(Duration::from_secs(20));
        }

        unsafe extern "C-unwind" fn release(info: *const libc::c_void) {
            drop(unsafe { Box::from_raw(info.cast_mut().cast::<Pause>()) });
        }

        let (entered_tx, entered_rx) = mpsc::channel::<()>();
        let (resume_tx, resume_rx) = mpsc::channel();
        let mut context = cf::CFRunLoopSourceContext {
            version: 0,
            info: Box::into_raw(Box::new((entered_tx, resume_rx))).cast(),
            retain: None,
            release: Some(release),
            copyDescription: None,
            equal: None,
            hash: None,
            schedule: None,
            cancel: None,
            perform: Some(perform),
        };
        let runloop = &watcher.runloop.as_ref().expect("running watcher").runloop;
        let source = unsafe {
            cf::CFRunLoopSource::new(cf::kCFAllocatorDefault, isize::MIN, &mut context)
                .expect("pause source")
        };
        runloop.add_source(Some(&source), unsafe { cf::kCFRunLoopDefaultMode });
        source.signal();
        runloop.wake_up();
        entered_rx
            .recv_timeout(Duration::from_secs(10))
            .expect("runloop paused");
        resume_tx
    }

    #[test]
    fn queued_events_survive_watch_changes() {
        for change in [
            "watch",
            "rewatch",
            "unwatch",
            "update_paths",
            "failed_update_paths",
        ] {
            let dir = tempfile::tempdir().unwrap();
            let root = dir.path().canonicalize().unwrap();
            let a = root.join("a");
            let b = root.join("b");
            let c = root.join("c");
            for path in [&a, &b, &c] {
                std::fs::create_dir(path).unwrap();
            }

            let (tx, rx) = mpsc::channel();
            let mut watcher = FsEventWatcher::new(tx, Config::default()).unwrap();
            watcher
                .update_paths(vec![
                    PathOp::watch_recursive(a.clone()),
                    PathOp::watch_recursive(b.clone()),
                ])
                .unwrap();

            // A separate stream confirms that FSEvents has produced the pending
            // event while the stream under test cannot invoke its callback.
            let (observer_tx, observer_rx) = mpsc::channel();
            let mut observer = FsEventWatcher::new(observer_tx, Config::default()).unwrap();
            observer.watch(&a, RecursiveMode::Recursive).unwrap();
            let resume = pause_runloop(&watcher);
            let path = a.join("pending");
            std::fs::write(&path, "pending event").unwrap();
            wait_for_path(&observer_rx, &path);
            assert!(
                rx.try_iter()
                    .all(|event| !event.unwrap().paths.contains(&path))
            );

            let stop_flag = watcher.runloop.as_ref().unwrap().stop_flag.clone();
            thread::scope(|scope| {
                scope.spawn(move || {
                    let deadline = Instant::now() + Duration::from_secs(10);
                    while !stop_flag.load(Ordering::Acquire) && Instant::now() < deadline {
                        thread::sleep(Duration::from_millis(1));
                    }
                    let stopped = stop_flag.load(Ordering::Acquire);
                    resume.send(()).expect("resume runloop");
                    assert!(stopped, "{change} did not stop the old stream");
                });
                match change {
                    "watch" => watcher.watch(&c, RecursiveMode::Recursive).unwrap(),
                    "rewatch" => watcher.watch(&a, RecursiveMode::NonRecursive).unwrap(),
                    "unwatch" => watcher.unwatch(&b).unwrap(),
                    "update_paths" => watcher
                        .update_paths(vec![
                            PathOp::unwatch(b.clone()),
                            PathOp::watch_recursive(c.clone()),
                        ])
                        .unwrap(),
                    "failed_update_paths" => {
                        let err = watcher
                            .update_paths(vec![
                                PathOp::unwatch(b.clone()),
                                PathOp::unwatch(root.join("missing")),
                                PathOp::watch_recursive(c.clone()),
                            ])
                            .unwrap_err();
                        assert!(matches!(err.source.kind, ErrorKind::WatchNotFound));
                        assert_eq!(err.remaining.len(), 1);
                    }
                    _ => unreachable!(),
                }
            });
            wait_for_path(&rx, &path);
        }
    }

    #[test]
    fn events_during_stream_restart_are_replayed() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().canonicalize().unwrap();
        let (tx, rx) = mpsc::channel();
        let mut watcher = FsEventWatcher::new(tx, Config::default()).unwrap();
        watcher.watch(&root, RecursiveMode::Recursive).unwrap();

        let before = root.join("before");
        std::fs::write(&before, "before restart").unwrap();
        wait_for_path(&rx, &before);
        watcher.stop();
        assert!(watcher.since_when.is_some());

        let (observer_tx, observer_rx) = mpsc::channel();
        let mut observer = FsEventWatcher::new(observer_tx, Config::default()).unwrap();
        observer.watch(&root, RecursiveMode::Recursive).unwrap();
        let during = root.join("during");
        std::fs::write(&during, "during restart").unwrap();
        wait_for_path(&observer_rx, &during);

        watcher.run().unwrap();
        wait_for_path(&rx, &during);

        watcher.unwatch(&root).unwrap();
        assert_eq!(watcher.since_when, None);
    }

    #[test]
    fn unchanged_stream_roots_keep_pending_events() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().canonicalize().unwrap();
        let child = root.join("child");
        let nested = child.join("nested");
        let missing = root.join("missing");
        std::fs::create_dir_all(&nested).unwrap();

        let (tx, rx) = mpsc::channel();
        let mut watcher = FsEventWatcher::new(tx, Config::default()).unwrap();
        watcher.watch(&root, RecursiveMode::Recursive).unwrap();
        watcher.watch(&child, RecursiveMode::Recursive).unwrap();
        let thread_id = runloop_thread_id(&watcher);
        let resume = pause_runloop(&watcher);
        let pending = root.join("pending");
        std::fs::write(&pending, "pending event").unwrap();

        watcher.watch(&root, RecursiveMode::Recursive).unwrap();
        assert_eq!(runloop_thread_id(&watcher), thread_id);
        watcher.unwatch(&child).unwrap();
        assert_eq!(runloop_thread_id(&watcher), thread_id);
        assert!(matches!(
            watcher.unwatch(&missing).unwrap_err().kind,
            ErrorKind::WatchNotFound
        ));
        assert_eq!(runloop_thread_id(&watcher), thread_id);

        watcher.update_paths(vec![]).unwrap();
        assert_eq!(runloop_thread_id(&watcher), thread_id);
        watcher
            .update_paths(vec![PathOp::watch_recursive(child.clone())])
            .unwrap();
        assert_eq!(runloop_thread_id(&watcher), thread_id);
        watcher
            .update_paths(vec![PathOp::unwatch(child.clone())])
            .unwrap();
        assert_eq!(runloop_thread_id(&watcher), thread_id);

        let err = watcher
            .update_paths(vec![PathOp::watch_recursive(missing.clone())])
            .unwrap_err();
        assert!(matches!(err.source.kind, ErrorKind::PathNotFound));
        assert_eq!(runloop_thread_id(&watcher), thread_id);

        // Successful operations before an error must still update the callback,
        // even when the native stream's roots have not changed.
        let err = watcher
            .update_paths(vec![
                PathOp::Watch(
                    root.clone(),
                    WatchPathConfig::new(RecursiveMode::NonRecursive),
                ),
                PathOp::unwatch(missing),
                PathOp::watch_recursive(child),
            ])
            .unwrap_err();
        assert!(matches!(err.source.kind, ErrorKind::WatchNotFound));
        assert_eq!(err.remaining.len(), 1);
        assert_eq!(runloop_thread_id(&watcher), thread_id);
        let recursive_info = watcher
            .runloop
            .as_ref()
            .unwrap()
            .recursive_info
            .lock()
            .unwrap();
        assert!(!recursive_info[&root].is_recursive);
        assert!(watch_for(&recursive_info, &nested.join("entry")).is_none());
        drop(recursive_info);
        resume.send(()).expect("resume runloop");
        wait_for_path(&rx, &pending);

        let path = root.join("entry");
        std::fs::write(&path, "still watched").unwrap();
        wait_for_path(&rx, &path);
    }

    #[test]
    fn rewatching_a_replaced_root_restarts_the_stream() {
        for batch in [false, true] {
            let dir = tempfile::tempdir().unwrap();
            let parent = dir.path().canonicalize().unwrap();
            let root = parent.join("root");
            std::fs::create_dir(&root).unwrap();

            let (tx, rx) = mpsc::channel();
            let mut watcher = FsEventWatcher::new(tx, Config::default()).unwrap();
            watcher.watch(&root, RecursiveMode::Recursive).unwrap();
            let thread_id = runloop_thread_id(&watcher);

            std::fs::rename(&root, parent.join("old")).unwrap();
            std::fs::create_dir(&root).unwrap();
            if batch {
                watcher
                    .update_paths(vec![PathOp::watch_recursive(root.clone())])
                    .unwrap();
            } else {
                watcher.watch(&root, RecursiveMode::Recursive).unwrap();
            }
            assert_ne!(runloop_thread_id(&watcher), thread_id);

            let path = root.join("entry");
            std::fs::write(&path, "replacement root").unwrap();
            wait_for_path(&rx, &path);
        }
    }

    #[test]
    fn rewatching_same_path_replaces_recursive_info() {
        let dir = tempfile::tempdir().unwrap();
        let mut watcher = FsEventWatcher::new(|_| {}, Config::default()).unwrap();

        watcher
            .append_path(dir.path(), RecursiveMode::Recursive)
            .expect("watch recursively");
        watcher
            .append_path(dir.path(), RecursiveMode::NonRecursive)
            .expect("rewatch non-recursively");

        let watched = watcher.watched_paths().expect("watched paths");
        assert_eq!(
            watched,
            vec![(dir.path().to_path_buf(), RecursiveMode::NonRecursive)]
        );
        assert_eq!(watcher.stream_paths().iter().count(), 1);
    }

    #[test]
    fn only_recursive_ancestors_cover_nested_watches() {
        let dir = tempfile::tempdir().unwrap();
        let child = dir.path().join("child");
        let grandchild = child.join("first").join("second").join("grandchild");
        std::fs::create_dir_all(&grandchild).unwrap();

        let mut watcher = FsEventWatcher::new(|_| {}, Config::default()).unwrap();
        for (path, mode) in [
            (dir.path(), RecursiveMode::Recursive),
            (child.as_path(), RecursiveMode::NonRecursive),
            (grandchild.as_path(), RecursiveMode::Recursive),
        ] {
            watcher.append_path(path, mode).expect("watch");
        }

        assert_eq!(watcher.stream_paths().iter().count(), 1);
        assert_eq!(watcher.watched_paths().expect("watched paths").len(), 3);

        watcher.remove_path(dir.path()).expect("unwatch parent");
        // The remaining non-recursive child would discard changes to the deeper
        // watch's intermediate ancestors, so both watches need stream roots.
        assert_eq!(watcher.stream_paths().iter().count(), 2);

        watcher.remove_path(&child).expect("unwatch child");
        assert_eq!(watcher.stream_paths().iter().count(), 1);
    }

    #[test]
    fn sibling_watches_each_get_a_stream_root() {
        let dir = tempfile::tempdir().unwrap();
        let first = dir.path().join("first");
        let second = dir.path().join("second");
        std::fs::create_dir(&first).unwrap();
        std::fs::create_dir(&second).unwrap();

        let mut watcher = FsEventWatcher::new(|_| {}, Config::default()).unwrap();
        watcher
            .append_path(&first, RecursiveMode::Recursive)
            .expect("watch first");
        watcher
            .append_path(&second, RecursiveMode::Recursive)
            .expect("watch second");

        assert_eq!(watcher.stream_paths().iter().count(), 2);
    }

    #[test]
    fn covering_watch_keeps_receiving_outside_the_nested_watch() {
        let tmpdir = testdir();
        let child = tmpdir.path().join("child");
        let sibling = tmpdir.path().join("sibling");
        std::fs::create_dir(&child).expect("create child");
        std::fs::create_dir(&sibling).expect("create sibling");

        let (mut watcher, mut rx) = watcher();
        watcher.watch_recursively(&tmpdir);
        let thread_id = runloop_thread_id(&watcher.watcher);
        watcher.watch_recursively(&child);
        assert_eq!(runloop_thread_id(&watcher.watcher), thread_id);

        let path = sibling.join("entry");
        std::fs::File::create_new(&path).expect("create");

        rx.wait_unordered([expected(path).create_file()]);
    }

    #[test]
    fn covered_watch_updates_callback_without_restarting_stream() {
        let tmpdir = testdir();
        let root = tmpdir.path().join("root");
        let child = root.join("child");
        let nested = child.join("nested");
        let alias = tmpdir.path().join("alias");
        std::fs::create_dir_all(&nested).expect("create dirs");
        std::os::unix::fs::symlink(&child, &alias).expect("symlink");

        let (mut watcher, mut rx) = watcher();
        watcher.watch_recursively(&root);
        let thread_id = runloop_thread_id(&watcher.watcher);
        watcher.watch_recursively(&alias);
        assert_eq!(runloop_thread_id(&watcher.watcher), thread_id);

        std::fs::File::create_new(nested.join("first")).expect("create");
        rx.wait_unordered([expected(alias.join("nested/first")).create_file()]);

        watcher.watch_nonrecursively(&alias);
        assert_eq!(runloop_thread_id(&watcher.watcher), thread_id);
        assert!(
            watcher
                .watcher
                .watched_paths()
                .expect("watched paths")
                .contains(&(alias.clone(), RecursiveMode::NonRecursive))
        );

        std::fs::File::create_new(child.join("immediate")).expect("create");
        std::fs::File::create_new(nested.join("second")).expect("create");
        rx.wait_unordered([
            expected(alias.join("immediate")).create_file(),
            // The non-recursive alias no longer matches deeper events, but the
            // original recursive root still covers them.
            expected(nested.join("second")).create_file(),
        ]);
    }

    #[test]
    fn non_recursive_parent_does_not_skip_stream_restart() {
        let tmpdir = testdir();
        let child = tmpdir.path().join("child");
        let nested = child.join("nested");
        std::fs::create_dir_all(&nested).expect("create dirs");

        let (mut watcher, mut rx) = watcher();
        watcher.watch_nonrecursively(&tmpdir);
        let thread_id = runloop_thread_id(&watcher.watcher);
        watcher.watch_recursively(&child);
        assert_ne!(runloop_thread_id(&watcher.watcher), thread_id);

        let path = nested.join("entry");
        std::fs::File::create_new(&path).expect("create");
        rx.wait_unordered([expected(path).create_file()]);
    }

    #[test]
    fn failed_watch_keeps_stream_running() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();
        watcher.watch_recursively(&tmpdir);
        let thread_id = runloop_thread_id(&watcher.watcher);

        let err = watcher
            .watcher
            .watch(&tmpdir.path().join("missing"), RecursiveMode::Recursive)
            .expect_err("watch missing path");
        assert!(matches!(err.kind, ErrorKind::PathNotFound));
        assert_eq!(runloop_thread_id(&watcher.watcher), thread_id);

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");
        rx.wait_unordered([expected(path).create_file()]);
    }

    #[test]
    fn nested_watch_keeps_receiving_after_unwatching_its_parent() {
        let tmpdir = testdir();
        let child = tmpdir.path().join("child");
        std::fs::create_dir(&child).expect("create dir");

        let (mut watcher, mut rx) = watcher();
        watcher.watch_recursively(&tmpdir);
        watcher.watch_recursively(&child);
        watcher
            .watcher
            .unwatch(tmpdir.path())
            .expect("unwatch parent");

        let path = child.join("entry");
        std::fs::File::create_new(&path).expect("create");

        rx.wait_unordered([expected(path).create_file()]);
    }

    #[test]
    fn stop_does_not_wait_for_runloop_to_be_waiting() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::mpsc;
        use std::time::Instant;

        // Regression test for a shutdown hang where `stop()` waited for
        // `CFRunLoopIsWaiting()` to become true. If the runloop is busy (e.g. under
        // heavy event load), it may never enter a waiting state.

        let dir = tempfile::tempdir().unwrap();

        let (tx, _rx) = mpsc::channel::<crate::Result<Event>>();
        let mut watcher = FsEventWatcher::new(tx, Default::default()).unwrap();
        watcher.watch(dir.path(), RecursiveMode::Recursive).unwrap();

        let runloop = watcher
            .runloop
            .as_ref()
            .expect("watcher to be running")
            .runloop
            .clone();
        let mode = unsafe { cf::kCFRunLoopDefaultMode.expect("default runloop mode") };

        // Keep the runloop continuously "busy" by creating a source that signals itself in its
        // perform callback.
        struct SourceHammer {
            source: *const cf::CFRunLoopSource,
            fires: AtomicUsize,
        }

        unsafe extern "C-unwind" fn hammer_source(info: *mut std::ffi::c_void) {
            let Some(hammer) = (info as *const SourceHammer).as_ref() else {
                return;
            };
            hammer.fires.fetch_add(1, Ordering::Relaxed);

            // Signal the source again so the runloop has more work to do.
            let Some(source) = (hammer.source as *const cf::CFRunLoopSource).as_ref() else {
                return;
            };
            source.signal();
        }

        let mut hammer = Box::new(SourceHammer {
            source: std::ptr::null(),
            fires: AtomicUsize::new(0),
        });

        let mut ctx = cf::CFRunLoopSourceContext {
            version: 0,
            info: (&mut *hammer as *mut SourceHammer).cast(),
            retain: None,
            release: None,
            copyDescription: None,
            equal: None,
            hash: None,
            schedule: None,
            cancel: None,
            perform: Some(hammer_source),
        };

        let source = unsafe {
            cf::CFRunLoopSource::new(cf::kCFAllocatorDefault, 0, &mut ctx)
                .expect("source to be created")
        };
        hammer.source = cf::CFRetained::as_ptr(&source).as_ptr();

        runloop.add_source(Some(&source), Some(mode));
        source.signal();
        runloop.wake_up();

        // Ensure our setup actually made the runloop busy.
        let setup_start = Instant::now();
        while hammer.fires.load(Ordering::Relaxed) == 0
            && setup_start.elapsed() < Duration::from_secs(1)
        {
            std::thread::yield_now();
        }
        assert!(
            hammer.fires.load(Ordering::Relaxed) > 0,
            "runloop source never fired; test setup failed"
        );

        let (done_tx, done_rx) = mpsc::channel::<()>();
        std::thread::spawn(move || {
            drop(watcher);
            let _ = done_tx.send(());
        });

        // If shutdown regresses, this would hang indefinitely; keep the test bounded.
        done_rx
            .recv_timeout(Duration::from_secs(5))
            .expect("dropping FsEventWatcher timed out (possible shutdown hang)");

        // No cleanup: The source is owned by the runloop; removing sources cross-thread can be
        // sensitive on some systems. Dropping the last reference to the runloop will release it.
    }

    #[test]
    fn test_fsevent_watcher_drop() {
        use super::*;
        use std::time::Duration;

        let dir = tempfile::tempdir().unwrap();

        let (tx, rx) = std::sync::mpsc::channel();

        {
            let mut watcher = FsEventWatcher::new(tx, Default::default()).unwrap();
            watcher.watch(dir.path(), RecursiveMode::Recursive).unwrap();
            thread::sleep(Duration::from_millis(2000));
            println!("is running -> {}", watcher.is_running());

            thread::sleep(Duration::from_millis(1000));
            watcher.unwatch(dir.path()).unwrap();
            println!("is running -> {}", watcher.is_running());
        }

        thread::sleep(Duration::from_millis(1000));

        for res in rx {
            let e = res.unwrap();
            println!("debug => {:?} {:?}", e.kind, e.paths);
        }

        println!("in test: {} works", file!());
    }

    #[test]
    fn test_steam_context_info_send_and_sync() {
        fn check_send<T: Send + Sync>() {}
        check_send::<StreamContextInfo>();
    }

    /// Drives `callback_impl` with the given watch table and raw `(path, flags)` events,
    /// returning everything it delivered. Paths are bytes because FSEvents reports
    /// whatever the filesystem holds, which need not be UTF-8.
    fn run_callback(
        recursive_info: HashMap<PathBuf, WatchInfo>,
        events: &[(&[u8], u32)],
    ) -> Vec<crate::Result<Event>> {
        use std::ffi::CString;
        use std::ptr;

        let (tx, rx) = std::sync::mpsc::channel::<crate::Result<Event>>();
        let event_handler: Arc<Mutex<dyn EventHandler>> = Arc::new(Mutex::new(tx));

        let context = Box::new(StreamContextInfo {
            event_handler,
            recursive_info: Arc::new(Mutex::new(recursive_info)),
            event_kinds: EventKindMask::ALL,
        });
        let context_ptr = Box::into_raw(context) as *mut libc::c_void;

        let c_paths: Vec<CString> = events
            .iter()
            .map(|(path, _)| CString::new(*path).expect("cstring"))
            .collect();
        let path_ptrs: Vec<_> = c_paths.iter().map(|p| p.as_ptr()).collect();
        let event_paths = NonNull::new(path_ptrs.as_ptr() as *mut libc::c_void).unwrap();

        let flags_arr: Vec<fs::FSEventStreamEventFlags> = events
            .iter()
            .map(|(_, flags)| *flags as fs::FSEventStreamEventFlags)
            .collect();
        let event_flags =
            NonNull::new(flags_arr.as_ptr() as *mut fs::FSEventStreamEventFlags).unwrap();

        let ids_arr: Vec<fs::FSEventStreamEventId> = (0..events.len())
            .map(|i| i as fs::FSEventStreamEventId)
            .collect();
        let event_ids = NonNull::new(ids_arr.as_ptr() as *mut fs::FSEventStreamEventId).unwrap();

        let res = std::panic::catch_unwind(|| unsafe {
            callback_impl(
                ptr::null(),
                context_ptr,
                events.len(),
                event_paths,
                event_flags,
                event_ids,
            );
        });
        unsafe {
            drop(Box::from_raw(context_ptr as *mut StreamContextInfo));
        }
        assert!(res.is_ok(), "callback_impl should not panic");

        rx.try_iter().collect()
    }

    #[test]
    fn callback_impl_delivers_rescan_for_ancestor_recovery_paths() {
        let mut recursive_info = HashMap::new();
        recursive_info.insert(
            PathBuf::from("/watched/a"),
            WatchInfo {
                is_recursive: true,
                reported_path: PathBuf::from("/watched/a"),
            },
        );
        recursive_info.insert(
            PathBuf::from("/watched/b"),
            WatchInfo {
                is_recursive: false,
                reported_path: PathBuf::from("/watched/b"),
            },
        );

        let recovery_flags = [
            StreamFlags::MUST_SCAN_SUBDIRS.bits(),
            (StreamFlags::MUST_SCAN_SUBDIRS | StreamFlags::USER_DROPPED).bits(),
            (StreamFlags::MUST_SCAN_SUBDIRS | StreamFlags::KERNEL_DROPPED).bits(),
        ];
        for flags in recovery_flags {
            for path in [
                b"/watched/a".as_slice(),
                b"/watched".as_slice(),
                b"/".as_slice(),
            ] {
                let events = run_callback(recursive_info.clone(), &[(path, flags)]);
                let path_buf = PathBuf::from(std::str::from_utf8(path).unwrap());
                assert!(
                    events.iter().any(|event| {
                        let event = event.as_ref().expect("callback event");
                        event.need_rescan() && event.paths == [path_buf.clone()]
                    }),
                    "missing rescan for {} flags {flags:#x}: {events:?}",
                    String::from_utf8_lossy(path)
                );
            }
        }

        let ordinary = (StreamFlags::ITEM_MODIFIED | StreamFlags::IS_FILE).bits();
        for (path, expected) in [
            (b"/watched".as_slice(), false),
            (b"/".as_slice(), false),
            (b"/watched/a/deep/file".as_slice(), true),
            (b"/watched/b/file".as_slice(), true),
            (b"/watched/b/deep/file".as_slice(), false),
        ] {
            let events = run_callback(recursive_info.clone(), &[(path, ordinary)]);
            assert_eq!(
                !events.is_empty(),
                expected,
                "ordinary path {}",
                String::from_utf8_lossy(path)
            );
            assert!(
                events
                    .iter()
                    .all(|event| !event.as_ref().expect("callback event").need_rescan())
            );
        }
    }

    #[test]
    fn callback_impl_handles_non_utf8_paths_without_panicking() {
        use std::os::unix::ffi::OsStrExt;

        let mut recursive_info = HashMap::new();
        recursive_info.insert(
            PathBuf::from("/tmp"),
            WatchInfo {
                is_recursive: true,
                reported_path: PathBuf::from("/tmp"),
            },
        );

        let bytes = b"/tmp/\xff";
        let events = run_callback(
            recursive_info,
            &[(bytes.as_slice(), StreamFlags::ITEM_CREATED.bits())],
        );

        let event = events
            .first()
            .expect("expected event")
            .as_ref()
            .expect("expected Ok(Event)");
        assert!(
            event.kind.is_create(),
            "expected create event, got {event:?}"
        );
        assert_eq!(event.paths.len(), 1);
        assert_eq!(event.paths[0].as_os_str().as_bytes(), bytes);
    }

    #[test]
    fn callback_impl_ignores_unknown_flag_bits_without_panicking() {
        // Include an unknown bit so the old `from_bits(...).unwrap_or_else(panic!)` behavior
        // would have panicked. New behavior should tolerate it.
        let unknown_mask = !StreamFlags::all().bits();
        let unknown_bit = unknown_mask & unknown_mask.wrapping_neg();
        assert_ne!(unknown_bit, 0, "StreamFlags unexpectedly uses all bits");
        let raw_flag = StreamFlags::ITEM_CREATED.bits() | unknown_bit;
        assert!(
            StreamFlags::from_bits(raw_flag).is_none(),
            "raw_flag must include an unknown bit for this test to be meaningful"
        );

        let mut recursive_info = HashMap::new();
        recursive_info.insert(
            PathBuf::from("/tmp"),
            WatchInfo {
                is_recursive: true,
                reported_path: PathBuf::from("/tmp"),
            },
        );

        let events = run_callback(recursive_info, &[(b"/tmp/file".as_slice(), raw_flag)]);

        let event = events
            .first()
            .expect("expected event")
            .as_ref()
            .expect("expected Ok(Event)");
        assert!(
            event.kind.is_create(),
            "expected create event, got {event:?}"
        );
    }

    #[test]
    fn translate_flags_ignores_is_file_only_events() {
        assert!(translate_flags(StreamFlags::IS_FILE, true).is_empty());
        assert!(
            translate_flags(StreamFlags::IS_FILE | StreamFlags::ITEM_CLONED, true).is_empty(),
            "type-only clone flags should not produce events"
        );
    }

    #[test]
    fn translate_flags_sets_clone_info_for_file_events() {
        let create = translate_flags(
            StreamFlags::ITEM_CREATED | StreamFlags::IS_FILE | StreamFlags::ITEM_CLONED,
            true,
        );
        assert_eq!(create.len(), 1);
        assert_eq!(create[0].kind, EventKind::Create(CreateKind::File));
        assert_eq!(create[0].info(), Some("is: clone"));

        let modify = translate_flags(
            StreamFlags::INODE_META_MOD
                | StreamFlags::ITEM_MODIFIED
                | StreamFlags::IS_FILE
                | StreamFlags::ITEM_CLONED,
            true,
        );
        assert_eq!(modify.len(), 2);
        assert!(
            modify
                .iter()
                .any(|e| matches!(e.kind, EventKind::Modify(ModifyKind::Metadata(_))))
        );
        assert!(
            modify
                .iter()
                .any(|e| matches!(e.kind, EventKind::Modify(ModifyKind::Data(_))))
        );
        assert!(
            modify.iter().all(|e| e.info() == Some("is: clone")),
            "all events should be annotated as clone-related: {modify:?}"
        );
    }

    #[test]
    fn translate_flags_does_not_override_existing_info() {
        let evs = translate_flags(
            StreamFlags::ROOT_CHANGED
                | StreamFlags::ITEM_REMOVED
                | StreamFlags::IS_FILE
                | StreamFlags::ITEM_CLONED,
            true,
        );
        assert_eq!(evs.len(), 1);
        assert_eq!(evs[0].info(), Some("root changed"));
    }

    #[test]
    fn does_not_crash_with_empty_path() {
        let mut watcher = FsEventWatcher::new(|_| {}, Default::default()).unwrap();

        let watch_result = watcher.watch(Path::new(""), RecursiveMode::Recursive);
        assert!(
            matches!(
                watch_result,
                Err(Error {
                    kind: ErrorKind::PathNotFound,
                    paths: _
                })
            ),
            "actual: {watch_result:#?}"
        );

        let unwatch_result = watcher.unwatch(Path::new(""));
        assert!(
            matches!(
                unwatch_result,
                Err(Error {
                    kind: ErrorKind::WatchNotFound,
                    paths: _
                })
            ),
            "actual: {unwatch_result:#?}"
        );
    }

    #[test]
    fn create_file() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();
        watcher.watch_recursively(&tmpdir);

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        rx.wait_unordered([expected(path).create_file()]);
    }

    #[test]
    fn write_file() {
        let tmpdir = testdir();

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        let (mut watcher, mut rx) = watcher();

        watcher.watch_recursively(&tmpdir);

        std::fs::write(&path, b"123").expect("write");

        rx.wait_unordered([expected(&path).modify_data_content()]);
    }

    #[test]
    fn chmod_file() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        let file = std::fs::File::create_new(&path).expect("create");
        let mut permissions = file.metadata().expect("metadata").permissions();
        permissions.set_readonly(true);

        watcher.watch_recursively(&tmpdir);
        file.set_permissions(permissions).expect("set_permissions");

        rx.wait_unordered([expected(&path).modify_meta_owner()]);
    }

    #[test]
    fn rename_file() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        watcher.watch_recursively(&tmpdir);
        let new_path = tmpdir.path().join("renamed");

        std::fs::rename(&path, &new_path).expect("rename");

        rx.wait_unordered([expected(path).rename_any(), expected(new_path).rename_any()]);
    }

    #[test]
    fn delete_file() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();
        let file = tmpdir.path().join("file");
        std::fs::write(&file, "").expect("write");

        watcher.watch_nonrecursively(&tmpdir);

        std::fs::remove_file(&file).expect("remove");

        rx.wait_unordered([expected(&file).remove_file()]);
    }

    #[test]
    fn delete_self_file() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();
        let file = tmpdir.path().join("file");
        std::fs::write(&file, "").expect("write");

        watcher.watch_nonrecursively(&file);

        std::fs::remove_file(&file).expect("remove");

        rx.wait_unordered([expected(file).remove_file()]);
    }

    #[test]
    fn delete_self_dir() {
        let tmpdir = testdir();
        let dir = tmpdir.path().join("dir");
        std::fs::create_dir(&dir).expect("create");

        let (mut watcher, mut rx) = watcher();
        watcher.watch_nonrecursively(&dir);

        std::fs::remove_dir(&dir).expect("remove");

        rx.wait_unordered([expected(&dir).remove_folder()]);
    }

    #[test]
    fn create_write_overwrite() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();
        let overwritten_file = tmpdir.path().join("overwritten_file");
        let overwriting_file = tmpdir.path().join("overwriting_file");
        std::fs::write(&overwritten_file, "123").expect("write1");

        watcher.watch_nonrecursively(&tmpdir);

        std::fs::File::create(&overwriting_file).expect("create");
        std::fs::write(&overwriting_file, "321").expect("write2");
        std::fs::rename(&overwriting_file, &overwritten_file).expect("rename");

        rx.wait_unordered([
            expected(&overwriting_file).create(),
            expected(&overwriting_file).modify_data_content().multiple(),
            expected(&overwriting_file).rename_any(),
            expected(&overwritten_file).rename_any(),
        ]);
    }

    #[test]
    fn create_dir() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();
        watcher.watch_recursively(&tmpdir);

        let path = tmpdir.path().join("entry");
        std::fs::create_dir(&path).expect("create");

        rx.wait_unordered([expected(&path).create_folder()]);
    }

    #[test]
    fn chmod_dir() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::create_dir(&path).expect("create_dir");
        let mut permissions = std::fs::metadata(&path).expect("metadata").permissions();
        permissions.set_readonly(true);

        watcher.watch_recursively(&tmpdir);
        std::fs::set_permissions(&path, permissions).expect("set_permissions");

        rx.wait_unordered([expected(&path).modify_meta_owner()]);
    }

    #[test]
    fn rename_dir() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        let new_path = tmpdir.path().join("new_path");
        std::fs::create_dir(&path).expect("create_dir");

        watcher.watch_recursively(&tmpdir);
        std::fs::rename(&path, &new_path).expect("rename");

        rx.wait_ordered([
            expected(&path).rename_any(),
            expected(&new_path).rename_any(),
        ]);
    }

    #[test]
    fn delete_dir() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::create_dir(&path).expect("create_dir");

        watcher.watch_recursively(&tmpdir);
        std::fs::remove_dir(&path).expect("remove");

        rx.wait_unordered([expected(path).remove_folder()]);
    }

    #[test]
    fn rename_dir_twice() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        let new_path = tmpdir.path().join("new_path");
        let new_path2 = tmpdir.path().join("new_path2");
        std::fs::create_dir(&path).expect("create_dir");

        watcher.watch_recursively(&tmpdir);
        std::fs::rename(&path, &new_path).expect("rename");
        std::fs::rename(&new_path, &new_path2).expect("rename2");

        rx.wait_unordered([
            expected(&path).rename_any(),
            expected(&new_path).rename_any(),
            expected(&new_path2).rename_any(),
        ]);
    }

    #[test]
    fn move_out_of_watched_dir() {
        let tmpdir = testdir();
        let subdir = tmpdir.path().join("subdir");
        let (mut watcher, mut rx) = watcher();

        let path = subdir.join("entry");
        std::fs::create_dir_all(&subdir).expect("create_dir_all");
        std::fs::File::create_new(&path).expect("create");

        watcher.watch_recursively(&subdir);
        let new_path = tmpdir.path().join("entry");

        std::fs::rename(&path, &new_path).expect("rename");

        rx.wait_unordered([expected(path).rename_any()]);
    }

    #[test]
    #[ignore = "https://github.com/notify-rs/notify/issues/729"]
    fn create_write_write_rename_write_remove() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let file1 = tmpdir.path().join("entry");
        let file2 = tmpdir.path().join("entry2");
        std::fs::File::create_new(&file2).expect("create file2");
        let new_path = tmpdir.path().join("renamed");

        watcher.watch_recursively(&tmpdir);
        std::fs::write(&file1, "123").expect("write 1");
        std::fs::write(&file2, "321").expect("write 2");
        std::fs::rename(&file1, &new_path).expect("rename");
        std::fs::write(&new_path, b"1").expect("write 3");
        std::fs::remove_file(&new_path).expect("remove");

        rx.wait_ordered([
            expected(&file1).create_file(),
            expected(&file1).modify_data_content(),
            expected(&file2).modify_data_content(),
            expected(&file1).rename_any(),
            expected(&new_path).rename_any(),
            expected(&new_path).modify_data_content(),
            expected(&new_path).remove_file(),
        ]);
    }

    #[test]
    fn rename_twice() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        watcher.watch_recursively(&tmpdir);
        let new_path1 = tmpdir.path().join("renamed1");
        let new_path2 = tmpdir.path().join("renamed2");

        std::fs::rename(&path, &new_path1).expect("rename1");
        std::fs::rename(&new_path1, &new_path2).expect("rename2");

        rx.wait_unordered([
            expected(&path).rename_any(),
            expected(&new_path1).rename_any(),
            expected(&new_path2).rename_any(),
        ]);
    }

    #[test]
    fn set_file_mtime() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        let file = std::fs::File::create_new(&path).expect("create");

        watcher.watch_recursively(&tmpdir);

        file.set_modified(
            std::time::SystemTime::now()
                .checked_sub(Duration::from_secs(60 * 60))
                .expect("time"),
        )
        .expect("set_time");

        rx.wait_unordered([expected(&path).modify_meta_any()]);
    }

    #[test]
    fn write_file_non_recursive_watch() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        watcher.watch_nonrecursively(&path);

        std::fs::write(&path, b"123").expect("write");

        rx.wait_unordered([expected(path).modify_data_content()]);
    }

    #[test]
    fn write_to_a_hardlink_pointed_to_the_watched_file_triggers_an_event() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let subdir = tmpdir.path().join("subdir");
        let file = subdir.join("file");
        let hardlink = tmpdir.path().join("hardlink");

        std::fs::create_dir(&subdir).expect("create");
        std::fs::write(&file, "").expect("file");
        std::fs::hard_link(&file, &hardlink).expect("hardlink");

        watcher.watch_nonrecursively(&file);

        std::fs::write(&hardlink, "123123").expect("write to the hard link");

        rx.wait_unordered([expected(file).modify_data_content()]);
    }

    #[test]
    fn recursive_creation() {
        let tmpdir = testdir();
        let nested1 = tmpdir.path().join("1");
        let nested2 = tmpdir.path().join("1/2");
        let nested3 = tmpdir.path().join("1/2/3");
        let nested4 = tmpdir.path().join("1/2/3/4");
        let nested5 = tmpdir.path().join("1/2/3/4/5");
        let nested6 = tmpdir.path().join("1/2/3/4/5/6");
        let nested7 = tmpdir.path().join("1/2/3/4/5/6/7");
        let nested8 = tmpdir.path().join("1/2/3/4/5/6/7/8");
        let nested9 = tmpdir.path().join("1/2/3/4/5/6/7/8/9");

        let (mut watcher, mut rx) = watcher();

        watcher.watch_recursively(&tmpdir);

        std::fs::create_dir_all(&nested9).expect("create_dir_all");

        rx.wait_ordered([
            expected(&nested1).create_folder(),
            expected(&nested2).create_folder(),
            expected(&nested3).create_folder(),
            expected(&nested4).create_folder(),
            expected(&nested5).create_folder(),
            expected(&nested6).create_folder(),
            expected(&nested7).create_folder(),
            expected(&nested8).create_folder(),
            expected(&nested9).create_folder(),
        ]);
    }

    #[test]
    fn fsevent_watcher_respects_event_kind_mask() {
        use crate::Watcher;
        use notify_types::event::EventKindMask;

        let tmpdir = testdir();
        let (tx, rx) = std::sync::mpsc::channel();

        // Create watcher with CREATE-only mask (no MODIFY events)
        let config = Config::default().with_event_kinds(EventKindMask::CREATE);

        let mut watcher = FsEventWatcher::new(tx, config).expect("create watcher");
        watcher
            .watch(tmpdir.path(), crate::RecursiveMode::Recursive)
            .expect("watch");

        let path = tmpdir.path().join("test_file");

        // Create a file - should generate CREATE event
        std::fs::File::create_new(&path).expect("create");

        // Small delay to let events propagate
        std::thread::sleep(Duration::from_millis(100));

        // Modify the file - should NOT generate event (filtered by mask)
        std::fs::write(&path, "modified content").expect("write modified");

        std::thread::sleep(Duration::from_millis(100));

        // Collect all events
        let events: Vec<_> = rx.try_iter().filter_map(|r| r.ok()).collect();

        // Should have CREATE event
        assert!(
            events.iter().any(|e| e.kind.is_create()),
            "Expected CREATE event, got: {:?}",
            events
        );

        // Should NOT have MODIFY event (filtered out)
        assert!(
            !events.iter().any(|e| e.kind.is_modify()),
            "Should not receive MODIFY events with CREATE-only mask, got: {:?}",
            events
        );
    }

    // Replaces a test that watched 4097 paths to provoke an `FSEventStreamStart` failure
    // (https://github.com/fsnotify/fsevents/issues/48). That path count is exactly what
    // closes fd 0, so the test corrupted the process it ran in and needed a `catch_unwind`
    // around its own cleanup to stay green.
    #[test]
    fn refuses_more_paths_than_fsevents_can_carry() {
        let budget = fsevents_path_budget().expect("path budget");
        if budget > 4096 {
            eprintln!("skipping: RLIMIT_NOFILE leaves a budget of {budget} paths");
            return;
        }

        let tmpdir = testdir();
        let (mut watcher, _rx) = watcher();

        let mut paths = Vec::new();
        for i in 0..=budget {
            let path = tmpdir.path().join(format!("dir_{i}"));
            std::fs::create_dir(&path).expect("create_dir");
            paths.push(PathOp::Watch(
                path,
                WatchPathConfig::new(RecursiveMode::NonRecursive),
            ));
        }

        let err = watcher
            .watcher
            .update_paths(paths)
            .expect_err("watching more paths than the budget must fail");
        assert!(
            matches!(err.source.kind, ErrorKind::MaxFilesWatch),
            "expected MaxFilesWatch, got {err:?}"
        );
    }

    #[test]
    fn path_budget_is_shared_across_live_streams() {
        static ACTIVE_PATHS: AtomicUsize = AtomicUsize::new(0);

        let first = FseventsPathReservation::acquire(&ACTIVE_PATHS, 15, 21)
            .expect("first stream must fit within the budget");
        let active_path_count = FseventsPathReservation::acquire(&ACTIVE_PATHS, 15, 21)
            .expect_err("the combined path count must exceed the budget");
        assert_eq!(active_path_count, 15);

        drop(first);

        let second = FseventsPathReservation::acquire(&ACTIVE_PATHS, 15, 21)
            .expect("stopping the first stream must release its paths");
        drop(second);
        assert_eq!(ACTIVE_PATHS.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn rename_then_remove_remove_event_must_be_the_last_one() {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        watcher.watch_recursively(&tmpdir);
        let new_path1 = tmpdir.path().join("renamed1");
        let new_path2 = tmpdir.path().join("renamed2");

        std::fs::rename(&path, &new_path1).expect("rename1");
        std::fs::rename(&new_path1, &new_path2).expect("rename2");

        std::fs::remove_file(&new_path2).expect("remove_file");

        loop {
            let ev = rx.recv();
            if matches!(ev.kind, EventKind::Remove(RemoveKind::File)) {
                assert_eq!(&ev.paths, &[new_path2]);
                break;
            }
        }

        rx.ensure_empty();
    }
}