arr/crates/arr-daemon/src/reaper.rs

//! Removes arr torrents only after Transmission says their seeding obligation
//! is complete. The library/import state is deliberately not consulted (§7.3).

use std::collections::HashSet;

use arr_db::Db;
use arr_dl::TransmissionClient;

use crate::grab::label_for_root;
use crate::reconcile::{Action, ActionFuture, Outcome};

#[derive(Debug)]
pub struct ReaperAction {
    transmission: TransmissionClient,
}

impl ReaperAction {
    #[must_use]
    pub fn new(transmission: TransmissionClient) -> Self {
        Self { transmission }
    }

    async fn tick(&self, labels: &HashSet<String>) -> Result<Vec<Outcome>, arr_dl::Error> {
        let torrents = self.transmission.list_torrents().await?;
        let mut outcomes = Vec::new();
        for torrent in torrents {
            if !torrent.is_finished || !torrent.labels.iter().any(|label| labels.contains(label)) {
                continue;
            }
            self.transmission.remove_torrent(torrent.id, true).await?;
            outcomes.push(Outcome::new(
                format!("torrent {} finished seeding", torrent.hash),
                format!("removed torrent {} and its download data", torrent.id),
            ));
        }
        Ok(outcomes)
    }
}

impl Action for ReaperAction {
    fn name(&self) -> &'static str {
        "torrent-reaper"
    }

    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
        Box::pin(async move {
            let labels = sqlx::query_as::<_, (String, String)>("SELECT kind, audience FROM roots")
                .fetch_all(database.pool())
                .await
                .map_err(|error| Box::new(error) as crate::reconcile::ActionError)?
                .into_iter()
                .map(|(kind, audience)| label_for_root(&kind, &audience))
                .collect::<HashSet<_>>();
            self.tick(&labels)
                .await
                .map_err(|error| Box::new(error) as crate::reconcile::ActionError)
        })
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use std::sync::{Arc, Mutex};

    use serde_json::{json, Value};
    use wiremock::matchers::any;
    use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};

    use super::*;

    #[derive(Clone)]
    struct Transmission {
        torrents: Value,
        removed: Arc<Mutex<Vec<Value>>>,
    }

    impl Respond for Transmission {
        fn respond(&self, request: &Request) -> ResponseTemplate {
            let body: Value = serde_json::from_slice(&request.body).unwrap();
            let arguments = match body["method"].as_str().unwrap() {
                "torrent-get" => json!({"torrents": self.torrents}),
                "torrent-remove" => {
                    self.removed.lock().unwrap().push(body["arguments"].clone());
                    json!({})
                }
                method => panic!("unexpected method {method}"),
            };
            ResponseTemplate::new(200)
                .insert_header("x-transmission-session-id", "session")
                .set_body_json(json!({"result": "success", "arguments": arguments}))
        }
    }

    fn torrent(id: i64, label: &str, finished: bool) -> Value {
        json!({
            "id": id, "name": format!("torrent-{id}"), "hashString": format!("hash-{id}"),
            "status": if finished { 0 } else { 6 }, "percentDone": 1.0,
            "downloadDir": "/downloads", "labels": [label], "isFinished": finished
        })
    }

    #[tokio::test]
    async fn removes_only_finished_arr_torrents_with_data() {
        let server = MockServer::start().await;
        let removed = Arc::new(Mutex::new(Vec::new()));
        Mock::given(any())
            .respond_with(Transmission {
                torrents: json!([
                    torrent(1, "movies-main", false),
                    torrent(2, "movies-main", true),
                    torrent(3, "radarr", true)
                ]),
                removed: Arc::clone(&removed),
            })
            .mount(&server)
            .await;
        let action = ReaperAction::new(TransmissionClient::new(&server.uri()).unwrap());

        let labels = HashSet::from(["movies-main".to_owned(), "movies-kids".to_owned()]);
        let outcomes = action.tick(&labels).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        assert_eq!(
            *removed.lock().unwrap(),
            [json!({"ids": [2], "delete-local-data": true})]
        );
    }

    #[tokio::test]
    async fn finished_torrent_does_not_need_a_grab_or_import_row() {
        let server = MockServer::start().await;
        let removed = Arc::new(Mutex::new(Vec::new()));
        Mock::given(any())
            .respond_with(Transmission {
                torrents: json!([torrent(9, "movies-kids", true)]),
                removed: Arc::clone(&removed),
            })
            .mount(&server)
            .await;
        let action = ReaperAction::new(TransmissionClient::new(&server.uri()).unwrap());

        let labels = HashSet::from(["movies-kids".to_owned()]);
        action.tick(&labels).await.unwrap();

        assert_eq!(removed.lock().unwrap().len(), 1);
    }
}