feat(arr): expose live downloads

Cargo.lock

@@ -38,6 +38,7 @@ version = "0.1.0"
 dependencies = [
  "arr-core",
  "arr-db",
+ "arr-dl",
  "arr-indexer",
  "arr-meta",
  "arr-parse",

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 }

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<i64>,
+    pub error: Option<String>,
+}
+
+#[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<AppState>) -> Result<Json<Vec<Download>>, 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<Grab> =
+        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);
+    }
+}

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<AppState> {
     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))

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<RwLock<Option<(Instant, Vec<arr_dl::Torrent>)>>>;
+
 /// 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<JellyfinClient>,
     syncer: Syncer,
+    transmission: Option<TransmissionClient>,
+    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,

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<AppState, Error> {
     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> {
         &notifier,
         &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)

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<i64>,
+    /// Transmission's error message, when the torrent has errored.
+    pub error: Option<String>,
     pub download_dir: PathBuf,
     pub labels: Vec<String>,
     /// 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<i64>,
+    #[serde(rename = "errorString", default)]
+    error: String,
     #[serde(rename = "downloadDir")]
     download_dir: PathBuf,
     #[serde(default)]
@@ -411,6 +425,9 @@ impl From<RpcTorrent> 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,