mini_static/

watcher.rs

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};

use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};

use crate::reload::ChangeType;

/// A change event broadcast when a watched file is added, modified, or removed.
#[derive(Clone, Debug)]
pub struct ChangeEvent {
    /// The path to the file that changed, relative to the watched root.
    pub path: PathBuf,
    /// The type of change (CSS, script, HTML, or other).
    pub change_type: ChangeType,
}

/// Broadcasts file change events to multiple subscribers.
///
/// A single broadcaster can have many subscribers (e.g., multiple browser clients
/// connected via SSE). When a file changes, all active subscribers are notified.
/// If a subscriber's channel is full or closed, that subscriber is removed.
#[derive(Clone)]
pub struct Broadcaster {
    senders: Arc<Mutex<Vec<UnboundedSender<ChangeEvent>>>>,
}

impl Broadcaster {
    /// Create a new broadcaster with no subscribers.
    pub fn new() -> Self {
        Broadcaster {
            senders: Arc::new(Mutex::new(Vec::new())),
        }
    }

    /// Broadcast a change event to all active subscribers.
    ///
    /// Removes any subscribers whose channels are closed or full.
    pub fn broadcast(&self, event: ChangeEvent) {
        let mut senders = self.senders.lock().unwrap();
        senders.retain(|sender| sender.send(event.clone()).is_ok());
    }

    /// Subscribe to change events.
    ///
    /// Returns a receiver that will yield each broadcasted change event.
    pub fn subscribe(&self) -> UnboundedReceiver<ChangeEvent> {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        self.senders.lock().unwrap().push(tx);
        rx
    }

    /// Get the current number of active subscribers.
    #[cfg(test)]
    pub fn subscriber_count(&self) -> usize {
        self.senders.lock().unwrap().len()
    }
}

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

/// Start watching a directory for file changes.
///
/// Spawns a background task that periodically polls the directory tree for
/// modifications using mtime. When changes are detected, broadcasts them to all
/// active subscribers via the given `broadcaster`.
///
/// The poll interval is bounded to prevent busy-waiting (per architecture principle A2).
/// Polls every 500ms — a balance between responsiveness and system load.
///
/// # Panics
///
/// Panics if the async task cannot be spawned (e.g., no runtime available).
pub fn start_watching(dir: Arc<PathBuf>, broadcaster: Broadcaster) {
    tokio::spawn(async move {
        let mut mtimes: HashMap<PathBuf, SystemTime> = HashMap::new();
        let mut first_pass = true;
        let mut interval = tokio::time::interval(Duration::from_millis(500));
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

        loop {
            interval.tick().await;

            let entries = walk_dir(dir.as_path()).await.unwrap_or_default();
            let mut current = HashMap::new();

            for path in entries {
                if let Ok(meta) = tokio::fs::metadata(&path).await {
                    if let Ok(mtime) = meta.modified() {
                        current.insert(path.clone(), mtime);

                        if !first_pass {
                            let is_new = !mtimes.contains_key(&path);
                            let changed = mtimes.get(&path).is_none_or(|old| *old != mtime);
                            if is_new || changed {
                                let change_type = ChangeType::from_path(&path);
                                broadcaster.broadcast(ChangeEvent { path, change_type });
                            }
                        }
                    }
                }
            }

            mtimes = current;
            first_pass = false;
        }
    });
}

/// Recursively walk a directory tree and return all file paths.
async fn walk_dir(dir: &Path) -> std::io::Result<Vec<PathBuf>> {
    let mut files = Vec::new();
    let mut dirs = vec![dir.to_path_buf()];

    while let Some(dir) = dirs.pop() {
        let mut rd = tokio::fs::read_dir(&dir).await?;
        while let Some(entry) = rd.next_entry().await? {
            let path = entry.path();
            if entry.file_type().await?.is_dir() {
                dirs.push(path);
            } else {
                files.push(path);
            }
        }
    }

    Ok(files)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use tempfile::TempDir;
    use tokio::time::{sleep, timeout};

    #[tokio::test]
    async fn broadcaster_delivers_events_to_subscribers() {
        let broadcaster = Broadcaster::new();
        let mut rx = broadcaster.subscribe();

        let event = ChangeEvent {
            path: PathBuf::from("style.css"),
            change_type: ChangeType::Css,
        };
        broadcaster.broadcast(event.clone());

        let received = timeout(Duration::from_secs(1), rx.recv())
            .await
            .expect("timeout")
            .expect("channel closed");
        assert_eq!(received.path, event.path);
        assert_eq!(received.change_type, event.change_type);
    }

    #[tokio::test]
    async fn broadcaster_tracks_subscriber_count() {
        let broadcaster = Broadcaster::new();
        assert_eq!(broadcaster.subscriber_count(), 0);

        let _rx1 = broadcaster.subscribe();
        assert_eq!(broadcaster.subscriber_count(), 1);

        let _rx2 = broadcaster.subscribe();
        assert_eq!(broadcaster.subscriber_count(), 2);

        drop(_rx1);
        broadcaster.broadcast(ChangeEvent {
            path: PathBuf::from("file.js"),
            change_type: ChangeType::Script,
        });
        assert_eq!(broadcaster.subscriber_count(), 1);
    }

    #[tokio::test]
    async fn watcher_detects_file_changes() {
        let temp = TempDir::new().unwrap();
        let dir_path = Arc::new(temp.path().to_path_buf());

        let broadcaster = Broadcaster::new();
        let mut rx = broadcaster.subscribe();

        start_watching(Arc::clone(&dir_path), broadcaster);

        sleep(Duration::from_millis(600)).await;

        fs::write(dir_path.join("new_file.js"), "console.log('hello');").unwrap();

        let event = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timeout")
            .expect("channel closed");
        assert_eq!(event.path.file_name().unwrap(), "new_file.js");
        assert_eq!(event.change_type, ChangeType::Script);
    }

    #[tokio::test]
    async fn watcher_detects_file_modifications() {
        let temp = TempDir::new().unwrap();
        let dir_path = Arc::new(temp.path().to_path_buf());
        let file_path = dir_path.join("style.css");
        fs::write(&file_path, "body { color: red; }").unwrap();

        let broadcaster = Broadcaster::new();
        let mut rx = broadcaster.subscribe();

        start_watching(Arc::clone(&dir_path), broadcaster);

        sleep(Duration::from_millis(600)).await;

        fs::write(&file_path, "body { color: blue; }").unwrap();

        let event = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timeout")
            .expect("channel closed");
        assert_eq!(event.path.file_name().unwrap(), "style.css");
        assert_eq!(event.change_type, ChangeType::Css);
    }

    #[tokio::test]
    async fn watcher_ignores_changes_in_first_pass() {
        let temp = TempDir::new().unwrap();
        let dir_path = Arc::new(temp.path().to_path_buf());
        fs::write(dir_path.join("existing.html"), "<h1>Hello</h1>").unwrap();

        let broadcaster = Broadcaster::new();
        let mut rx = broadcaster.subscribe();

        start_watching(Arc::clone(&dir_path), broadcaster);

        sleep(Duration::from_millis(600)).await;

        let result = timeout(Duration::from_millis(100), rx.recv()).await;
        assert!(
            result.is_err(),
            "first pass should not emit events for existing files"
        );
    }
}