Magnet upload + resolution

Cargo.toml

@@ -17,6 +17,7 @@ askama = "0.16"
 askama_web = { version = "0.16", features = ["axum-0.8"] }
 axum = { version = "0.8.9", features = ["macros"] }
 axum-extra = { version = "0.12.6", features = ["cookie"] }
+axum_typed_multipart = { version = "0.16.5", default-features = false }
 # UTF-8 paths for easier String/PathBuf interop
 camino = { version = "1.2.2", features = ["serde1"] }
 # Date/time management
@@ -36,8 +37,10 @@ env_logger = "0.11.10"
 # Interactions with the torrent client
 # Comment/uncomment below for development version
 # hightorrent_api = { path = "../hightorrent_api" }
-hightorrent_api = { git = "https://github.com/angrynode/hightorrent_api" }
+hightorrent_api = { git = "https://github.com/angrynode/hightorrent_api", branch = "feat-sea-orm", features = [ "sea_orm" ] }
 # hightorrent_api = "0.2"
+# rqbit torrent client to resolve magnets
+librqbit = { git = "https://github.com/ikatson/rqbit" }
 log = "0.4.29"
 # SQLite ORM
 sea-orm = { version = "2.0.0-rc.38", features = [ "runtime-tokio", "debug-print", "sqlx-sqlite"] }
@@ -47,6 +50,7 @@ sea-orm-migration = { version = "2.0.0-rc.38" }
 serde = { version = "1.0.228", features = ["derive", "rc"] }
 # (De)serialization for operations log
 serde_json = { version = "1" }
+serde_with = "3.20.0"
 # Error declaration/context
 snafu = "0.9"
 # Serve static assets directly from the binary

src/config.rs

@@ -88,6 +88,9 @@ pub struct AppConfig {
 
     #[serde(default = "AppConfig::default_log_path")]
     pub log_path: Utf8PathBuf,
+
+    #[serde(default = "AppConfig::default_rqbit_path")]
+    pub rqbit_path: Utf8PathBuf,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -127,6 +130,10 @@ impl AppConfig {
         Self::config_dir().join("operations.log")
     }
 
+    pub fn default_rqbit_path() -> Utf8PathBuf {
+        Self::config_dir().join("rqbit")
+    }
+
     pub async fn load_from_xdg() -> Result<Self, ConfigError> {
         let config_dir = Self::config_dir();
         create_dir_all(&config_dir)

src/database/category.rs

@@ -6,7 +6,7 @@ use snafu::prelude::*;
 use std::str::FromStr;
 
 use crate::database::operator::DatabaseOperator;
-use crate::database::{content_folder, operation::*};
+use crate::database::{content_folder, magnet, operation::*, torrent};
 use crate::extractors::normalized_path::*;
 use crate::extractors::user::User;
 use crate::routes::category::CategoryForm;
@@ -29,6 +29,10 @@ pub struct Model {
     pub path: NormalizedPathAbsolute,
     #[sea_orm(has_many)]
     pub content_folders: HasMany<content_folder::Entity>,
+    #[sea_orm(has_many)]
+    pub magnets: HasMany<magnet::Entity>,
+    #[sea_orm(has_many)]
+    pub torrents: HasMany<torrent::Entity>,
 }
 
 #[async_trait::async_trait]

src/database/content_folder.rs

@@ -7,8 +7,10 @@ use snafu::prelude::*;
 use std::str::FromStr;
 
 use crate::database::category::{self, CategoryError};
+use crate::database::magnet;
 use crate::database::operation::{Operation, OperationId, OperationLog, OperationType, Table};
 use crate::database::operator::DatabaseOperator;
+use crate::database::torrent;
 use crate::extractors::normalized_path::{NormalizedPathAbsolute, NormalizedPathComponent};
 use crate::extractors::user::User;
 use crate::routes::content_folder::ContentFolderForm;
@@ -43,6 +45,10 @@ pub struct Model {
     pub parent_id: Option<i32>,
     #[sea_orm(self_ref, relation_enum = "Parent", from = "parent_id", to = "id")]
     pub parent: HasOne<Entity>,
+    #[sea_orm(has_many)]
+    pub magnets: HasMany<magnet::Entity>,
+    #[sea_orm(has_many)]
+    pub torrents: HasMany<torrent::Entity>,
 }
 
 #[async_trait::async_trait]

src/database/magnet.rs

@@ -0,0 +1,292 @@
+use chrono::Utc;
+use hightorrent_api::hightorrent::{MagnetLink, MagnetLinkError, TorrentID};
+use sea_orm::entity::prelude::*;
+use sea_orm::*;
+use snafu::prelude::*;
+
+use crate::database::operation::*;
+use crate::database::operator::DatabaseOperator;
+use crate::database::torrent::TorrentError;
+use crate::database::{category, content_folder};
+use crate::extractors::user::User;
+use crate::routes::magnet::MagnetForm;
+use crate::state::AppState;
+use crate::state::logger::LoggerError;
+
+/// A category to store associated files.
+///
+/// Each category has a name and an associated path on disk, where
+/// symlinks to the content will be created.
+#[sea_orm::model]
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
+#[sea_orm(table_name = "magnet")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub torrent_id: TorrentID,
+    pub link: MagnetLink,
+    pub name: String,
+    pub category_id: i32,
+    #[sea_orm(belongs_to, from = "category_id", to = "id")]
+    pub category: HasOne<category::Entity>,
+    pub content_folder_id: Option<i32>,
+    #[sea_orm(belongs_to, from = "content_folder_id", to = "id")]
+    pub content_folder: HasOne<content_folder::Entity>,
+}
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for ActiveModel {}
+
+#[derive(Debug, Snafu)]
+#[snafu(visibility(pub))]
+pub enum MagnetError {
+    #[snafu(display("The magnet is invalid"))]
+    InvalidMagnet { source: MagnetLinkError },
+    #[snafu(display("Database error"))]
+    DB { source: sea_orm::DbErr },
+    #[snafu(display("The magnet (ID: {id}) does not exist"))]
+    NotFound { id: i32 },
+    #[snafu(display("The magnet (TorrentID: {id}) does not exist"))]
+    NotFoundTorrentID { id: TorrentID },
+    #[snafu(display("Failed to save the operation log"))]
+    Logger { source: LoggerError },
+    #[snafu(display("Requested category not found"))]
+    NoSuchCategory { id: i32 },
+    #[snafu(display("Requested content folder not found"))]
+    NoSuchContentFolder { id: i32 },
+    #[snafu(display("The magnet is already uploaded"))]
+    DuplicateMagnet,
+    #[snafu(display("This magnet is already known in the system with the full torrent"))]
+    DuplicateTorrent,
+    #[snafu(display("Failed to read the torrent list"))]
+    Torrent { source: TorrentError },
+}
+
+#[derive(Clone, Debug)]
+pub struct MagnetOperator {
+    pub state: AppState,
+    pub user: Option<User>,
+}
+
+impl MagnetOperator {
+    pub fn new(state: AppState, user: Option<User>) -> Self {
+        Self { state, user }
+    }
+
+    pub fn db(&self) -> DatabaseOperator {
+        DatabaseOperator {
+            state: self.state.clone(),
+            user: self.user.clone(),
+        }
+    }
+
+    /// List magnets with related category/content_folder
+    ///
+    /// Should not fail, unless SQLite was corrupted for some reason.
+    pub async fn list_with_related(&self) -> Result<Vec<ModelEx>, MagnetError> {
+        Entity::load()
+            .with(category::Entity)
+            .with(content_folder::Entity)
+            .all(&self.state.database)
+            .await
+            .context(DBSnafu)
+    }
+
+    /// List magnets
+    ///
+    /// Should not fail, unless SQLite was corrupted for some reason.
+    pub async fn list(&self) -> Result<Vec<Model>, MagnetError> {
+        Entity::find()
+            .all(&self.state.database)
+            .await
+            .context(DBSnafu)
+    }
+
+    /// List magnets in a specific category (without recursing)
+    ///
+    /// Should not fail, unless SQLite was corrupted for some reason.
+    pub async fn list_for_category(&self, category_id: i32) -> Result<Vec<Model>, MagnetError> {
+        // TODO: optimization
+        Ok(self
+            .list()
+            .await?
+            .into_iter()
+            .filter(|x| x.category_id == category_id && x.content_folder_id.is_none())
+            .collect())
+    }
+
+    /// List magnets in a specific content_folder (without recursing)
+    ///
+    /// Should not fail, unless SQLite was corrupted for some reason.
+    pub async fn list_for_content_folder(
+        &self,
+        content_folder_id: i32,
+    ) -> Result<Vec<Model>, MagnetError> {
+        // TODO: optimization
+        Ok(self
+            .list()
+            .await?
+            .into_iter()
+            .filter(|x| x.content_folder_id == Some(content_folder_id))
+            .collect())
+    }
+
+    /// Count magnets
+    ///
+    /// Should not fail, unless SQLite was corrupted for some reason.
+    pub async fn count(&self) -> Result<usize, MagnetError> {
+        // TODO: there may be a faster sea_orm operation for this
+        Ok(self.list().await?.len())
+    }
+
+    pub async fn get(&self, id: i32) -> Result<Model, MagnetError> {
+        let db = &self.state.database;
+
+        Entity::find_by_id(id)
+            .one(db)
+            .await
+            .context(DBSnafu)?
+            .ok_or(MagnetError::NotFound { id })
+    }
+
+    pub async fn get_by_torrent_id(&self, id: &TorrentID) -> Result<Model, MagnetError> {
+        let db = &self.state.database;
+
+        Entity::find()
+            .filter(Column::TorrentId.eq(id.clone()))
+            .one(db)
+            .await
+            .context(DBSnafu)?
+            .ok_or(MagnetError::NotFoundTorrentID { id: id.clone() })
+    }
+
+    /// Delete an uploaded magnet
+    pub async fn delete(&self, id: i32) -> Result<String, MagnetError> {
+        let db = &self.state.database;
+
+        let uploaded_magnet = Entity::find_by_id(id)
+            .one(db)
+            .await
+            .context(DBSnafu)?
+            .ok_or(MagnetError::NotFound { id })?;
+
+        let clone: Model = uploaded_magnet.clone();
+        uploaded_magnet.delete(db).await.context(DBSnafu)?;
+
+        let operation_log = OperationLog {
+            user: self.user.clone(),
+            date: Utc::now(),
+            table: Table::Magnet,
+            operation: OperationType::Delete,
+            operation_id: OperationId {
+                object_id: clone.id,
+                name: clone.name.to_owned(),
+            },
+            operation_form: None,
+        };
+
+        self.state
+            .logger
+            .write(operation_log)
+            .await
+            .context(LoggerSnafu)?;
+
+        Ok(clone.name)
+    }
+
+    /// Create a new uploaded magnet
+    ///
+    /// Fails if:
+    ///
+    /// - the magnet is invalid
+    pub async fn create(&self, f: &MagnetForm) -> Result<Model, MagnetError> {
+        let magnet = MagnetLink::new(&f.magnet).context(InvalidMagnetSnafu)?;
+
+        // Check duplicates
+        let list = self.list().await?;
+
+        if list.iter().any(|x| x.torrent_id == magnet.id()) {
+            // The magnet is already known
+            return Err(MagnetError::DuplicateMagnet);
+        }
+
+        // Check for duplicates in the torrent table, so we don't
+        // even have to resolve the magnet.
+        let list = self.db().torrent().list().await.context(TorrentSnafu)?;
+        if list.iter().any(|x| x.torrent_id == magnet.id()) {
+            // The magnet is already known as a torrent
+            return Err(MagnetError::DuplicateTorrent);
+        }
+
+        // Verify that the requested category/content_folder exist
+        let _category = self
+            .db()
+            .category()
+            .find_by_id(f.category_id)
+            .await
+            .map_err(|_e| MagnetError::NoSuchCategory { id: f.category_id })?;
+
+        if let Some(content_folder_id) = f.content_folder_id {
+            let _content_folder = self
+                .db()
+                .content_folder()
+                .find_by_id(content_folder_id)
+                .await
+                .map_err(|_e| MagnetError::NoSuchContentFolder {
+                    id: content_folder_id,
+                })?;
+        }
+
+        let model = ActiveModel {
+            torrent_id: Set(magnet.id()),
+            link: Set(magnet.clone()),
+            name: Set(magnet.name().to_string()),
+            category_id: Set(f.category_id),
+            content_folder_id: Set(f.content_folder_id),
+            ..Default::default()
+        }
+        .save(&self.state.database)
+        .await
+        .context(DBSnafu)?;
+
+        // Now that the magnet has been summoned into the DB,
+        // we should let the resolver know about it.
+        self.state
+            .resolver
+            .send(magnet.clone())
+            .expect("resolver sender channel has been closed");
+
+        // Should not fail
+        let model = model.try_into_model().unwrap();
+
+        let operation_log = OperationLog {
+            user: self.user.clone(),
+            date: Utc::now(),
+            table: Table::Magnet,
+            operation: OperationType::Create,
+            operation_id: OperationId {
+                object_id: model.id.to_owned(),
+                name: model.name.to_string(),
+            },
+            operation_form: Some(Operation::Magnet(f.clone())),
+        };
+
+        self.state
+            .logger
+            .write(operation_log)
+            .await
+            .context(LoggerSnafu)?;
+
+        Ok(model)
+    }
+
+    /// Removes (and cancels resolution) for a given magnet,
+    ///
+    /// if it was previously known. Has no effect otherwise.
+    pub async fn cancel_and_remove(&self, torrent_id: &TorrentID) {
+        if let Ok(magnet) = self.get_by_torrent_id(torrent_id).await {
+            // TODO: should we error here?
+            self.delete(magnet.id).await.unwrap();
+        }
+    }
+}

src/database/mod.rs

@@ -1,5 +1,7 @@
 // sea_orm example: https://github.com/SeaQL/sea-orm/blob/master/examples/axum_example/
 pub mod category;
 pub mod content_folder;
+pub mod magnet;
 pub mod operation;
 pub mod operator;
+pub mod torrent;