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");
	}
}