diff --git a/Cargo.lock b/Cargo.lock index 9ac95d8..db8e502 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -38,6 +38,7 @@ version = "0.1.0" dependencies = [ "arr-core", "arr-db", + "arr-dl", "arr-indexer", "arr-meta", "arr-parse", diff --git a/crates/arr-api/Cargo.toml b/crates/arr-api/Cargo.toml index 9608ff3..346a6bf 100644 --- a/crates/arr-api/Cargo.toml +++ b/crates/arr-api/Cargo.toml @@ -8,6 +8,7 @@ publish = false [dependencies] arr-core = { workspace = true } +arr-dl = { workspace = true } arr-db = { workspace = true } arr-indexer = { workspace = true } arr-meta = { workspace = true } diff --git a/crates/arr-api/src/downloads.rs b/crates/arr-api/src/downloads.rs new file mode 100644 index 0000000..b69a202 --- /dev/null +++ b/crates/arr-api/src/downloads.rs @@ -0,0 +1,241 @@ +//! The live download snapshot, joined to arr-owned grabs only. + +use std::time::{Duration, Instant}; + +use arr_dl::{Torrent, TorrentState}; +use axum::extract::State; +use axum::Json; +use serde::Serialize; +use utoipa::ToSchema; + +use crate::movies::ApiError; +use crate::state::AppState; + +const SNAPSHOT_TTL: Duration = Duration::from_secs(15); + +#[derive(Debug, Clone, Copy, Serialize, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum DownloadPhase { + Downloading, + Stalled, + Errored, + Seeding, + Queued, +} + +#[derive(Debug, Clone, Serialize, ToSchema)] +pub struct Download { + pub target_kind: String, + pub target_id: i64, + pub infohash: String, + pub name: String, + pub phase: DownloadPhase, + pub progress: f64, + pub download_rate: u64, + pub eta: Option, + pub error: Option, +} + +#[utoipa::path( + get, + path = "/api/downloads", + tag = "system", + responses( + (status = 200, description = "Live downloads started by arr", body = [Download]), + (status = 503, description = "Transmission or database unavailable", body = crate::movies::ErrorBody) + ) +)] +pub(crate) async fn list(State(state): State) -> Result>, ApiError> { + let database = state.database().ok_or(ApiError::Unavailable)?; + let transmission = state + .transmission() + .ok_or(ApiError::Upstream("transmission"))?; + + let torrents = { + let mut cache = state.download_snapshot().write().await; + if let Some((at, torrents)) = cache.as_ref() { + if at.elapsed() < SNAPSHOT_TTL { + torrents.clone() + } else { + let torrents = transmission + .list_torrents() + .await + .map_err(|_| ApiError::Upstream("transmission"))?; + *cache = Some((Instant::now(), torrents.clone())); + torrents + } + } else { + let torrents = transmission + .list_torrents() + .await + .map_err(|_| ApiError::Upstream("transmission"))?; + *cache = Some((Instant::now(), torrents.clone())); + torrents + } + }; + + let grabs: Vec = + sqlx::query_as("SELECT target_kind, target_id, infohash FROM grabs ORDER BY id") + .fetch_all(database.pool()) + .await?; + + let mut downloads = Vec::new(); + for grab in grabs { + let Some(torrent) = torrents + .iter() + .find(|torrent| torrent.hash.eq_ignore_ascii_case(&grab.infohash)) + else { + continue; + }; + downloads.push(Download { + target_kind: grab.target_kind, + target_id: grab.target_id, + infohash: torrent.hash.clone(), + name: torrent.name.clone(), + phase: phase(torrent), + progress: torrent.progress, + download_rate: torrent.download_rate, + eta: torrent.eta, + error: torrent.error.clone(), + }); + } + + Ok(Json(downloads)) +} + +#[derive(Debug, sqlx::FromRow)] +struct Grab { + target_kind: String, + target_id: i64, + infohash: String, +} + +fn phase(torrent: &Torrent) -> DownloadPhase { + if torrent.error.is_some() { + DownloadPhase::Errored + } else { + match torrent.state { + TorrentState::Seeding => DownloadPhase::Seeding, + TorrentState::Downloading if torrent.download_rate == 0 => DownloadPhase::Stalled, + TorrentState::Downloading => DownloadPhase::Downloading, + _ => DownloadPhase::Queued, + } + } +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use arr_db::Db; + use axum::extract::State; + use serde_json::json; + use wiremock::matchers::{body_partial_json, method}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::{list, phase, DownloadPhase}; + use crate::state::{AppState, Upstreams}; + + #[test] + fn torrent_states_become_the_download_phases() { + let torrent = + |state: arr_dl::TorrentState, rate: u64, error: Option<&str>| arr_dl::Torrent { + id: 1, + name: "name".into(), + hash: "hash".into(), + state, + progress: 0.5, + download_rate: rate, + eta: None, + error: error.map(str::to_owned), + download_dir: PathBuf::from("/downloads"), + labels: Vec::new(), + is_finished: false, + }; + + assert_eq!( + phase(&torrent(arr_dl::TorrentState::Downloading, 10, None)), + DownloadPhase::Downloading + ); + assert_eq!( + phase(&torrent(arr_dl::TorrentState::Downloading, 0, None)), + DownloadPhase::Stalled + ); + assert_eq!( + phase(&torrent(arr_dl::TorrentState::Seeding, 0, None)), + DownloadPhase::Seeding + ); + assert_eq!( + phase(&torrent( + arr_dl::TorrentState::Downloading, + 10, + Some("disk full") + )), + DownloadPhase::Errored + ); + assert_eq!( + phase(&torrent(arr_dl::TorrentState::QueuedToDownload, 0, None)), + DownloadPhase::Queued + ); + } + + #[tokio::test] + async fn endpoint_joins_only_arr_grabs_and_reuses_the_snapshot() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with( + ResponseTemplate::new(409).insert_header("x-transmission-session-id", "session"), + ) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(body_partial_json(json!({"method": "torrent-get"}))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", + "arguments": {"torrents": [{ + "id": 1, "name": "grabbed", "hashString": "ABC", + "status": 4, "percentDone": 0.4, "rateDownload": 123, + "eta": 60, "errorString": "", "downloadDir": "/downloads", + "labels": [], "isFinished": false + }]} + }))) + .expect(1) + .mount(&server) + .await; + + let directory = tempfile::tempdir().expect("tempdir"); + let database = Db::connect(directory.path().join("arr.db")) + .await + .expect("database"); + database.migrate().await.expect("migrate"); + sqlx::query( + "INSERT INTO releases (id, indexer_id, guid, name, size, download_url, parsed) + VALUES (1, 1, 'guid', 'release', 1, 'magnet:?x', '{}')", + ) + .execute(database.pool()) + .await + .expect("release"); + sqlx::query( + "INSERT INTO grabs (release_id, target_kind, target_id, infohash) + VALUES (1, 'movie', 42, 'abc')", + ) + .execute(database.pool()) + .await + .expect("grab"); + + let state = AppState::new(Upstreams::new("unused".into(), server.uri())) + .expect("state") + .with_database(database) + .with_transmission(arr_dl::TransmissionClient::new(&server.uri()).expect("client")); + let first = list(State(state.clone())).await.expect("first").0; + let second = list(State(state)).await.expect("second").0; + assert_eq!(first.len(), 1); + assert_eq!(first[0].target_kind, "movie"); + assert_eq!(first[0].target_id, 42); + assert_eq!(first[0].infohash, "ABC"); + assert_eq!(first[0].phase, DownloadPhase::Downloading); + assert_eq!(first[0].download_rate, 123); + assert_eq!(second.len(), 1); + } +} diff --git a/crates/arr-api/src/lib.rs b/crates/arr-api/src/lib.rs index 9924a15..977498a 100644 --- a/crates/arr-api/src/lib.rs +++ b/crates/arr-api/src/lib.rs @@ -6,6 +6,7 @@ //! carrying a `#[utoipa::path]` annotation. A handler added without one fails //! to compile, and the gate in DESIGN.md §12 fails with it. +mod downloads; mod health; pub mod jellyfin; mod metadata; @@ -31,6 +32,7 @@ use utoipa_axum::router::OpenApiRouter; use utoipa_axum::routes; use utoipa_scalar::{Scalar, Servable}; +pub use downloads::{Download, DownloadPhase}; pub use health::{Check, Health, HealthReport, Status}; pub use metadata::{MetadataTrailer, MovieMetadata, SeriesMetadata}; pub use movies::{ @@ -92,6 +94,7 @@ struct ApiDoc; fn api_router() -> OpenApiRouter { OpenApiRouter::with_openapi(ApiDoc::openapi()) .routes(routes!(health::health)) + .routes(routes!(downloads::list)) .routes(routes!(movies::list, movies::create)) .routes(routes!(movies::get, movies::update, movies::delete)) .routes(routes!(movies::search)) diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index 5b2fb53..e93c265 100644 --- a/crates/arr-api/src/state.rs +++ b/crates/arr-api/src/state.rs @@ -3,12 +3,14 @@ use std::ffi::{OsStr, OsString}; use std::sync::{atomic::AtomicU64, Arc}; -use std::time::Duration; +use std::time::{Duration, Instant}; use arr_db::Db; +use arr_dl::TransmissionClient; use arr_probe::Extractor; use arr_subs::{Backend, OpenAiEndpoint, Provider, Syncer}; use tokio::sync::mpsc; +use tokio::sync::RwLock; use crate::jellyfin::JellyfinClient; @@ -20,6 +22,8 @@ pub const DEFAULT_TMDB_URL: &str = "https://api.themoviedb.org/3"; /// `arr-probe`'s extractor, which is where it is actually run. pub const DEFAULT_FFMPEG_BINARY: &str = "ffmpeg"; +type DownloadSnapshot = Arc)>>>; + /// How long an upstream has to answer a probe before it counts as /// unreachable. Health is polled by a human waiting on a page. const PROBE_TIMEOUT: Duration = Duration::from_secs(3); @@ -105,6 +109,8 @@ pub struct AppState { extractor: Extractor, jellyfin: Option, syncer: Syncer, + transmission: Option, + download_snapshot: DownloadSnapshot, } /// Work explicitly requested through the movie API. @@ -183,6 +189,8 @@ impl AppState { extractor: Extractor::default(), jellyfin: None, syncer: Syncer::default(), + transmission: None, + download_snapshot: Arc::new(RwLock::new(None)), }) } @@ -193,6 +201,14 @@ impl AppState { self } + /// Attach the shared Transmission client used by the API's live download + /// snapshot endpoint. + #[must_use] + pub fn with_transmission(mut self, transmission: TransmissionClient) -> Self { + self.transmission = Some(transmission); + self + } + /// Attach the subtitle providers this deployment has credentials for /// (`DESIGN.md` §15). /// @@ -352,6 +368,14 @@ impl AppState { self.database.as_ref() } + pub(crate) fn transmission(&self) -> Option<&TransmissionClient> { + self.transmission.as_ref() + } + + pub(crate) fn download_snapshot(&self) -> &DownloadSnapshot { + &self.download_snapshot + } + pub(crate) fn send_movie_command( &self, command: MovieCommand, diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 7219482..55fcc43 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -118,6 +118,7 @@ fn api_state( database: &Db, jellyfin: arr_api::jellyfin::JellyfinClient, translators: &Translators, + transmission: &arr_dl::TransmissionClient, ) -> Result { let mut upstreams = Upstreams::new(config.prowlarr_url.clone(), config.transmission_url.clone()) @@ -129,6 +130,7 @@ fn api_state( let mut state = AppState::new(upstreams)? .with_database(database.clone()) + .with_transmission(transmission.clone()) .with_subtitle_providers(subtitle_providers( config.opensubtitles_api_key.clone(), config.opensubtitles_username.clone(), @@ -151,6 +153,7 @@ fn api_state( Ok(state) } +#[allow(clippy::too_many_lines)] async fn run() -> Result<(), Error> { let config = Config::load()?; let database = Db::connect(&config.database_path).await?; @@ -181,9 +184,6 @@ async fn run() -> Result<(), Error> { ¬ifier, &translators, )?; - // Issue #176: the on-demand half of the metadata lane needs its own - // handle — the sweep's `SeriesRefreshAction` is owned by `ReconcileLoop`, - // and the compat shim takes the other clone below. let metadata_tmdb = tmdb.clone(); // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and @@ -193,8 +193,13 @@ async fn run() -> Result<(), Error> { compat = compat.with_tmdb(tmdb); } - let state = api_state(&config, &database, api_jellyfin, &translators)?; - + let state = api_state( + &config, + &database, + api_jellyfin, + &translators, + &transmission, + )?; let app = arr_api::router(state.clone()) .merge(arr_compat::router(compat)) .fallback(web::serve) diff --git a/crates/arr-dl/src/lib.rs b/crates/arr-dl/src/lib.rs index 7a8827b..73c306d 100644 --- a/crates/arr-dl/src/lib.rs +++ b/crates/arr-dl/src/lib.rs @@ -68,6 +68,13 @@ pub struct Torrent { pub hash: String, pub state: TorrentState, pub progress: f64, + /// Download rate in bytes per second. + pub download_rate: u64, + /// Estimated seconds until complete, or `None` when Transmission cannot + /// estimate it. + pub eta: Option, + /// Transmission's error message, when the torrent has errored. + pub error: Option, pub download_dir: PathBuf, pub labels: Vec, /// Transmission has stopped this torrent because its configured seeding @@ -230,7 +237,8 @@ impl TransmissionClient { json!({ "fields": [ "id", "name", "hashString", "status", "percentDone", - "downloadDir", "labels", "isFinished" + "rateDownload", "eta", "errorString", "downloadDir", + "labels", "isFinished" ] }), ) @@ -384,6 +392,12 @@ struct RpcTorrent { status: u8, #[serde(rename = "percentDone")] progress: f64, + #[serde(rename = "rateDownload", default)] + download_rate: u64, + #[serde(default)] + eta: Option, + #[serde(rename = "errorString", default)] + error: String, #[serde(rename = "downloadDir")] download_dir: PathBuf, #[serde(default)] @@ -411,6 +425,9 @@ impl From for Torrent { hash: value.hash, state, progress: value.progress, + download_rate: value.download_rate, + eta: value.eta.filter(|eta| *eta >= 0), + error: (!value.error.is_empty()).then_some(value.error), download_dir: value.download_dir, labels: value.labels, is_finished: value.is_finished,