use std::io::Read as _;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use flate2::read::GzDecoder;
use notify::{Event, EventKind, RecursiveMode, Watcher};
use tokio::sync::mpsc;
use tracing::{error, info, warn};
use crate::queue_manager::QueueManager;
const MAX_WATCHED_NZB_BYTES: usize = 100 * 1024 * 1024;
pub struct DirWatcher {
watch_dir: PathBuf,
queue_manager: Arc<QueueManager>,
}
impl DirWatcher {
pub fn new(watch_dir: PathBuf, queue_manager: Arc<QueueManager>) -> Self {
Self {
watch_dir,
queue_manager,
}
}
pub async fn run(self) {
info!(dir = %self.watch_dir.display(), "Starting directory watcher");
if let Err(e) = std::fs::create_dir_all(&self.watch_dir) {
error!(error = %e, "Failed to create watch directory");
return;
}
self.process_existing_files().await;
let (tx, mut rx) = mpsc::channel(100);
let _watcher = {
let tx = tx.clone();
let mut watcher =
notify::recommended_watcher(move |res: Result<Event, notify::Error>| {
if let Ok(event) = res {
let _ = tx.blocking_send(event);
}
})
.expect("Failed to create file watcher");
watcher
.watch(&self.watch_dir, RecursiveMode::NonRecursive)
.expect("Failed to watch directory");
watcher };
while let Some(event) = rx.recv().await {
match event.kind {
EventKind::Create(_) | EventKind::Modify(_) => {
for path in &event.paths {
if Self::is_nzb_file(path) {
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
self.process_file(path).await;
}
}
}
_ => {}
}
}
}
fn is_nzb_file(path: &Path) -> bool {
path.extension().is_some_and(|ext| ext == "nzb") || Self::is_gz_nzb(path)
}
fn is_gz_nzb(path: &Path) -> bool {
path.to_str().is_some_and(|s| s.ends_with(".nzb.gz"))
}
async fn process_existing_files(&self) {
let entries = match std::fs::read_dir(&self.watch_dir) {
Ok(e) => e,
Err(e) => {
warn!(error = %e, "Failed to read watch directory");
return;
}
};
for entry in entries.flatten() {
let path = entry.path();
if Self::is_nzb_file(&path) {
self.process_file(&path).await;
}
}
}
async fn process_file(&self, path: &Path) {
info!(file = %path.display(), "Processing NZB from watch directory");
let raw_data = match Self::read_limited(path) {
Ok(d) => d,
Err(e) => {
warn!(error = %e, file = %path.display(), "Failed to read NZB file");
return;
}
};
let data = if Self::is_gz_nzb(path) {
let decoder = GzDecoder::new(raw_data.as_slice());
let mut decompressed = Vec::new();
if let Err(error) = decoder
.take((MAX_WATCHED_NZB_BYTES as u64).saturating_add(1))
.read_to_end(&mut decompressed)
{
warn!(error = %error, file = %path.display(), "Failed to decompress watched NZB");
return;
}
if decompressed.len() > MAX_WATCHED_NZB_BYTES {
warn!(file = %path.display(), limit = MAX_WATCHED_NZB_BYTES, "Decompressed watched NZB exceeds the input limit");
return;
}
decompressed
} else {
raw_data
};
let name = if Self::is_gz_nzb(path) {
path.file_name()
.and_then(|name| name.to_str())
.and_then(|name| name.strip_suffix(".nzb.gz"))
.unwrap_or("unknown")
.to_string()
} else {
path.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown")
.to_string()
};
match crate::nzb_core::nzb_parser::parse_nzb(&name, &data) {
Ok(mut job) => {
job.work_dir = self.queue_manager.incomplete_dir().join(&job.id);
job.output_dir = self.queue_manager.complete_dir().join(&job.name);
if let Err(e) = std::fs::create_dir_all(&job.work_dir) {
error!(error = %e, "Failed to create work directory");
return;
}
info!(name = %job.name, id = %job.id, "Auto-enqueuing NZB from watch dir");
if let Err(e) = self.queue_manager.add_job(job, Some(data)) {
error!(error = %e, "Failed to enqueue NZB");
return;
}
let processed_dir = self.watch_dir.join("processed");
let _ = std::fs::create_dir_all(&processed_dir);
let dest = processed_dir.join(path.file_name().unwrap_or_default());
if let Err(_e) = std::fs::rename(path, &dest) {
if let Err(e2) =
std::fs::copy(path, &dest).and_then(|_| std::fs::remove_file(path))
{
warn!(error = %e2, "Failed to move processed NZB file");
}
}
}
Err(e) => {
warn!(error = %e, file = %path.display(), "Failed to parse NZB from watch dir");
}
}
}
fn read_limited(path: &Path) -> std::io::Result<Vec<u8>> {
let metadata = std::fs::symlink_metadata(path)?;
if metadata.file_type().is_symlink() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"watched NZB symlinks are not supported",
));
}
if metadata.len() > MAX_WATCHED_NZB_BYTES as u64 {
return Err(std::io::Error::new(
std::io::ErrorKind::FileTooLarge,
format!("watched NZB exceeds the {MAX_WATCHED_NZB_BYTES} byte limit"),
));
}
let file = std::fs::File::open(path)?;
let mut data = Vec::new();
file.take((MAX_WATCHED_NZB_BYTES as u64).saturating_add(1))
.read_to_end(&mut data)?;
if data.len() > MAX_WATCHED_NZB_BYTES {
return Err(std::io::Error::new(
std::io::ErrorKind::FileTooLarge,
format!("watched NZB exceeds the {MAX_WATCHED_NZB_BYTES} byte limit"),
));
}
Ok(data)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn recognizes_plain_and_gzipped_nzb_paths_case_sensitively() {
assert!(DirWatcher::is_nzb_file(Path::new("release.nzb")));
assert!(DirWatcher::is_nzb_file(Path::new("release.nzb.gz")));
assert!(!DirWatcher::is_nzb_file(Path::new("release.NZB")));
assert!(!DirWatcher::is_nzb_file(Path::new("release.txt")));
}
#[test]
fn bounded_file_reader_rejects_oversized_input() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("oversized.nzb");
std::fs::write(&path, vec![b'x'; MAX_WATCHED_NZB_BYTES + 1]).unwrap();
let error = DirWatcher::read_limited(&path).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::FileTooLarge);
}
}