fix(targets): add multiple library matching (#253)

Co-authored-by: Copilot <[email protected]>

Author: dan-online

.gitignore

@@ -15,4 +15,5 @@ data
 *.sh
 
 # Nix
-result
+result
+shell.nix

crates/service/src/settings/targets/autopulse.rs

@@ -54,12 +54,12 @@ impl Autopulse {
 
 impl TargetProcess for Autopulse {
     async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
-        let mut succeded = Vec::new();
+        let mut succeeded = Vec::new();
 
         for ev in evs {
             match self.scan(ev).await {
                 Ok(()) => {
-                    succeded.push(ev.id.clone());
+                    succeeded.push(ev.id.clone());
                     debug!("file scanned: {}", ev.get_path(&self.rewrite));
                 }
                 Err(e) => {
@@ -68,6 +68,6 @@ impl TargetProcess for Autopulse {
             }
         }
 
-        Ok(succeded)
+        Ok(succeeded)
     }
 }

crates/service/src/settings/targets/command.rs

@@ -83,16 +83,16 @@ impl Command {
 
 impl TargetProcess for Command {
     async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
-        let mut succeded = Vec::new();
+        let mut succeeded = Vec::new();
 
         for ev in evs {
             if let Err(e) = self.run(ev).await {
                 error!("failed to process '{}': {}", ev.get_path(&self.rewrite), e);
             } else {
-                succeded.push(ev.id.clone());
+                succeeded.push(ev.id.clone());
             }
         }
 
-        Ok(succeded)
+        Ok(succeeded)
     }
 }

crates/service/src/settings/targets/emby.rs

@@ -129,20 +129,21 @@ impl Emby {
         Ok(res.json().await?)
     }
 
-    fn get_library(&self, libraries: &[Library], path: &str) -> Option<Library> {
+    fn get_libraries(&self, libraries: &[Library], path: &str) -> Vec<Library> {
         let ev_path = Path::new(path);
+        let mut matched: Vec<Library> = vec![];
 
         for library in libraries {
             for location in &library.locations {
                 let path = Path::new(location);
 
                 if ev_path.starts_with(path) {
-                    return Some(library.clone());
+                    matched.push(library.clone());
                 }
             }
         }
 
-        None
+        matched
     }
 
     async fn _get_item(&self, library: &Library, path: &str) -> anyhow::Result<Option<Item>> {
@@ -171,7 +172,7 @@ impl Emby {
 
         let res = client.get(url).perform().await?;
 
-        // Possibly uneeded unless we can use streams
+        // Possibly unneeded unless we can use streams
         let bytes = res.bytes().await?;
 
         let mut json_reader = JsonStreamReader::new(Cursor::new(bytes));
@@ -352,7 +353,7 @@ impl TargetProcess for Emby {
             .await
             .context("failed to fetch libraries")?;
 
-        let mut succeded = Vec::new();
+        let mut succeeded: HashMap<String, bool> = HashMap::new();
 
         let mut to_find = HashMap::new();
         let mut to_refresh = Vec::new();
@@ -362,10 +363,15 @@ impl TargetProcess for Emby {
             for ev in evs {
                 let ev_path = ev.get_path(&self.rewrite);
 
-                if let Some(library) = self.get_library(&libraries, &ev_path) {
-                    to_find.entry(library).or_insert_with(Vec::new).push(*ev);
-                } else {
+                let matched_libraries = self.get_libraries(&libraries, &ev_path);
+
+                if matched_libraries.is_empty() {
                     error!("failed to find library for file: {}", ev_path);
+                    continue;
+                }
+
+                for library in matched_libraries {
+                    to_find.entry(library).or_insert_with(Vec::new).push(*ev);
                 }
             }
 
@@ -388,10 +394,11 @@ impl TargetProcess for Emby {
                 match self.refresh_item(&item).await {
                     Ok(()) => {
                         debug!("refreshed item: {}", item.id);
-                        succeded.push(ev.id.clone());
+                        *succeeded.entry(ev.id.clone()).or_insert(true) &= true;
                     }
                     Err(e) => {
                         error!("failed to refresh item: {}", e);
+                        succeeded.insert(ev.id.clone(), false);
                     }
                 }
             }
@@ -400,15 +407,27 @@ impl TargetProcess for Emby {
         }
 
         if !to_scan.is_empty() {
-            self.scan(&to_scan).await.context("failed to scan files")?;
+            match self.scan(&to_scan).await {
+                Ok(()) => {
+                    for ev in &to_scan {
+                        debug!("scanned file: {}", ev.file_path);
+
+                        *succeeded.entry(ev.id.clone()).or_insert(true) &= true;
+                    }
+                }
+                Err(e) => {
+                    error!("failed to scan items: {}", e);
 
-            for file in &to_scan {
-                debug!("scanned file: {}", file.file_path);
+                    for ev in &to_scan {
+                        succeeded.insert(ev.id.clone(), false);
+                    }
+                }
             }
         }
 
-        succeded.extend(to_scan.iter().map(|ev| ev.id.clone()));
-
-        Ok(succeded)
+        Ok(succeeded
+            .iter()
+            .filter_map(|(k, v)| if *v { Some(k.clone()) } else { None })
+            .collect())
     }
 }

crates/service/src/settings/targets/fileflows.rs

@@ -130,7 +130,7 @@ impl FileFlows {
         Ok(files.first().cloned())
     }
 
-    async fn reprocess_library_filse(&self, evs: Vec<&FileFlowsLibraryFile>) -> anyhow::Result<()> {
+    async fn reprocess_library_file(&self, evs: Vec<&FileFlowsLibraryFile>) -> anyhow::Result<()> {
         let client = self.get_client()?;
 
         let url = get_url(&self.url)?.join("api/library-file/reprocess")?;
@@ -274,7 +274,7 @@ impl TargetProcess for FileFlows {
 
             if !processed.is_empty() {
                 match self
-                    .reprocess_library_filse(
+                    .reprocess_library_file(
                         processed
                             .iter()
                             .filter_map(|(_, file)| file.as_ref())

crates/service/src/settings/targets/plex.rs

@@ -6,7 +6,10 @@ use autopulse_database::models::ScanEvent;
 use autopulse_utils::{get_url, squash_directory, what_is, PathType};
 use reqwest::header;
 use serde::Deserialize;
-use std::path::Path;
+use std::{
+    collections::{HashMap, HashSet},
+    path::Path,
+};
 use tracing::{debug, error, trace};
 
 #[derive(Deserialize, Clone)]
@@ -152,7 +155,7 @@ impl Plex {
         Ok(libraries.media_container.directory.unwrap())
     }
 
-    fn get_library(&self, libraries: &[Library], path: &str) -> Option<Library> {
+    fn get_libraries(&self, libraries: &[Library], path: &str) -> Vec<Library> {
         let ev_path = Path::new(path);
         let mut matches: Vec<(usize, &Library)> = vec![];
 
@@ -165,13 +168,13 @@ impl Plex {
             }
         }
 
+        // Sort the best matched library first
         matches.sort_by(|(len_a, _), (len_b, _)| len_b.cmp(len_a));
 
-        // Return the most specific match
         matches
             .into_iter()
-            .next()
             .map(|(_, library)| library.clone())
+            .collect()
     }
 
     async fn get_episodes(&self, key: &str) -> anyhow::Result<LibraryResponse> {
@@ -398,13 +401,27 @@ impl TargetProcess for Plex {
     async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
         let libraries = self.libraries().await.context("failed to get libraries")?;
 
-        let mut succeeded = Vec::new();
+        let mut succeeded: HashMap<String, bool> = HashMap::new();
 
         for ev in evs {
+            let succeeded_entry = succeeded.entry(ev.id.clone()).or_insert(true);
+
             let ev_path = ev.get_path(&self.rewrite);
+            let matched_libraries = self.get_libraries(&libraries, &ev_path);
+
+            if matched_libraries.is_empty() {
+                error!("no matching library for {ev_path}");
+
+                *succeeded_entry = false;
+
+                continue;
+            }
+
+            let mut processed_items = HashSet::new();
 
-            if let Some(library) = self.get_library(&libraries, &ev_path) {
+            for library in matched_libraries {
                 trace!("found library '{}' for {ev_path}", library.title);
+
                 match self.scan(ev, &library).await {
                     Ok(()) => {
                         debug!("scanned '{}'", ev_path);
@@ -417,7 +434,8 @@ impl TargetProcess for Plex {
                                             "failed to find items for file: '{}', leaving at scan",
                                             ev_path
                                         );
-                                        succeeded.push(ev.id.clone());
+
+                                        *succeeded_entry = true;
                                     } else {
                                         trace!("found items for file '{}'", ev_path);
 
@@ -426,6 +444,14 @@ impl TargetProcess for Plex {
                                         for item in items {
                                             let mut item_success = true;
 
+                                            if processed_items.contains(&item.key) {
+                                                debug!(
+                                                    "already processed item '{}' earlier, skipping",
+                                                    item.key
+                                                );
+                                                continue;
+                                            }
+
                                             if self.refresh {
                                                 match self.refresh_item(&item.key).await {
                                                     Ok(()) => {
@@ -459,10 +485,12 @@ impl TargetProcess for Plex {
                                             if !item_success {
                                                 all_success = false;
                                             }
+
+                                            processed_items.insert(item.key);
                                         }
 
                                         if all_success {
-                                            succeeded.push(ev.id.clone());
+                                            *succeeded_entry &= true;
                                         }
                                     }
                                 }
@@ -471,19 +499,20 @@ impl TargetProcess for Plex {
                                 }
                             };
                         } else {
-                            succeeded.push(ev.id.clone());
+                            *succeeded_entry &= true;
                         }
                     }
                     Err(e) => {
                         error!("failed to scan file '{}': {}", ev_path, e);
                     }
                 }
-            } else {
-                error!("failed to find library for file: {}", ev_path);
             }
         }
 
-        Ok(succeeded)
+        Ok(succeeded
+            .into_iter()
+            .filter_map(|(k, v)| if v { Some(k) } else { None })
+            .collect())
     }
 }
 
@@ -545,8 +574,8 @@ mod tests {
         }];
 
         let path = "/media/movies/Inception.mkv";
-        let library = plex.get_library(&libraries, path).unwrap();
-        assert!(library.key == "library_key_movies");
+        let libraries = plex.get_libraries(&libraries, path);
+        assert!(libraries[0].key == "library_key_movies");
 
         let nested_libraries = [
             Library {
@@ -567,7 +596,8 @@ mod tests {
 
         let path = "/media/movies/4k/Inception.mkv";
 
-        let library = plex.get_library(&nested_libraries, path).unwrap();
-        assert!(library.key == "library_key_movies_4k");
+        let libraries = plex.get_libraries(&nested_libraries, path);
+        assert!(libraries[0].key == "library_key_movies_4k");
+        assert!(libraries[1].key == "library_key_movies");
     }
 }