diff --git a/Cargo.lock b/Cargo.lock index 4e13b2f..121aab5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -81,10 +81,25 @@ dependencies = [ [[package]] name = "arr-dl" version = "0.1.0" +dependencies = [ + "base64", + "reqwest", + "serde", + "serde_json", + "thiserror", + "tokio", + "url", + "wiremock", +] [[package]] name = "arr-e2e" version = "0.1.0" +dependencies = [ + "arr-dl", + "tokio", + "uuid", +] [[package]] name = "arr-indexer" @@ -2475,6 +2490,18 @@ dependencies = [ "utoipa", ] +[[package]] +name = "uuid" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f053576934f05a761a402421fbbe3d425d9366f75f978806a037b3ca481abecc" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "serde_core", + "wasm-bindgen", +] + [[package]] name = "valuable" version = "0.1.1" diff --git a/Cargo.toml b/Cargo.toml index a04b997..0ef8df5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,7 @@ axum = "0.8" tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time", "net", "io-util", "fs", "signal", "process"] } tower-http = { version = "0.6", features = ["trace"] } reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +base64 = "0.22" # Persistence sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "sqlite", "macros", "migrate", "chrono", "json"] } @@ -51,6 +52,7 @@ clap = { version = "4.5", features = ["derive", "env"] } thiserror = "2" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } +url = "2" uuid = { version = "1", features = ["v4", "serde"] } # Test-only diff --git a/crates/arr-dl/Cargo.toml b/crates/arr-dl/Cargo.toml index b3683fa..0a00a4c 100644 --- a/crates/arr-dl/Cargo.toml +++ b/crates/arr-dl/Cargo.toml @@ -7,6 +7,16 @@ repository.workspace = true publish = false [dependencies] +base64 = { workspace = true } +reqwest = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true } +url = { workspace = true } + +[dev-dependencies] +wiremock = { workspace = true } [lints] workspace = true diff --git a/crates/arr-dl/src/lib.rs b/crates/arr-dl/src/lib.rs index b31c26d..f373614 100644 --- a/crates/arr-dl/src/lib.rs +++ b/crates/arr-dl/src/lib.rs @@ -1 +1,460 @@ -//! arr-dl — see DESIGN.md. +//! Transmission RPC client for download lifecycle operations. + +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use base64::Engine as _; +use reqwest::{StatusCode, Url}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use tokio::sync::RwLock; + +const SESSION_HEADER: &str = "x-transmission-session-id"; +const MAX_SESSION_NEGOTIATIONS: usize = 4; +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + +/// A magnet URI or the bytes of a `.torrent` file. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum TorrentSource { + Magnet(String), + Metainfo(Vec), +} + +/// Settings applied atomically when a torrent is added. +#[derive(Clone, Debug, PartialEq)] +pub struct AddTorrent { + pub source: TorrentSource, + pub label: String, + pub download_dir: PathBuf, + pub seed_ratio_limit: f64, + /// Transmission expresses its idle limit in minutes. + pub seed_idle_limit_minutes: u64, +} + +/// The identity returned by Transmission for an added or existing torrent. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AddedTorrent { + pub id: i64, + pub name: String, + pub hash: String, + pub was_duplicate: bool, +} + +/// Transmission's torrent activity state. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TorrentState { + Stopped, + QueuedToVerify, + Verifying, + QueuedToDownload, + Downloading, + QueuedToSeed, + Seeding, + Unknown(u8), +} + +/// Current torrent state, read directly from Transmission. +#[derive(Clone, Debug, PartialEq)] +pub struct Torrent { + pub id: i64, + pub name: String, + pub hash: String, + pub state: TorrentState, + pub progress: f64, + pub download_dir: PathBuf, + pub labels: Vec, +} + +/// A Transmission RPC failure. +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("invalid Transmission RPC URL: {0}")] + InvalidUrl(#[from] url::ParseError), + #[error("download directory is not valid UTF-8")] + NonUtf8DownloadDir, + #[error("could not build Transmission HTTP client: {0}")] + BuildClient(reqwest::Error), + #[error("Transmission request failed: {0}")] + Http(#[from] reqwest::Error), + #[error("Transmission did not return a session ID after {0} attempts")] + SessionNegotiation(usize), + #[error("Transmission RPC failed: {0}")] + Rpc(String), + #[error("invalid Transmission response: {0}")] + InvalidResponse(String), +} + +/// Client for the unauthenticated Transmission JSON-RPC endpoint. +#[derive(Clone, Debug)] +pub struct TransmissionClient { + endpoint: Url, + http: reqwest::Client, + session_id: Arc>>, +} + +impl TransmissionClient { + /// Create a client for a full RPC URL, such as + /// `http://host:9091/transmission/rpc`. + /// + /// # Errors + /// + /// Returns an error when `endpoint` is not a valid URL. + pub fn new(endpoint: &str) -> Result { + Ok(Self { + endpoint: Url::parse(endpoint)?, + http: reqwest::Client::builder() + .connect_timeout(CONNECT_TIMEOUT) + .timeout(REQUEST_TIMEOUT) + .build() + .map_err(Error::BuildClient)?, + session_id: Arc::new(RwLock::new(None)), + }) + } + + /// Add a torrent with its label, destination, and tracker seed limits. + /// + /// # Errors + /// + /// Returns an error for transport failures or rejected/malformed RPC + /// responses. + pub async fn add_torrent(&self, request: AddTorrent) -> Result { + let download_dir = request + .download_dir + .to_str() + .ok_or(Error::NonUtf8DownloadDir)? + .to_owned(); + let label = request.label; + let seed_ratio_limit = request.seed_ratio_limit; + let seed_idle_limit = request.seed_idle_limit_minutes; + let mut arguments = json!({ + "download-dir": download_dir, + "labels": [label], + "seedRatioLimit": seed_ratio_limit, + "seedRatioMode": 1, + "seedIdleLimit": seed_idle_limit, + "seedIdleMode": 1, + }); + + match request.source { + TorrentSource::Magnet(uri) => arguments["filename"] = json!(uri), + TorrentSource::Metainfo(bytes) => { + arguments["metainfo"] = + json!(base64::engine::general_purpose::STANDARD.encode(bytes)); + } + } + + let arguments = self.call("torrent-add", arguments).await?; + let (torrent, was_duplicate) = if let Some(value) = arguments.get("torrent-added") { + (value.clone(), false) + } else if let Some(value) = arguments.get("torrent-duplicate") { + (value.clone(), true) + } else { + return Err(Error::InvalidResponse( + "torrent-add omitted torrent-added and torrent-duplicate".into(), + )); + }; + let torrent: RpcAddedTorrent = serde_json::from_value(torrent) + .map_err(|error| Error::InvalidResponse(error.to_string()))?; + + if was_duplicate { + self.call( + "torrent-set", + json!({ + "ids": [torrent.id], + "labels": [label], + "seedRatioLimit": seed_ratio_limit, + "seedRatioMode": 1, + "seedIdleLimit": seed_idle_limit, + "seedIdleMode": 1, + }), + ) + .await?; + self.call( + "torrent-set-location", + json!({"ids": [torrent.id], "location": download_dir, "move": true}), + ) + .await?; + } + + Ok(AddedTorrent { + id: torrent.id, + name: torrent.name, + hash: torrent.hash, + was_duplicate, + }) + } + + /// List torrents with their current state and completion progress. + /// + /// No torrent state is cached: every result is authoritative from + /// Transmission, including the first call after process startup. + /// + /// # Errors + /// + /// Returns an error for transport failures or rejected/malformed RPC + /// responses. + pub async fn list_torrents(&self) -> Result, Error> { + let arguments = self + .call( + "torrent-get", + json!({ + "fields": [ + "id", "name", "hashString", "status", "percentDone", + "downloadDir", "labels" + ] + }), + ) + .await?; + let response: RpcTorrentList = serde_json::from_value(arguments) + .map_err(|error| Error::InvalidResponse(error.to_string()))?; + + Ok(response.torrents.into_iter().map(Torrent::from).collect()) + } + + /// Remove one torrent, optionally deleting its downloaded data. + /// + /// # Errors + /// + /// Returns an error for transport failures or rejected RPC responses. + pub async fn remove_torrent(&self, id: i64, delete_data: bool) -> Result<(), Error> { + self.call( + "torrent-remove", + json!({"ids": [id], "delete-local-data": delete_data}), + ) + .await?; + Ok(()) + } + + async fn call(&self, method: &str, arguments: Value) -> Result { + let body = RpcRequest { method, arguments }; + + for _ in 0..MAX_SESSION_NEGOTIATIONS { + let session_id = self.session_id.read().await.clone(); + let mut request = self.http.post(self.endpoint.clone()).json(&body); + if let Some(session_id) = session_id { + request = request.header(SESSION_HEADER, session_id); + } + + let response = request.send().await?; + if response.status() == StatusCode::CONFLICT { + let session_id = response + .headers() + .get(SESSION_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| { + Error::InvalidResponse("409 response omitted session ID".into()) + })?; + *self.session_id.write().await = Some(session_id.to_owned()); + continue; + } + + let response = response.error_for_status()?.json::().await?; + if response.result != "success" { + return Err(Error::Rpc(response.result)); + } + return Ok(response.arguments); + } + + Err(Error::SessionNegotiation(MAX_SESSION_NEGOTIATIONS)) + } +} + +#[derive(Debug, Serialize)] +struct RpcRequest<'a> { + method: &'a str, + arguments: Value, +} + +#[derive(Debug, Deserialize)] +struct RpcResponse { + result: String, + arguments: Value, +} + +#[derive(Debug, Deserialize)] +struct RpcAddedTorrent { + id: i64, + name: String, + #[serde(rename = "hashString")] + hash: String, +} + +#[derive(Debug, Deserialize)] +struct RpcTorrentList { + torrents: Vec, +} + +#[derive(Debug, Deserialize)] +struct RpcTorrent { + id: i64, + name: String, + #[serde(rename = "hashString")] + hash: String, + status: u8, + #[serde(rename = "percentDone")] + progress: f64, + #[serde(rename = "downloadDir")] + download_dir: PathBuf, + #[serde(default)] + labels: Vec, +} + +impl From for Torrent { + fn from(value: RpcTorrent) -> Self { + let state = match value.status { + 0 => TorrentState::Stopped, + 1 => TorrentState::QueuedToVerify, + 2 => TorrentState::Verifying, + 3 => TorrentState::QueuedToDownload, + 4 => TorrentState::Downloading, + 5 => TorrentState::QueuedToSeed, + 6 => TorrentState::Seeding, + status => TorrentState::Unknown(status), + }; + + Self { + id: value.id, + name: value.name, + hash: value.hash, + state, + progress: value.progress, + download_dir: value.download_dir, + labels: value.labels, + } + } +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use serde_json::json; + use wiremock::matchers::{body_partial_json, method}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::{AddTorrent, Error, TorrentSource, TorrentState, TransmissionClient}; + + #[tokio::test] + async fn retries_the_session_handshake() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with( + ResponseTemplate::new(409).insert_header("x-transmission-session-id", "session"), + ) + .with_priority(1) + .up_to_n_times(1) + .expect(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": []} + }))) + .with_priority(2) + .expect(1) + .mount(&server) + .await; + + let client = TransmissionClient::new(&server.uri()).expect("client"); + assert!(client.list_torrents().await.expect("list").is_empty()); + } + + #[tokio::test] + async fn duplicate_reapplies_label_location_and_seed_limits() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(body_partial_json(json!({"method": "torrent-add"}))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", + "arguments": {"torrent-duplicate": { + "id": 7, "name": "duplicate", "hashString": "abc" + }} + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(body_partial_json(json!({ + "method": "torrent-set", + "arguments": { + "ids": [7], "labels": ["movies-main"], + "seedRatioLimit": 1.5, "seedIdleLimit": 60 + } + }))) + .respond_with(success()) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(body_partial_json(json!({ + "method": "torrent-set-location", + "arguments": {"ids": [7], "location": "/downloads", "move": true} + }))) + .respond_with(success()) + .expect(1) + .mount(&server) + .await; + + let client = TransmissionClient::new(&server.uri()).expect("client"); + let added = client + .add_torrent(AddTorrent { + source: TorrentSource::Magnet("magnet:?xt=urn:btih:abc".into()), + label: "movies-main".into(), + download_dir: PathBuf::from("/downloads"), + seed_ratio_limit: 1.5, + seed_idle_limit_minutes: 60, + }) + .await + .expect("duplicate"); + + assert!(added.was_duplicate); + } + + #[tokio::test] + #[cfg(unix)] + async fn rejects_non_utf8_download_directory() { + use std::os::unix::ffi::OsStringExt as _; + + let client = TransmissionClient::new("http://127.0.0.1:1").expect("client"); + let error = client + .add_torrent(AddTorrent { + source: TorrentSource::Magnet("magnet:?xt=urn:btih:abc".into()), + label: "movies-main".into(), + download_dir: PathBuf::from(std::ffi::OsString::from_vec(vec![0xff])), + seed_ratio_limit: 1.5, + seed_idle_limit_minutes: 60, + }) + .await + .expect_err("invalid path"); + + assert!(matches!(error, Error::NonUtf8DownloadDir)); + } + + #[tokio::test] + async fn preserves_unknown_status_without_hiding_the_list() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", + "arguments": {"torrents": [{ + "id": 1, "name": "future", "hashString": "abc", "status": 99, + "percentDone": 0.5, "downloadDir": "/downloads", "labels": [] + }]} + }))) + .expect(1) + .mount(&server) + .await; + + let client = TransmissionClient::new(&server.uri()).expect("client"); + let torrents = client.list_torrents().await.expect("list"); + assert_eq!(torrents[0].state, TorrentState::Unknown(99)); + } + + fn success() -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(json!({"result": "success", "arguments": {}})) + } +} diff --git a/crates/arr-e2e/Cargo.toml b/crates/arr-e2e/Cargo.toml index 7d5b576..3f2fe6d 100644 --- a/crates/arr-e2e/Cargo.toml +++ b/crates/arr-e2e/Cargo.toml @@ -8,5 +8,10 @@ publish = false [dependencies] +[dev-dependencies] +arr-dl = { workspace = true } +tokio = { workspace = true } +uuid = { workspace = true } + [lints] workspace = true diff --git a/crates/arr-e2e/src/lib.rs b/crates/arr-e2e/src/lib.rs index a9f88f7..f509fc0 100644 --- a/crates/arr-e2e/src/lib.rs +++ b/crates/arr-e2e/src/lib.rs @@ -5,9 +5,61 @@ #[cfg(test)] mod tests { - /// Placeholder, same reason as the one in `arr-core`: `cargo nextest` - /// fails a zero-test run, and `--no-tests=pass` would let a broken filter - /// go green later. Replaced by the real harness in the e2e issue. - #[test] - fn e2e_crate_builds() {} + use std::path::PathBuf; + + use arr_dl::{AddTorrent, TorrentSource, TransmissionClient}; + + #[tokio::test] + async fn transmission_add_list_and_remove() { + let endpoint = std::env::var("TRANSMISSION_RPC_URL") + .unwrap_or_else(|_| "http://127.0.0.1:9091/transmission/rpc".into()); + let client = TransmissionClient::new(&endpoint).expect("valid endpoint"); + let name = format!("arr-e2e-{}", uuid::Uuid::new_v4()); + let metainfo = torrent_with_name(&name); + let download_dir = PathBuf::from("/tmp/arr-e2e"); + let request = || AddTorrent { + source: TorrentSource::Metainfo(metainfo.clone()), + label: "movies-main".into(), + download_dir: download_dir.clone(), + seed_ratio_limit: 1.5, + seed_idle_limit_minutes: 60, + }; + + let first = client.add_torrent(request()).await.expect("add torrent"); + assert!(!first.was_duplicate); + + let torrents = client.list_torrents().await.expect("list torrents"); + let listed = torrents + .iter() + .find(|torrent| torrent.id == first.id) + .expect("added torrent is authoritative in list"); + assert_eq!(listed.name, name); + assert_eq!(listed.hash, first.hash); + assert_eq!(listed.download_dir, download_dir); + assert_eq!(listed.labels, ["movies-main"]); + assert!((0.0..=1.0).contains(&listed.progress)); + + client + .remove_torrent(first.id, false) + .await + .expect("remove without data"); + + let second = client.add_torrent(request()).await.expect("add again"); + client + .remove_torrent(second.id, true) + .await + .expect("remove with data"); + + let torrents = client.list_torrents().await.expect("list after remove"); + assert!(torrents.iter().all(|torrent| torrent.id != second.id)); + } + + fn torrent_with_name(name: &str) -> Vec { + let piece_hash = [0_u8; 20]; + let mut bytes = format!("d4:infod6:lengthi1e4:name{}:{name}", name.len()).into_bytes(); + bytes.extend_from_slice(b"12:piece lengthi16384e6:pieces20:"); + bytes.extend_from_slice(&piece_hash); + bytes.extend_from_slice(b"ee"); + bytes + } }