Move torrents to completed after finish

crates/torad/src/db.rs

@@ -32,7 +32,9 @@ pub struct TorrentRow {
     pub info_hash: String,
     pub name: Option<String>,
     pub source: String,
-    pub output_path: String,
+    pub output_path: Option<String>,
+    pub download_folder: Option<String>,
+    pub move_after_completion_folder: Option<String>,
     pub total_bytes: Option<i64>,
     pub downloaded_bytes: i64,
     pub state: TorrentState,
@@ -43,17 +45,20 @@ pub async fn insert_pending(
     pool: &PgPool,
     info_hash: &str,
     source: &str,
-    output_path: &str,
+    download_folder: Option<&str>,
+    move_after_completion_folder: Option<&str>,
 ) -> Result<TorrentRow> {
     let row = sqlx::query_as::<_, TorrentRow>(
-        "INSERT INTO torrents (info_hash, source, output_path)
-         VALUES ($1, $2, $3)
+        "INSERT INTO torrents (info_hash, source, download_folder, move_after_completion_folder)
+         VALUES ($1, $2, $3, $4)
          ON CONFLICT (info_hash) DO UPDATE SET updated_at = now()
-         RETURNING id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message",
+         RETURNING id, info_hash, name, source, output_path, download_folder, \
+                   move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message",
     )
     .bind(info_hash)
     .bind(source)
-    .bind(output_path)
+    .bind(download_folder)
+    .bind(move_after_completion_folder)
     .fetch_one(pool)
     .await?;
     Ok(row)
@@ -61,7 +66,8 @@ pub async fn insert_pending(
 
 pub async fn get(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
     let row = sqlx::query_as::<_, TorrentRow>(
-        "SELECT id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message
+        "SELECT id, info_hash, name, source, output_path, download_folder, \
+                move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message
          FROM torrents WHERE id = $1",
     )
     .bind(id)
@@ -72,7 +78,8 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
 
 pub async fn list(pool: &PgPool) -> Result<Vec<TorrentRow>> {
     let rows = sqlx::query_as::<_, TorrentRow>(
-        "SELECT id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message
+        "SELECT id, info_hash, name, source, output_path, download_folder, \
+                move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message
          FROM torrents ORDER BY added_at DESC",
     )
     .fetch_all(pool)
@@ -83,7 +90,8 @@ pub async fn list(pool: &PgPool) -> Result<Vec<TorrentRow>> {
 pub async fn delete(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
     let row = sqlx::query_as::<_, TorrentRow>(
         "DELETE FROM torrents WHERE id = $1
-         RETURNING id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message",
+         RETURNING id, info_hash, name, source, output_path, download_folder, \
+                   move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message",
     )
     .bind(id)
     .fetch_optional(pool)
@@ -93,7 +101,8 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
 
 pub async fn list_pending(pool: &PgPool) -> Result<Vec<TorrentRow>> {
     let rows = sqlx::query_as::<_, TorrentRow>(
-        "SELECT id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message
+        "SELECT id, info_hash, name, source, output_path, download_folder, \
+                move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message
          FROM torrents WHERE state = 'pending' ORDER BY added_at",
     )
     .fetch_all(pool)
@@ -119,6 +128,18 @@ pub async fn set_state(pool: &PgPool, id: Uuid, state: TorrentState) -> Result<(
     Ok(())
 }
 
+/// Used by the poller to record the actual output path once librqbit
+/// resolves metadata and creates the directory (or after a post-completion
+/// move). Leaves all other columns untouched.
+pub async fn update_output_path(pool: &PgPool, id: Uuid, output_path: &str) -> Result<()> {
+    sqlx::query("UPDATE torrents SET output_path = $2, updated_at = now() WHERE id = $1")
+        .bind(id)
+        .bind(output_path)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
 pub struct Progress<'a> {
     pub name: Option<&'a str>,
     pub total_bytes: i64,
@@ -142,7 +163,8 @@ pub async fn update_progress(
              updated_at = now(),
              completed_at = CASE WHEN $5 = 'finished' THEN now() ELSE completed_at END
          WHERE id = $1
-         RETURNING id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message",
+         RETURNING id, info_hash, name, source, output_path, download_folder, \
+                   move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message",
     )
     .bind(id)
     .bind(progress.name)

crates/torad/src/torrents.rs

@@ -1,5 +1,5 @@
 use std::collections::{HashMap, HashSet};
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
@@ -11,10 +11,10 @@ use sqlx::PgPool;
 use tokio::sync::{Mutex, broadcast};
 use tonic::{Request, Response, Status};
 use tora_proto::{
-    AddRequest, AddResponse, AddedInfo, ListRequest, ListResponse, Notification, NotificationKind,
-    NotificationsRequest, PauseRequest, PauseResponse, ProgressMark, RemoveRequest, RemoveResponse,
-    ResumeRequest, ResumeResponse, State as ProtoState, StateChanged, StatusRequest, TorrentStatus,
-    Torrents,
+    AddRequest, AddResponse, AddedInfo, FileState, FilesRequest, FilesResponse, ListRequest,
+    ListResponse, Notification, NotificationKind, NotificationsRequest, PauseRequest,
+    PauseResponse, ProgressMark, RemoveRequest, RemoveResponse, ResumeRequest, ResumeResponse,
+    State as ProtoState, StateChanged, StatusRequest, TorrentFile, TorrentStatus, Torrents,
 };
 use tracing::{error, info, warn};
 use uuid::Uuid;
@@ -45,7 +45,7 @@ pub struct TorrentManager {
     session: Arc<Session>,
     default_output_dir: PathBuf,
     source: SourceResolver,
-    tracked: Mutex<HashMap<Uuid, Arc<ManagedTorrent>>>,
+    pub(crate) tracked: Mutex<HashMap<Uuid, Arc<ManagedTorrent>>>,
     adding: Mutex<HashSet<Uuid>>,
     events: broadcast::Sender<Notification>,
     poll_cache: Mutex<HashMap<Uuid, PollCache>>,
@@ -87,14 +87,28 @@ impl TorrentManager {
         let _ = self.events.send(notification);
     }
 
-    pub async fn add(&self, source: &str, output_dir: Option<&str>) -> Result<TorrentRow> {
+    pub async fn add(
+        &self,
+        source: &str,
+        download_folder: Option<&str>,
+        move_after_completion_folder: Option<&str>,
+    ) -> Result<TorrentRow> {
         let resolved = self.source.resolve(source).await?;
         let stored_source = resolved.rewritten_source.as_deref().unwrap_or(source);
-        let output_path = output_dir
+        let download_folder = download_folder
             .filter(|s| !s.is_empty())
-            .map(str::to_string)
-            .unwrap_or_else(|| self.default_output_dir.display().to_string());
-        db::insert_pending(&self.pool, &resolved.info_hash, stored_source, &output_path).await
+            .map(str::to_string);
+        let move_after_completion_folder = move_after_completion_folder
+            .filter(|s| !s.is_empty())
+            .map(str::to_string);
+        db::insert_pending(
+            &self.pool,
+            &resolved.info_hash,
+            stored_source,
+            download_folder.as_deref(),
+            move_after_completion_folder.as_deref(),
+        )
+        .await
     }
 
     pub async fn get_status(&self, id: Uuid) -> Result<Option<TorrentStatus>> {
@@ -149,7 +163,9 @@ impl TorrentManager {
             }
 
             if delete_files {
-                let _ = std::fs::remove_dir_all(&row.output_path);
+                if let Some(path) = &row.output_path {
+                    let _ = std::fs::remove_dir_all(path);
+                }
             }
         }
         db::delete(&self.pool, id).await
@@ -237,8 +253,13 @@ impl TorrentManager {
         for row in to_start {
             let manager = Arc::clone(&self);
             tokio::spawn(async move {
+                // If the caller provided an explicit download_folder, pass it
+                // straight to librqbit (which uses it verbatim — flat layout
+                // for multi-file torrents). Otherwise leave output_folder
+                // unset so librqbit creates <download_dir>/<torrent_name>/
+                // itself for multi-file torrents.
                 let options = AddTorrentOptions {
-                    output_folder: Some(row.output_path.clone()),
+                    output_folder: row.download_folder.clone(),
                     overwrite: true,
                     ..Default::default()
                 };
@@ -318,7 +339,7 @@ impl TorrentManager {
                 state: final_state,
                 error_message: stats.error.as_deref(),
             };
-            let row = match db::update_progress(&self.pool, id, progress).await {
+            let mut row = match db::update_progress(&self.pool, id, progress).await {
                 Ok(row) => row,
                 Err(err) => {
                     error!(id = %id, error = %err, "failed to persist torrent progress");
@@ -326,6 +347,61 @@ impl TorrentManager {
                 }
             };
 
+            // Backfill output_path once metadata resolves (handle.name()
+            // becomes Some). Skipped if the user set download_folder
+            // explicitly — in that case the path IS the download_folder
+            // and we set it as soon as we know anything.
+            if row.output_path.is_none() {
+                let computed = row.download_folder.clone().or_else(|| {
+                    name.as_ref().map(|n| {
+                        self.default_output_dir
+                            .join(n)
+                            .to_string_lossy()
+                            .into_owned()
+                    })
+                });
+                if let Some(p) = computed {
+                    if let Err(err) = db::update_output_path(&self.pool, id, &p).await {
+                        warn!(id = %id, error = %err, "failed to backfill output_path");
+                    } else {
+                        row.output_path = Some(p);
+                    }
+                }
+            }
+
+            // On transition into Finished, fire-and-forget the move into the
+            // completed directory. Spawned so a slow cross-filesystem copy
+            // doesn't stall subsequent per-torrent work in this tick.
+            if state_changed && final_state == TorrentState::Finished {
+                if let Some(current) = row.output_path.clone() {
+                    let target = row.move_after_completion_folder.clone().unwrap_or_else(|| {
+                        let basename = Path::new(&current)
+                            .file_name()
+                            .map(|n| n.to_string_lossy().into_owned())
+                            .unwrap_or_else(|| id.to_string());
+                        self.default_output_dir
+                            .join("completed")
+                            .join(&basename)
+                            .to_string_lossy()
+                            .into_owned()
+                    });
+                    let pool = self.pool.clone();
+                    tokio::spawn(async move {
+                        match move_path(&current, &target).await {
+                            Ok(()) => {
+                                if let Err(err) = db::update_output_path(&pool, id, &target).await {
+                                    error!(id = %id, error = %err, "post-move output_path update failed");
+                                }
+                                info!(id = %id, from = %current, to = %target, "torrent moved after completion");
+                            }
+                            Err(err) => {
+                                warn!(id = %id, from = %current, to = %target, error = %err, "post-completion move failed; leaving at original path");
+                            }
+                        }
+                    });
+                }
+            }
+
             let status = row_to_status(row, live);
 
             if state_changed {
@@ -482,6 +558,48 @@ fn now_millis() -> u64 {
         .unwrap_or(0)
 }
 
+/// Moves a file or directory from `from` to `to`. Tries an atomic rename
+/// first; on EXDEV (cross-device) falls back to a recursive copy + delete.
+/// Parent of `to` is created if missing.
+async fn move_path(from: &str, to: &str) -> std::io::Result<()> {
+    let from = PathBuf::from(from);
+    let to = PathBuf::from(to);
+    tokio::task::spawn_blocking(move || -> std::io::Result<()> {
+        if let Some(parent) = to.parent() {
+            std::fs::create_dir_all(parent)?;
+        }
+        match std::fs::rename(&from, &to) {
+            Ok(()) => Ok(()),
+            Err(e) if e.raw_os_error() == Some(18) => {
+                // EXDEV: cross-device rename — fall back.
+                copy_tree(&from, &to)?;
+                if from.is_dir() {
+                    std::fs::remove_dir_all(&from)?;
+                } else {
+                    std::fs::remove_file(&from)?;
+                }
+                Ok(())
+            }
+            Err(e) => Err(e),
+        }
+    })
+    .await
+    .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?
+}
+
+fn copy_tree(from: &Path, to: &Path) -> std::io::Result<()> {
+    if from.is_dir() {
+        std::fs::create_dir_all(to)?;
+        for entry in std::fs::read_dir(from)? {
+            let entry = entry?;
+            copy_tree(&entry.path(), &to.join(entry.file_name()))?;
+        }
+    } else {
+        std::fs::copy(from, to)?;
+    }
+    Ok(())
+}
+
 async fn persist_error(pool: &PgPool, id: Uuid, message: &str) {
     let progress = db::Progress {
         name: None,
@@ -558,7 +676,7 @@ fn row_to_status(row: TorrentRow, live: LiveTorrentStats) -> TorrentStatus {
         info_hash: row.info_hash,
         name: row.name.unwrap_or_default(),
         source: row.source,
-        output_path: row.output_path,
+        output_path: row.output_path.unwrap_or_default(),
         total_bytes: row.total_bytes.unwrap_or(0) as u64,
         downloaded_bytes: row.downloaded_bytes as u64,
         state: proto_state(row.state) as i32,
@@ -598,10 +716,13 @@ fn resolve_err(e: ResolveError) -> Status {
 impl Torrents for GrpcTorrents {
     async fn add(&self, request: Request<AddRequest>) -> Result<Response<AddResponse>, Status> {
         let req = request.into_inner();
-        let output_dir = (!req.output_dir.is_empty()).then_some(req.output_dir.as_str());
+        let download_folder =
+            (!req.download_folder.is_empty()).then_some(req.download_folder.as_str());
+        let move_after_completion_folder = (!req.move_after_completion_folder.is_empty())
+            .then_some(req.move_after_completion_folder.as_str());
         let row = self
             .manager
-            .add(&req.magnet, output_dir)
+            .add(&req.magnet, download_folder, move_after_completion_folder)
             .await
             .map_err(|err| Status::internal(err.to_string()))?;
         let id_str = row.id.to_string();
@@ -782,6 +903,66 @@ impl Torrents for GrpcTorrents {
 
         Ok(Response::new(ReceiverStream::new(rx_stream)))
     }
+
+    async fn files(
+        &self,
+        request: Request<FilesRequest>,
+    ) -> Result<Response<FilesResponse>, Status> {
+        let id = self
+            .manager
+            .resolve_id(&request.into_inner().id)
+            .await
+            .map_err(resolve_err)?;
+
+        let (file_infos, file_progress) = {
+            let tracked = self.manager.tracked.lock().await;
+            let Some(handle) = tracked.get(&id) else {
+                return Err(Status::not_found(
+                    "torrent not found or metadata not yet resolved",
+                ));
+            };
+            let infos = handle
+                .with_metadata(|m| m.file_infos.clone())
+                .map_err(|e| Status::failed_precondition(format!("metadata not resolved: {e}")))?;
+            let progress = handle.stats().file_progress;
+            (infos, progress)
+        };
+
+        Ok(Response::new(FilesResponse {
+            files: build_file_list(&file_infos, &file_progress),
+        }))
+    }
+}
+
+/// Pure mapping from librqbit's per-file metadata + per-file verified-byte
+/// counters into the wire `TorrentFile` shape. Extracted from the `files`
+/// handler so it can be unit-tested without a live librqbit session.
+fn build_file_list(
+    file_infos: &[librqbit::file_info::FileInfo],
+    file_progress: &[u64],
+) -> Vec<TorrentFile> {
+    file_infos
+        .iter()
+        .enumerate()
+        .map(|(i, fi)| {
+            let length = fi.len;
+            let downloaded = file_progress.get(i).copied().unwrap_or(0);
+            TorrentFile {
+                name: fi.relative_filename.to_string_lossy().into_owned(),
+                length,
+                downloaded_bytes: downloaded,
+                state: derive_file_state(downloaded, length) as i32,
+            }
+        })
+        .collect()
+}
+
+fn derive_file_state(downloaded: u64, length: u64) -> FileState {
+    if length > 0 && downloaded >= length {
+        FileState::Ready
+    } else {
+        FileState::InProgress
+    }
 }
 
 #[cfg(test)]
@@ -972,4 +1153,76 @@ mod tests {
         let out = compute_diff(&mut cache, TorrentState::Downloading, 1000, 1, 1000);
         assert_eq!(out.2, 100);
     }
+
+    // ---- build_file_list / derive_file_state ----
+
+    fn fi(name: &str, len: u64) -> librqbit::file_info::FileInfo {
+        use librqbit_core::torrent_metainfo::FileDetailsAttrs;
+        librqbit::file_info::FileInfo {
+            relative_filename: std::path::PathBuf::from(name),
+            offset_in_torrent: 0,
+            piece_range: 0..0,
+            attrs: FileDetailsAttrs::default(),
+            len,
+        }
+    }
+
+    #[test]
+    fn derive_file_state_ready_when_downloaded_at_least_length() {
+        assert_eq!(derive_file_state(100, 100), FileState::Ready);
+        assert_eq!(derive_file_state(150, 100), FileState::Ready);
+    }
+
+    #[test]
+    fn derive_file_state_in_progress_below_length() {
+        assert_eq!(derive_file_state(0, 100), FileState::InProgress);
+        assert_eq!(derive_file_state(99, 100), FileState::InProgress);
+    }
+
+    #[test]
+    fn derive_file_state_in_progress_when_length_zero() {
+        // Defensive: length 0 shouldn't happen in practice but if it does we
+        // shouldn't claim READY — there's nothing to download.
+        assert_eq!(derive_file_state(0, 0), FileState::InProgress);
+    }
+
+    #[test]
+    fn build_file_list_zips_infos_with_progress_by_index() {
+        let infos = vec![fi("a.flac", 100), fi("b.flac", 200), fi("c.flac", 300)];
+        let progress = vec![100, 50, 0];
+        let out = build_file_list(&infos, &progress);
+        assert_eq!(out.len(), 3);
+        assert_eq!(out[0].name, "a.flac");
+        assert_eq!(out[0].length, 100);
+        assert_eq!(out[0].downloaded_bytes, 100);
+        assert_eq!(out[0].state, FileState::Ready as i32);
+        assert_eq!(out[1].downloaded_bytes, 50);
+        assert_eq!(out[1].state, FileState::InProgress as i32);
+        assert_eq!(out[2].downloaded_bytes, 0);
+        assert_eq!(out[2].state, FileState::InProgress as i32);
+    }
+
+    #[test]
+    fn build_file_list_handles_short_progress_vector() {
+        // librqbit shouldn't return a short file_progress, but be defensive.
+        let infos = vec![fi("a.flac", 100), fi("b.flac", 200)];
+        let progress = vec![100];
+        let out = build_file_list(&infos, &progress);
+        assert_eq!(out[1].downloaded_bytes, 0);
+        assert_eq!(out[1].state, FileState::InProgress as i32);
+    }
+
+    #[test]
+    fn build_file_list_preserves_subdir_in_name() {
+        let mut info = fi("subdir/track.flac", 100);
+        info.relative_filename = std::path::PathBuf::from("Artist - Album/01-track.flac");
+        let out = build_file_list(&[info], &[100]);
+        assert_eq!(out[0].name, "Artist - Album/01-track.flac");
+    }
+
+    #[test]
+    fn build_file_list_empty_inputs_returns_empty() {
+        let out = build_file_list(&[], &[]);
+        assert!(out.is_empty());
+    }
 }