Skip to main content

max / audiofiles

23.5 KB · 611 lines History Blame Raw
1 //! `DirectBackend` long-running operations: import, analysis, export, and VFS
2 //! mirror sync. These spawn workers and report progress through event channels
3 //! rather than blocking, and each carries a matching cancel path.
4
5 use super::{
6 AnalysisConfig, BackendError, BackendEvent, BackendResult, CleanupCommand, DirectBackend,
7 EditCommand, EditEvent, EditOperation, ExportCommand, ExportConfigDesc, ExportItemDesc,
8 ForgeEvent, ForgeImportOutcome, ForgePending, ImportCommand, ImportEvent, ImportStrategyDesc,
9 ImportedFolderDesc, OperationsBackend, Path, PathBuf, SearchEvent, WorkerCommand, WorkerEvent,
10 forge_import_chop, forge_import_conform, resolve_import_strategy,
11 };
12
13 impl OperationsBackend for DirectBackend {
14 // --- VFS Mirror ---
15
16 fn sync_vfs_mirror(&self, mirror_root: &Path) -> BackendResult<(usize, usize, usize)> {
17 let db = self.db.lock();
18 let config = audiofiles_core::vfs_mirror::MirrorConfig {
19 mirror_root: mirror_root.to_path_buf(),
20 store_root: self.store.root().to_path_buf(),
21 };
22 let stats = audiofiles_core::vfs_mirror::sync_mirror(&db, &config)?;
23 Ok((
24 stats.dirs_created,
25 stats.links_created,
26 stats.entries_removed,
27 ))
28 }
29
30 // --- Long-running operations ---
31
32 fn start_import(&self, source: &Path, strategy: ImportStrategyDesc) -> BackendResult<()> {
33 let import_strategy = resolve_import_strategy(strategy);
34 let handle = self.respawn_import_worker()?;
35 handle.send(ImportCommand::ImportDirectory {
36 source: source.to_path_buf(),
37 strategy: import_strategy,
38 });
39 *self.import_worker.lock() = Some(handle);
40 Ok(())
41 }
42
43 fn start_file_import(
44 &self,
45 paths: &[PathBuf],
46 strategy: ImportStrategyDesc,
47 ) -> BackendResult<()> {
48 let import_strategy = resolve_import_strategy(strategy);
49 let handle = self.respawn_import_worker()?;
50 handle.send(ImportCommand::ImportFiles {
51 paths: paths.to_vec(),
52 strategy: import_strategy,
53 });
54 *self.import_worker.lock() = Some(handle);
55 Ok(())
56 }
57
58 fn start_analysis(
59 &self,
60 sample_hashes: Vec<(String, String)>,
61 config: AnalysisConfig,
62 ) -> BackendResult<()> {
63 let samples: Vec<(String, String, PathBuf)> = {
64 let db = self.db.lock();
65 sample_hashes
66 .into_iter()
67 .filter_map(|(hash, ext)| {
68 let path =
69 audiofiles_core::store::resolve_file_path(&self.store, &db, &hash, &ext)
70 .ok()?;
71 if path.exists() {
72 Some((hash, ext, path))
73 } else {
74 None
75 }
76 })
77 .collect()
78 };
79
80 // Cancel any existing analysis worker before starting a new one
81 if let Some(old) = self.analysis_worker.lock().take() {
82 old.send(WorkerCommand::Cancel);
83 drop(old);
84 }
85
86 let handle = audiofiles_core::analysis::worker::spawn_worker()
87 .map_err(|e| BackendError::Other(format!("failed to spawn analysis worker: {e}")))?;
88 handle.send(WorkerCommand::AnalyzeBatch { samples, config });
89 *self.analysis_worker.lock() = Some(handle);
90 Ok(())
91 }
92
93 fn start_export(
94 &self,
95 items: Vec<ExportItemDesc>,
96 config: ExportConfigDesc,
97 ) -> BackendResult<()> {
98 let mut config = config;
99 let mut items = items;
100
101 #[cfg(feature = "device-profiles")]
102 self.resolve_device_profile(&mut config, &mut items);
103
104 let store_root = self.store.root().to_path_buf();
105 // Cancel any existing export worker before starting a new one
106 if let Some(old) = self.export_worker.lock().take() {
107 old.send(ExportCommand::Cancel);
108 drop(old);
109 }
110
111 let handle = crate::export::spawn_export_worker(store_root)
112 .map_err(|e| BackendError::Other(format!("failed to spawn export worker: {e}")))?;
113 handle.send(ExportCommand::Export { items, config });
114 *self.export_worker.lock() = Some(handle);
115 Ok(())
116 }
117
118 fn cancel_import(&self) -> BackendResult<()> {
119 if let Some(worker) = self.import_worker.lock().take() {
120 worker.send(ImportCommand::Cancel);
121 }
122 Ok(())
123 }
124
125 fn cancel_analysis(&self) -> BackendResult<()> {
126 if let Some(worker) = self.analysis_worker.lock().take() {
127 worker.send(WorkerCommand::Cancel);
128 }
129 Ok(())
130 }
131
132 fn cancel_export(&self) -> BackendResult<()> {
133 if let Some(worker) = self.export_worker.lock().take() {
134 worker.send(ExportCommand::Cancel);
135 }
136 Ok(())
137 }
138
139 fn start_edit(&self, hash: &str, ext: &str, operation: EditOperation) -> BackendResult<()> {
140 let path = {
141 let db = self.db.lock();
142 audiofiles_core::store::resolve_file_path(&self.store, &db, hash, ext)?
143 };
144 if !path.exists() {
145 return Err(BackendError::Core(
146 audiofiles_core::error::CoreError::SampleNotFound(hash.to_string()),
147 ));
148 }
149
150 // Cancel any existing edit worker before starting a new one
151 if let Some(old) = self.edit_worker.lock().take() {
152 old.send(EditCommand::Cancel);
153 drop(old);
154 }
155
156 let handle = audiofiles_core::edit::worker::spawn_edit_worker()
157 .map_err(|e| BackendError::Other(format!("failed to spawn edit worker: {e}")))?;
158 handle.send(EditCommand::Edit {
159 hash: hash.to_string(),
160 ext: ext.to_string(),
161 path,
162 operation,
163 });
164 *self.edit_worker.lock() = Some(handle);
165 Ok(())
166 }
167
168 fn cancel_edit(&self) -> BackendResult<()> {
169 if let Some(worker) = self.edit_worker.lock().take() {
170 worker.send(EditCommand::Cancel);
171 }
172 Ok(())
173 }
174
175 fn start_cleanup(&self) -> BackendResult<()> {
176 // Cancel any existing cleanup worker to prevent concurrent cleanups
177 if let Some(worker) = self.cleanup_worker.lock().take() {
178 worker.send(CleanupCommand::Cancel);
179 }
180
181 let db_path = self.data_dir.join("audiofiles.db");
182 let store_root = self.store.root().to_path_buf();
183
184 let handle = crate::cleanup::spawn_cleanup_worker(db_path, store_root)
185 .map_err(|e| BackendError::Other(format!("failed to spawn cleanup worker: {e}")))?;
186 handle.send(CleanupCommand::RemoveOrphans);
187 *self.cleanup_worker.lock() = Some(handle);
188 Ok(())
189 }
190
191 fn cancel_cleanup(&self) -> BackendResult<()> {
192 if let Some(worker) = self.cleanup_worker.lock().take() {
193 worker.send(CleanupCommand::Cancel);
194 }
195 Ok(())
196 }
197
198 fn record_edit_history(
199 &self,
200 source_hash: &str,
201 result_hash: &str,
202 operation: &EditOperation,
203 ) -> BackendResult<()> {
204 let db = self.db.lock();
205 let op_name = operation.display_name();
206 let params_json = serde_json::to_string(operation)
207 .map_err(|e| BackendError::Other(format!("serialize edit params: {e}")))?;
208 db.conn()
209 .execute(
210 "INSERT INTO edit_history (source_hash, result_hash, operation, params_json)
211 VALUES (?1, ?2, ?3, ?4)",
212 rusqlite::params![source_hash, result_hash, op_name, params_json],
213 )
214 .map_err(audiofiles_core::error::CoreError::Db)?;
215 Ok(())
216 }
217
218 fn delete_edit_history(&self, source_hash: &str, result_hash: &str) -> BackendResult<()> {
219 let db = self.db.lock();
220 db.conn()
221 .execute(
222 "DELETE FROM edit_history
223 WHERE id = (
224 SELECT id FROM edit_history
225 WHERE source_hash = ?1 AND result_hash = ?2
226 ORDER BY id DESC
227 LIMIT 1
228 )",
229 rusqlite::params![source_hash, result_hash],
230 )
231 .map_err(audiofiles_core::error::CoreError::Db)?;
232 Ok(())
233 }
234
235 fn storage_stats(&self) -> BackendResult<crate::backend::StorageStats> {
236 let db = self.db.lock();
237 let (sample_count, total_bytes) = db
238 .storage_stats()
239 .map_err(|e| BackendError::Other(e.to_string()))?;
240 let db_path = self.data_dir.join("audiofiles.db");
241 let db_bytes = std::fs::metadata(&db_path).map_or(0, |m| m.len());
242 Ok(crate::backend::StorageStats {
243 sample_count,
244 total_bytes,
245 db_bytes,
246 })
247 }
248
249 fn vfs_storage_stats(&self, vfs_id: audiofiles_core::VfsId) -> BackendResult<(u64, u64)> {
250 let db = self.db.lock();
251 db.vfs_storage_stats(vfs_id.as_i64())
252 .map_err(|e| BackendError::Other(e.to_string()))
253 }
254
255 fn poll_events(&self) -> Vec<BackendEvent> {
256 let mut events = Vec::new();
257
258 // Poll import worker
259 if let Some(ref worker) = *self.import_worker.lock() {
260 while let Some(event) = worker.try_recv() {
261 match event {
262 ImportEvent::WalkProgress { count, total_bytes } => {
263 events.push(BackendEvent::ImportWalkProgress { count, total_bytes });
264 }
265 ImportEvent::WalkComplete { total, total_bytes } => {
266 events.push(BackendEvent::ImportWalkComplete { total, total_bytes });
267 }
268 ImportEvent::Progress {
269 completed,
270 total,
271 current_name,
272 } => {
273 events.push(BackendEvent::ImportProgress {
274 completed,
275 total,
276 current_name,
277 });
278 }
279 ImportEvent::FileError { path, error } => {
280 events.push(BackendEvent::ImportFileError { path, error });
281 }
282 ImportEvent::Complete {
283 imported,
284 total_files,
285 errors,
286 duplicates,
287 folders,
288 } => {
289 events.push(BackendEvent::ImportComplete {
290 imported,
291 total_files,
292 errors,
293 duplicates,
294 folders: folders
295 .into_iter()
296 .map(|f| ImportedFolderDesc {
297 name: f.name,
298 samples: f.samples,
299 })
300 .collect(),
301 });
302 }
303 }
304 }
305 }
306
307 // Poll analysis worker
308 if let Some(ref worker) = *self.analysis_worker.lock() {
309 while let Some(event) = worker.try_recv() {
310 match event {
311 WorkerEvent::Progress {
312 completed,
313 total,
314 current_name,
315 } => {
316 events.push(BackendEvent::AnalysisProgress {
317 completed,
318 total,
319 current_name,
320 });
321 }
322 WorkerEvent::SampleDone {
323 result,
324 suggestions,
325 } => {
326 events.push(BackendEvent::AnalysisSampleDone {
327 result,
328 suggestions,
329 });
330 }
331 WorkerEvent::SampleError { hash, error } => {
332 events.push(BackendEvent::AnalysisSampleError { hash, error });
333 }
334 WorkerEvent::BatchComplete => {
335 events.push(BackendEvent::AnalysisBatchComplete);
336 }
337 }
338 }
339 }
340
341 // Poll export worker
342 if let Some(ref worker) = *self.export_worker.lock() {
343 while let Some(event) = worker.try_recv() {
344 match event {
345 crate::export::ExportEvent::Progress {
346 completed,
347 total,
348 current_name,
349 } => {
350 events.push(BackendEvent::ExportProgress {
351 completed,
352 total,
353 current_name,
354 });
355 }
356 crate::export::ExportEvent::Complete { total, errors } => {
357 events.push(BackendEvent::ExportComplete { total, errors });
358 }
359 }
360 }
361 }
362
363 // Poll cleanup worker
364 if let Some(ref worker) = *self.cleanup_worker.lock() {
365 while let Some(event) = worker.try_recv() {
366 match event {
367 crate::cleanup::CleanupEvent::Progress {
368 completed,
369 total,
370 current_name,
371 } => {
372 events.push(BackendEvent::CleanupProgress {
373 completed,
374 total,
375 current_name,
376 });
377 }
378 crate::cleanup::CleanupEvent::Complete { removed, errors } => {
379 events.push(BackendEvent::CleanupComplete { removed, errors });
380 }
381 }
382 }
383 }
384
385 // Poll loose-files worker
386 if let Some(ref worker) = *self.loose_files_worker.lock() {
387 use crate::loose_files_worker::LooseFilesEvent as Lfe;
388 while let Some(event) = worker.try_recv() {
389 events.push(match event {
390 Lfe::IntegrityResult { missing } => {
391 BackendEvent::LooseFilesIntegrity { missing }
392 }
393 Lfe::RelocateResult {
394 relocated,
395 still_missing,
396 } => BackendEvent::LooseFilesRelocated {
397 relocated,
398 still_missing,
399 },
400 Lfe::PurgeResult { purged } => BackendEvent::LooseFilesPurged { purged },
401 Lfe::VerifyResult { checked, corrupt } => {
402 BackendEvent::StoreIntegrity { checked, corrupt }
403 }
404 Lfe::Failed { message } => BackendEvent::LooseFilesFailed { message },
405 });
406 }
407 }
408
409 // Poll edit worker
410 if let Some(ref worker) = *self.edit_worker.lock() {
411 while let Some(event) = worker.try_recv() {
412 match event {
413 EditEvent::Started { hash } => {
414 events.push(BackendEvent::EditStarted { hash });
415 }
416 EditEvent::Complete {
417 source_hash,
418 result_path,
419 operation,
420 } => {
421 events.push(BackendEvent::EditComplete {
422 source_hash,
423 result_path,
424 operation,
425 });
426 }
427 EditEvent::Error { hash, error } => {
428 events.push(BackendEvent::EditError { hash, error });
429 }
430 }
431 }
432 }
433
434 // Poll forge worker. Collect events first (releasing the worker lock),
435 // then run the import stage here on the GUI thread, the worker only did
436 // the CPU work (decode/slice/conform/encode → temp files), so the DB
437 // write is fast and never blocked the render thread.
438 let forge_events: Vec<ForgeEvent> = match *self.forge_worker.lock() {
439 Some(ref worker) => {
440 let mut evs = Vec::new();
441 while let Some(ev) = worker.try_recv() {
442 evs.push(ev);
443 }
444 evs
445 }
446 None => Vec::new(),
447 };
448 for ev in forge_events {
449 match ev {
450 ForgeEvent::Started => events.push(BackendEvent::ForgeStarted),
451 ForgeEvent::Error { error } => {
452 self.forge_pending.lock().take();
453 events.push(BackendEvent::ForgeError { error });
454 }
455 ForgeEvent::ChopComplete { staging } => {
456 let pending = self.forge_pending.lock().take();
457 if let Some(ForgePending::Chop { vfs_id, parent_id }) = pending {
458 // Import off the GUI thread on its own WAL connection so the
459 // blob copies + DB writes never hold the GUI's db.lock().
460 let data_dir = self.data_dir.clone();
461 let root = self.store.root().to_path_buf();
462 let job = audiofiles_core::worker_runtime::spawn_job(
463 "forge-import-chop",
464 || ForgeImportOutcome::Error {
465 error: "forge chop import panicked".to_string(),
466 },
467 move || forge_import_chop(&data_dir, &root, vfs_id, parent_id, staging),
468 );
469 match job {
470 Ok(handle) => *self.forge_import_job.lock() = Some(handle),
471 Err(e) => events.push(BackendEvent::ForgeError {
472 error: e.to_string(),
473 }),
474 }
475 } else {
476 // No matching pending context (cancelled/superseded),
477 // drop the staged temp files rather than leak them.
478 for (_, p) in &staging.slices {
479 let _ = std::fs::remove_file(p);
480 }
481 }
482 }
483 ForgeEvent::ConformComplete { staging } => {
484 let pending = self.forge_pending.lock().take();
485 if let Some(ForgePending::Conform {
486 vfs_id,
487 parent_id,
488 sample_rate,
489 bit_depth,
490 }) = pending
491 {
492 // Import off the GUI thread (see ChopComplete).
493 let data_dir = self.data_dir.clone();
494 let root = self.store.root().to_path_buf();
495 let job = audiofiles_core::worker_runtime::spawn_job(
496 "forge-import-conform",
497 || ForgeImportOutcome::Error {
498 error: "forge conform import panicked".to_string(),
499 },
500 move || {
501 forge_import_conform(
502 &data_dir,
503 &root,
504 vfs_id,
505 parent_id,
506 sample_rate,
507 bit_depth,
508 staging,
509 )
510 },
511 );
512 match job {
513 Ok(handle) => *self.forge_import_job.lock() = Some(handle),
514 Err(e) => events.push(BackendEvent::ForgeError {
515 error: e.to_string(),
516 }),
517 }
518 } else {
519 let _ = std::fs::remove_file(&staging.temp_path);
520 }
521 }
522 ForgeEvent::PreviewComplete { marks } => {
523 events.push(BackendEvent::ChopPreviewComplete { marks });
524 }
525 }
526 }
527
528 // Drain the off-thread forge import job (chop/conform slices imported on
529 // their own connection), emitting its completion on a later frame.
530 let import_outcome = {
531 let guard = self.forge_import_job.lock();
532 let outcome = guard
533 .as_ref()
534 .and_then(audiofiles_core::worker_runtime::JobHandle::try_recv);
535 drop(guard);
536 if outcome.is_some() {
537 *self.forge_import_job.lock() = None;
538 }
539 outcome
540 };
541 if let Some(outcome) = import_outcome {
542 match outcome {
543 ForgeImportOutcome::Chop { slice_count } => {
544 events.push(BackendEvent::ForgeChopComplete { slice_count });
545 }
546 ForgeImportOutcome::Conform {
547 sample_rate,
548 bit_depth,
549 overshoot,
550 } => events.push(BackendEvent::ForgeConformComplete {
551 sample_rate,
552 bit_depth,
553 overshoot,
554 }),
555 ForgeImportOutcome::Error { error } => {
556 events.push(BackendEvent::ForgeError { error });
557 }
558 }
559 }
560
561 // Poll search worker (similarity / near-duplicate VP-tree build + query).
562 let search_events: Vec<SearchEvent> = match *self.search_worker.lock() {
563 Some(ref worker) => {
564 let mut evs = Vec::new();
565 while let Some(ev) = worker.try_recv() {
566 evs.push(ev);
567 }
568 evs
569 }
570 None => Vec::new(),
571 };
572 for ev in search_events {
573 match ev {
574 SearchEvent::SimilarResults { source, hashes } => {
575 events.push(BackendEvent::SimilarResults { source, hashes });
576 }
577 SearchEvent::NearDuplicateResults { source, hashes } => {
578 events.push(BackendEvent::NearDuplicateResults { source, hashes });
579 }
580 SearchEvent::Error { error } => {
581 events.push(BackendEvent::SearchError { error });
582 }
583 }
584 }
585
586 // Poll classifier worker (one-shot job).
587 let job_done = {
588 let guard = self.classifier_worker.lock();
589 match guard
590 .as_ref()
591 .and_then(crate::classifier_worker::ClassifierHandle::try_recv)
592 {
593 Some(crate::classifier_worker::ClassifierWorkerEvent::Done(result)) => {
594 events.push(BackendEvent::ClassifierJobDone(result));
595 true
596 }
597 Some(crate::classifier_worker::ClassifierWorkerEvent::Failed(error)) => {
598 events.push(BackendEvent::ClassifierJobFailed(error));
599 true
600 }
601 None => false,
602 }
603 };
604 if job_done {
605 *self.classifier_worker.lock() = None; // join the finished thread
606 }
607
608 events
609 }
610 }
611