Skip to main content

max / audiofiles

23.0 KB · 706 lines History Blame Raw
1 //! Background folder import worker: walks directories and imports audio files off the GUI thread.
2 //!
3 //! Mirrors the pattern in `audiofiles_core::analysis::worker` — the worker runs in a
4 //! dedicated thread with its own DB connection, communicating via channels. Between files
5 //! it checks for cancellation, keeping the UI responsive during large imports.
6
7 use std::fs;
8 use std::path::{Path, PathBuf};
9 use std::sync::{mpsc, Mutex};
10 use std::thread;
11 use std::time::{Duration, Instant};
12
13 use tracing::{error, instrument, warn};
14
15 use audiofiles_core::db::Database;
16 use audiofiles_core::error::CoreError;
17 use audiofiles_core::store::SampleStore;
18 use audiofiles_core::vfs::{self, NodeType};
19 use audiofiles_core::{NodeId, VfsId};
20
21 /// Check whether a path has an audio file extension.
22 fn is_audio_file(path: &Path) -> bool {
23 audiofiles_core::util::is_audio_file(path)
24 }
25
26 /// Check whether a directory should be skipped during traversal.
27 fn is_skipped_dir(path: &Path) -> bool {
28 audiofiles_core::util::is_macos_metadata_dir(path)
29 }
30
31 /// How imported files should be organized in the VFS.
32 pub enum ImportStrategy {
33 /// All links in one directory, no subdirs created.
34 Flat { vfs_id: VfsId, parent_id: Option<NodeId> },
35 /// Create a new VFS, preserve source directory structure.
36 NewVfs { vfs_name: String },
37 /// Merge with structure into an existing VFS.
38 MergeIntoVfs { vfs_id: VfsId, parent_id: Option<NodeId> },
39 }
40
41 /// A top-level source folder and its imported samples.
42 #[derive(Clone)]
43 pub struct ImportedFolder {
44 pub name: String,
45 pub samples: Vec<(String, String)>, // (hash, ext)
46 }
47
48 /// Result of importing a single file into the VFS.
49 enum ImportFileResult {
50 Imported(String, String), // (hash, ext)
51 Duplicate, // NameConflict — link already existed
52 }
53
54 /// Command sent from the GUI thread to the import worker.
55 pub enum ImportCommand {
56 /// Import all audio files from `source` using the given strategy.
57 ImportDirectory {
58 source: PathBuf,
59 strategy: ImportStrategy,
60 },
61 /// Cancel the current import.
62 Cancel,
63 /// Shut down the worker thread.
64 Shutdown,
65 }
66
67 /// Event sent from the import worker back to the GUI thread.
68 pub enum ImportEvent {
69 /// Throttled progress emitted during the pre-walk so the UI can show a
70 /// running file count instead of an indeterminate spinner (m-12). Fires
71 /// at most every ~100ms.
72 WalkProgress { count: usize, total_bytes: u64 },
73 /// Pre-walk finished — we now know the total file count and size.
74 WalkComplete { total: usize, total_bytes: u64 },
75 /// One file was processed.
76 Progress {
77 completed: usize,
78 total: usize,
79 current_name: String,
80 },
81 /// A single file failed to import.
82 FileError { path: String, error: String },
83 /// The entire import is done.
84 Complete {
85 /// `(hash, file_extension)` pairs for analysis flow.
86 imported: Vec<(String, String)>,
87 total_files: usize,
88 errors: usize,
89 duplicates: usize,
90 /// Top-level folders with their samples (empty for flat imports).
91 folders: Vec<ImportedFolder>,
92 },
93 }
94
95 /// Handle for communicating with the background import worker.
96 ///
97 /// The receiver is wrapped in a `Mutex` so `BrowserState` remains `Sync` (required by nih-plug).
98 /// Only the GUI thread actually calls `try_recv`, so contention is zero.
99 pub struct ImportHandle {
100 cmd_tx: mpsc::Sender<ImportCommand>,
101 event_rx: Mutex<mpsc::Receiver<ImportEvent>>,
102 _thread: Option<thread::JoinHandle<()>>,
103 }
104
105 impl ImportHandle {
106 /// Poll for the next event without blocking.
107 pub fn try_recv(&self) -> Option<ImportEvent> {
108 self.event_rx.lock().ok()?.try_recv().ok()
109 }
110
111 /// Send a command to the worker.
112 pub fn send(&self, cmd: ImportCommand) {
113 let _ = self.cmd_tx.send(cmd);
114 }
115 }
116
117 impl Drop for ImportHandle {
118 fn drop(&mut self) {
119 let _ = self.cmd_tx.send(ImportCommand::Shutdown);
120 if let Some(handle) = self._thread.take() {
121 let _ = handle.join();
122 }
123 }
124 }
125
126 /// Spawn the background import worker thread.
127 ///
128 /// The worker opens its own `Database` and `SampleStore` to avoid Mutex contention
129 /// with the GUI thread. Returns a handle for sending commands and polling events.
130 #[instrument(skip_all)]
131 pub fn spawn_import_worker(db_path: PathBuf, store_root: PathBuf) -> std::io::Result<ImportHandle> {
132 let (cmd_tx, cmd_rx) = mpsc::channel::<ImportCommand>();
133 let (event_tx, event_rx) = mpsc::channel::<ImportEvent>();
134
135 let thread = thread::Builder::new()
136 .name("import-worker".to_string())
137 .spawn(move || {
138 worker_loop(cmd_rx, event_tx, &db_path, &store_root);
139 })?;
140
141 Ok(ImportHandle {
142 cmd_tx,
143 event_rx: Mutex::new(event_rx),
144 _thread: Some(thread),
145 })
146 }
147
148 /// Recursively count audio files and sum their sizes under `dir`.
149 /// Checks for cancellation between entries. Returns `None` if cancelled.
150 #[instrument(skip_all)]
151 fn count_audio_files(
152 dir: &Path,
153 cmd_rx: &mpsc::Receiver<ImportCommand>,
154 event_tx: &mpsc::Sender<ImportEvent>,
155 ) -> Option<(usize, u64)> {
156 let mut count = 0;
157 let mut total_bytes = 0u64;
158 let mut stack = vec![dir.to_path_buf()];
159 let mut last_emit = Instant::now();
160 let emit_interval = Duration::from_millis(100);
161
162 while let Some(current) = stack.pop() {
163 // Check for cancel
164 if let Ok(ImportCommand::Cancel) | Ok(ImportCommand::Shutdown) = cmd_rx.try_recv() {
165 return None;
166 }
167
168 let entries = match fs::read_dir(&current) {
169 Ok(e) => e,
170 Err(e) => {
171 tracing::warn!(dir = %current.display(), "Failed to read directory during pre-walk: {e}");
172 continue;
173 }
174 };
175
176 for entry in entries.flatten() {
177 let path = entry.path();
178 if path.is_dir() {
179 if !is_skipped_dir(&path) {
180 stack.push(path);
181 }
182 } else if is_audio_file(&path) {
183 count += 1;
184 if let Ok(meta) = fs::metadata(&path) {
185 total_bytes += meta.len();
186 }
187 if last_emit.elapsed() >= emit_interval {
188 let _ = event_tx.send(ImportEvent::WalkProgress { count, total_bytes });
189 last_emit = Instant::now();
190 }
191 }
192 }
193 }
194
195 Some((count, total_bytes))
196 }
197
198 /// Import a single file into store + VFS, returning the result.
199 fn import_single_file(
200 path: &Path,
201 vfs_id: VfsId,
202 parent_id: Option<NodeId>,
203 store: &SampleStore,
204 db: &Database,
205 loose_files: bool,
206 ) -> Result<ImportFileResult, CoreError> {
207 let hash = if loose_files {
208 store.import_loose_files(path, db)?
209 } else {
210 store.import(path, db)?
211 };
212 let name = audiofiles_core::util::get_filename(path, "unknown");
213 let ext = audiofiles_core::util::get_extension(path);
214
215 match vfs::create_sample_link(db, vfs_id, parent_id, &name, &hash) {
216 Ok(_) => Ok(ImportFileResult::Imported(hash, ext)),
217 Err(CoreError::NameConflict(_)) => Ok(ImportFileResult::Duplicate),
218 Err(e) => {
219 if let CoreError::Db(ref sqlite_err) = e {
220 if sqlite_err.to_string().contains("UNIQUE") {
221 return Ok(ImportFileResult::Duplicate);
222 }
223 }
224 Err(e)
225 }
226 }
227 }
228
229 /// Shared mutable state and dependencies for the import functions.
230 ///
231 /// Bundles the store, DB, channels, and progress counters so that
232 /// `import_directory_recursive`, `import_directory_flat`, and `import_structured`
233 /// don't each need 12 parameters.
234 struct ImportContext<'a> {
235 store: &'a SampleStore,
236 db: &'a Database,
237 event_tx: &'a mpsc::Sender<ImportEvent>,
238 cmd_rx: &'a mpsc::Receiver<ImportCommand>,
239 completed: &'a mut usize,
240 total: usize,
241 errors: &'a mut usize,
242 duplicates: &'a mut usize,
243 imported: &'a mut Vec<(String, String)>,
244 loose_files: bool,
245 }
246
247 impl ImportContext<'_> {
248 /// Check for a cancellation command without blocking.
249 fn is_cancelled(&self) -> bool {
250 matches!(
251 self.cmd_rx.try_recv(),
252 Ok(ImportCommand::Cancel) | Ok(ImportCommand::Shutdown)
253 )
254 }
255
256 /// Send a progress event for the current file.
257 fn send_progress(&self, name: String) {
258 let _ = self.event_tx.send(ImportEvent::Progress {
259 completed: *self.completed,
260 total: self.total,
261 current_name: name,
262 });
263 }
264
265 /// Import a single audio file, updating counters and sending error events.
266 fn process_file(&mut self, path: &Path, vfs_id: VfsId, parent_id: Option<NodeId>) {
267 let name = audiofiles_core::util::get_filename(path, "unknown");
268 self.send_progress(name);
269
270 match import_single_file(path, vfs_id, parent_id, self.store, self.db, self.loose_files) {
271 Ok(ImportFileResult::Imported(hash, ext)) => {
272 self.imported.push((hash, ext));
273 *self.completed += 1;
274 }
275 Ok(ImportFileResult::Duplicate) => {
276 *self.duplicates += 1;
277 *self.completed += 1;
278 }
279 Err(e) => {
280 *self.errors += 1;
281 let _ = self.event_tx.send(ImportEvent::FileError {
282 path: path.display().to_string(),
283 error: e.to_string(),
284 });
285 }
286 }
287 }
288 }
289
290 /// Recursively import a directory with structure, sending progress events.
291 /// Returns `true` if cancelled.
292 fn import_directory_recursive(
293 dir: &Path,
294 vfs_id: VfsId,
295 parent_id: Option<NodeId>,
296 ctx: &mut ImportContext<'_>,
297 ) -> bool {
298 let entries = match fs::read_dir(dir) {
299 Ok(e) => e,
300 Err(_) => {
301 *ctx.errors += 1;
302 return false; // not cancelled
303 }
304 };
305
306 let mut paths: Vec<PathBuf> = entries.flatten().map(|e| e.path()).collect();
307 paths.sort();
308
309 for path in paths {
310 if ctx.is_cancelled() {
311 return true;
312 }
313
314 if path.is_dir() {
315 if is_skipped_dir(&path) {
316 continue;
317 }
318 let dir_name = audiofiles_core::util::get_filename(&path, "folder");
319
320 let dir_node_id =
321 match vfs::create_directory(ctx.db, vfs_id, parent_id, &dir_name) {
322 Ok(id) => Some(id),
323 Err(CoreError::NameConflict(_)) => {
324 vfs::list_children(ctx.db, vfs_id, parent_id)
325 .unwrap_or_default()
326 .iter()
327 .find(|n| {
328 n.name == dir_name && n.node_type == NodeType::Directory
329 })
330 .map(|n| n.id)
331 }
332 Err(_) => {
333 *ctx.errors += 1;
334 continue;
335 }
336 };
337
338 let cancelled = import_directory_recursive(
339 &path,
340 vfs_id,
341 dir_node_id.or(parent_id),
342 ctx,
343 );
344 if cancelled {
345 return true;
346 }
347 } else if path.is_file() && is_audio_file(&path) {
348 ctx.process_file(&path, vfs_id, parent_id);
349 }
350 }
351
352 false // not cancelled
353 }
354
355 /// Flat import: walk source tree recursively but put all VFS links into a single directory.
356 /// Returns `true` if cancelled.
357 fn import_directory_flat(
358 dir: &Path,
359 vfs_id: VfsId,
360 parent_id: Option<NodeId>,
361 ctx: &mut ImportContext<'_>,
362 ) -> bool {
363 let entries = match fs::read_dir(dir) {
364 Ok(e) => e,
365 Err(_) => {
366 *ctx.errors += 1;
367 return false;
368 }
369 };
370
371 let mut paths: Vec<PathBuf> = entries.flatten().map(|e| e.path()).collect();
372 paths.sort();
373
374 for path in paths {
375 if ctx.is_cancelled() {
376 return true;
377 }
378
379 if path.is_dir() {
380 if is_skipped_dir(&path) {
381 continue;
382 }
383 // Recurse into subdirs but still put all links at the same flat level
384 let cancelled = import_directory_flat(&path, vfs_id, parent_id, ctx);
385 if cancelled {
386 return true;
387 }
388 } else if path.is_file() && is_audio_file(&path) {
389 ctx.process_file(&path, vfs_id, parent_id);
390 }
391 }
392
393 false
394 }
395
396 /// Structured import: iterate source dir's immediate children, import each top-level
397 /// subdirectory via `import_directory_recursive`, tracking `ImportedFolder` per top-level dir.
398 /// Files directly in the source root are imported without a folder grouping.
399 /// Returns `(cancelled, folders)`.
400 fn import_structured(
401 source: &Path,
402 vfs_id: VfsId,
403 parent_id: Option<NodeId>,
404 ctx: &mut ImportContext<'_>,
405 ) -> (bool, Vec<ImportedFolder>) {
406 let entries = match fs::read_dir(source) {
407 Ok(e) => e,
408 Err(_) => {
409 *ctx.errors += 1;
410 return (false, Vec::new());
411 }
412 };
413
414 let mut paths: Vec<PathBuf> = entries.flatten().map(|e| e.path()).collect();
415 paths.sort();
416
417 let mut folders = Vec::new();
418
419 for path in paths {
420 if ctx.is_cancelled() {
421 return (true, folders);
422 }
423
424 if path.is_dir() {
425 if is_skipped_dir(&path) {
426 continue;
427 }
428 let dir_name = audiofiles_core::util::get_filename(&path, "folder");
429
430 let dir_node_id =
431 match vfs::create_directory(ctx.db, vfs_id, parent_id, &dir_name) {
432 Ok(id) => Some(id),
433 Err(CoreError::NameConflict(_)) => {
434 vfs::list_children(ctx.db, vfs_id, parent_id)
435 .unwrap_or_default()
436 .iter()
437 .find(|n| {
438 n.name == dir_name && n.node_type == NodeType::Directory
439 })
440 .map(|n| n.id)
441 }
442 Err(_) => {
443 *ctx.errors += 1;
444 continue;
445 }
446 };
447
448 // Track where this folder's samples start in the imported list
449 let folder_start = ctx.imported.len();
450
451 let cancelled = import_directory_recursive(
452 &path,
453 vfs_id,
454 dir_node_id.or(parent_id),
455 ctx,
456 );
457
458 // Collect samples imported within this top-level folder
459 let folder_samples: Vec<(String, String)> =
460 ctx.imported[folder_start..].to_vec();
461
462 if !folder_samples.is_empty() {
463 folders.push(ImportedFolder {
464 name: dir_name,
465 samples: folder_samples,
466 });
467 }
468
469 if cancelled {
470 return (true, folders);
471 }
472 } else if path.is_file() && is_audio_file(&path) {
473 // Root-level files — import directly, no folder grouping
474 ctx.process_file(&path, vfs_id, parent_id);
475 }
476 }
477
478 (false, folders)
479 }
480
481 fn worker_loop(
482 cmd_rx: mpsc::Receiver<ImportCommand>,
483 event_tx: mpsc::Sender<ImportEvent>,
484 db_path: &Path,
485 store_root: &Path,
486 ) {
487 // Open our own DB connection and store
488 let db = match Database::open(db_path) {
489 Ok(db) => db,
490 Err(e) => {
491 let _ = event_tx.send(ImportEvent::Complete {
492 imported: Vec::new(),
493 total_files: 0,
494 errors: 1,
495 duplicates: 0,
496 folders: Vec::new(),
497 });
498 error!("Import worker failed to open DB: {e}");
499 return;
500 }
501 };
502
503 let store = match SampleStore::new(store_root) {
504 Ok(s) => s,
505 Err(e) => {
506 let _ = event_tx.send(ImportEvent::Complete {
507 imported: Vec::new(),
508 total_files: 0,
509 errors: 1,
510 duplicates: 0,
511 folders: Vec::new(),
512 });
513 error!("Import worker failed to open store: {e}");
514 return;
515 }
516 };
517
518 while let Ok(cmd) = cmd_rx.recv() {
519 match cmd {
520 ImportCommand::Shutdown => break,
521 ImportCommand::Cancel => continue,
522 ImportCommand::ImportDirectory { source, strategy } => {
523 // Resolve strategy to concrete (vfs_id, parent_id, flat)
524 let (vfs_id, parent_id, flat) = match strategy {
525 ImportStrategy::Flat { vfs_id, parent_id } => (vfs_id, parent_id, true),
526 ImportStrategy::NewVfs { vfs_name } => {
527 match vfs::create_vfs(&db, &vfs_name) {
528 Ok(id) => (id, None, false),
529 Err(e) => {
530 let _ = event_tx.send(ImportEvent::Complete {
531 imported: Vec::new(),
532 total_files: 0,
533 errors: 1,
534 duplicates: 0,
535 folders: Vec::new(),
536 });
537 error!("Failed to create VFS '{vfs_name}': {e}");
538 continue;
539 }
540 }
541 }
542 ImportStrategy::MergeIntoVfs { vfs_id, parent_id } => {
543 (vfs_id, parent_id, false)
544 }
545 };
546
547 // Phase 1: pre-walk to count audio files and sum sizes
548 let (total, total_bytes) = match count_audio_files(&source, &cmd_rx, &event_tx) {
549 Some(result) => result,
550 None => {
551 let _ = event_tx.send(ImportEvent::Complete {
552 imported: Vec::new(),
553 total_files: 0,
554 errors: 0,
555 duplicates: 0,
556 folders: Vec::new(),
557 });
558 continue;
559 }
560 };
561
562 let _ = event_tx.send(ImportEvent::WalkComplete { total, total_bytes });
563
564 // Check if loose-files mode is enabled for this vault
565 let loose_files = db
566 .conn()
567 .query_row(
568 "SELECT value FROM user_config WHERE key = 'loose_files'",
569 [],
570 |row| row.get::<_, String>(0),
571 )
572 .ok()
573 .is_some_and(|v| v == "1");
574
575 // Phase 2: import files with progress
576 let mut completed = 0usize;
577 let mut errors = 0usize;
578 let mut duplicates = 0usize;
579 let mut imported = Vec::new();
580
581 let mut ctx = ImportContext {
582 store: &store,
583 db: &db,
584 event_tx: &event_tx,
585 cmd_rx: &cmd_rx,
586 completed: &mut completed,
587 total,
588 errors: &mut errors,
589 duplicates: &mut duplicates,
590 imported: &mut imported,
591 loose_files,
592 };
593
594 let (cancelled, folders) = if flat {
595 let c = import_directory_flat(
596 &source, vfs_id, parent_id, &mut ctx,
597 );
598 (c, Vec::new())
599 } else {
600 import_structured(
601 &source, vfs_id, parent_id, &mut ctx,
602 )
603 };
604
605 let total_files = if cancelled { completed } else { total };
606
607 // Checkpoint WAL after large import to keep -shm file fresh
608 // and avoid stale memory-mapped state on macOS.
609 if let Err(e) = db.wal_checkpoint() {
610 warn!("WAL checkpoint after import failed: {e}");
611 }
612
613 let _ = event_tx.send(ImportEvent::Complete {
614 imported,
615 total_files,
616 errors,
617 duplicates,
618 folders,
619 });
620 }
621 }
622 }
623 }
624
625 #[cfg(test)]
626 mod tests {
627 use super::*;
628
629 #[test]
630 fn is_audio_file_recognises_extensions() {
631 assert!(is_audio_file(Path::new("kick.wav")));
632 assert!(is_audio_file(Path::new("pad.FLAC")));
633 assert!(is_audio_file(Path::new("song.mp3")));
634 assert!(is_audio_file(Path::new("loop.ogg")));
635 assert!(is_audio_file(Path::new("strings.aiff")));
636 assert!(is_audio_file(Path::new("brass.AIF")));
637 assert!(!is_audio_file(Path::new("readme.txt")));
638 assert!(!is_audio_file(Path::new("photo.png")));
639 assert!(!is_audio_file(Path::new("noext")));
640 }
641
642 #[test]
643 fn import_command_variants_constructible() {
644 let _import = ImportCommand::ImportDirectory {
645 source: PathBuf::from("/tmp/samples"),
646 strategy: ImportStrategy::MergeIntoVfs {
647 vfs_id: VfsId::from(1),
648 parent_id: None,
649 },
650 };
651 let _cancel = ImportCommand::Cancel;
652 let _shutdown = ImportCommand::Shutdown;
653 }
654
655 #[test]
656 fn import_event_variants_constructible() {
657 let _walk = ImportEvent::WalkComplete { total: 42, total_bytes: 1024 };
658 let _walk_progress = ImportEvent::WalkProgress { count: 17, total_bytes: 512 };
659 let _progress = ImportEvent::Progress {
660 completed: 5,
661 total: 42,
662 current_name: "kick.wav".to_string(),
663 };
664 let _err = ImportEvent::FileError {
665 path: "/tmp/bad.wav".to_string(),
666 error: "decode failed".to_string(),
667 };
668 let _done = ImportEvent::Complete {
669 imported: vec![("abc".to_string(), "wav".to_string())],
670 total_files: 1,
671 errors: 0,
672 duplicates: 0,
673 folders: vec![],
674 };
675 }
676
677 #[test]
678 fn import_strategy_variants_constructible() {
679 let _flat = ImportStrategy::Flat {
680 vfs_id: VfsId::from(1),
681 parent_id: None,
682 };
683 let _new = ImportStrategy::NewVfs {
684 vfs_name: "Test".to_string(),
685 };
686 let _merge = ImportStrategy::MergeIntoVfs {
687 vfs_id: VfsId::from(1),
688 parent_id: Some(NodeId::from(5)),
689 };
690 }
691
692 #[test]
693 fn spawn_and_drop_does_not_hang() {
694 let dir = tempfile::TempDir::new().unwrap();
695 let db_path = dir.path().join("test.db");
696 let store_root = dir.path().join("store");
697
698 // Create the DB so worker can open it
699 let _db = Database::open(&db_path).unwrap();
700
701 let handle = spawn_import_worker(db_path, store_root).unwrap();
702 assert!(handle.try_recv().is_none());
703 drop(handle); // Should send Shutdown and join cleanly
704 }
705 }
706