Move torrents to completed after finish

Cargo.lock

@@ -3200,12 +3200,14 @@ name = "tora"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "chrono",
  "clap",
  "colored",
  "hyper-util",
  "nix",
  "serde_json",
  "tokio",
+ "tokio-stream",
  "tonic",
  "tonic-health",
  "tora-proto",

crates/proto/src/generated/torrent/torrent.tonic.rs

@@ -173,6 +173,37 @@ pub mod torrents_client {
                 .insert(GrpcMethod::new("torrent.Torrents", "Resume"));
             self.inner.unary(req, path, codec).await
         }
+        pub async fn notifications(
+            &mut self,
+            request: impl tonic::IntoRequest<super::NotificationsRequest>,
+        ) -> std::result::Result<
+            tonic::Response<tonic::codec::Streaming<super::Notification>>,
+            tonic::Status,
+        > {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/torrent.Torrents/Notifications");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("torrent.Torrents", "Notifications"));
+            self.inner.server_streaming(req, path, codec).await
+        }
+        pub async fn files(
+            &mut self,
+            request: impl tonic::IntoRequest<super::FilesRequest>,
+        ) -> std::result::Result<tonic::Response<super::FilesResponse>, tonic::Status> {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/torrent.Torrents/Files");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("torrent.Torrents", "Files"));
+            self.inner.unary(req, path, codec).await
+        }
     }
 }
 /// Generated server implementations.
@@ -212,6 +243,19 @@ pub mod torrents_server {
             &self,
             request: tonic::Request<super::ResumeRequest>,
         ) -> std::result::Result<tonic::Response<super::ResumeResponse>, tonic::Status>;
+        /// Server streaming response type for the Notifications method.
+        type NotificationsStream: tonic::codegen::tokio_stream::Stream<
+                Item = std::result::Result<super::Notification, tonic::Status>,
+            > + std::marker::Send
+            + 'static;
+        async fn notifications(
+            &self,
+            request: tonic::Request<super::NotificationsRequest>,
+        ) -> std::result::Result<tonic::Response<Self::NotificationsStream>, tonic::Status>;
+        async fn files(
+            &self,
+            request: tonic::Request<super::FilesRequest>,
+        ) -> std::result::Result<tonic::Response<super::FilesResponse>, tonic::Status>;
     }
     #[derive(Debug)]
     pub struct TorrentsServer<T> {
@@ -508,6 +552,87 @@ pub mod torrents_server {
                     };
                     Box::pin(fut)
                 }
+                "/torrent.Torrents/Notifications" => {
+                    #[allow(non_camel_case_types)]
+                    struct NotificationsSvc<T: Torrents>(pub Arc<T>);
+                    impl<T: Torrents>
+                        tonic::server::ServerStreamingService<super::NotificationsRequest>
+                        for NotificationsSvc<T>
+                    {
+                        type Response = super::Notification;
+                        type ResponseStream = T::NotificationsStream;
+                        type Future =
+                            BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::NotificationsRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as Torrents>::notifications(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = NotificationsSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.server_streaming(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/torrent.Torrents/Files" => {
+                    #[allow(non_camel_case_types)]
+                    struct FilesSvc<T: Torrents>(pub Arc<T>);
+                    impl<T: Torrents> tonic::server::UnaryService<super::FilesRequest> for FilesSvc<T> {
+                        type Response = super::FilesResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::FilesRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as Torrents>::files(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = FilesSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 _ => Box::pin(async move {
                     let mut response = http::Response::new(tonic::body::Body::default());
                     let headers = response.headers_mut();

crates/proto/src/lib.rs

@@ -2,9 +2,12 @@ pub mod torrent {
     include!("generated/torrent/torrent.rs");
 }
 
+pub use torrent::notification as notification_detail;
 pub use torrent::{
-    AddRequest, AddResponse, ListRequest, ListResponse, PauseRequest, PauseResponse, RemoveRequest,
-    RemoveResponse, ResumeRequest, ResumeResponse, State, StatusRequest, TorrentStatus,
+    AddRequest, AddResponse, AddedInfo, FileState, FilesRequest, FilesResponse, ListRequest,
+    ListResponse, Notification, NotificationKind, NotificationsRequest, PauseRequest,
+    PauseResponse, ProgressMark, RemoveRequest, RemoveResponse, RemovedInfo, ResumeRequest,
+    ResumeResponse, State, StateChanged, StatusRequest, TorrentFile, TorrentStatus,
     torrents_client::TorrentsClient,
     torrents_server::{Torrents, TorrentsServer},
 };

crates/tora/Cargo.toml

@@ -5,12 +5,14 @@ edition = "2024"
 
 [dependencies]
 anyhow = "1.0.103"
+chrono = { version = "0.4", default-features = false, features = ["clock"] }
 clap = { version = "4.6.1", features = ["derive", "env"] }
 colored = "3"
 hyper-util = { version = "0.1.20", features = ["tokio"] }
 nix = { version = "0.31", features = ["signal"] }
 serde_json = "1.0.140"
-tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "net"] }
+tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "net", "signal"] }
+tokio-stream = "0.1.18"
 tonic = "0.14.6"
 tonic-health = "0.14.6"
 tora-proto = { version = "0.1.0", path = "../proto" }

crates/tora/src/cli.rs

@@ -1,6 +1,7 @@
 use std::path::PathBuf;
 
 use clap::{Parser, Subcommand, ValueEnum};
+use tora_proto::NotificationKind;
 
 #[derive(Parser)]
 #[command(
@@ -43,6 +44,31 @@ pub enum SortKey {
     Added,
 }
 
+/// CLI-facing mirror of `NotificationKind` so `clap`'s `ValueEnum` derive can
+/// enumerate the values. Mapped back to the proto enum via `to_proto`.
+#[derive(Copy, Clone, Debug, ValueEnum)]
+pub enum KindFilter {
+    /// `TORRENT_ADDED`
+    Added,
+    /// `TORRENT_REMOVED`
+    Removed,
+    /// `STATE_CHANGED` (includes DOWNLOADING <-> STALE, * -> FINISHED, etc.)
+    StateChanged,
+    /// `PROGRESS` (one event per integer percent advance)
+    Progress,
+}
+
+impl KindFilter {
+    pub fn to_proto(self) -> i32 {
+        match self {
+            KindFilter::Added => NotificationKind::TorrentAdded as i32,
+            KindFilter::Removed => NotificationKind::TorrentRemoved as i32,
+            KindFilter::StateChanged => NotificationKind::StateChanged as i32,
+            KindFilter::Progress => NotificationKind::Progress as i32,
+        }
+    }
+}
+
 #[derive(Subcommand)]
 pub enum Command {
     /// Check whether torad is reachable and healthy.
@@ -51,9 +77,13 @@ pub enum Command {
     Download {
         /// Magnet link, HTTP(S) URL, or path to a .torrent file.
         source: String,
-        /// Directory to save the torrent into. Defaults to torad's configured download directory.
+        /// Folder to download into. Defaults to torad's `<download_dir>/<torrent_name>/`.
         #[arg(long)]
-        output_dir: Option<String>,
+        download_folder: Option<String>,
+        /// Folder to move the torrent into once it finishes downloading.
+        /// Defaults to torad's `<download_dir>/completed/<torrent_name>/`.
+        #[arg(long)]
+        move_after_completion_folder: Option<String>,
     },
     /// Show the status of a single torrent.
     Status {
@@ -95,4 +125,15 @@ pub enum Command {
     },
     /// Stop the torad daemon.
     Stop,
+    /// Stream torrent notifications live from torad. Runs until interrupted
+    /// (Ctrl-C) or the daemon disconnects.
+    Watch {
+        /// Filter to these torrent ids (UUIDs or unique prefixes). Repeatable.
+        /// Omit to observe all torrents.
+        #[arg(long = "id", value_name = "ID")]
+        ids: Vec<String>,
+        /// Only emit these event kinds. Repeatable. Omit for all kinds.
+        #[arg(long = "kind", value_enum)]
+        kinds: Vec<KindFilter>,
+    },
 }

crates/tora/src/main.rs

@@ -6,17 +6,18 @@ use std::time::Duration;
 
 use anyhow::{Context, Result};
 use clap::Parser;
-use cli::{Cli, Command, SortKey};
+use cli::{Cli, Command, KindFilter, SortKey};
 use hyper_util::rt::TokioIo;
 use output::Formatter;
 use tokio::net::UnixStream;
+use tokio_stream::StreamExt;
 use tonic::transport::{Channel, Endpoint, Uri};
 use tonic_health::pb::HealthCheckRequest;
 use tonic_health::pb::health_check_response::ServingStatus;
 use tonic_health::pb::health_client::HealthClient;
 use tora_proto::{
-    AddRequest, ListRequest, PauseRequest, RemoveRequest, ResumeRequest, StatusRequest,
-    TorrentsClient,
+    AddRequest, ListRequest, NotificationsRequest, PauseRequest, RemoveRequest, ResumeRequest,
+    StatusRequest, TorrentsClient,
 };
 use tower::service_fn;
 
@@ -40,7 +41,20 @@ async fn main() -> Result<()> {
 
     match cli.command {
         Command::Health => health(&socket, f).await,
-        Command::Download { source, output_dir } => download(&socket, source, output_dir, f).await,
+        Command::Download {
+            source,
+            download_folder,
+            move_after_completion_folder,
+        } => {
+            download(
+                &socket,
+                source,
+                download_folder,
+                move_after_completion_folder,
+                f,
+            )
+            .await
+        }
         Command::Status { id } => status(&socket, id, f).await,
         Command::List { sort } => list(&socket, sort, f).await,
         Command::Remove { id, delete_files } => remove(&socket, id, delete_files, f).await,
@@ -51,6 +65,7 @@ async fn main() -> Result<()> {
             download_dir,
         } => start_daemon(&socket, &pid_file, &database_url, download_dir, f).await,
         Command::Stop => stop_daemon(&socket, &pid_file, f).await,
+        Command::Watch { ids, kinds } => watch(&socket, ids, kinds, f).await,
     }
 }
 
@@ -99,7 +114,8 @@ async fn health(socket: &Path, f: &dyn Formatter) -> Result<()> {
 async fn download(
     socket: &Path,
     source: String,
-    output_dir: Option<String>,
+    download_folder: Option<String>,
+    move_after_completion_folder: Option<String>,
     f: &dyn Formatter,
 ) -> Result<()> {
     let channel = connect(socket).await?;
@@ -109,7 +125,8 @@ async fn download(
         RPC_TIMEOUT,
         client.add(AddRequest {
             magnet: source,
-            output_dir: output_dir.unwrap_or_default(),
+            download_folder: download_folder.unwrap_or_default(),
+            move_after_completion_folder: move_after_completion_folder.unwrap_or_default(),
         }),
     )
     .await
@@ -269,3 +286,44 @@ async fn stop_daemon(socket: &Path, pid_file: &Path, f: &dyn Formatter) -> Resul
         Err(err) => Err(err).context("failed to send SIGTERM to torad"),
     }
 }
+
+async fn watch(
+    socket: &Path,
+    ids: Vec<String>,
+    kinds: Vec<KindFilter>,
+    f: &dyn Formatter,
+) -> Result<()> {
+    let channel = connect(socket).await?;
+    let mut client = TorrentsClient::new(channel);
+
+    let request = NotificationsRequest {
+        ids,
+        kinds: kinds.iter().map(|k| k.to_proto()).collect(),
+    };
+
+    let stream = client
+        .notifications(request)
+        .await
+        .context("Notifications RPC failed")?
+        .into_inner();
+
+    tokio::pin!(stream);
+
+    let stdout = std::io::stdout();
+    while let Some(message) = stream.next().await {
+        match message {
+            Ok(notification) => {
+                let line = f.watch(&notification)?;
+                let mut guard = stdout.lock();
+                use std::io::Write;
+                writeln!(guard, "{line}")?;
+                guard.flush()?;
+            }
+            Err(status) => {
+                anyhow::bail!("stream error: {status}");
+            }
+        }
+    }
+
+    Ok(())
+}

crates/tora/src/output.rs

@@ -1,8 +1,12 @@
 use std::fmt::Write;
 
 use anyhow::Result;
+use chrono::{Local, TimeZone};
 use colored::Colorize;
-use tora_proto::{State, TorrentStatus};
+use tora_proto::{
+    Notification, NotificationKind, ProgressMark, State, StateChanged, TorrentStatus,
+    notification_detail::Detail,
+};
 
 use crate::cli::OutputFormat;
 
@@ -18,6 +22,7 @@ pub trait Formatter {
     fn resume(&self) -> Result<String>;
     fn start(&self) -> Result<String>;
     fn stop(&self) -> Result<String>;
+    fn watch(&self, event: &Notification) -> Result<String>;
 }
 
 pub fn from_format(format: OutputFormat) -> Box<dyn Formatter> {
@@ -77,6 +82,10 @@ impl Formatter for LongFormatter {
     fn stop(&self) -> Result<String> {
         Ok(format!("  {} torad stopped", "●".yellow()))
     }
+
+    fn watch(&self, event: &Notification) -> Result<String> {
+        Ok(format_event_long(event))
+    }
 }
 
 struct JsonFormatter;
@@ -130,6 +139,10 @@ impl Formatter for JsonFormatter {
             &serde_json::json!({ "stopped": true }),
         )?)
     }
+
+    fn watch(&self, event: &Notification) -> Result<String> {
+        Ok(serde_json::to_string(event)?)
+    }
 }
 
 fn format_status_long(status: &TorrentStatus) -> String {
@@ -216,6 +229,99 @@ fn format_status_long(status: &TorrentStatus) -> String {
     out
 }
 
+fn format_event_long(event: &Notification) -> String {
+    let time = Local
+        .timestamp_millis_opt(event.observed_at_unix_millis as i64)
+        .single()
+        .map(|t| t.format("%H:%M:%S").to_string())
+        .unwrap_or_else(|| "??".to_string());
+
+    let kind = NotificationKind::try_from(event.kind).unwrap_or(NotificationKind::Unspecified);
+    let torrent = event.torrent.as_ref();
+    let id = torrent.map(|t| short_id(&t.id)).unwrap_or_default();
+    let name = torrent
+        .map(|t| display_name(&t.name))
+        .unwrap_or_else(|| "(unknown)".to_string());
+    let state = torrent
+        .and_then(|t| State::try_from(t.state).ok())
+        .unwrap_or(State::Unspecified);
+
+    let id_w = 8usize;
+    let name_w = 30usize;
+
+    let (icon, suffix) = match (&kind, event.detail.as_ref()) {
+        (NotificationKind::TorrentAdded, _) => ("+".green(), String::new()),
+        (NotificationKind::TorrentRemoved, Some(Detail::Removed(r))) => {
+            let note = if r.files_deleted {
+                " (files deleted)"
+            } else {
+                ""
+            };
+            ("−".red(), format!("{}{}", colored_state(state), note.red()))
+        }
+        (NotificationKind::TorrentRemoved, _) => ("−".red(), colored_state(state).to_string()),
+        (
+            NotificationKind::StateChanged,
+            Some(Detail::StateChanged(StateChanged { current, .. })),
+        ) => {
+            let cur = State::try_from(*current).unwrap_or(State::Unspecified);
+            let icon = state_icon(cur);
+            (icon, colored_state(cur).to_string())
+        }
+        (
+            NotificationKind::Progress,
+            Some(Detail::Progress(ProgressMark {
+                percent,
+                bytes_delta,
+            })),
+        ) => {
+            let delta = humanize_bytes(*bytes_delta);
+            let pct_bar = format_pct_inline(*percent);
+            (
+                "↑".yellow(),
+                format!(
+                    "{} {} {}",
+                    pct_bar.yellow(),
+                    format!("{}%", percent).bold(),
+                    format!("(Δ {delta})").dimmed()
+                ),
+            )
+        }
+        (_, _) => ("·".dimmed(), colored_state(state).to_string()),
+    };
+
+    format!(
+        "  {}  {}  {}  {}  {}",
+        time.dimmed(),
+        icon,
+        pad(&id, id_w).dimmed(),
+        pad(&name, name_w),
+        suffix,
+    )
+}
+
+fn state_icon(state: State) -> colored::ColoredString {
+    match state {
+        State::Finished => "✓".green(),
+        State::Downloading => "▶".yellow(),
+        State::Paused => "⏸".cyan(),
+        State::Error => "✗".red(),
+        State::Stale => "⚠".red(),
+        State::Pending | State::Unspecified => "·".dimmed(),
+    }
+}
+
+fn format_pct_inline(percent: u32) -> String {
+    let filled = (percent as usize * BAR_WIDTH) / 100;
+    let empty = BAR_WIDTH - filled;
+    format!(
+        "{}{} {:>3}%",
+        "█".repeat(filled),
+        "░".repeat(empty),
+        percent
+    )
+}
+
 fn format_list_long(torrents: &[TorrentStatus]) -> String {
     if torrents.is_empty() {
         return "  no torrents".dimmed().to_string();
@@ -301,6 +407,7 @@ fn colored_state(state: State) -> colored::ColoredString {
         State::Downloading => s.yellow(),
         State::Paused => s.cyan(),
         State::Error => s.red().bold(),
+        State::Stale => s.red(),
         State::Pending | State::Unspecified => s.dimmed(),
     }
 }
@@ -312,6 +419,7 @@ fn pad_colored(s: &str, width: usize, state: State) -> colored::ColoredString {
         State::Downloading => padded.yellow(),
         State::Paused => padded.cyan(),
         State::Error => padded.red().bold(),
+        State::Stale => padded.red(),
         State::Pending | State::Unspecified => padded.dimmed(),
     }
 }

crates/torad/src/db.rs

@@ -23,6 +23,7 @@ pub enum TorrentState {
     Paused,
     Finished,
     Error,
+    Stale,
 }
 
 #[derive(Debug, Clone, sqlx::FromRow)]
@@ -31,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,
@@ -42,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)
@@ -60,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)
@@ -71,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)
@@ -82,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)
@@ -92,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)
@@ -118,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,
@@ -126,8 +148,12 @@ pub struct Progress<'a> {
     pub error_message: Option<&'a str>,
 }
 
-pub async fn update_progress(pool: &PgPool, id: Uuid, progress: Progress<'_>) -> Result<()> {
-    sqlx::query(
+pub async fn update_progress(
+    pool: &PgPool,
+    id: Uuid,
+    progress: Progress<'_>,
+) -> Result<TorrentRow> {
+    let row = sqlx::query_as::<_, TorrentRow>(
         "UPDATE torrents
          SET name = COALESCE($2, name),
              total_bytes = $3,
@@ -136,7 +162,9 @@ pub async fn update_progress(pool: &PgPool, id: Uuid, progress: Progress<'_>) ->
              error_message = $6,
              updated_at = now(),
              completed_at = CASE WHEN $5 = 'finished' THEN now() ELSE completed_at END
-         WHERE id = $1",
+         WHERE id = $1
+         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)
@@ -144,7 +172,7 @@ pub async fn update_progress(pool: &PgPool, id: Uuid, progress: Progress<'_>) ->
     .bind(progress.downloaded_bytes)
     .bind(progress.state)
     .bind(progress.error_message)
-    .execute(pool)
+    .fetch_one(pool)
     .await?;
-    Ok(())
+    Ok(row)
 }

db/schema.sql

@@ -3,7 +3,8 @@ CREATE TYPE torrent_state AS ENUM (
     'downloading',
     'paused',
     'finished',
-    'error'
+    'error',
+    'stale'
 );
 
 CREATE TABLE torrents (
@@ -11,7 +12,15 @@ CREATE TABLE torrents (
     info_hash TEXT NOT NULL UNIQUE,
     name TEXT,
     source TEXT NOT NULL,
-    output_path TEXT NOT NULL,
+    -- Set by the poller once librqbit resolves metadata and creates the
+    -- output directory. NULL until then (magnet links take time to resolve).
+    output_path TEXT,
+    -- Per-torrent override of the download location. NULL = use the daemon
+    -- default (<download_dir>/<torrent_name>/ for multi-file torrents).
+    download_folder TEXT,
+    -- Per-torrent override of the post-completion move target. NULL = use
+    -- the daemon default (<download_dir>/completed/<torrent_name>/).
+    move_after_completion_folder TEXT,
     total_bytes BIGINT,
     downloaded_bytes BIGINT NOT NULL DEFAULT 0,
     state torrent_state NOT NULL DEFAULT 'pending',

proto/torrent.proto

@@ -15,12 +15,25 @@ service Torrents {
   rpc Pause(PauseRequest) returns (PauseResponse);
   // Resume a paused torrent.
   rpc Resume(ResumeRequest) returns (ResumeResponse);
+  // Subscribe to a live stream of torrent notifications. Events are delivered
+  // from the subscribe point onward — no history is replayed. New subscribers
+  // only see events that fire after they connect.
+  rpc Notifications(NotificationsRequest) returns (stream Notification);
+
+  // List the files inside a torrent with their download progress. Returns
+  // NOT_FOUND if the torrent is unknown or its metadata has not resolved yet
+  // (e.g. a magnet link still fetching peers).
+  rpc Files(FilesRequest) returns (FilesResponse);
 }
 
 message AddRequest {
   string magnet = 1;
-  // Optional output directory override. Empty means use torad's default.
-  string output_dir = 2;
+  // Folder to download into. Empty = daemon default
+  // (<download_dir>/<torrent_name>/ for multi-file torrents).
+  string download_folder = 2;
+  // Folder to move the torrent into after completion. Empty = daemon default
+  // (<download_dir>/completed/<torrent_name>/).
+  string move_after_completion_folder = 3;
 }
 
 message AddResponse {
@@ -56,6 +69,94 @@ message ResumeRequest {
 
 message ResumeResponse {}
 
+message NotificationsRequest {
+  // If non-empty, only events for these torrent ids (or id prefixes, as
+  // accepted by `Status`/`Remove`) are delivered.
+  repeated string ids = 1;
+  // If non-empty, only these event kinds are delivered. An empty list means
+  // all kinds.
+  repeated NotificationKind kinds = 2;
+}
+
+message FilesRequest {
+  string id = 1;
+}
+
+enum FileState {
+  FILE_STATE_UNSPECIFIED = 0;
+  // File is partially downloaded.
+  IN_PROGRESS = 1;
+  // File's full byte range has been verified.
+  READY = 2;
+}
+
+message TorrentFile {
+  // Path relative to the torrent root, as declared in the .torrent metadata.
+  // May contain forward slashes for multi-file torrents.
+  string name = 1;
+  uint64 length = 2;
+  uint64 downloaded_bytes = 3;
+  FileState state = 4;
+}
+
+message FilesResponse {
+  repeated TorrentFile files = 1;
+}
+
+message Notification {
+  // Wall-clock time the daemon observed the change, in milliseconds since the
+  // Unix epoch.
+  uint64 observed_at_unix_millis = 1;
+  // What changed. Use this to dispatch; use the `oneof detail` for
+  // kind-specific payload.
+  NotificationKind kind = 2;
+  // Full snapshot of the torrent *after* the change. Always populated so
+  // consumers don't need to fan out to `Status` for context.
+  TorrentStatus torrent = 3;
+  oneof detail {
+    StateChanged state_changed = 10;
+    ProgressMark progress = 11;
+    AddedInfo added = 12;
+    RemovedInfo removed = 13;
+  }
+}
+
+enum NotificationKind {
+  NOTIFICATION_KIND_UNSPECIFIED = 0;
+  // Emitted once when a torrent is added via `Add`.
+  TORRENT_ADDED = 1;
+  // Emitted once when a torrent is removed via `Remove`.
+  TORRENT_REMOVED = 2;
+  // Emitted on any `State` transition, including PAUSED <-> DOWNLOADING,
+  // DOWNLOADING <-> STALE, * -> FINISHED, * -> ERROR.
+  STATE_CHANGED = 3;
+  // Emitted when the integer download percentage changes (rate-limited).
+  // No event is sent if the percentage did not advance since the last tick.
+  PROGRESS = 4;
+}
+
+message StateChanged {
+  State previous = 1;
+  State current = 2;
+}
+
+message ProgressMark {
+  // Current integer percentage 0..=100.
+  uint32 percent = 1;
+  // Bytes downloaded since the previous PROGRESS event for this torrent.
+  uint64 bytes_delta = 2;
+}
+
+message AddedInfo {
+  // Source string as provided to `Add` (magnet, URL, or file path).
+  string source = 1;
+}
+
+message RemovedInfo {
+  // True if `Remove` was called with `delete_files = true`.
+  bool files_deleted = 1;
+}
+
 enum State {
   STATE_UNSPECIFIED = 0;
   PENDING = 1;
@@ -63,6 +164,12 @@ enum State {
   PAUSED = 3;
   FINISHED = 4;
   ERROR = 5;
+  // Live per librqbit, but the daemon has detected a stall: no byte progress
+  // for several ticks AND zero connected peers. The torrent is still
+  // considered live by the underlying engine; this state is an application-
+  // level overlay. Recovers to DOWNLOADING automatically when bytes advance
+  // or a peer connects.
+  STALE = 6;
 }
 
 message TorrentStatus {