lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0

//! Watch management for filesystem events.
//!
//! This module handles the registration and management of filesystem watches.

use alloc::boxed::Box;
use alloc::collections::BTreeMap;
use alloc::string::{String, ToString};
use alloc::vec::Vec;

use lazy_static::lazy_static;
use spin::Mutex;

use super::types::{EventMask, EventType, FsEvent, NotifyError, WatchDescriptor, WatchOptions};

// ═══════════════════════════════════════════════════════════════════════════════
// WATCH CALLBACK
// ═══════════════════════════════════════════════════════════════════════════════

/// Callback function type for watch events.
pub type WatchCallback = Box<dyn Fn(&FsEvent) + Send + Sync>;

// ═══════════════════════════════════════════════════════════════════════════════
// WATCH
// ═══════════════════════════════════════════════════════════════════════════════

/// A registered watch on a path.
pub struct Watch {
    /// Watch descriptor
    pub descriptor: WatchDescriptor,
    /// Dataset being watched
    pub dataset: String,
    /// Path being watched
    pub path: String,
    /// Event mask (which events to watch)
    pub mask: EventMask,
    /// Watch options
    pub options: WatchOptions,
    /// Callback function
    callback: Option<WatchCallback>,
    /// Number of events received
    pub event_count: u64,
    /// Whether watch is active
    pub active: bool,
}

impl Watch {
    /// Create a new watch.
    pub fn new(
        descriptor: WatchDescriptor,
        dataset: &str,
        path: &str,
        mask: EventMask,
        options: WatchOptions,
    ) -> Self {
        Self {
            descriptor,
            dataset: dataset.into(),
            path: path.into(),
            mask,
            options,
            callback: None,
            event_count: 0,
            active: true,
        }
    }

    /// Set the callback function.
    pub fn with_callback(mut self, callback: WatchCallback) -> Self {
        self.callback = Some(callback);
        self
    }

    /// Check if this watch matches an event.
    pub fn matches(&self, event: &FsEvent) -> bool {
        if !self.active {
            return false;
        }

        // Check dataset
        if self.dataset != event.dataset {
            return false;
        }

        // Check event type mask
        if !self.mask.contains(event.event_type) {
            return false;
        }

        // Check path
        if self.options.recursive {
            // For recursive watches, path must be under watched path
            event.path.starts_with(&self.path)
                || event.path == self.path
                || (event.path.starts_with(&self.path)
                    && event.path.as_bytes().get(self.path.len()) == Some(&b'/'))
        } else {
            // For non-recursive, path must be exact or immediate child
            if event.path == self.path {
                !self.options.exclude_self
            } else if let Some(parent) = event.path.rsplit_once('/') {
                parent.0 == self.path
            } else {
                false
            }
        }
    }

    /// Invoke the callback with an event.
    pub fn invoke(&mut self, event: &FsEvent) {
        if let Some(ref callback) = self.callback {
            callback(event);
        }
        self.event_count += 1;

        // Handle oneshot
        if self.options.oneshot {
            self.active = false;
        }
    }

    /// Deactivate the watch.
    pub fn deactivate(&mut self) {
        self.active = false;
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// WATCH REGISTRY
// ═══════════════════════════════════════════════════════════════════════════════

lazy_static! {
    /// Global registry of watches.
    static ref WATCHES: Mutex<WatchRegistry> = Mutex::new(WatchRegistry::new());
}

/// Registry of all active watches.
pub struct WatchRegistry {
    /// Watches by descriptor ID
    watches: BTreeMap<u64, Watch>,
    /// Next descriptor ID
    next_id: u64,
    /// Maximum number of watches allowed
    max_watches: usize,
}

impl WatchRegistry {
    /// Create a new watch registry.
    pub fn new() -> Self {
        Self {
            watches: BTreeMap::new(),
            next_id: 1,
            max_watches: 65536, // 64K watches default
        }
    }

    /// Set the maximum number of watches.
    pub fn set_max_watches(&mut self, max: usize) {
        self.max_watches = max;
    }

    /// Add a new watch.
    pub fn add(
        &mut self,
        dataset: &str,
        path: &str,
        mask: EventMask,
        options: WatchOptions,
        callback: Option<WatchCallback>,
    ) -> Result<WatchDescriptor, NotifyError> {
        if self.watches.len() >= self.max_watches {
            return Err(NotifyError::TooManyWatches);
        }

        let id = self.next_id;
        self.next_id += 1;

        let descriptor = WatchDescriptor::new(id);
        let mut watch = Watch::new(descriptor, dataset, path, mask, options);
        if let Some(cb) = callback {
            watch = watch.with_callback(cb);
        }

        self.watches.insert(id, watch);

        Ok(descriptor)
    }

    /// Remove a watch.
    pub fn remove(&mut self, descriptor: WatchDescriptor) -> Result<(), NotifyError> {
        if self.watches.remove(&descriptor.id()).is_none() {
            return Err(NotifyError::WatchNotFound(descriptor.id()));
        }
        Ok(())
    }

    /// Get a watch by descriptor.
    pub fn get(&self, descriptor: WatchDescriptor) -> Option<&Watch> {
        self.watches.get(&descriptor.id())
    }

    /// Get a mutable watch by descriptor.
    pub fn get_mut(&mut self, descriptor: WatchDescriptor) -> Option<&mut Watch> {
        self.watches.get_mut(&descriptor.id())
    }

    /// Find all watches that match an event.
    pub fn matching_watches(&self, event: &FsEvent) -> Vec<WatchDescriptor> {
        self.watches
            .values()
            .filter(|w| w.matches(event))
            .map(|w| w.descriptor)
            .collect()
    }

    /// Get the number of watches.
    pub fn count(&self) -> usize {
        self.watches.len()
    }

    /// Get the number of active watches.
    pub fn active_count(&self) -> usize {
        self.watches.values().filter(|w| w.active).count()
    }

    /// List all watches for a dataset.
    pub fn list_by_dataset(&self, dataset: &str) -> Vec<WatchDescriptor> {
        self.watches
            .values()
            .filter(|w| w.dataset == dataset)
            .map(|w| w.descriptor)
            .collect()
    }

    /// Remove all watches for a dataset.
    pub fn remove_by_dataset(&mut self, dataset: &str) -> usize {
        let to_remove: Vec<u64> = self
            .watches
            .iter()
            .filter(|(_, w)| w.dataset == dataset)
            .map(|(id, _)| *id)
            .collect();

        let count = to_remove.len();
        for id in to_remove {
            self.watches.remove(&id);
        }
        count
    }

    /// Remove all inactive (oneshot triggered) watches.
    pub fn cleanup_inactive(&mut self) -> usize {
        let to_remove: Vec<u64> = self
            .watches
            .iter()
            .filter(|(_, w)| !w.active)
            .map(|(id, _)| *id)
            .collect();

        let count = to_remove.len();
        for id in to_remove {
            self.watches.remove(&id);
        }
        count
    }

    /// Clear all watches.
    pub fn clear(&mut self) {
        self.watches.clear();
    }
}

impl Default for WatchRegistry {
    fn default() -> Self {
        Self::new()
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// PUBLIC API
// ═══════════════════════════════════════════════════════════════════════════════

/// Add a watch on a path.
///
/// # Arguments
/// * `dataset` - Dataset to watch
/// * `path` - Path within dataset
/// * `mask` - Event types to watch
/// * `options` - Watch options
/// * `callback` - Optional callback function
///
/// # Returns
/// Watch descriptor on success.
pub fn add_watch(
    dataset: &str,
    path: &str,
    mask: EventMask,
    options: WatchOptions,
    callback: Option<WatchCallback>,
) -> Result<WatchDescriptor, NotifyError> {
    let mut registry = WATCHES.lock();
    registry.add(dataset, path, mask, options, callback)
}

/// Add a simple watch with default options.
pub fn watch(
    dataset: &str,
    path: &str,
    events: &[EventType],
    callback: WatchCallback,
) -> Result<WatchDescriptor, NotifyError> {
    let mask = EventMask::from_events(events);
    add_watch(dataset, path, mask, WatchOptions::default(), Some(callback))
}

/// Add a recursive watch.
pub fn watch_recursive(
    dataset: &str,
    path: &str,
    events: &[EventType],
    callback: WatchCallback,
) -> Result<WatchDescriptor, NotifyError> {
    let mask = EventMask::from_events(events);
    add_watch(
        dataset,
        path,
        mask,
        WatchOptions::recursive(),
        Some(callback),
    )
}

/// Remove a watch.
pub fn remove_watch(descriptor: WatchDescriptor) -> Result<(), NotifyError> {
    let mut registry = WATCHES.lock();
    registry.remove(descriptor)
}

/// Get watch information.
pub fn get_watch_info(descriptor: WatchDescriptor) -> Option<WatchInfo> {
    let registry = WATCHES.lock();
    registry.get(descriptor).map(|w| WatchInfo {
        descriptor: w.descriptor,
        dataset: w.dataset.clone(),
        path: w.path.clone(),
        mask: w.mask,
        recursive: w.options.recursive,
        active: w.active,
        event_count: w.event_count,
    })
}

/// Get the number of active watches.
pub fn watch_count() -> usize {
    let registry = WATCHES.lock();
    registry.count()
}

/// Information about a watch (for querying).
#[derive(Debug, Clone)]
pub struct WatchInfo {
    /// Watch descriptor
    pub descriptor: WatchDescriptor,
    /// Dataset being watched
    pub dataset: String,
    /// Path being watched
    pub path: String,
    /// Event mask
    pub mask: EventMask,
    /// Whether recursive
    pub recursive: bool,
    /// Whether active
    pub active: bool,
    /// Number of events received
    pub event_count: u64,
}

/// Dispatch an event to matching watches.
///
/// This is called by the event emission system.
pub(crate) fn dispatch_to_watches(event: &FsEvent) {
    let mut registry = WATCHES.lock();

    // Find matching watches first to avoid borrow issues
    let matching: Vec<u64> = registry
        .watches
        .iter()
        .filter(|(_, w)| w.matches(event))
        .map(|(id, _)| *id)
        .collect();

    // Invoke callbacks
    for id in matching {
        if let Some(watch) = registry.watches.get_mut(&id) {
            watch.invoke(event);
        }
    }

    // Cleanup inactive oneshot watches
    registry.cleanup_inactive();
}

// ═══════════════════════════════════════════════════════════════════════════════
// TESTS
// ═══════════════════════════════════════════════════════════════════════════════

#[cfg(test)]
mod tests {
    use super::*;
    use alloc::sync::Arc;
    use core::sync::atomic::{AtomicU64, Ordering};

    #[test]
    fn test_watch_matches_exact() {
        let watch = Watch::new(
            WatchDescriptor::new(1),
            "tank/data",
            "/path/to",
            EventMask::ALL,
            WatchOptions::default(),
        );

        // Immediate child should match
        let event = FsEvent::new(EventType::Create, "tank/data", "/path/to/file.txt");
        assert!(watch.matches(&event));

        // Exact path should match (if not exclude_self)
        let event = FsEvent::new(EventType::Create, "tank/data", "/path/to");
        assert!(watch.matches(&event));

        // Wrong dataset
        let event = FsEvent::new(EventType::Create, "other/pool", "/path/to/file.txt");
        assert!(!watch.matches(&event));

        // Deeper path without recursive
        let event = FsEvent::new(EventType::Create, "tank/data", "/path/to/sub/file.txt");
        assert!(!watch.matches(&event));
    }

    #[test]
    fn test_watch_matches_recursive() {
        let watch = Watch::new(
            WatchDescriptor::new(1),
            "tank/data",
            "/path/to",
            EventMask::ALL,
            WatchOptions::recursive(),
        );

        // Should match deep paths
        let event = FsEvent::new(EventType::Create, "tank/data", "/path/to/sub/deep/file.txt");
        assert!(watch.matches(&event));
    }

    #[test]
    fn test_watch_mask_filter() {
        let watch = Watch::new(
            WatchDescriptor::new(1),
            "tank",
            "/path",
            EventMask::from_events(&[EventType::Create, EventType::Delete]),
            WatchOptions::default(),
        );

        let create = FsEvent::new(EventType::Create, "tank", "/path/file.txt");
        assert!(watch.matches(&create));

        let modify = FsEvent::new(EventType::Modify, "tank", "/path/file.txt");
        assert!(!watch.matches(&modify));
    }

    #[test]
    fn test_watch_oneshot() {
        let mut watch = Watch::new(
            WatchDescriptor::new(1),
            "tank",
            "/path",
            EventMask::ALL,
            WatchOptions::oneshot(),
        );

        let event = FsEvent::new(EventType::Create, "tank", "/path/file.txt");
        assert!(watch.matches(&event));

        watch.invoke(&event);
        assert!(!watch.active);
        assert!(!watch.matches(&event)); // Should not match after oneshot triggered
    }

    #[test]
    fn test_registry_add_remove() {
        let mut registry = WatchRegistry::new();

        let wd = registry
            .add(
                "tank",
                "/path",
                EventMask::ALL,
                WatchOptions::default(),
                None,
            )
            .unwrap();

        assert_eq!(registry.count(), 1);

        registry.remove(wd).unwrap();
        assert_eq!(registry.count(), 0);
    }

    #[test]
    fn test_registry_max_watches() {
        let mut registry = WatchRegistry::new();
        registry.set_max_watches(2);

        registry
            .add(
                "tank",
                "/path1",
                EventMask::ALL,
                WatchOptions::default(),
                None,
            )
            .unwrap();
        registry
            .add(
                "tank",
                "/path2",
                EventMask::ALL,
                WatchOptions::default(),
                None,
            )
            .unwrap();

        let result = registry.add(
            "tank",
            "/path3",
            EventMask::ALL,
            WatchOptions::default(),
            None,
        );
        assert!(matches!(result, Err(NotifyError::TooManyWatches)));
    }

    #[test]
    fn test_registry_matching_watches() {
        let mut registry = WatchRegistry::new();

        let wd1 = registry
            .add(
                "tank",
                "/path1",
                EventMask::from_events(&[EventType::Create]),
                WatchOptions::default(),
                None,
            )
            .unwrap();

        let wd2 = registry
            .add(
                "tank",
                "/path2",
                EventMask::from_events(&[EventType::Create]),
                WatchOptions::default(),
                None,
            )
            .unwrap();

        let event = FsEvent::new(EventType::Create, "tank", "/path1/file.txt");
        let matches = registry.matching_watches(&event);

        assert_eq!(matches.len(), 1);
        assert_eq!(matches[0], wd1);
    }

    #[test]
    fn test_registry_remove_by_dataset() {
        let mut registry = WatchRegistry::new();

        registry
            .add(
                "tank",
                "/path1",
                EventMask::ALL,
                WatchOptions::default(),
                None,
            )
            .unwrap();
        registry
            .add(
                "tank",
                "/path2",
                EventMask::ALL,
                WatchOptions::default(),
                None,
            )
            .unwrap();
        registry
            .add(
                "other",
                "/path3",
                EventMask::ALL,
                WatchOptions::default(),
                None,
            )
            .unwrap();

        let removed = registry.remove_by_dataset("tank");
        assert_eq!(removed, 2);
        assert_eq!(registry.count(), 1);
    }

    #[test]
    fn test_watch_with_callback() {
        let counter = Arc::new(AtomicU64::new(0));
        let counter_clone = counter.clone();

        let callback: WatchCallback = Box::new(move |_event| {
            counter_clone.fetch_add(1, Ordering::SeqCst);
        });

        let mut watch = Watch::new(
            WatchDescriptor::new(1),
            "tank",
            "/path",
            EventMask::ALL,
            WatchOptions::default(),
        )
        .with_callback(callback);

        let event = FsEvent::new(EventType::Create, "tank", "/path/file.txt");
        watch.invoke(&event);
        watch.invoke(&event);

        assert_eq!(counter.load(Ordering::SeqCst), 2);
        assert_eq!(watch.event_count, 2);
    }
}