Transmission RPC client (#52)

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"

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

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

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<u8>),
+}
+
+/// 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<String>,
+}
+
+/// 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<RwLock<Option<String>>>,
+}
+
+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<Self, Error> {
+        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<AddedTorrent, Error> {
+        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<Vec<Torrent>, 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<Value, Error> {
+        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::<RpcResponse>().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<RpcTorrent>,
+}
+
+#[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<String>,
+}
+
+impl From<RpcTorrent> 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": {}}))
+    }
+}

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

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`
+    use std::path::PathBuf;
-    /// 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.
+    use arr_dl::{AddTorrent, TorrentSource, TransmissionClient};
-    #[test]
+
-    fn e2e_crate_builds() {}
+    #[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<u8> {
+        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
+    }
 }