diff --git a/crates/arr-indexer/Cargo.toml b/crates/arr-indexer/Cargo.toml index c90106f..7503622 100644 --- a/crates/arr-indexer/Cargo.toml +++ b/crates/arr-indexer/Cargo.toml @@ -12,9 +12,9 @@ quick-xml.workspace = true reqwest.workspace = true serde.workspace = true thiserror.workspace = true +tokio.workspace = true [dev-dependencies] -tokio.workspace = true wiremock.workspace = true [lints] diff --git a/crates/arr-indexer/src/search.rs b/crates/arr-indexer/src/search.rs index bbfd455..e8e6d6a 100644 --- a/crates/arr-indexer/src/search.rs +++ b/crates/arr-indexer/src/search.rs @@ -1,4 +1,4 @@ -use std::time::SystemTime; +use std::{borrow::Cow, time::SystemTime}; use chrono::DateTime; use quick_xml::{ @@ -8,6 +8,7 @@ use quick_xml::{ }; use reqwest::StatusCode; use thiserror::Error; +use tokio::task::JoinSet; use crate::ProwlarrClient; @@ -38,7 +39,7 @@ pub struct SearchRelease { pub size: Option, pub seeders: Option, pub publish_date: Option, - pub download_url: Option, + pub download_url: String, } /// Results from one indexer in a multi-indexer search. @@ -61,6 +62,8 @@ pub enum SearchError { code: Option, description: String, }, + #[error("search task did not complete")] + Task, } impl ProwlarrClient { @@ -70,19 +73,37 @@ impl ProwlarrClient { indexer_ids: &[i64], request: &SearchRequest, ) -> Vec { - let mut searches = Vec::with_capacity(indexer_ids.len()); - for &indexer_id in indexer_ids { - let (releases, error) = match self.search_indexer(indexer_id, request).await { - Ok(releases) => (releases, None), - Err(error) => (Vec::new(), Some(error)), - }; - searches.push(IndexerSearch { - indexer_id, - releases, - error, + let mut tasks = JoinSet::new(); + for (position, &indexer_id) in indexer_ids.iter().enumerate() { + let client = self.clone(); + let request = request.clone(); + tasks.spawn(async move { + ( + position, + indexer_id, + client.search_indexer(indexer_id, &request).await, + ) }); } + + let mut searches = vec![None; indexer_ids.len()]; + while let Some(result) = tasks.join_next().await { + if let Ok((position, indexer_id, result)) = result { + searches[position] = Some(search_result(indexer_id, result)); + } + } + searches + .into_iter() + .enumerate() + .map(|(position, search)| { + search.unwrap_or_else(|| IndexerSearch { + indexer_id: indexer_ids[position], + releases: Vec::new(), + error: Some(SearchError::Task), + }) + }) + .collect() } /// Runs one search against one indexer's Torznab endpoint. @@ -171,12 +192,9 @@ struct ReleaseBuilder { impl ReleaseBuilder { fn finish(self, indexer_id: i64) -> Option { - let name = self.title.filter(|title| !title.is_empty())?; - let download_url = self.enclosure_url.or(self.link); - let guid = self - .guid - .filter(|guid| !guid.is_empty()) - .or_else(|| download_url.clone())?; + let name = cleaned(self.title)?; + let download_url = cleaned(self.enclosure_url.or(self.link))?; + let guid = cleaned(self.guid).unwrap_or_else(|| download_url.clone()); Some(SearchRelease { indexer_id, @@ -190,16 +208,15 @@ impl ReleaseBuilder { } fn set_text(&mut self, field: Field, value: &str) { - let value = value.trim(); match field { Field::Title => append_text(&mut self.title, value), Field::Guid => append_text(&mut self.guid, value), Field::Link => append_text(&mut self.link, value), Field::PublishDate => { - self.publish_date = parse_date(value).or(self.publish_date); + self.publish_date = parse_date(value.trim()).or(self.publish_date); } - Field::Size => self.size = value.parse().ok().or(self.size), - Field::Seeders => self.seeders = value.parse().ok().or(self.seeders), + Field::Size => self.size = value.trim().parse().ok().or(self.size), + Field::Seeders => self.seeders = value.trim().parse().ok().or(self.seeders), } } } @@ -216,7 +233,6 @@ enum Field { fn parse_releases(indexer_id: i64, body: &[u8]) -> Result, SearchError> { let mut reader = Reader::from_reader(body); - reader.config_mut().trim_text(true); let mut root = None; let mut item = None; let mut field = None; @@ -241,7 +257,7 @@ fn parse_releases(indexer_id: i64, body: &[u8]) -> Result, Se item = Some(ReleaseBuilder::default()); field = None; } else if let Some(builder) = item.as_mut() { - read_element_attributes(&reader, builder, &element)?; + read_element_attributes(&reader, builder, &element); field = field_for(name); } } @@ -254,7 +270,7 @@ fn parse_releases(indexer_id: i64, body: &[u8]) -> Result, Se } } if let Some(builder) = item.as_mut() { - read_element_attributes(&reader, builder, &element)?; + read_element_attributes(&reader, builder, &element); } } Event::Text(text) => { @@ -296,11 +312,11 @@ fn read_element_attributes( reader: &Reader<&[u8]>, builder: &mut ReleaseBuilder, element: &BytesStart<'_>, -) -> Result<(), SearchError> { +) { match element.local_name().as_ref() { b"attr" => { - let name = attribute(reader, element, b"name")?; - let value = attribute(reader, element, b"value")?; + let name = attribute(reader, element, b"name"); + let value = attribute(reader, element, b"value"); match name.as_deref() { Some("size") => { builder.size = value.and_then(|value| value.parse().ok()).or(builder.size); @@ -314,34 +330,31 @@ fn read_element_attributes( } } b"enclosure" => { - if let Some(url) = attribute(reader, element, b"url")? { + if let Some(url) = attribute(reader, element, b"url") { builder.enclosure_url = Some(url); } if builder.size.is_none() { builder.size = - attribute(reader, element, b"length")?.and_then(|value| value.parse().ok()); + attribute(reader, element, b"length").and_then(|value| value.parse().ok()); } } _ => {} } - Ok(()) } -fn attribute( - reader: &Reader<&[u8]>, - element: &BytesStart<'_>, - name: &[u8], -) -> Result, SearchError> { +fn attribute(reader: &Reader<&[u8]>, element: &BytesStart<'_>, name: &[u8]) -> Option { for attribute in element.attributes() { - let attribute = attribute.map_err(|_| SearchError::InvalidResponse)?; + let Ok(attribute) = attribute else { + continue; + }; if attribute.key.local_name().as_ref() == name { return attribute .decode_and_unescape_value(reader.decoder()) - .map(|value| Some(value.into_owned())) - .map_err(|_| SearchError::InvalidResponse); + .ok() + .map(Cow::into_owned); } } - Ok(None) + None } fn field_for(name: LocalName<'_>) -> Option { @@ -357,14 +370,8 @@ fn field_for(name: LocalName<'_>) -> Option { } fn parse_torznab_error(reader: &Reader<&[u8]>, element: &BytesStart<'_>) -> SearchError { - let code = attribute(reader, element, b"code") - .ok() - .flatten() - .and_then(|code| code.parse().ok()); - let description = attribute(reader, element, b"description") - .ok() - .flatten() - .unwrap_or_default(); + let code = attribute(reader, element, b"code").and_then(|code| code.parse().ok()); + let description = attribute(reader, element, b"description").unwrap_or_default(); SearchError::Torznab { code, description } } @@ -375,6 +382,27 @@ fn append_text(target: &mut Option, value: &str) { target.get_or_insert_with(String::new).push_str(value); } +fn cleaned(value: Option) -> Option { + value + .map(|value| value.trim().to_owned()) + .filter(|value| !value.is_empty()) +} + +fn search_result( + indexer_id: i64, + result: Result, SearchError>, +) -> IndexerSearch { + let (releases, error) = match result { + Ok(releases) => (releases, None), + Err(error) => (Vec::new(), Some(error)), + }; + IndexerSearch { + indexer_id, + releases, + error, + } +} + fn parse_date(value: &str) -> Option { DateTime::parse_from_rfc2822(value) .or_else(|_| DateTime::parse_from_rfc3339(value)) @@ -384,7 +412,7 @@ fn parse_date(value: &str) -> Option { #[cfg(test)] mod tests { - use super::{SearchError, SearchRequest}; + use super::{parse_releases, SearchError, SearchRequest}; use crate::ProwlarrClient; use wiremock::{ matchers::{method, path, query_param, query_param_is_missing}, @@ -484,6 +512,15 @@ mod tests { assert_eq!(searches.len(), 4); assert_eq!(searches[0].releases[0].indexer_id, 3); + assert_eq!( + searches[0].releases[0].name, + "Example.Movie.2026.2160p.WEB-DL.DDP5.1.H.265-GROUP" + ); + assert_eq!(searches[0].releases[0].guid, "bhd-redacted-1"); + assert_eq!( + searches[0].releases[0].download_url, + "https://indexer.invalid/download/bhd-redacted-1" + ); assert_eq!(searches[0].releases[0].size, Some(23_622_320_128)); assert_eq!(searches[0].releases[0].seeders, Some(41)); assert!(searches[0].releases[0].publish_date.is_some()); @@ -491,17 +528,58 @@ mod tests { assert_eq!(searches[1].releases.len(), 2); assert_eq!(searches[1].releases[0].size, Some(4_294_967_296)); assert_eq!(searches[1].releases[0].seeders, Some(7)); + assert_eq!( + searches[1].releases[0].download_url, + "https://indexer.invalid/download/tl-redacted-1?token=redacted" + ); assert_eq!(searches[1].releases[1].size, None); assert_eq!(searches[1].releases[1].publish_date, None); assert_eq!(searches[2].releases[0].size, Some(9_126_805_504)); assert_eq!(searches[2].releases[0].seeders, Some(18)); + assert_eq!( + searches[2].releases[0].guid, + "https://indexer.invalid/details/ipt-redacted-1" + ); + assert_eq!( + searches[2].releases[0].download_url, + "https://indexer.invalid/download/ipt-redacted-1" + ); assert_eq!(searches[2].error, None); assert!(searches[3].releases.is_empty()); assert_eq!(searches[3].error, Some(SearchError::InvalidResponse)); } + #[test] + fn preserves_split_text_and_skips_undownloadable_items() { + let body = br#" + + + + Movie <![CDATA[& More]]> 2026 + split-title + https://indexer.invalid/details/split-title + + + + No download + no-download + + + + "#; + + let releases = parse_releases(3, body).expect("feed structure is valid"); + + assert_eq!(releases.len(), 1); + assert_eq!(releases[0].name, "Movie & More 2026"); + assert_eq!( + releases[0].download_url, + "https://indexer.invalid/download/split-title" + ); + } + async fn mount_request( server: &MockServer, id: i64, diff --git a/crates/arr-indexer/tests/fixtures/torrentleech-search.xml b/crates/arr-indexer/tests/fixtures/torrentleech-search.xml index 0334784..6e6c4b1 100644 --- a/crates/arr-indexer/tests/fixtures/torrentleech-search.xml +++ b/crates/arr-indexer/tests/fixtures/torrentleech-search.xml @@ -5,7 +5,7 @@ Example Show S02E03 1080p WEB H264-GROUP tl-redacted-1 - + 2026-08-22T11:15:00Z