feat: add deduplication request and progress tracking for audio library

Author: Losses

messages/library_manage.proto

@@ -61,9 +61,30 @@ message AnalyzeAudioLibraryResponse {
     int32 total = 2;
 }
 
+
+// [DART-SIGNAL]
+message DeduplicateAudioLibraryRequest {
+    string path = 1;
+    float similarityThreshold = 2;
+    float workloadFactor = 3;
+}
+
+// [RUST-SIGNAL]
+message DeduplicateAudioLibraryProgress {
+    string path = 1;
+    int32 progress = 2;
+    int32 total = 3;
+}
+
+// [RUST-SIGNAL]
+message DeduplicateAudioLibraryResponse {
+    string path = 1;
+}
+
 enum CancelTaskType {
   AnalyzeAudioLibrary = 0;
   ScanAudioLibrary = 1;
+  DeduplicateAudioLibrary = 2;
 }
 
 // [DART-SIGNAL]

native/hub/src/backends/remote/mod.rs

@@ -273,6 +273,7 @@ impl WebSocketDartBridge {
                     task_tokens: Arc::new(Mutex::new(TaskTokens {
                         scan_token: None,
                         analyze_token: None,
+                        deduplicate_token: None,
                     })),
                     player: Arc::new(Mutex::new(MockPlayer {})),
                     sfx_player,

native/hub/src/handlers/library_manage.rs

@@ -2,6 +2,10 @@ use std::path::Path;
 use std::sync::Arc;
 
 use anyhow::{Context, Result};
+use database::actions::fingerprint::compare_all_pairs;
+use database::actions::fingerprint::compute_file_fingerprints;
+use database::actions::fingerprint::mark_duplicate_files;
+use database::actions::fingerprint::Configuration;
 use log::{debug, info, warn};
 use tokio::{sync::Mutex, task};
 use tokio_util::sync::CancellationToken;
@@ -279,6 +283,131 @@ impl Signal for AnalyzeAudioLibraryRequest {
     }
 }
 
+impl ParamsExtractor for DeduplicateAudioLibraryRequest {
+    type Params = (
+        Arc<MainDbConnection>,
+        Arc<Mutex<TaskTokens>>,
+        Arc<dyn Broadcaster>,
+    );
+
+    fn extract_params(&self, all_params: &GlobalParams) -> Self::Params {
+        (
+            Arc::clone(&all_params.main_db),
+            Arc::clone(&all_params.task_tokens),
+            Arc::clone(&all_params.broadcaster),
+        )
+    }
+}
+
+impl Signal for DeduplicateAudioLibraryRequest {
+    type Params = (
+        Arc<MainDbConnection>,
+        Arc<Mutex<TaskTokens>>,
+        Arc<dyn Broadcaster>,
+    );
+    type Response = ();
+
+    async fn handle(
+        &self,
+        (main_db, task_tokens, broadcaster): Self::Params,
+        _session: Option<Session>,
+        dart_signal: &Self,
+    ) -> Result<Option<Self::Response>> {
+        let mut tokens = task_tokens.lock().await;
+        if let Some(token) = tokens.deduplicate_token.take() {
+            token.cancel();
+        }
+
+        let new_token = CancellationToken::new();
+        tokens.deduplicate_token = Some(new_token.clone());
+        drop(tokens);
+
+        let request = dart_signal.clone();
+        let request_path = Arc::new(request.path.clone());
+        let batch_size = determine_batch_size(request.workload_factor);
+        let config = Configuration::default();
+        let similarity_threshold = request.similarity_threshold;
+
+        let request_path_clone = request_path.clone();
+        task::spawn_blocking(move || {
+            let rt = tokio::runtime::Runtime::new().unwrap();
+
+            let request_path_clone = request_path_clone.clone();
+            rt.block_on(async {
+                // Stage 1: Compute fingerprints (0% - 33%)
+                let broadcaster_clone = Arc::clone(&broadcaster);
+                let progress_path = request_path_clone.to_string();
+
+                let request_path_clone = request_path_clone.to_string();
+                compute_file_fingerprints(
+                    &main_db,
+                    Path::new(&request_path_clone),
+                    batch_size,
+                    move |cur, total| {
+                        let progress = cur as f32 / total as f32 * 0.33;
+
+                        broadcaster_clone.broadcast(&DeduplicateAudioLibraryProgress {
+                            path: progress_path.clone(),
+                            progress: (progress * 100.0) as i32,
+                            total: 100,
+                        })
+                    },
+                    Some(new_token.clone()),
+                )
+                .await?;
+
+                // Stage 2: Compare all pairs (33% - 66%)
+                let broadcaster_clone = Arc::clone(&broadcaster);
+                let progress_path = request_path_clone.to_string();
+
+                compare_all_pairs(
+                    &main_db,
+                    batch_size,
+                    move |cur, total| {
+                        let progress = 0.33 + cur as f32 / total as f32 * 0.33;
+
+                        broadcaster_clone.broadcast(&DeduplicateAudioLibraryProgress {
+                            path: progress_path.clone(),
+                            progress: (progress * 100.0) as i32,
+                            total: 100,
+                        });
+                    },
+                    &config,
+                    Some(Arc::new(new_token.clone())),
+                    1000,
+                )
+                .await?;
+
+                // Stage 3: Mark duplicates (66% - 100%)
+                if !new_token.is_cancelled() {
+                    let broadcaster_clone = Arc::clone(&broadcaster);
+                    let progress_path = request_path_clone.to_string();
+
+                    mark_duplicate_files(&main_db, similarity_threshold, move |cur, total| {
+                        let progress = 0.66 + cur as f32 / total as f32 * 0.34;
+
+                        broadcaster_clone.broadcast(&DeduplicateAudioLibraryProgress {
+                            path: progress_path.clone(),
+                            progress: (progress * 100.0) as i32,
+                            total: 100,
+                        });
+                    })
+                    .await?;
+                }
+
+                let request_path_clone = request_path_clone.to_string();
+                broadcaster.broadcast(&DeduplicateAudioLibraryResponse {
+                    path: request_path_clone,
+                });
+
+                Ok::<(), anyhow::Error>(())
+            })
+        });
+
+        Ok(Some(()))
+    }
+}
+
 impl ParamsExtractor for CancelTaskRequest {
     type Params = (Arc<Mutex<TaskTokens>>,);
 

native/hub/src/utils/broadcastable.rs

@@ -7,6 +7,10 @@ use crate::utils::RinfRustSignal;
 broadcastable!(ScanAudioLibraryProgress, ScanAudioLibraryResponse);
 broadcastable!(SetMediaLibraryPathResponse);
 broadcastable!(AnalyzeAudioLibraryProgress, AnalyzeAudioLibraryResponse);
+broadcastable!(
+    DeduplicateAudioLibraryProgress,
+    DeduplicateAudioLibraryResponse
+);
 broadcastable!(
     PlaybackStatus,
     ScrobbleServiceStatusUpdated,

native/hub/src/utils/mod.rs

@@ -70,6 +70,7 @@ pub async fn initialize_databases(
 pub struct TaskTokens {
     pub scan_token: Option<CancellationToken>,
     pub analyze_token: Option<CancellationToken>,
+    pub deduplicate_token: Option<CancellationToken>,
 }
 
 #[derive(Debug, Clone, Copy)]