feat: Include notifications via NTFY

Author: dormant-user

Cargo.toml

@@ -29,7 +29,7 @@ rustdoc-args = ["--document-private-items"]
 utoipa = { version = "5.5.0", features = ["actix_extras"] }
 utoipa-swagger-ui = { version = "9.0.2", features = ["actix-web"] }
 actix-web = "4.13.0"
-reqwest = { version = "0.13.3", features = ["blocking", "cookies", "json", "form"] }
+reqwest = { version = "0.13.3", features = ["cookies", "json", "form"] }
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.149"
 actix-rt = "2.11.0"

src/lib.rs

@@ -1,7 +1,7 @@
 #![allow(rustdoc::bare_urls)]
 #![doc = include_str!("../README.md")]
 
-use actix_web::{web, App, HttpResponse, HttpServer};
+use actix_web::{web, App, HttpServer};
 use std::{collections::HashMap, sync::Arc};
 use tokio::sync::RwLock;
 
@@ -12,6 +12,7 @@ mod rsync;
 mod settings;
 mod squire;
 mod swagger;
+mod ntfy;
 
 /// Contains entrypoint and initializer settings to trigger the asynchronous `HTTPServer`
 ///

src/ntfy.rs

@@ -0,0 +1,40 @@
+use reqwest::Client;
+use crate::settings;
+
+pub async fn send(config: &settings::Config, title: &str, data: &str) -> bool {
+    let client = Client::builder().build().unwrap();
+    let url = format!("{}/{}", config.ntfy_url, config.ntfy_topic);
+    log::info!("Client url: {}", &url);
+
+    let mut request = client
+        .post(&url)
+        .header("X-Title", title)
+        .header("Content-Type", "application/x-www-form-urlencoded")
+        .body(data.to_string());
+
+    let username = config.ntfy_username.to_string();
+    let password = config.ntfy_password.to_string();
+    if !username.is_empty() && !password.is_empty() {
+        request = request.basic_auth(username, Some(password));
+    }
+
+    match request.send().await {
+        Ok(resp) => match resp.error_for_status() {
+            Ok(resp) => {
+                match resp.text().await {
+                    Ok(body) => log::debug!("ntfy response: {}", body),
+                    Err(e) => log::error!("Failed to read response body: {}", e),
+                }
+                true
+            }
+            Err(e) => {
+                log::error!("HTTP error: {}", e);
+                false
+            }
+        },
+        Err(e) => {
+            log::error!("Request failed: {}", e);
+            false
+        }
+    }
+}

src/settings.rs

@@ -20,11 +20,19 @@ pub struct Config {
     pub port: u16,
     pub apikey: String,
     pub workers: usize,
+
     pub qbit_api: String,
+
     pub username: String,
     pub password: String,
+
     pub utc_logger: bool,
     pub log_level: log::LevelFilter,
+
+    pub ntfy_url: String,
+    pub ntfy_topic: String,
+    pub ntfy_username: String,
+    pub ntfy_password: String,
 }
 
 fn startup_error(msg: &str) {
@@ -76,7 +84,7 @@ impl Config {
         let mut qbit_api = squire::get_env_var("qbit_api", Some("http://localhost:8080/"));
         let username = squire::get_env_var("username", None);
         let password = squire::get_env_var("password", None);
-        qbit_api = qbit_api.strip_suffix("/").unwrap_or_default().to_string();
+        qbit_api = qbit_api.strip_suffix("/").unwrap_or(&qbit_api).to_string();
 
         let utc_logger = squire::get_env_var("utc_logger", Some("true")) == "true";
         let default_log_level = squire::get_env_var("log_level", Some("info"));
@@ -92,6 +100,14 @@ impl Config {
             }
         };
 
+        let mut ntfy_url = squire::get_env_var("ntfy_url", None);
+        let mut ntfy_topic = squire::get_env_var("ntfy_topic", None);
+        let ntfy_username = squire::get_env_var("ntfy_username", None);
+        let ntfy_password = squire::get_env_var("ntfy_password", None);
+
+        ntfy_url = ntfy_url.strip_suffix("/").unwrap_or(&ntfy_url).to_string();
+        ntfy_topic = ntfy_topic.strip_prefix("/").unwrap_or(&ntfy_topic).to_string();
+
         Self {
             host,
             port,
@@ -102,6 +118,10 @@ impl Config {
             password,
             utc_logger,
             log_level,
+            ntfy_url,
+            ntfy_topic,
+            ntfy_username,
+            ntfy_password,
         }
     }
 }

src/squire.rs

@@ -1,4 +1,4 @@
-use crate::{qb, rsync, settings};
+use crate::{ntfy, qb, rsync, settings};
 use regex::Regex;
 use reqwest::Client;
 use serde_json::Value;
@@ -181,21 +181,42 @@ pub fn spawn_worker(
                 };
 
                 match entry.status {
-                    settings::Status::Copying(_) | settings::Status::Completed => continue,
+                    settings::Status::Copying(_) => continue,
+
+                    settings::Status::Completed => {
+                        let config_cloned = config.clone();
+                        let name_clone = entry.name.clone();
+                        let entry_clone = entry.rsync.clone();
+                        tokio::spawn(async move {
+                            let _ = ntfy::send(
+                                &config_cloned, "RuTorrent: Transfer Complete",
+                                format!("{} has been transferred to {}", name_clone, entry_clone.unwrap().host).as_str()
+                            ).await;
+                        });
+                        db.remove(&hash);
+                    }
 
                     settings::Status::Downloading(_) => {
                         entry.status = settings::Status::Downloading(progress);
 
                         if progress >= 1.0 {
                             if let Some(target) = entry.rsync.clone() {
+                                let config_cloned = config.clone();
+                                let name_clone = entry.name.clone();
+                                tokio::spawn(async move {
+                                    let _ = ntfy::send(
+                                        &config_cloned, "RuTorrent: Download Complete",
+                                        format!("{} has been downloaded", name_clone).as_str()
+                                    ).await;
+                                });
+
                                 log::info!("Download complete → rsync: {}", entry.name);
 
                                 entry.status = settings::Status::Copying(0.0);
 
                                 let state_clone = state.clone();
                                 let hash_clone = hash.clone();
                                 let name_clone = entry.name.clone();
-
                                 tokio::spawn(async move {
                                     rsync::run(
                                         state_clone,
@@ -230,16 +251,18 @@ pub fn spawn_worker(
 ///
 /// Returns the resolved env var or a default string.
 pub fn get_env_var(key: &str, default: Option<&str>) -> String {
-    let lower = std::env::var(key.to_lowercase());
-    let upper = std::env::var(key.to_uppercase());
-    let default = default.unwrap_or_default();
-    lower
-        .unwrap_or(upper.unwrap_or(default.to_string()))
-        .to_string()
+    if let Ok(upper) = std::env::var(key.to_uppercase()) {
+        return upper;
+    }
+    if let Ok(lower) = std::env::var(key.to_lowercase()) {
+        return lower;
+    }
+    default.unwrap_or_default().to_string()
 }
 
 /// Load dotenv file using the env var `env_file` or `ENV_FILE`
 pub fn load_env_file() {
+    // TODO: dotenv does not load env vars with $ sign
     let env_file = get_env_var("env_file", Some(".env"));
     let env_file_path = std::env::current_dir().unwrap_or_default().join(env_file);
     let _ = dotenv::from_path(env_file_path.as_path());