From 95f7bb6e52975cc444dd93677ae6290ea6a8a658 Mon Sep 17 00:00:00 2001 From: Jamie Pine <32987599+jamiepine@users.noreply.github.com> Date: Fri, 1 Apr 2022 07:41:42 -0700 Subject: [PATCH] better rust formating --- core/.rustfmt.toml | 20 +- core/src/client/mod.rs | 96 +++--- core/src/crypto/encryption.rs | 8 +- core/src/db/migrate.rs | 207 ++++++------ core/src/db/mod.rs | 22 +- core/src/file/checksum.rs | 102 +++--- core/src/file/explorer/mod.rs | 45 ++- core/src/file/indexer.rs | 433 +++++++++++++------------ core/src/file/mod.rs | 138 ++++---- core/src/file/thumb.rs | 170 +++++----- core/src/file/watcher.rs | 32 +- core/src/job/jobs.rs | 258 +++++++-------- core/src/job/mod.rs | 8 +- core/src/job/worker.rs | 298 ++++++++--------- core/src/lib.rs | 579 ++++++++++++++++------------------ core/src/library/loader.rs | 114 +++---- core/src/p2p/listener.rs | 68 ++-- core/src/state/client.rs | 136 ++++---- core/src/sys/locations.rs | 314 +++++++++--------- core/src/sys/mod.rs | 16 +- core/src/sys/volumes.rs | 130 ++++---- 21 files changed, 1576 insertions(+), 1618 deletions(-) diff --git a/core/.rustfmt.toml b/core/.rustfmt.toml index 7f591ca55..3901c3cce 100644 --- a/core/.rustfmt.toml +++ b/core/.rustfmt.toml @@ -1,7 +1,15 @@ -hard_tabs = true -tab_spaces = 4 -match_block_trailing_comma = true -max_width = 90 +max_width = 100 +hard_tabs = false +tab_spaces = 2 newline_style = "Unix" -use_field_init_shorthand = true -use_try_shorthand = true \ No newline at end of file +use_small_heuristics = "Default" +reorder_imports = true +reorder_modules = true +remove_nested_parens = true +edition = "2021" +merge_derives = true +use_try_shorthand = false +use_field_init_shorthand = false +force_explicit_abi = true +# normalize_comments = true +normalize_doc_attributes = true \ No newline at end of file diff --git a/core/src/client/mod.rs b/core/src/client/mod.rs index 9437fd7cb..e298586ee 100644 --- a/core/src/client/mod.rs +++ b/core/src/client/mod.rs @@ -1,69 +1,69 @@ use crate::{ - prisma::{self, Client}, - state, Core, + prisma::{self, Client}, + state, Core, }; use std::env; use thiserror::Error; #[derive(Error, Debug)] pub enum ClientError { - #[error("Database error")] - DatabaseError(#[from] prisma::QueryError), + #[error("Database error")] + DatabaseError(#[from] prisma::QueryError), } pub enum Platform { - Unknown = 0, - Windows, - MacOS, - Linux, - IOS, - Android, + Unknown = 0, + Windows, + MacOS, + Linux, + IOS, + Android, } pub async fn create(core: &Core) -> Result<(), ClientError> { - println!("Creating client..."); - let mut config = state::client::get(); + println!("Creating client..."); + let mut config = state::client::get(); - let db = &core.database; + let db = &core.database; - let hostname = match hostname::get() { - Ok(hostname) => hostname.to_str().unwrap_or_default().to_owned(), - Err(_) => "unknown".to_owned(), - }; + let hostname = match hostname::get() { + Ok(hostname) => hostname.to_str().unwrap_or_default().to_owned(), + Err(_) => "unknown".to_owned(), + }; - let platform = match env::consts::OS { - "windows" => Platform::Windows, - "macos" => Platform::MacOS, - "linux" => Platform::Linux, - _ => Platform::Unknown, - }; + let platform = match env::consts::OS { + "windows" => Platform::Windows, + "macos" => Platform::MacOS, + "linux" => Platform::Linux, + _ => Platform::Unknown, + }; - let client = match db - .client() - .find_unique(Client::uuid().equals(config.client_uuid.clone())) - .exec() - .await? - { - Some(client) => client, - None => { - db.client() - .create_one( - Client::uuid().set(config.client_uuid.clone()), - Client::name().set(hostname.clone()), - vec![ - Client::platform().set(platform as i32), - Client::online().set(true), - ], - ) - .exec() - .await? - }, - }; + let client = match db + .client() + .find_unique(Client::uuid().equals(config.client_uuid.clone())) + .exec() + .await? + { + Some(client) => client, + None => { + db.client() + .create_one( + Client::uuid().set(config.client_uuid.clone()), + Client::name().set(hostname.clone()), + vec![ + Client::platform().set(platform as i32), + Client::online().set(true), + ], + ) + .exec() + .await? + } + }; - config.client_name = hostname; - config.save(); + config.client_name = hostname; + config.save(); - println!("Client: {:?}", &client); + println!("Client: {:?}", &client); - Ok(()) + Ok(()) } diff --git a/core/src/crypto/encryption.rs b/core/src/crypto/encryption.rs index 78b744561..9771faf01 100644 --- a/core/src/crypto/encryption.rs +++ b/core/src/crypto/encryption.rs @@ -6,8 +6,8 @@ use ts_rs::TS; #[derive(Debug, Clone, Copy, Serialize, Deserialize, TS, Eq, PartialEq, IntEnum)] #[ts(export)] pub enum EncryptionAlgorithm { - None = 0, - AES128 = 1, - AES192 = 2, - AES256 = 3, + None = 0, + AES128 = 1, + AES192 = 2, + AES256 = 3, } diff --git a/core/src/db/migrate.rs b/core/src/db/migrate.rs index 5937ffe08..cf5e9cb68 100644 --- a/core/src/db/migrate.rs +++ b/core/src/db/migrate.rs @@ -6,129 +6,124 @@ use include_dir::{include_dir, Dir}; use std::ffi::OsStr; use std::io::BufReader; -const INIT_MIGRATION: &str = - include_str!("../../prisma/migrations/migration_table/migration.sql"); +const INIT_MIGRATION: &str = include_str!("../../prisma/migrations/migration_table/migration.sql"); static MIGRATIONS_DIR: Dir = include_dir!("$CARGO_MANIFEST_DIR/prisma/migrations"); pub async fn run_migrations(db_url: &str) -> Result<()> { - let client = prisma::new_client_with_url(&format!("file:{}", &db_url)).await?; + let client = prisma::new_client_with_url(&format!("file:{}", &db_url)).await?; - match client - ._query_raw::( - "SELECT name FROM sqlite_master WHERE type='table' AND name='_migrations'", - ) - .await - { - Ok(data) => { - if data.len() == 0 { - println!("Migration table does not exist"); - // execute migration - match client._execute_raw(INIT_MIGRATION).await { - Ok(_) => {}, - Err(e) => { - println!("Failed to create migration table: {}", e); - }, - }; + match client + ._query_raw::( + "SELECT name FROM sqlite_master WHERE type='table' AND name='_migrations'", + ) + .await + { + Ok(data) => { + if data.len() == 0 { + println!("Migration table does not exist"); + // execute migration + match client._execute_raw(INIT_MIGRATION).await { + Ok(_) => {} + Err(e) => { + println!("Failed to create migration table: {}", e); + } + }; - let value: Vec = client - ._query_raw("SELECT name FROM sqlite_master WHERE type='table' AND name='_migrations'") - .await - .unwrap(); + let value: Vec = client + ._query_raw("SELECT name FROM sqlite_master WHERE type='table' AND name='_migrations'") + .await + .unwrap(); - println!("Migration table created: {:?}", value); - } else { - println!("Migration table exists: {:?}", data); - } + println!("Migration table created: {:?}", value); + } else { + println!("Migration table exists: {:?}", data); + } - let mut migration_subdirs = MIGRATIONS_DIR - .dirs() - .filter(|subdir| { - subdir - .path() - .file_name() - .map(|name| name != OsStr::new("migration_table")) - .unwrap_or(false) - }) - .collect::>(); + let mut migration_subdirs = MIGRATIONS_DIR + .dirs() + .filter(|subdir| { + subdir + .path() + .file_name() + .map(|name| name != OsStr::new("migration_table")) + .unwrap_or(false) + }) + .collect::>(); - migration_subdirs.sort_by(|a, b| { - let a_name = a.path().file_name().unwrap().to_str().unwrap(); - let b_name = b.path().file_name().unwrap().to_str().unwrap(); + migration_subdirs.sort_by(|a, b| { + let a_name = a.path().file_name().unwrap().to_str().unwrap(); + let b_name = b.path().file_name().unwrap().to_str().unwrap(); - let a_time = a_name[..14].parse::().unwrap(); - let b_time = b_name[..14].parse::().unwrap(); + let a_time = a_name[..14].parse::().unwrap(); + let b_time = b_name[..14].parse::().unwrap(); - a_time.cmp(&b_time) - }); + a_time.cmp(&b_time) + }); - for subdir in migration_subdirs { - println!("{:?}", subdir.path()); - let migration_file = subdir - .get_file(subdir.path().join("./migration.sql")) - .unwrap(); - let migration_sql = migration_file.contents_utf8().unwrap(); + for subdir in migration_subdirs { + println!("{:?}", subdir.path()); + let migration_file = subdir + .get_file(subdir.path().join("./migration.sql")) + .unwrap(); + let migration_sql = migration_file.contents_utf8().unwrap(); - let digest = sha256_digest(BufReader::new(migration_file.contents()))?; - // create a lowercase hash from - let checksum = HEXLOWER.encode(digest.as_ref()); - let name = subdir.path().file_name().unwrap().to_str().unwrap(); + let digest = sha256_digest(BufReader::new(migration_file.contents()))?; + // create a lowercase hash from + let checksum = HEXLOWER.encode(digest.as_ref()); + let name = subdir.path().file_name().unwrap().to_str().unwrap(); - // get existing migration by checksum, if it doesn't exist run the migration - let existing_migration = client - .migration() - .find_unique(Migration::checksum().equals(checksum.clone())) - .exec() - .await?; + // get existing migration by checksum, if it doesn't exist run the migration + let existing_migration = client + .migration() + .find_unique(Migration::checksum().equals(checksum.clone())) + .exec() + .await?; - if existing_migration.is_none() { - println!("Running migration: {}", name); + if existing_migration.is_none() { + println!("Running migration: {}", name); - let steps = migration_sql.split(";").collect::>(); - let steps = &steps[0..steps.len() - 1]; + let steps = migration_sql.split(";").collect::>(); + let steps = &steps[0..steps.len() - 1]; - client - .migration() - .create_one( - Migration::name().set(name.to_string()), - Migration::checksum().set(checksum.clone()), - vec![], - ) - .exec() - .await?; + client + .migration() + .create_one( + Migration::name().set(name.to_string()), + Migration::checksum().set(checksum.clone()), + vec![], + ) + .exec() + .await?; - for (i, step) in steps.iter().enumerate() { - match client._execute_raw(&format!("{};", step)).await { - Ok(_) => { - println!("Step {} ran successfully", i); - client - .migration() - .find_unique( - Migration::checksum().equals(checksum.clone()), - ) - .update(vec![ - Migration::steps_applied().set(i as i32 + 1) - ]) - .exec() - .await?; - }, - Err(e) => { - println!("Error running migration: {}", name); - println!("{}", e); - break; - }, - } - } + for (i, step) in steps.iter().enumerate() { + match client._execute_raw(&format!("{};", step)).await { + Ok(_) => { + println!("Step {} ran successfully", i); + client + .migration() + .find_unique(Migration::checksum().equals(checksum.clone())) + .update(vec![Migration::steps_applied().set(i as i32 + 1)]) + .exec() + .await?; + } + Err(e) => { + println!("Error running migration: {}", name); + println!("{}", e); + break; + } + } + } - println!("Migration {} recorded successfully", name); - } else { - println!("Migration {} already exists", name); - } - } - }, - Err(err) => { - panic!("Failed to check migration table existence: {:?}", err); - }, - } + println!("Migration {} recorded successfully", name); + } else { + println!("Migration {} already exists", name); + } + } + } + Err(err) => { + panic!("Failed to check migration table existence: {:?}", err); + } + } - Ok(()) + Ok(()) } diff --git a/core/src/db/mod.rs b/core/src/db/mod.rs index 02486a77b..f9e09a62d 100644 --- a/core/src/db/mod.rs +++ b/core/src/db/mod.rs @@ -5,22 +5,22 @@ pub mod migrate; #[derive(Error, Debug)] pub enum DatabaseError { - #[error("Failed to connect to database")] - MissingConnection, - #[error("Unable find current_library in the client config")] - MalformedConfig, - #[error("Unable to initialize the Prisma client")] - ClientError(#[from] prisma::NewClientError), + #[error("Failed to connect to database")] + MissingConnection, + #[error("Unable find current_library in the client config")] + MalformedConfig, + #[error("Unable to initialize the Prisma client")] + ClientError(#[from] prisma::NewClientError), } pub async fn create_connection() -> Result { - let config = state::client::get(); + let config = state::client::get(); - let current_library = config.get_current_library(); + let current_library = config.get_current_library(); - let path = current_library.library_path.clone(); + let path = current_library.library_path.clone(); - let client = prisma::new_client_with_url(&format!("file:{}", &path)).await?; + let client = prisma::new_client_with_url(&format!("file:{}", &path)).await?; - Ok(client) + Ok(client) } diff --git a/core/src/file/checksum.rs b/core/src/file/checksum.rs index 45d7032e3..e812c2532 100644 --- a/core/src/file/checksum.rs +++ b/core/src/file/checksum.rs @@ -10,67 +10,67 @@ static SAMPLE_COUNT: u64 = 4; static SAMPLE_SIZE: u64 = 10000; pub fn partial_checksum(path: &str, size: u64) -> Result { - // open file reference - let file = File::open(path)?; + // open file reference + let file = File::open(path)?; - let mut context = Context::new(&SHA256); + let mut context = Context::new(&SHA256); - // include the file size in the checksum - context.update(&size.to_le_bytes()); + // include the file size in the checksum + context.update(&size.to_le_bytes()); - // if size is small enough, just read the whole thing - if SAMPLE_COUNT * SAMPLE_SIZE > size { - let mut buf = vec![0u8; size.try_into()?]; - file.read_exact_at(&mut buf, 0)?; - context.update(&buf); - } else { - // loop over samples - for i in 0..SAMPLE_COUNT { - let start_point = (size / SAMPLE_COUNT) * i; - let mut buf = vec![0u8; SAMPLE_SIZE.try_into()?]; - file.read_exact_at(&mut buf, start_point)?; - context.update(&buf); - } - // sample end of file - let mut buf = vec![0u8; SAMPLE_SIZE.try_into()?]; - file.read_exact_at(&mut buf, size - SAMPLE_SIZE)?; - context.update(&buf); - } + // if size is small enough, just read the whole thing + if SAMPLE_COUNT * SAMPLE_SIZE > size { + let mut buf = vec![0u8; size.try_into()?]; + file.read_exact_at(&mut buf, 0)?; + context.update(&buf); + } else { + // loop over samples + for i in 0..SAMPLE_COUNT { + let start_point = (size / SAMPLE_COUNT) * i; + let mut buf = vec![0u8; SAMPLE_SIZE.try_into()?]; + file.read_exact_at(&mut buf, start_point)?; + context.update(&buf); + } + // sample end of file + let mut buf = vec![0u8; SAMPLE_SIZE.try_into()?]; + file.read_exact_at(&mut buf, size - SAMPLE_SIZE)?; + context.update(&buf); + } - let digest = context.finish(); - let hex = HEXLOWER.encode(digest.as_ref()); + let digest = context.finish(); + let hex = HEXLOWER.encode(digest.as_ref()); - Ok(hex) + Ok(hex) } pub fn full_checksum(path: &str) -> Result { - // read file as buffer and convert to digest - let mut reader = BufReader::new(File::open(path).unwrap()); - let mut context = Context::new(&SHA256); - let mut buffer = [0; 1024]; - loop { - let count = reader.read(&mut buffer)?; - if count == 0 { - break; - } - context.update(&buffer[..count]); - } - let digest = context.finish(); - // create a lowercase hash from - let hex = HEXLOWER.encode(digest.as_ref()); + // read file as buffer and convert to digest + let mut reader = BufReader::new(File::open(path).unwrap()); + let mut context = Context::new(&SHA256); + let mut buffer = [0; 1024]; + loop { + let count = reader.read(&mut buffer)?; + if count == 0 { + break; + } + context.update(&buffer[..count]); + } + let digest = context.finish(); + // create a lowercase hash from + let hex = HEXLOWER.encode(digest.as_ref()); - Ok(hex) + Ok(hex) } pub fn sha256_digest(mut reader: R) -> io::Result { - let mut context = Context::new(&SHA256); - let mut buffer = [0; 1024]; - loop { - let count = reader.read(&mut buffer)?; - if count == 0 { - break; - } - context.update(&buffer[..count]); - } - Ok(context.finish()) + let mut context = Context::new(&SHA256); + let mut buffer = [0; 1024]; + loop { + let count = reader.read(&mut buffer)?; + if count == 0 { + break; + } + context.update(&buffer[..count]); + } + Ok(context.finish()) } diff --git a/core/src/file/explorer/mod.rs b/core/src/file/explorer/mod.rs index e596d5a29..ab7758b2d 100644 --- a/core/src/file/explorer/mod.rs +++ b/core/src/file/explorer/mod.rs @@ -2,32 +2,29 @@ use crate::{file::DirectoryWithContents, prisma::FilePath, CoreContext}; use super::FileError; -pub async fn open_dir( - ctx: &CoreContext, - path: &str, -) -> Result { - let db = &ctx.database; +pub async fn open_dir(ctx: &CoreContext, path: &str) -> Result { + let db = &ctx.database; - println!("getting files... {:?}", &path); + println!("getting files... {:?}", &path); - let directory = db - .file_path() - .find_first(vec![ - FilePath::materialized_path().equals(path.into()), - FilePath::is_dir().equals(true), - ]) - .exec() - .await? - .ok_or(FileError::FileNotFound(path.to_string()))?; + let directory = db + .file_path() + .find_first(vec![ + FilePath::materialized_path().equals(path.into()), + FilePath::is_dir().equals(true), + ]) + .exec() + .await? + .ok_or(FileError::FileNotFound(path.to_string()))?; - let files = db - .file_path() - .find_many(vec![FilePath::parent_id().equals(directory.id)]) - .exec() - .await?; + let files = db + .file_path() + .find_many(vec![FilePath::parent_id().equals(directory.id)]) + .exec() + .await?; - Ok(DirectoryWithContents { - directory: directory.into(), - contents: files.into_iter().map(|l| l.into()).collect(), - }) + Ok(DirectoryWithContents { + directory: directory.into(), + contents: files.into_iter().map(|l| l.into()).collect(), + }) } diff --git a/core/src/file/indexer.rs b/core/src/file/indexer.rs index 010c20ec6..0c1341abc 100644 --- a/core/src/file/indexer.rs +++ b/core/src/file/indexer.rs @@ -14,282 +14,275 @@ use super::checksum::partial_checksum; #[derive(Debug)] pub struct IndexerJob { - pub path: String, + pub path: String, } #[async_trait::async_trait] impl Job for IndexerJob { - async fn run(&self, ctx: WorkerContext) -> Result<()> { - let core_ctx = ctx.core_ctx.clone(); - scan_path(&core_ctx, self.path.as_str(), move |p| { - ctx.progress( - p.iter() - .map(|p| match p.clone() { - ScanProgress::ChunkCount(c) => JobReportUpdate::TaskCount(c), - ScanProgress::SavedChunks(p) => { - JobReportUpdate::CompletedTaskCount(p) - }, - ScanProgress::Message(m) => JobReportUpdate::Message(m), - }) - .collect(), - ) - }) - .await?; - Ok(()) - } + async fn run(&self, ctx: WorkerContext) -> Result<()> { + let core_ctx = ctx.core_ctx.clone(); + scan_path(&core_ctx, self.path.as_str(), move |p| { + ctx.progress( + p.iter() + .map(|p| match p.clone() { + ScanProgress::ChunkCount(c) => JobReportUpdate::TaskCount(c), + ScanProgress::SavedChunks(p) => JobReportUpdate::CompletedTaskCount(p), + ScanProgress::Message(m) => JobReportUpdate::Message(m), + }) + .collect(), + ) + }) + .await?; + Ok(()) + } } #[derive(Clone)] pub enum ScanProgress { - ChunkCount(usize), - SavedChunks(usize), - Message(String), + ChunkCount(usize), + SavedChunks(usize), + Message(String), } // creates a vector of valid path buffers from a directory pub async fn scan_path( - ctx: &CoreContext, - path: &str, - on_progress: impl Fn(Vec) + Send + Sync + 'static, + ctx: &CoreContext, + path: &str, + on_progress: impl Fn(Vec) + Send + Sync + 'static, ) -> Result<()> { - let db = &ctx.database; - let path = path.to_string(); + let db = &ctx.database; + let path = path.to_string(); - let location = create_location(&ctx, &path).await?; + let location = create_location(&ctx, &path).await?; - // query db to highers id, so we can increment it for the new files indexed - #[derive(Deserialize, Serialize, Debug)] - struct QueryRes { - id: Option, - } - // grab the next id so we can increment in memory for batch inserting - let first_file_id = match db - ._query_raw::(r#"SELECT MAX(id) id FROM file_paths"#) - .await - { - Ok(rows) => rows[0].id.unwrap_or(0), - Err(e) => Err(anyhow!("Error querying for next file id: {}", e))?, - }; + // query db to highers id, so we can increment it for the new files indexed + #[derive(Deserialize, Serialize, Debug)] + struct QueryRes { + id: Option, + } + // grab the next id so we can increment in memory for batch inserting + let first_file_id = match db + ._query_raw::(r#"SELECT MAX(id) id FROM file_paths"#) + .await + { + Ok(rows) => rows[0].id.unwrap_or(0), + Err(e) => Err(anyhow!("Error querying for next file id: {}", e))?, + }; - //check is path is a directory - if !PathBuf::from(&path).is_dir() { - return Err(anyhow::anyhow!("{} is not a directory", &path)); - } - let dir_path = path.clone(); + //check is path is a directory + if !PathBuf::from(&path).is_dir() { + return Err(anyhow::anyhow!("{} is not a directory", &path)); + } + let dir_path = path.clone(); - // spawn a dedicated thread to scan the directory for performance - let (paths, scan_start, on_progress) = tokio::task::spawn_blocking(move || { - // store every valid path discovered - let mut paths: Vec<(PathBuf, i32, Option)> = Vec::new(); - // store a hashmap of directories to their file ids for fast lookup - let mut dirs: HashMap = HashMap::new(); - // begin timer for logging purposes - let scan_start = Instant::now(); + // spawn a dedicated thread to scan the directory for performance + let (paths, scan_start, on_progress) = tokio::task::spawn_blocking(move || { + // store every valid path discovered + let mut paths: Vec<(PathBuf, i32, Option)> = Vec::new(); + // store a hashmap of directories to their file ids for fast lookup + let mut dirs: HashMap = HashMap::new(); + // begin timer for logging purposes + let scan_start = Instant::now(); - let mut next_file_id = first_file_id; - let mut get_id = || { - next_file_id += 1; - next_file_id - }; - // walk through directory recursively - for entry in WalkDir::new(&dir_path).into_iter().filter_entry(|dir| { - let approved = !is_hidden(dir) - && !is_app_bundle(dir) - && !is_node_modules(dir) - && !is_library(dir); - approved - }) { - // extract directory entry or log and continue if failed - let entry = match entry { - Ok(entry) => entry, - Err(e) => { - println!("Error reading file {}", e); - continue; - }, - }; - let path = entry.path(); + let mut next_file_id = first_file_id; + let mut get_id = || { + next_file_id += 1; + next_file_id + }; + // walk through directory recursively + for entry in WalkDir::new(&dir_path).into_iter().filter_entry(|dir| { + let approved = + !is_hidden(dir) && !is_app_bundle(dir) && !is_node_modules(dir) && !is_library(dir); + approved + }) { + // extract directory entry or log and continue if failed + let entry = match entry { + Ok(entry) => entry, + Err(e) => { + println!("Error reading file {}", e); + continue; + } + }; + let path = entry.path(); - let parent_path = path - .parent() - .unwrap_or(Path::new("")) - .to_str() - .unwrap_or(""); - let parent_dir_id = dirs.get(&*parent_path); + let parent_path = path + .parent() + .unwrap_or(Path::new("")) + .to_str() + .unwrap_or(""); + let parent_dir_id = dirs.get(&*parent_path); - on_progress(vec![ - ScanProgress::Message(format!("Found: {:?}", &path)), - ScanProgress::ChunkCount(paths.len() / 100), - ]); + on_progress(vec![ + ScanProgress::Message(format!("Found: {:?}", &path)), + ScanProgress::ChunkCount(paths.len() / 100), + ]); - let file_id = get_id(); - paths.push((path.to_owned(), file_id, parent_dir_id.cloned())); + let file_id = get_id(); + paths.push((path.to_owned(), file_id, parent_dir_id.cloned())); - if entry.file_type().is_dir() { - let _path = match path.to_str() { - Some(path) => path.to_owned(), - None => continue, - }; - dirs.insert(_path, file_id); - } - } - (paths, scan_start, on_progress) - }) - .await - .unwrap(); + if entry.file_type().is_dir() { + let _path = match path.to_str() { + Some(path) => path.to_owned(), + None => continue, + }; + dirs.insert(_path, file_id); + } + } + (paths, scan_start, on_progress) + }) + .await + .unwrap(); - let db_write_start = Instant::now(); - let scan_read_time = scan_start.elapsed(); + let db_write_start = Instant::now(); + let scan_read_time = scan_start.elapsed(); - for (i, chunk) in paths.chunks(100).enumerate() { - on_progress(vec![ - ScanProgress::SavedChunks(i as usize), - ScanProgress::Message(format!( - "Writing {} files to db at chunk {}", - chunk.len(), - i - )), - ]); + for (i, chunk) in paths.chunks(100).enumerate() { + on_progress(vec![ + ScanProgress::SavedChunks(i as usize), + ScanProgress::Message(format!( + "Writing {} files to db at chunk {}", + chunk.len(), + i + )), + ]); - // vector to store active models - let mut files: Vec = Vec::new(); - for (file_path, file_id, parent_dir_id) in chunk { - files.push( - match prepare_values(&file_path, *file_id, &location, parent_dir_id) { - Ok(file) => file, - Err(e) => { - println!( - "Error creating file model from path {:?}: {}", - file_path, e - ); - continue; - }, - }, - ); - } - let raw_sql = format!( - r#" + // vector to store active models + let mut files: Vec = Vec::new(); + for (file_path, file_id, parent_dir_id) in chunk { + files.push( + match prepare_values(&file_path, *file_id, &location, parent_dir_id) { + Ok(file) => file, + Err(e) => { + println!("Error creating file model from path {:?}: {}", file_path, e); + continue; + } + }, + ); + } + let raw_sql = format!( + r#" INSERT INTO file_paths (id, is_dir, location_id, materialized_path, name, extension, parent_id, date_created, temp_checksum) VALUES {} "#, - files.join(", ") - ); - // println!("{}", raw_sql); - let count = db._execute_raw(&raw_sql).await; - println!("Inserted {:?} records", count); - } - println!( - "scan of {:?} completed in {:?}. {:?} files found. db write completed in {:?}", - &path, - scan_read_time, - paths.len(), - db_write_start.elapsed() - ); - Ok(()) + files.join(", ") + ); + // println!("{}", raw_sql); + let count = db._execute_raw(&raw_sql).await; + println!("Inserted {:?} records", count); + } + println!( + "scan of {:?} completed in {:?}. {:?} files found. db write completed in {:?}", + &path, + scan_read_time, + paths.len(), + db_write_start.elapsed() + ); + Ok(()) } // reads a file at a path and creates an ActiveModel with metadata fn prepare_values( - file_path: &PathBuf, - id: i32, - location: &LocationResource, - parent_id: &Option, + file_path: &PathBuf, + id: i32, + location: &LocationResource, + parent_id: &Option, ) -> Result { - let metadata = fs::metadata(&file_path)?; - let location_path = location.path.as_ref().unwrap().as_str(); - // let size = metadata.len(); - let name = extract_name(file_path.file_stem()); - let extension = extract_name(file_path.extension()); + let metadata = fs::metadata(&file_path)?; + let location_path = location.path.as_ref().unwrap().as_str(); + // let size = metadata.len(); + let name = extract_name(file_path.file_stem()); + let extension = extract_name(file_path.extension()); - let materialized_path = match file_path.to_str() { - Some(p) => p - .clone() - .strip_prefix(&location_path) - // .and_then(|p| p.strip_suffix(format!("{}{}", name, extension).as_str())) - .unwrap_or_default(), - None => return Err(anyhow!("{}", file_path.to_str().unwrap_or_default())), - }; + let materialized_path = match file_path.to_str() { + Some(p) => p + .clone() + .strip_prefix(&location_path) + // .and_then(|p| p.strip_suffix(format!("{}{}", name, extension).as_str())) + .unwrap_or_default(), + None => return Err(anyhow!("{}", file_path.to_str().unwrap_or_default())), + }; - let partial_checksum = { - if !metadata.is_dir() { - partial_checksum(&file_path.to_str().unwrap(), metadata.len()).unwrap() - } else { - "".to_string() - } - }; + let partial_checksum = { + if !metadata.is_dir() { + partial_checksum(&file_path.to_str().unwrap(), metadata.len()).unwrap() + } else { + "".to_string() + } + }; - let date_created: DateTime = metadata.created().unwrap().into(); - let parsed_date_created = date_created.to_rfc3339_opts(SecondsFormat::Millis, true); + let date_created: DateTime = metadata.created().unwrap().into(); + let parsed_date_created = date_created.to_rfc3339_opts(SecondsFormat::Millis, true); - let values = format!( - "({}, {}, {}, \"{}\", \"{}\", \"{}\", {},\"{}\", \"{}\")", - id, - metadata.is_dir(), - location.id, - materialized_path, - name, - extension, - parent_id - .clone() - .map(|id| format!("\"{}\"", &id)) - .unwrap_or("NULL".to_string()), - parsed_date_created, - partial_checksum - ); + let values = format!( + "({}, {}, {}, \"{}\", \"{}\", \"{}\", {},\"{}\", \"{}\")", + id, + metadata.is_dir(), + location.id, + materialized_path, + name, + extension, + parent_id + .clone() + .map(|id| format!("\"{}\"", &id)) + .unwrap_or("NULL".to_string()), + parsed_date_created, + partial_checksum + ); - println!("{}", values); + println!("{}", values); - Ok(values) + Ok(values) } // extract name from OsStr returned by PathBuff fn extract_name(os_string: Option<&OsStr>) -> String { - os_string - .unwrap_or_default() - .to_str() - .unwrap_or_default() - .to_owned() + os_string + .unwrap_or_default() + .to_str() + .unwrap_or_default() + .to_owned() } fn is_hidden(entry: &DirEntry) -> bool { - entry - .file_name() - .to_str() - .map(|s| s.starts_with(".")) - .unwrap_or(false) + entry + .file_name() + .to_str() + .map(|s| s.starts_with(".")) + .unwrap_or(false) } fn is_library(entry: &DirEntry) -> bool { - entry - .path() - .to_str() - // make better this is shit - .map(|s| s.contains("/Library/")) - .unwrap_or(false) + entry + .path() + .to_str() + // make better this is shit + .map(|s| s.contains("/Library/")) + .unwrap_or(false) } fn is_node_modules(entry: &DirEntry) -> bool { - entry - .file_name() - .to_str() - .map(|s| s.contains("node_modules")) - .unwrap_or(false) + entry + .file_name() + .to_str() + .map(|s| s.contains("node_modules")) + .unwrap_or(false) } fn is_app_bundle(entry: &DirEntry) -> bool { - let is_dir = entry.metadata().unwrap().is_dir(); - let contains_dot = entry - .file_name() - .to_str() - .map(|s| s.contains(".app") | s.contains(".bundle")) - .unwrap_or(false); + let is_dir = entry.metadata().unwrap().is_dir(); + let contains_dot = entry + .file_name() + .to_str() + .map(|s| s.contains(".app") | s.contains(".bundle")) + .unwrap_or(false); - let is_app_bundle = is_dir && contains_dot; - // if is_app_bundle { - // let path_buff = entry.path(); - // let path = path_buff.to_str().unwrap(); + let is_app_bundle = is_dir && contains_dot; + // if is_app_bundle { + // let path_buff = entry.path(); + // let path = path_buff.to_str().unwrap(); - // self::path(&path, ); - // } + // self::path(&path, ); + // } - is_app_bundle + is_app_bundle } diff --git a/core/src/file/mod.rs b/core/src/file/mod.rs index 57324f85e..8dc4906d3 100644 --- a/core/src/file/mod.rs +++ b/core/src/file/mod.rs @@ -4,8 +4,8 @@ use thiserror::Error; use ts_rs::TS; use crate::{ - crypto::encryption::EncryptionAlgorithm, - prisma::{self, FileData, FilePathData}, + crypto::encryption::EncryptionAlgorithm, + prisma::{self, FileData, FilePathData}, }; pub mod checksum; pub mod explorer; @@ -17,98 +17,98 @@ pub mod watcher; #[derive(Debug, Clone, Serialize, Deserialize, TS)] #[ts(export)] pub struct File { - pub id: i32, - pub partial_checksum: String, - pub checksum: Option, - pub size_in_bytes: String, - pub encryption: EncryptionAlgorithm, - pub file_type: FileType, - #[ts(type = "string")] - pub date_created: chrono::DateTime, - #[ts(type = "string")] - pub date_modified: chrono::DateTime, - #[ts(type = "string")] - pub date_indexed: chrono::DateTime, - pub ipfs_id: Option, - pub file_paths: Vec, + pub id: i32, + pub partial_checksum: String, + pub checksum: Option, + pub size_in_bytes: String, + pub encryption: EncryptionAlgorithm, + pub file_type: FileType, + #[ts(type = "string")] + pub date_created: chrono::DateTime, + #[ts(type = "string")] + pub date_modified: chrono::DateTime, + #[ts(type = "string")] + pub date_indexed: chrono::DateTime, + pub ipfs_id: Option, + pub file_paths: Vec, } // A physical file path #[derive(Debug, Clone, Serialize, Deserialize, TS)] #[ts(export)] pub struct FilePath { - pub id: i32, - pub is_dir: bool, - pub location_id: i32, - pub materialized_path: String, - pub file_id: Option, - pub parent_id: Option, - #[ts(type = "string")] - pub date_indexed: chrono::DateTime, - pub permissions: Option, + pub id: i32, + pub is_dir: bool, + pub location_id: i32, + pub materialized_path: String, + pub file_id: Option, + pub parent_id: Option, + #[ts(type = "string")] + pub date_indexed: chrono::DateTime, + pub permissions: Option, } #[repr(i32)] #[derive(Debug, Clone, Copy, Serialize, Deserialize, TS, Eq, PartialEq, IntEnum)] #[ts(export)] pub enum FileType { - Unknown = 0, - Directory = 1, - Package = 2, - Archive = 3, - Image = 4, - Video = 5, - Audio = 6, - Plaintext = 7, - Alias = 8, + Unknown = 0, + Directory = 1, + Package = 2, + Archive = 3, + Image = 4, + Video = 5, + Audio = 6, + Plaintext = 7, + Alias = 8, } impl Into for FileData { - fn into(self) -> File { - File { - id: self.id, - partial_checksum: self.partial_checksum, - checksum: self.checksum, - size_in_bytes: self.size_in_bytes.to_string(), - encryption: EncryptionAlgorithm::from_int(self.encryption).unwrap(), - file_type: FileType::Unknown, - date_created: self.date_created, - date_modified: self.date_modified, - date_indexed: self.date_indexed, - ipfs_id: self.ipfs_id, - file_paths: vec![], - } - } + fn into(self) -> File { + File { + id: self.id, + partial_checksum: self.partial_checksum, + checksum: self.checksum, + size_in_bytes: self.size_in_bytes.to_string(), + encryption: EncryptionAlgorithm::from_int(self.encryption).unwrap(), + file_type: FileType::Unknown, + date_created: self.date_created, + date_modified: self.date_modified, + date_indexed: self.date_indexed, + ipfs_id: self.ipfs_id, + file_paths: vec![], + } + } } impl Into for FilePathData { - fn into(self) -> FilePath { - FilePath { - id: self.id, - is_dir: self.is_dir, - materialized_path: self.materialized_path, - file_id: self.file_id, - parent_id: self.parent_id, - location_id: self.location_id, - date_indexed: self.date_indexed, - permissions: self.permissions, - } - } + fn into(self) -> FilePath { + FilePath { + id: self.id, + is_dir: self.is_dir, + materialized_path: self.materialized_path, + file_id: self.file_id, + parent_id: self.parent_id, + location_id: self.location_id, + date_indexed: self.date_indexed, + permissions: self.permissions, + } + } } #[derive(Serialize, Deserialize, TS, Debug)] #[ts(export)] pub struct DirectoryWithContents { - pub directory: FilePath, - pub contents: Vec, + pub directory: FilePath, + pub contents: Vec, } #[derive(Error, Debug)] pub enum FileError { - #[error("Directory not found (path: {0:?})")] - DirectoryNotFound(String), - #[error("File not found (path: {0:?})")] - FileNotFound(String), - #[error("Database error")] - DatabaseError(#[from] prisma::QueryError), + #[error("Directory not found (path: {0:?})")] + DirectoryNotFound(String), + #[error("File not found (path: {0:?})")] + FileNotFound(String), + #[error("Database error")] + DatabaseError(#[from] prisma::QueryError), } diff --git a/core/src/file/thumb.rs b/core/src/file/thumb.rs index 670e77347..744f47977 100644 --- a/core/src/file/thumb.rs +++ b/core/src/file/thumb.rs @@ -3,9 +3,9 @@ use crate::prisma::FilePathData; use crate::state::client; use crate::sys; use crate::{ - job::{jobs::Job, worker::WorkerContext}, - prisma::FilePath, - CoreContext, + job::{jobs::Job, worker::WorkerContext}, + prisma::FilePath, + CoreContext, }; use anyhow::Result; use image::*; @@ -16,7 +16,7 @@ use webp::*; #[derive(Debug)] pub struct ThumbnailJob { - pub location_id: i32, + pub location_id: i32, } static THUMBNAIL_SIZE_FACTOR: f32 = 0.2; @@ -25,110 +25,100 @@ static CACHE_DIR_NAME: &str = "thumbnails"; #[async_trait::async_trait] impl Job for ThumbnailJob { - async fn run(&self, ctx: WorkerContext) -> Result<()> { - let config = client::get(); - let core_ctx = ctx.core_ctx.clone(); + async fn run(&self, ctx: WorkerContext) -> Result<()> { + let config = client::get(); + let core_ctx = ctx.core_ctx.clone(); - let location = sys::locations::get_location(&core_ctx, self.location_id).await?; + let location = sys::locations::get_location(&core_ctx, self.location_id).await?; - fs::create_dir_all( - Path::new(&config.data_path) - .join(CACHE_DIR_NAME) - .join(format!("{}", self.location_id)), - ) - .unwrap(); + fs::create_dir_all( + Path::new(&config.data_path) + .join(CACHE_DIR_NAME) + .join(format!("{}", self.location_id)), + ) + .unwrap(); - let root_path = location.path.unwrap(); + let root_path = location.path.unwrap(); - let image_files = get_images(&core_ctx, self.location_id).await?; + let image_files = get_images(&core_ctx, self.location_id).await?; - let location_id = location.id.clone(); + let location_id = location.id.clone(); - println!("Found {:?} files", image_files.len()); + println!("Found {:?} files", image_files.len()); - tokio::task::spawn_blocking(move || { - ctx.progress(vec![ - JobReportUpdate::TaskCount(image_files.len()), - JobReportUpdate::Message(format!( - "Preparing to process {} files", - image_files.len() - )), - ]); + tokio::task::spawn_blocking(move || { + ctx.progress(vec![ + JobReportUpdate::TaskCount(image_files.len()), + JobReportUpdate::Message(format!("Preparing to process {} files", image_files.len())), + ]); - for (i, image_file) in image_files.iter().enumerate() { - ctx.progress(vec![JobReportUpdate::Message(format!( - "Processing {}", - image_file.materialized_path.clone() - ))]); - let path = format!("{}{}", root_path, image_file.materialized_path); - let checksum = image_file.temp_checksum.as_ref().unwrap(); + for (i, image_file) in image_files.iter().enumerate() { + ctx.progress(vec![JobReportUpdate::Message(format!( + "Processing {}", + image_file.materialized_path.clone() + ))]); + let path = format!("{}{}", root_path, image_file.materialized_path); + let checksum = image_file.temp_checksum.as_ref().unwrap(); - generate_thumbnail(&path, checksum, location_id).unwrap(); + generate_thumbnail(&path, checksum, location_id).unwrap(); - ctx.progress(vec![JobReportUpdate::CompletedTaskCount(i + 1)]); - } - }) - .await - .unwrap(); + ctx.progress(vec![JobReportUpdate::CompletedTaskCount(i + 1)]); + } + }) + .await + .unwrap(); - Ok(()) - } + Ok(()) + } } -pub fn generate_thumbnail( - file_path: &str, - file_hash: &str, - location_id: i32, -) -> Result<()> { - let config = client::get(); - // Using `image` crate, open the included .jpg file - let img = image::open(file_path).unwrap(); - let (w, h) = img.dimensions(); - // Optionally, resize the existing photo and convert back into DynamicImage - let img: DynamicImage = image::DynamicImage::ImageRgba8(imageops::resize( - &img, - (w as f32 * THUMBNAIL_SIZE_FACTOR) as u32, - (h as f32 * THUMBNAIL_SIZE_FACTOR) as u32, - imageops::FilterType::Triangle, - )); - // Create the WebP encoder for the above image - let encoder: Encoder = Encoder::from_image(&img).unwrap(); +pub fn generate_thumbnail(file_path: &str, file_hash: &str, location_id: i32) -> Result<()> { + let config = client::get(); + // Using `image` crate, open the included .jpg file + let img = image::open(file_path).unwrap(); + let (w, h) = img.dimensions(); + // Optionally, resize the existing photo and convert back into DynamicImage + let img: DynamicImage = image::DynamicImage::ImageRgba8(imageops::resize( + &img, + (w as f32 * THUMBNAIL_SIZE_FACTOR) as u32, + (h as f32 * THUMBNAIL_SIZE_FACTOR) as u32, + imageops::FilterType::Triangle, + )); + // Create the WebP encoder for the above image + let encoder: Encoder = Encoder::from_image(&img).unwrap(); - // Encode the image at a specified quality 0-100 - let webp: WebPMemory = encoder.encode(THUMBNAIL_QUALITY); - // Define and write the WebP-encoded file to a given path - let output_path = Path::new(&config.data_path) - .join(CACHE_DIR_NAME) - .join(format!("{}", location_id)) - .join(file_hash) - .with_extension("webp"); + // Encode the image at a specified quality 0-100 + let webp: WebPMemory = encoder.encode(THUMBNAIL_QUALITY); + // Define and write the WebP-encoded file to a given path + let output_path = Path::new(&config.data_path) + .join(CACHE_DIR_NAME) + .join(format!("{}", location_id)) + .join(file_hash) + .with_extension("webp"); - println!("Writing to {}", output_path.display()); + println!("Writing to {}", output_path.display()); - std::fs::write(&output_path, &*webp).unwrap(); + std::fs::write(&output_path, &*webp).unwrap(); - Ok(()) + Ok(()) } -pub async fn get_images( - ctx: &CoreContext, - location_id: i32, -) -> Result> { - let image_files = ctx - .database - .file_path() - .find_many(vec![ - FilePath::location_id().equals(location_id), - or!( - FilePath::extension().equals("png".to_string()), - FilePath::extension().equals("jpeg".to_string()), - FilePath::extension().equals("jpg".to_string()), - FilePath::extension().equals("gif".to_string()), - FilePath::extension().equals("webp".to_string()), - ), - ]) - .exec() - .await?; +pub async fn get_images(ctx: &CoreContext, location_id: i32) -> Result> { + let image_files = ctx + .database + .file_path() + .find_many(vec![ + FilePath::location_id().equals(location_id), + or!( + FilePath::extension().equals("png".to_string()), + FilePath::extension().equals("jpeg".to_string()), + FilePath::extension().equals("jpg".to_string()), + FilePath::extension().equals("gif".to_string()), + FilePath::extension().equals("webp".to_string()), + ), + ]) + .exec() + .await?; - Ok(image_files) + Ok(image_files) } diff --git a/core/src/file/watcher.rs b/core/src/file/watcher.rs index 0b60e86cd..7f74bf721 100644 --- a/core/src/file/watcher.rs +++ b/core/src/file/watcher.rs @@ -1,25 +1,25 @@ use std::path::Path; use hotwatch::{ - blocking::{Flow, Hotwatch}, - Event, + blocking::{Flow, Hotwatch}, + Event, }; pub fn watch_dir(path: &str) { - let mut hotwatch = Hotwatch::new().expect("hotwatch failed to initialize!"); - hotwatch - .watch(&path, |event: Event| { - if let Event::Write(path) = event { - println!("{:?} changed!", path); - // Flow::Exit - Flow::Continue - } else { - Flow::Continue - } - }) - .expect("failed to watch file!"); + let mut hotwatch = Hotwatch::new().expect("hotwatch failed to initialize!"); + hotwatch + .watch(&path, |event: Event| { + if let Event::Write(path) = event { + println!("{:?} changed!", path); + // Flow::Exit + Flow::Continue + } else { + Flow::Continue + } + }) + .expect("failed to watch file!"); - hotwatch.run(); + hotwatch.run(); - println!("watching directory {:?}", Path::new(&path)); + println!("watching directory {:?}", Path::new(&path)); } diff --git a/core/src/job/jobs.rs b/core/src/job/jobs.rs index cf595d8a9..5ca259a25 100644 --- a/core/src/job/jobs.rs +++ b/core/src/job/jobs.rs @@ -1,10 +1,10 @@ use super::{ - worker::{Worker, WorkerContext}, - JobError, + worker::{Worker, WorkerContext}, + JobError, }; use crate::{ - prisma::{self, JobData}, - CoreContext, + prisma::{self, JobData}, + CoreContext, }; use anyhow::Result; use int_enum::IntEnum; @@ -18,164 +18,166 @@ const MAX_WORKERS: usize = 4; #[async_trait::async_trait] pub trait Job: Send + Sync + Debug { - async fn run(&self, ctx: WorkerContext) -> Result<()>; + async fn run(&self, ctx: WorkerContext) -> Result<()>; } // jobs struct is maintained by the core pub struct Jobs { - // job_queue: Vec>, - // workers are spawned when jobs are picked off the queue - running_workers: HashMap>>, + // job_queue: Vec>, + // workers are spawned when jobs are picked off the queue + running_workers: HashMap>>, } impl Jobs { - pub fn new() -> Self { - Self { - // job_queue: vec![], - running_workers: HashMap::new(), - } - } - pub async fn ingest(&mut self, ctx: &CoreContext, job: Box) { - // create worker to process job - let worker = Worker::new(job); - let id = worker.id(); + pub fn new() -> Self { + Self { + // job_queue: vec![], + running_workers: HashMap::new(), + } + } + pub async fn ingest(&mut self, ctx: &CoreContext, job: Box) { + // create worker to process job + let worker = Worker::new(job); + let id = worker.id(); - if self.running_workers.len() < MAX_WORKERS { - let wrapped_worker = Arc::new(Mutex::new(worker)); + if self.running_workers.len() < MAX_WORKERS { + let wrapped_worker = Arc::new(Mutex::new(worker)); - Worker::spawn(wrapped_worker.clone(), ctx).await; + Worker::spawn(wrapped_worker.clone(), ctx).await; - self.running_workers.insert(id, wrapped_worker); - } - } - pub fn complete(&mut self, job_id: String) { - // remove worker from running workers - self.running_workers.remove(&job_id); - } - pub async fn get_running(&self) -> Vec { - let mut ret = vec![]; + self.running_workers.insert(id, wrapped_worker); + } + } + pub fn complete(&mut self, job_id: String) { + // remove worker from running workers + self.running_workers.remove(&job_id); + } + pub async fn get_running(&self) -> Vec { + let mut ret = vec![]; - for worker in self.running_workers.values() { - let worker = worker.lock().await; - ret.push(worker.job_report.clone()); - } - ret - } - pub async fn get_history(ctx: &CoreContext) -> Result, JobError> { - let db = &ctx.database; - let jobs = db - .job() - .find_many(vec![or![ - prisma::Job::status().equals(JobStatus::Completed.int_value()), - prisma::Job::status().equals(JobStatus::Failed.int_value()), - prisma::Job::status().equals(JobStatus::Canceled.int_value()), - prisma::Job::status().equals(JobStatus::Queued.int_value()), - ]]) - .exec() - .await?; + for worker in self.running_workers.values() { + let worker = worker.lock().await; + ret.push(worker.job_report.clone()); + } + ret + } + pub async fn get_history(ctx: &CoreContext) -> Result, JobError> { + let db = &ctx.database; + let jobs = db + .job() + .find_many(vec![or![ + prisma::Job::status().equals(JobStatus::Completed.int_value()), + prisma::Job::status().equals(JobStatus::Failed.int_value()), + prisma::Job::status().equals(JobStatus::Canceled.int_value()), + prisma::Job::status().equals(JobStatus::Queued.int_value()), + ]]) + .exec() + .await?; - Ok(jobs.into_iter().map(|j| j.into()).collect()) - } + Ok(jobs.into_iter().map(|j| j.into()).collect()) + } } #[derive(Debug)] pub enum JobReportUpdate { - TaskCount(usize), - CompletedTaskCount(usize), - Message(String), - SecondsElapsed(u64), + TaskCount(usize), + CompletedTaskCount(usize), + Message(String), + SecondsElapsed(u64), } #[derive(Debug, Serialize, Deserialize, TS, Clone)] #[ts(export)] pub struct JobReport { - pub id: String, - // client_id: i32, - #[ts(type = "string")] - pub date_created: chrono::DateTime, - #[ts(type = "string")] - pub date_modified: chrono::DateTime, + pub id: String, + // client_id: i32, + #[ts(type = "string")] + pub date_created: chrono::DateTime, + #[ts(type = "string")] + pub date_modified: chrono::DateTime, - pub status: JobStatus, - pub task_count: i32, - pub completed_task_count: i32, + pub status: JobStatus, + pub task_count: i32, + pub completed_task_count: i32, - pub message: String, - // pub percentage_complete: f64, - #[ts(type = "string")] - pub seconds_elapsed: i32, + pub message: String, + // pub percentage_complete: f64, + #[ts(type = "string")] + pub seconds_elapsed: i32, } // convert database struct into a resource struct impl Into for JobData { - fn into(self) -> JobReport { - JobReport { - id: self.id, - // client_id: self.client_id, - status: JobStatus::from_int(self.status).unwrap(), - task_count: self.task_count, - completed_task_count: self.completed_task_count, - date_created: self.date_created, - date_modified: self.date_modified, - message: String::new(), - seconds_elapsed: self.seconds_elapsed, - } - } + fn into(self) -> JobReport { + JobReport { + id: self.id, + // client_id: self.client_id, + status: JobStatus::from_int(self.status).unwrap(), + task_count: self.task_count, + completed_task_count: self.completed_task_count, + date_created: self.date_created, + date_modified: self.date_modified, + message: String::new(), + seconds_elapsed: self.seconds_elapsed, + } + } } impl JobReport { - pub fn new(uuid: String) -> Self { - Self { - id: uuid, - // client_id: 0, - date_created: chrono::Utc::now(), - date_modified: chrono::Utc::now(), - status: JobStatus::Queued, - task_count: 0, - completed_task_count: 0, - message: String::new(), - seconds_elapsed: 0, - } - } - pub async fn create(&self, ctx: &CoreContext) -> Result<(), JobError> { - // let config = client::get(); - ctx.database - .job() - .create_one( - prisma::Job::id().set(self.id.clone()), - prisma::Job::action().set(1), - // prisma::Job::clients().link(prisma::Client::id().equals(config.client_uuid)), - vec![], - ) - .exec() - .await?; - Ok(()) - } - pub async fn update(&self, ctx: &CoreContext) -> Result<(), JobError> { - // let config = client::get(); - ctx.database - .job() - .find_unique(prisma::Job::id().equals(self.id.clone())) - .update(vec![ - prisma::Job::status().set(self.status.int_value()), - prisma::Job::task_count().set(self.task_count), - prisma::Job::completed_task_count().set(self.completed_task_count), - prisma::Job::date_modified().set(chrono::Utc::now()), - prisma::Job::seconds_elapsed().set(self.seconds_elapsed), - ]) - .exec() - .await?; - Ok(()) - } + pub fn new(uuid: String) -> Self { + Self { + id: uuid, + // client_id: 0, + date_created: chrono::Utc::now(), + date_modified: chrono::Utc::now(), + status: JobStatus::Queued, + task_count: 0, + completed_task_count: 0, + message: String::new(), + seconds_elapsed: 0, + } + } + pub async fn create(&self, ctx: &CoreContext) -> Result<(), JobError> { + // let config = client::get(); + ctx + .database + .job() + .create_one( + prisma::Job::id().set(self.id.clone()), + prisma::Job::action().set(1), + // prisma::Job::clients().link(prisma::Client::id().equals(config.client_uuid)), + vec![], + ) + .exec() + .await?; + Ok(()) + } + pub async fn update(&self, ctx: &CoreContext) -> Result<(), JobError> { + // let config = client::get(); + ctx + .database + .job() + .find_unique(prisma::Job::id().equals(self.id.clone())) + .update(vec![ + prisma::Job::status().set(self.status.int_value()), + prisma::Job::task_count().set(self.task_count), + prisma::Job::completed_task_count().set(self.completed_task_count), + prisma::Job::date_modified().set(chrono::Utc::now()), + prisma::Job::seconds_elapsed().set(self.seconds_elapsed), + ]) + .exec() + .await?; + Ok(()) + } } #[repr(i32)] #[derive(Debug, Clone, Copy, Serialize, Deserialize, TS, Eq, PartialEq, IntEnum)] #[ts(export)] pub enum JobStatus { - Queued = 0, - Running = 1, - Completed = 2, - Canceled = 3, - Failed = 4, + Queued = 0, + Running = 1, + Completed = 2, + Canceled = 3, + Failed = 4, } diff --git a/core/src/job/mod.rs b/core/src/job/mod.rs index 5dc2773c5..a0f0383ed 100644 --- a/core/src/job/mod.rs +++ b/core/src/job/mod.rs @@ -8,10 +8,10 @@ pub mod worker; #[derive(Error, Debug)] pub enum JobError { - #[error("Failed to create job (job_id {job_id:?})")] - CreateFailure { job_id: String }, - #[error("Database error")] - DatabaseError(#[from] prisma::QueryError), + #[error("Failed to create job (job_id {job_id:?})")] + CreateFailure { job_id: String }, + #[error("Database error")] + DatabaseError(#[from] prisma::QueryError), } // pub struct JobContext { diff --git a/core/src/job/worker.rs b/core/src/job/worker.rs index 37bbcbbff..4e37d2753 100644 --- a/core/src/job/worker.rs +++ b/core/src/job/worker.rs @@ -2,185 +2,189 @@ use super::jobs::{JobReport, JobReportUpdate, JobStatus}; use crate::{ClientQuery, CoreContext, CoreEvent, InternalEvent, Job}; use std::{sync::Arc, time::Duration}; use tokio::{ - sync::{ - mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}, - Mutex, - }, - time::{sleep, Instant}, + sync::{ + mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}, + Mutex, + }, + time::{sleep, Instant}, }; // used to update the worker state from inside the worker thread pub enum WorkerEvent { - Progressed(Vec), - Completed, - Failed, + Progressed(Vec), + Completed, + Failed, } enum WorkerState { - Pending(Box, UnboundedReceiver), - Running, + Pending(Box, UnboundedReceiver), + Running, } #[derive(Clone)] pub struct WorkerContext { - pub uuid: String, - pub core_ctx: CoreContext, - pub sender: UnboundedSender, + pub uuid: String, + pub core_ctx: CoreContext, + pub sender: UnboundedSender, } impl WorkerContext { - pub fn progress(&self, updates: Vec) { - self.sender - .send(WorkerEvent::Progressed(updates)) - .unwrap_or(()); - } + pub fn progress(&self, updates: Vec) { + self + .sender + .send(WorkerEvent::Progressed(updates)) + .unwrap_or(()); + } } // a worker is a dedicated thread that runs a single job // once the job is complete the worker will exit pub struct Worker { - pub job_report: JobReport, - state: WorkerState, - worker_sender: UnboundedSender, + pub job_report: JobReport, + state: WorkerState, + worker_sender: UnboundedSender, } impl Worker { - pub fn new(job: Box) -> Self { - let (worker_sender, worker_receiver) = unbounded_channel(); - let uuid = uuid::Uuid::new_v4().to_string(); + pub fn new(job: Box) -> Self { + let (worker_sender, worker_receiver) = unbounded_channel(); + let uuid = uuid::Uuid::new_v4().to_string(); - Self { - state: WorkerState::Pending(job, worker_receiver), - job_report: JobReport::new(uuid), - worker_sender, - } - } - // spawns a thread and extracts channel sender to communicate with it - pub async fn spawn(worker: Arc>, ctx: &CoreContext) { - // we capture the worker receiver channel so state can be updated from inside the worker - let mut worker_mut = worker.lock().await; - // extract owned job and receiver from Self - let (job, worker_receiver) = - match std::mem::replace(&mut worker_mut.state, WorkerState::Running) { - WorkerState::Pending(job, worker_receiver) => { - worker_mut.state = WorkerState::Running; - (job, worker_receiver) - }, - WorkerState::Running => unreachable!(), - }; - let worker_sender = worker_mut.worker_sender.clone(); - let core_ctx = ctx.clone(); + Self { + state: WorkerState::Pending(job, worker_receiver), + job_report: JobReport::new(uuid), + worker_sender, + } + } + // spawns a thread and extracts channel sender to communicate with it + pub async fn spawn(worker: Arc>, ctx: &CoreContext) { + // we capture the worker receiver channel so state can be updated from inside the worker + let mut worker_mut = worker.lock().await; + // extract owned job and receiver from Self + let (job, worker_receiver) = + match std::mem::replace(&mut worker_mut.state, WorkerState::Running) { + WorkerState::Pending(job, worker_receiver) => { + worker_mut.state = WorkerState::Running; + (job, worker_receiver) + } + WorkerState::Running => unreachable!(), + }; + let worker_sender = worker_mut.worker_sender.clone(); + let core_ctx = ctx.clone(); - worker_mut.job_report.status = JobStatus::Running; + worker_mut.job_report.status = JobStatus::Running; - worker_mut.job_report.create(&ctx).await.unwrap_or(()); + worker_mut.job_report.create(&ctx).await.unwrap_or(()); - // spawn task to handle receiving events from the worker - tokio::spawn(Worker::track_progress( - worker.clone(), - worker_receiver, - ctx.clone(), - )); + // spawn task to handle receiving events from the worker + tokio::spawn(Worker::track_progress( + worker.clone(), + worker_receiver, + ctx.clone(), + )); - let uuid = worker_mut.job_report.id.clone(); - // spawn task to handle running the job - tokio::spawn(async move { - let worker_ctx = WorkerContext { - uuid, - core_ctx, - sender: worker_sender, - }; - let job_start = Instant::now(); + let uuid = worker_mut.job_report.id.clone(); + // spawn task to handle running the job + tokio::spawn(async move { + let worker_ctx = WorkerContext { + uuid, + core_ctx, + sender: worker_sender, + }; + let job_start = Instant::now(); - // track time - let sender = worker_ctx.sender.clone(); - tokio::spawn(async move { - loop { - let elapsed = job_start.elapsed().as_secs(); - sender - .send(WorkerEvent::Progressed(vec![ - JobReportUpdate::SecondsElapsed(elapsed), - ])) - .unwrap_or(()); - sleep(Duration::from_millis(1000)).await; - } - }); + // track time + let sender = worker_ctx.sender.clone(); + tokio::spawn(async move { + loop { + let elapsed = job_start.elapsed().as_secs(); + sender + .send(WorkerEvent::Progressed(vec![ + JobReportUpdate::SecondsElapsed(elapsed), + ])) + .unwrap_or(()); + sleep(Duration::from_millis(1000)).await; + } + }); - let result = job.run(worker_ctx.clone()).await; + let result = job.run(worker_ctx.clone()).await; - if let Err(e) = result { - println!("job failed {:?}", e); - worker_ctx.sender.send(WorkerEvent::Failed).unwrap_or(()); - } else { - // handle completion - worker_ctx.sender.send(WorkerEvent::Completed).unwrap_or(()); - } - worker_ctx - .core_ctx - .internal_sender - .send(InternalEvent::JobComplete(worker_ctx.uuid.clone())) - .unwrap_or(()); - }); - } + if let Err(e) = result { + println!("job failed {:?}", e); + worker_ctx.sender.send(WorkerEvent::Failed).unwrap_or(()); + } else { + // handle completion + worker_ctx.sender.send(WorkerEvent::Completed).unwrap_or(()); + } + worker_ctx + .core_ctx + .internal_sender + .send(InternalEvent::JobComplete(worker_ctx.uuid.clone())) + .unwrap_or(()); + }); + } - pub fn id(&self) -> String { - self.job_report.id.to_owned() - } + pub fn id(&self) -> String { + self.job_report.id.to_owned() + } - async fn track_progress( - worker: Arc>, - mut channel: UnboundedReceiver, - ctx: CoreContext, - ) { - while let Some(command) = channel.recv().await { - let mut worker = worker.lock().await; + async fn track_progress( + worker: Arc>, + mut channel: UnboundedReceiver, + ctx: CoreContext, + ) { + while let Some(command) = channel.recv().await { + let mut worker = worker.lock().await; - match command { - WorkerEvent::Progressed(changes) => { - // protect against updates if job is not running - if worker.job_report.status != JobStatus::Running { - continue; - }; - for change in changes { - match change { - JobReportUpdate::TaskCount(task_count) => { - worker.job_report.task_count = task_count as i32; - }, - JobReportUpdate::CompletedTaskCount(completed_task_count) => { - worker.job_report.completed_task_count = - completed_task_count as i32; - }, - JobReportUpdate::Message(message) => { - worker.job_report.message = message; - }, - JobReportUpdate::SecondsElapsed(seconds) => { - worker.job_report.seconds_elapsed = seconds as i32; - }, - } - } - ctx.emit(CoreEvent::InvalidateQueryDebounced( - ClientQuery::JobGetRunning, - )) - .await; - }, - WorkerEvent::Completed => { - worker.job_report.status = JobStatus::Completed; - worker.job_report.update(&ctx).await.unwrap_or(()); + match command { + WorkerEvent::Progressed(changes) => { + // protect against updates if job is not running + if worker.job_report.status != JobStatus::Running { + continue; + }; + for change in changes { + match change { + JobReportUpdate::TaskCount(task_count) => { + worker.job_report.task_count = task_count as i32; + } + JobReportUpdate::CompletedTaskCount(completed_task_count) => { + worker.job_report.completed_task_count = completed_task_count as i32; + } + JobReportUpdate::Message(message) => { + worker.job_report.message = message; + } + JobReportUpdate::SecondsElapsed(seconds) => { + worker.job_report.seconds_elapsed = seconds as i32; + } + } + } + ctx + .emit(CoreEvent::InvalidateQueryDebounced( + ClientQuery::JobGetRunning, + )) + .await; + } + WorkerEvent::Completed => { + worker.job_report.status = JobStatus::Completed; + worker.job_report.update(&ctx).await.unwrap_or(()); - ctx.emit(CoreEvent::InvalidateQuery(ClientQuery::JobGetRunning)) - .await; - ctx.emit(CoreEvent::InvalidateQuery(ClientQuery::JobGetHistory)) - .await; - break; - }, - WorkerEvent::Failed => { - worker.job_report.status = JobStatus::Failed; - worker.job_report.update(&ctx).await.unwrap_or(()); + ctx + .emit(CoreEvent::InvalidateQuery(ClientQuery::JobGetRunning)) + .await; + ctx + .emit(CoreEvent::InvalidateQuery(ClientQuery::JobGetHistory)) + .await; + break; + } + WorkerEvent::Failed => { + worker.job_report.status = JobStatus::Failed; + worker.job_report.update(&ctx).await.unwrap_or(()); - ctx.emit(CoreEvent::InvalidateQuery(ClientQuery::JobGetHistory)) - .await; - break; - }, - } - } - } + ctx + .emit(CoreEvent::InvalidateQuery(ClientQuery::JobGetHistory)) + .await; + break; + } + } + } + } } diff --git a/core/src/lib.rs b/core/src/lib.rs index f8157128c..b1584fecc 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -6,8 +6,8 @@ use state::client::ClientState; use std::{fs, path::Path, sync::Arc}; use thiserror::Error; use tokio::sync::{ - mpsc::{self, unbounded_channel, UnboundedReceiver, UnboundedSender}, - oneshot, + mpsc::{self, unbounded_channel, UnboundedReceiver, UnboundedSender}, + oneshot, }; use ts_rs::TS; @@ -29,288 +29,269 @@ pub mod util; // a wrapper around external input with a returning sender channel for core to respond pub struct ReturnableMessage> { - data: D, - return_sender: oneshot::Sender, + data: D, + return_sender: oneshot::Sender, } // core controller is passed to the client to communicate with the core which runs in a dedicated thread pub struct CoreController { - query_sender: UnboundedSender>, - command_sender: UnboundedSender>, + query_sender: UnboundedSender>, + command_sender: UnboundedSender>, } impl CoreController { - pub async fn query(&self, query: ClientQuery) -> Result { - // a one time use channel to send and await a response - let (sender, recv) = oneshot::channel(); - self.query_sender - .send(ReturnableMessage { - data: query, - return_sender: sender, - }) - .unwrap_or(()); - // wait for response and return - recv.await.unwrap() - } + pub async fn query(&self, query: ClientQuery) -> Result { + // a one time use channel to send and await a response + let (sender, recv) = oneshot::channel(); + self + .query_sender + .send(ReturnableMessage { + data: query, + return_sender: sender, + }) + .unwrap_or(()); + // wait for response and return + recv.await.unwrap() + } - pub async fn command( - &self, - command: ClientCommand, - ) -> Result { - let (sender, recv) = oneshot::channel(); - self.command_sender - .send(ReturnableMessage { - data: command, - return_sender: sender, - }) - .unwrap_or(()); + pub async fn command(&self, command: ClientCommand) -> Result { + let (sender, recv) = oneshot::channel(); + self + .command_sender + .send(ReturnableMessage { + data: command, + return_sender: sender, + }) + .unwrap_or(()); - recv.await.unwrap() - } + recv.await.unwrap() + } } #[derive(Debug)] pub enum InternalEvent { - JobIngest(Box), - JobComplete(String), + JobIngest(Box), + JobComplete(String), } #[derive(Clone)] pub struct CoreContext { - pub database: Arc, - pub event_sender: mpsc::Sender, - pub internal_sender: UnboundedSender, + pub database: Arc, + pub event_sender: mpsc::Sender, + pub internal_sender: UnboundedSender, } impl CoreContext { - pub fn spawn_job(&self, job: Box) { - self.internal_sender - .send(InternalEvent::JobIngest(job)) - .unwrap_or_else(|e| { - error!("Failed to spawn job. {:?}", e); - }); - } - pub async fn emit(&self, event: CoreEvent) { - self.event_sender.send(event).await.unwrap_or_else(|e| { - error!("Failed to emit event. {:?}", e); - }); - } + pub fn spawn_job(&self, job: Box) { + self + .internal_sender + .send(InternalEvent::JobIngest(job)) + .unwrap_or_else(|e| { + error!("Failed to spawn job. {:?}", e); + }); + } + pub async fn emit(&self, event: CoreEvent) { + self.event_sender.send(event).await.unwrap_or_else(|e| { + error!("Failed to emit event. {:?}", e); + }); + } } pub struct Core { - state: ClientState, - jobs: job::jobs::Jobs, - database: Arc, - // filetype_registry: library::TypeRegistry, - // extension_registry: library::ExtensionRegistry, + state: ClientState, + jobs: job::jobs::Jobs, + database: Arc, + // filetype_registry: library::TypeRegistry, + // extension_registry: library::ExtensionRegistry, - // global messaging channels - query_channel: ( - UnboundedSender>, - UnboundedReceiver>, - ), - command_channel: ( - UnboundedSender>, - UnboundedReceiver>, - ), - event_sender: mpsc::Sender, + // global messaging channels + query_channel: ( + UnboundedSender>, + UnboundedReceiver>, + ), + command_channel: ( + UnboundedSender>, + UnboundedReceiver>, + ), + event_sender: mpsc::Sender, - // a channel for child threads to send events back to the core - internal_channel: ( - UnboundedSender, - UnboundedReceiver, - ), + // a channel for child threads to send events back to the core + internal_channel: ( + UnboundedSender, + UnboundedReceiver, + ), } impl Core { - // create new instance of core, run startup tasks - pub async fn new( - mut data_dir: std::path::PathBuf, - ) -> (Core, mpsc::Receiver) { - let (event_sender, event_recv) = mpsc::channel(100); + // create new instance of core, run startup tasks + pub async fn new(mut data_dir: std::path::PathBuf) -> (Core, mpsc::Receiver) { + let (event_sender, event_recv) = mpsc::channel(100); - data_dir = data_dir.join("spacedrive"); - let data_dir = data_dir.to_str().unwrap(); - // create data directory if it doesn't exist - fs::create_dir_all(&data_dir).unwrap(); - // prepare basic client state - let mut state = - ClientState::new(data_dir, "diamond-mastering-space-dragon").unwrap(); - // load from disk - state - .read_disk() - .unwrap_or(error!("No client state found, creating new one...")); + data_dir = data_dir.join("spacedrive"); + let data_dir = data_dir.to_str().unwrap(); + // create data directory if it doesn't exist + fs::create_dir_all(&data_dir).unwrap(); + // prepare basic client state + let mut state = ClientState::new(data_dir, "diamond-mastering-space-dragon").unwrap(); + // load from disk + state + .read_disk() + .unwrap_or(error!("No client state found, creating new one...")); - state.save(); + state.save(); - let database = Arc::new(db::create_connection().await.unwrap()); + let database = Arc::new(db::create_connection().await.unwrap()); - let internal_channel = unbounded_channel::(); + let internal_channel = unbounded_channel::(); - let core = Core { - state, - query_channel: unbounded_channel(), - command_channel: unbounded_channel(), - jobs: Jobs::new(), - event_sender, - database, - internal_channel, - }; + let core = Core { + state, + query_channel: unbounded_channel(), + command_channel: unbounded_channel(), + jobs: Jobs::new(), + event_sender, + database, + internal_channel, + }; - (core, event_recv) - } + (core, event_recv) + } - pub fn get_context(&self) -> CoreContext { - CoreContext { - database: self.database.clone(), - event_sender: self.event_sender.clone(), - internal_sender: self.internal_channel.0.clone(), - } - } + pub fn get_context(&self) -> CoreContext { + CoreContext { + database: self.database.clone(), + event_sender: self.event_sender.clone(), + internal_sender: self.internal_channel.0.clone(), + } + } - pub fn get_controller(&self) -> CoreController { - CoreController { - query_sender: self.query_channel.0.clone(), - command_sender: self.command_channel.0.clone(), - } - } + pub fn get_controller(&self) -> CoreController { + CoreController { + query_sender: self.query_channel.0.clone(), + command_sender: self.command_channel.0.clone(), + } + } - pub async fn start(&mut self) { - let ctx = self.get_context(); - loop { - // listen on global messaging channels for incoming messages - tokio::select! { - Some(msg) = self.query_channel.1.recv() => { - let res = self.exec_query(msg.data).await; - msg.return_sender.send(res).unwrap_or(()); - } - Some(msg) = self.command_channel.1.recv() => { - let res = self.exec_command(msg.data).await; - msg.return_sender.send(res).unwrap_or(()); - } - Some(event) = self.internal_channel.1.recv() => { - match event { - InternalEvent::JobIngest(job) => { - self.jobs.ingest(&ctx, job).await; - }, - InternalEvent::JobComplete(id) => { - self.jobs.complete(id); - }, - } - } - } - } - } - // load library database + initialize client with db - pub async fn initializer(&self) { - if self.state.libraries.len() == 0 { - match library::loader::create(&self, None).await { - Ok(library) => info!("Created new library: {:?}", library), - Err(e) => info!("Error creating library: {:?}", e), - } - } else { - for library in self.state.libraries.iter() { - // init database for library - match library::loader::load(&library.library_path, &library.library_uuid) - .await - { - Ok(library) => info!("Loaded library: {:?}", library), - Err(e) => info!("Error loading library: {:?}", e), - } - } - } - // init client - match client::create(&self).await { - Ok(_) => info!("Spacedrive online"), - Err(e) => info!("Error initializing client: {:?}", e), - }; - } + pub async fn start(&mut self) { + let ctx = self.get_context(); + loop { + // listen on global messaging channels for incoming messages + tokio::select! { + Some(msg) = self.query_channel.1.recv() => { + let res = self.exec_query(msg.data).await; + msg.return_sender.send(res).unwrap_or(()); + } + Some(msg) = self.command_channel.1.recv() => { + let res = self.exec_command(msg.data).await; + msg.return_sender.send(res).unwrap_or(()); + } + Some(event) = self.internal_channel.1.recv() => { + match event { + InternalEvent::JobIngest(job) => { + self.jobs.ingest(&ctx, job).await; + }, + InternalEvent::JobComplete(id) => { + self.jobs.complete(id); + }, + } + } + } + } + } + // load library database + initialize client with db + pub async fn initializer(&self) { + if self.state.libraries.len() == 0 { + match library::loader::create(&self, None).await { + Ok(library) => info!("Created new library: {:?}", library), + Err(e) => info!("Error creating library: {:?}", e), + } + } else { + for library in self.state.libraries.iter() { + // init database for library + match library::loader::load(&library.library_path, &library.library_uuid).await { + Ok(library) => info!("Loaded library: {:?}", library), + Err(e) => info!("Error loading library: {:?}", e), + } + } + } + // init client + match client::create(&self).await { + Ok(_) => info!("Spacedrive online"), + Err(e) => info!("Error initializing client: {:?}", e), + }; + } - async fn exec_command( - &mut self, - cmd: ClientCommand, - ) -> Result { - info!("Core command: {:?}", cmd); - let ctx = self.get_context(); - Ok(match cmd { - // CRUD for locations - ClientCommand::LocCreate { path } => CoreResponse::LocCreate( - sys::locations::new_location_and_scan(&ctx, &path).await?, - ), - ClientCommand::LocUpdate { id: _, name: _ } => todo!(), - ClientCommand::LocDelete { id: _ } => todo!(), - // CRUD for files - ClientCommand::FileRead { id: _ } => todo!(), - // ClientCommand::FileEncrypt { id: _, algorithm: _ } => todo!(), - ClientCommand::FileDelete { id: _ } => todo!(), - // CRUD for tags - ClientCommand::TagCreate { name: _, color: _ } => todo!(), - ClientCommand::TagAssign { - file_id: _, - tag_id: _, - } => todo!(), - ClientCommand::TagDelete { id: _ } => todo!(), - // CRUD for libraries - ClientCommand::SysVolumeUnmount { id: _ } => todo!(), - ClientCommand::LibDelete { id: _ } => todo!(), - ClientCommand::TagUpdate { name: _, color: _ } => todo!(), - ClientCommand::GenerateThumbsForLocation { id } => { - ctx.spawn_job(Box::new(ThumbnailJob { location_id: id })); - CoreResponse::Success(()) - }, - ClientCommand::PurgeDatabase => { - println!("Purging database..."); - fs::remove_file(Path::new(&self.state.data_path).join("library.db")) - .unwrap(); - CoreResponse::Success(()) - }, - }) - } + async fn exec_command(&mut self, cmd: ClientCommand) -> Result { + info!("Core command: {:?}", cmd); + let ctx = self.get_context(); + Ok(match cmd { + // CRUD for locations + ClientCommand::LocCreate { path } => { + CoreResponse::LocCreate(sys::locations::new_location_and_scan(&ctx, &path).await?) + } + ClientCommand::LocUpdate { id: _, name: _ } => todo!(), + ClientCommand::LocDelete { id: _ } => todo!(), + // CRUD for files + ClientCommand::FileRead { id: _ } => todo!(), + // ClientCommand::FileEncrypt { id: _, algorithm: _ } => todo!(), + ClientCommand::FileDelete { id: _ } => todo!(), + // CRUD for tags + ClientCommand::TagCreate { name: _, color: _ } => todo!(), + ClientCommand::TagAssign { + file_id: _, + tag_id: _, + } => todo!(), + ClientCommand::TagDelete { id: _ } => todo!(), + // CRUD for libraries + ClientCommand::SysVolumeUnmount { id: _ } => todo!(), + ClientCommand::LibDelete { id: _ } => todo!(), + ClientCommand::TagUpdate { name: _, color: _ } => todo!(), + ClientCommand::GenerateThumbsForLocation { id } => { + ctx.spawn_job(Box::new(ThumbnailJob { location_id: id })); + CoreResponse::Success(()) + } + ClientCommand::PurgeDatabase => { + println!("Purging database..."); + fs::remove_file(Path::new(&self.state.data_path).join("library.db")).unwrap(); + CoreResponse::Success(()) + } + }) + } - // query sources of data - async fn exec_query(&self, query: ClientQuery) -> Result { - info!("Core query: {:?}", query); - let ctx = self.get_context(); - Ok(match query { - // return the client state from memory - ClientQuery::ClientGetState => { - CoreResponse::ClientGetState(self.state.clone()) - }, - // get system volumes without saving to library - ClientQuery::SysGetVolumes => { - CoreResponse::SysGetVolumes(sys::volumes::get_volumes()?) - }, - // get location from library - ClientQuery::SysGetLocation { id } => CoreResponse::SysGetLocation( - sys::locations::get_location(&ctx, id).await?, - ), - // return contents of a directory for the explorer - ClientQuery::LibGetExplorerDir { path, limit: _ } => { - CoreResponse::LibGetExplorerDir( - file::explorer::open_dir(&ctx, &path).await?, - ) - }, - ClientQuery::LibGetTags => todo!(), - ClientQuery::JobGetRunning => { - CoreResponse::JobGetRunning(self.jobs.get_running().await) - }, - // TODO: FIX THIS - ClientQuery::JobGetHistory => { - CoreResponse::JobGetHistory(Jobs::get_history(&ctx).await?) - }, - }) - } + // query sources of data + async fn exec_query(&self, query: ClientQuery) -> Result { + info!("Core query: {:?}", query); + let ctx = self.get_context(); + Ok(match query { + // return the client state from memory + ClientQuery::ClientGetState => CoreResponse::ClientGetState(self.state.clone()), + // get system volumes without saving to library + ClientQuery::SysGetVolumes => CoreResponse::SysGetVolumes(sys::volumes::get_volumes()?), + // get location from library + ClientQuery::SysGetLocation { id } => { + CoreResponse::SysGetLocation(sys::locations::get_location(&ctx, id).await?) + } + // return contents of a directory for the explorer + ClientQuery::LibGetExplorerDir { path, limit: _ } => { + CoreResponse::LibGetExplorerDir(file::explorer::open_dir(&ctx, &path).await?) + } + ClientQuery::LibGetTags => todo!(), + ClientQuery::JobGetRunning => CoreResponse::JobGetRunning(self.jobs.get_running().await), + // TODO: FIX THIS + ClientQuery::JobGetHistory => CoreResponse::JobGetHistory(Jobs::get_history(&ctx).await?), + }) + } - // pub fn queue(&mut self, job: JobResource) -> &mut JobResource { - // self.job_runner.queued_jobs.push(job); - // self.job_runner.queued_jobs.last_mut().unwrap() - // } - // send an event to the client - // async fn emit_event(&mut self, event: ClientEvent) {} + // pub fn queue(&mut self, job: JobResource) -> &mut JobResource { + // self.job_runner.queued_jobs.push(job); + // self.job_runner.queued_jobs.last_mut().unwrap() + // } + // send an event to the client + // async fn emit_event(&mut self, event: ClientEvent) {} - // pub async fn send(&self, event: CoreEvent) { - // // self.event_channel.1.send(event).await; - // } + // pub async fn send(&self, event: CoreEvent) { + // // self.event_channel.1.send(event).await; + // } } // represents an event this library can emit @@ -318,25 +299,25 @@ impl Core { #[serde(tag = "key", content = "params")] #[ts(export)] pub enum ClientCommand { - // Files - FileRead { id: i32 }, - // FileEncrypt { id: i32, algorithm: EncryptionAlgorithm }, - FileDelete { id: i32 }, - // Library - LibDelete { id: i32 }, - // Tags - TagCreate { name: String, color: String }, - TagUpdate { name: String, color: String }, - TagAssign { file_id: i32, tag_id: i32 }, - TagDelete { id: i32 }, - // Locations - LocCreate { path: String }, - LocUpdate { id: i32, name: Option }, - LocDelete { id: i32 }, - // System - SysVolumeUnmount { id: i32 }, - GenerateThumbsForLocation { id: i32 }, - PurgeDatabase, + // Files + FileRead { id: i32 }, + // FileEncrypt { id: i32, algorithm: EncryptionAlgorithm }, + FileDelete { id: i32 }, + // Library + LibDelete { id: i32 }, + // Tags + TagCreate { name: String, color: String }, + TagUpdate { name: String, color: String }, + TagAssign { file_id: i32, tag_id: i32 }, + TagDelete { id: i32 }, + // Locations + LocCreate { path: String }, + LocUpdate { id: i32, name: Option }, + LocDelete { id: i32 }, + // System + SysVolumeUnmount { id: i32 }, + GenerateThumbsForLocation { id: i32 }, + PurgeDatabase, } // represents an event this library can emit @@ -344,13 +325,13 @@ pub enum ClientCommand { #[serde(tag = "key", content = "params")] #[ts(export)] pub enum ClientQuery { - ClientGetState, - SysGetVolumes, - LibGetTags, - JobGetRunning, - JobGetHistory, - SysGetLocation { id: i32 }, - LibGetExplorerDir { path: String, limit: i32 }, + ClientGetState, + SysGetVolumes, + LibGetTags, + JobGetRunning, + JobGetHistory, + SysGetLocation { id: i32 }, + LibGetExplorerDir { path: String, limit: i32 }, } // represents an event this library can emit @@ -358,49 +339,49 @@ pub enum ClientQuery { #[serde(tag = "key", content = "data")] #[ts(export)] pub enum CoreEvent { - // most all events should be once of these two - InvalidateQuery(ClientQuery), - InvalidateQueryDebounced(ClientQuery), - InvalidateResource(CoreResource), - Log { message: String }, - DatabaseDisconnected { reason: Option }, + // most all events should be once of these two + InvalidateQuery(ClientQuery), + InvalidateQueryDebounced(ClientQuery), + InvalidateResource(CoreResource), + Log { message: String }, + DatabaseDisconnected { reason: Option }, } #[derive(Serialize, Deserialize, Debug, TS)] #[serde(tag = "key", content = "data")] #[ts(export)] pub enum CoreResponse { - Success(()), - SysGetVolumes(Vec), - SysGetLocation(sys::locations::LocationResource), - LibGetExplorerDir(file::DirectoryWithContents), - ClientGetState(ClientState), - LocCreate(sys::locations::LocationResource), - JobGetRunning(Vec), - JobGetHistory(Vec), + Success(()), + SysGetVolumes(Vec), + SysGetLocation(sys::locations::LocationResource), + LibGetExplorerDir(file::DirectoryWithContents), + ClientGetState(ClientState), + LocCreate(sys::locations::LocationResource), + JobGetRunning(Vec), + JobGetHistory(Vec), } #[derive(Error, Debug)] pub enum CoreError { - #[error("Query error")] - QueryError, - #[error("System error")] - SysError(#[from] sys::SysError), - #[error("File error")] - FileError(#[from] file::FileError), - #[error("Job error")] - JobError(#[from] job::JobError), - #[error("Database error")] - DatabaseError(#[from] prisma::QueryError), + #[error("Query error")] + QueryError, + #[error("System error")] + SysError(#[from] sys::SysError), + #[error("File error")] + FileError(#[from] file::FileError), + #[error("Job error")] + JobError(#[from] job::JobError), + #[error("Database error")] + DatabaseError(#[from] prisma::QueryError), } #[derive(Serialize, Deserialize, Debug, TS)] #[ts(export)] pub enum CoreResource { - Client, - Library, - Location(sys::locations::LocationResource), - File(file::File), - Job(JobReport), - Tag, + Client, + Library, + Location(sys::locations::LocationResource), + File(file::File), + Job(JobReport), + Tag, } diff --git a/core/src/library/loader.rs b/core/src/library/loader.rs index 6a551f87b..9ccf9448a 100644 --- a/core/src/library/loader.rs +++ b/core/src/library/loader.rs @@ -4,9 +4,9 @@ use uuid::Uuid; use crate::state::client::LibraryState; use crate::{ - db::migrate, - prisma::{Library, LibraryData}, - state, + db::migrate, + prisma::{Library, LibraryData}, + state, }; use crate::{prisma, Core}; @@ -15,84 +15,84 @@ pub static DEFAULT_NAME: &str = "My Library"; #[derive(Error, Debug)] pub enum LibraryError { - #[error("Database error")] - DatabaseError(#[from] prisma::QueryError), + #[error("Database error")] + DatabaseError(#[from] prisma::QueryError), } pub async fn get(core: &Core) -> Result { - let config = state::client::get(); - let db = &core.database; + let config = state::client::get(); + let db = &core.database; - let library_state = config.get_current_library(); + let library_state = config.get_current_library(); - println!("{:?}", library_state); + println!("{:?}", library_state); - // get library from db - let library = match db - .library() - .find_unique(Library::uuid().equals(library_state.library_uuid.clone())) - .exec() - .await? - { - Some(library) => Ok(library), - None => { - // update config library state to offline - // config.libraries + // get library from db + let library = match db + .library() + .find_unique(Library::uuid().equals(library_state.library_uuid.clone())) + .exec() + .await? + { + Some(library) => Ok(library), + None => { + // update config library state to offline + // config.libraries - Err(anyhow::anyhow!("library_not_found")) - }, - }; + Err(anyhow::anyhow!("library_not_found")) + } + }; - Ok(library.unwrap()) + Ok(library.unwrap()) } pub async fn load(library_path: &str, library_id: &str) -> Result<()> { - let mut config = state::client::get(); + let mut config = state::client::get(); - println!("Initializing library: {} {}", &library_id, library_path); + println!("Initializing library: {} {}", &library_id, library_path); - if config.current_library_uuid != library_id { - config.current_library_uuid = library_id.to_string(); - config.save(); - } - // create connection with library database & run migrations - migrate::run_migrations(&library_path).await?; - // if doesn't exist, mark as offline - Ok(()) + if config.current_library_uuid != library_id { + config.current_library_uuid = library_id.to_string(); + config.save(); + } + // create connection with library database & run migrations + migrate::run_migrations(&library_path).await?; + // if doesn't exist, mark as offline + Ok(()) } pub async fn create(core: &Core, name: Option) -> Result<()> { - let mut config = state::client::get(); + let mut config = state::client::get(); - let uuid = Uuid::new_v4().to_string(); + let uuid = Uuid::new_v4().to_string(); - println!("Creating library {:?}, UUID: {:?}", name, uuid); + println!("Creating library {:?}, UUID: {:?}", name, uuid); - let library_state = LibraryState { - library_uuid: uuid.clone(), - library_path: format!("{}/{}", config.data_path, LIBRARY_DB_NAME), - ..LibraryState::default() - }; + let library_state = LibraryState { + library_uuid: uuid.clone(), + library_path: format!("{}/{}", config.data_path, LIBRARY_DB_NAME), + ..LibraryState::default() + }; - migrate::run_migrations(&library_state.library_path).await?; + migrate::run_migrations(&library_state.library_path).await?; - config.libraries.push(library_state); + config.libraries.push(library_state); - config.current_library_uuid = uuid; + config.current_library_uuid = uuid; - config.save(); + config.save(); - let db = &core.database; + let db = &core.database; - let _library = db - .library() - .create_one( - Library::uuid().set(config.current_library_uuid), - Library::name().set(name.unwrap_or(DEFAULT_NAME.into())), - vec![], - ) - .exec() - .await; + let _library = db + .library() + .create_one( + Library::uuid().set(config.current_library_uuid), + Library::name().set(name.unwrap_or(DEFAULT_NAME.into())), + vec![], + ) + .exec() + .await; - Ok(()) + Ok(()) } diff --git a/core/src/p2p/listener.rs b/core/src/p2p/listener.rs index 3ed8132b4..71659b206 100644 --- a/core/src/p2p/listener.rs +++ b/core/src/p2p/listener.rs @@ -1,48 +1,48 @@ use futures::StreamExt; use libp2p::{ - identity, ping, - swarm::{Swarm, SwarmEvent}, - Multiaddr, PeerId, + identity, ping, + swarm::{Swarm, SwarmEvent}, + Multiaddr, PeerId, }; use std::error::Error; pub async fn listen(port: Option) -> Result<(), Box> { - let local_key = identity::Keypair::generate_ed25519(); - let local_peer_id = PeerId::from(local_key.public()); - println!("Local peer id: {:?}", local_peer_id); + let local_key = identity::Keypair::generate_ed25519(); + let local_peer_id = PeerId::from(local_key.public()); + println!("Local peer id: {:?}", local_peer_id); - let transport = libp2p::development_transport(local_key).await?; + let transport = libp2p::development_transport(local_key).await?; - // Create a ping network behavior. - // - // For illustrative purposes, the ping protocol is configured to - // keep the connection alive, so a continuous sequence of pings - // can be observed. - let behavior = ping::Behaviour::new(ping::Config::new().with_keep_alive(true)); + // Create a ping network behavior. + // + // For illustrative purposes, the ping protocol is configured to + // keep the connection alive, so a continuous sequence of pings + // can be observed. + let behavior = ping::Behaviour::new(ping::Config::new().with_keep_alive(true)); - let mut swarm = Swarm::new(transport, behavior, local_peer_id); + let mut swarm = Swarm::new(transport, behavior, local_peer_id); - // Tell the swarm to listen on all interfaces and a random, OS-assigned - // port. - swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?; + // Tell the swarm to listen on all interfaces and a random, OS-assigned + // port. + swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?; - // Dial the peer identified by the multi-address given as the second - // command-line argument, if any. + // Dial the peer identified by the multi-address given as the second + // command-line argument, if any. - if port.is_some() { - let addr = format!("{:?}{:?}", "/ip4/127.0.0.1/tcp/", port); - let remote: Multiaddr = addr.parse()?; - swarm.dial(remote)?; - println!("Dialed {}", addr) - } + if port.is_some() { + let addr = format!("{:?}{:?}", "/ip4/127.0.0.1/tcp/", port); + let remote: Multiaddr = addr.parse()?; + swarm.dial(remote)?; + println!("Dialed {}", addr) + } - loop { - match swarm.select_next_some().await { - SwarmEvent::NewListenAddr { address, .. } => { - println!("Listening on {:?}", address) - }, - SwarmEvent::Behaviour(event) => println!("{:?}", event), - _ => {}, - } - } + loop { + match swarm.select_next_some().await { + SwarmEvent::NewListenAddr { address, .. } => { + println!("Listening on {:?}", address) + } + SwarmEvent::Behaviour(event) => println!("{:?}", event), + _ => {} + } + } } diff --git a/core/src/state/client.rs b/core/src/state/client.rs index 2770a319e..1dd364934 100644 --- a/core/src/state/client.rs +++ b/core/src/state/client.rs @@ -10,18 +10,18 @@ use uuid::Uuid; #[derive(Debug, Serialize, Deserialize, Clone, Default, TS)] #[ts(export)] pub struct ClientState { - // client id is a uniquely generated UUID - pub client_uuid: String, - // client_name is the name of the device running the client - pub client_name: String, - // config path is stored as struct can exist only in memory during startup and be written to disk later without supplying path - pub data_path: String, - // the port this client uses to listen for incoming connections - pub tcp_port: u32, - // all the libraries loaded by this client - pub libraries: Vec, - // used to quickly find the default library - pub current_library_uuid: String, + // client id is a uniquely generated UUID + pub client_uuid: String, + // client_name is the name of the device running the client + pub client_name: String, + // config path is stored as struct can exist only in memory during startup and be written to disk later without supplying path + pub data_path: String, + // the port this client uses to listen for incoming connections + pub tcp_port: u32, + // all the libraries loaded by this client + pub libraries: Vec, + // used to quickly find the default library + pub current_library_uuid: String, } pub static CLIENT_STATE_CONFIG_NAME: &str = "client_state.json"; @@ -29,75 +29,75 @@ pub static CLIENT_STATE_CONFIG_NAME: &str = "client_state.json"; #[derive(Debug, Serialize, Deserialize, Clone, Default, TS)] #[ts(export)] pub struct LibraryState { - pub library_uuid: String, - pub library_path: String, - pub offline: bool, + pub library_uuid: String, + pub library_path: String, + pub offline: bool, } // global, thread-safe storage for client state lazy_static! { - static ref CONFIG: RwLock> = RwLock::new(None); + static ref CONFIG: RwLock> = RwLock::new(None); } pub fn get() -> ClientState { - match CONFIG.read() { - Ok(guard) => guard.clone().unwrap_or(ClientState::default()), - Err(_) => return ClientState::default(), - } + match CONFIG.read() { + Ok(guard) => guard.clone().unwrap_or(ClientState::default()), + Err(_) => return ClientState::default(), + } } impl ClientState { - pub fn new(data_path: &str, client_name: &str) -> Result { - let uuid = Uuid::new_v4().to_string(); - // create struct and assign defaults - let config = Self { - client_uuid: uuid, - data_path: data_path.to_string(), - client_name: client_name.to_string(), - ..Default::default() - }; - Ok(config) - } + pub fn new(data_path: &str, client_name: &str) -> Result { + let uuid = Uuid::new_v4().to_string(); + // create struct and assign defaults + let config = Self { + client_uuid: uuid, + data_path: data_path.to_string(), + client_name: client_name.to_string(), + ..Default::default() + }; + Ok(config) + } - pub fn save(&self) { - self.write_memory(); - // only write to disk if config path is set - if !&self.data_path.is_empty() { - let config_path = format!("{}/{}", &self.data_path, CLIENT_STATE_CONFIG_NAME); - let mut file = fs::File::create(config_path).unwrap(); - let json = serde_json::to_string(&self).unwrap(); - file.write_all(json.as_bytes()).unwrap(); - } - } + pub fn save(&self) { + self.write_memory(); + // only write to disk if config path is set + if !&self.data_path.is_empty() { + let config_path = format!("{}/{}", &self.data_path, CLIENT_STATE_CONFIG_NAME); + let mut file = fs::File::create(config_path).unwrap(); + let json = serde_json::to_string(&self).unwrap(); + file.write_all(json.as_bytes()).unwrap(); + } + } - pub fn read_disk(&mut self) -> Result<()> { - let config_path = format!("{}/{}", &self.data_path, CLIENT_STATE_CONFIG_NAME); - // open the file and parse json - let file = fs::File::open(config_path)?; - let reader = BufReader::new(file); - let data = serde_json::from_reader(reader)?; - // assign to self - *self = data; - Ok(()) - } + pub fn read_disk(&mut self) -> Result<()> { + let config_path = format!("{}/{}", &self.data_path, CLIENT_STATE_CONFIG_NAME); + // open the file and parse json + let file = fs::File::open(config_path)?; + let reader = BufReader::new(file); + let data = serde_json::from_reader(reader)?; + // assign to self + *self = data; + Ok(()) + } - fn write_memory(&self) { - let mut writeable = CONFIG.write().unwrap(); - *writeable = Some(self.clone()); - } + fn write_memory(&self) { + let mut writeable = CONFIG.write().unwrap(); + *writeable = Some(self.clone()); + } - pub fn get_current_library(&self) -> LibraryState { - match self - .libraries - .iter() - .find(|lib| lib.library_uuid == self.current_library_uuid) - { - Some(lib) => lib.clone(), - None => LibraryState::default(), - } - } + pub fn get_current_library(&self) -> LibraryState { + match self + .libraries + .iter() + .find(|lib| lib.library_uuid == self.current_library_uuid) + { + Some(lib) => lib.clone(), + None => LibraryState::default(), + } + } - pub fn get_current_library_db_path(&self) -> String { - format!("{}/library.db", &self.get_current_library().library_path) - } + pub fn get_current_library_db_path(&self) -> String { + format!("{}/library.db", &self.get_current_library().library_path) + } } diff --git a/core/src/sys/locations.rs b/core/src/sys/locations.rs index 71ff0cbdb..f65694cc2 100644 --- a/core/src/sys/locations.rs +++ b/core/src/sys/locations.rs @@ -1,9 +1,9 @@ use crate::{ - file::indexer::IndexerJob, - prisma::{FilePath, Location}, - state::client, - sys::{volumes, volumes::Volume}, - CoreContext, + file::indexer::IndexerJob, + prisma::{FilePath, Location}, + state::client, + sys::{volumes, volumes::Volume}, + CoreContext, }; use anyhow::Result; use log::info; @@ -19,40 +19,40 @@ use super::SysError; #[derive(Debug, Clone, Serialize, Deserialize, TS)] #[ts(export)] pub struct LocationResource { - pub id: i32, - pub name: Option, - pub path: Option, - pub total_capacity: Option, - pub available_capacity: Option, - pub is_removable: bool, - pub is_ejectable: bool, - pub is_root_filesystem: bool, - pub is_online: bool, - #[ts(type = "string")] - pub date_created: chrono::DateTime, + pub id: i32, + pub name: Option, + pub path: Option, + pub total_capacity: Option, + pub available_capacity: Option, + pub is_removable: bool, + pub is_ejectable: bool, + pub is_root_filesystem: bool, + pub is_online: bool, + #[ts(type = "string")] + pub date_created: chrono::DateTime, } impl Into for LocationData { - fn into(self) -> LocationResource { - LocationResource { - id: self.id, - name: self.name, - path: self.local_path, - total_capacity: self.total_capacity, - available_capacity: self.available_capacity, - is_removable: self.is_removable, - is_ejectable: self.is_ejectable, - is_root_filesystem: self.is_root_filesystem, - is_online: self.is_online, - date_created: self.date_created, - } - } + fn into(self) -> LocationResource { + LocationResource { + id: self.id, + name: self.name, + path: self.local_path, + total_capacity: self.total_capacity, + available_capacity: self.available_capacity, + is_removable: self.is_removable, + is_ejectable: self.is_ejectable, + is_root_filesystem: self.is_root_filesystem, + is_online: self.is_online, + date_created: self.date_created, + } + } } #[derive(Serialize, Deserialize, Default)] pub struct DotSpacedrive { - pub location_uuid: String, - pub library_uuid: String, + pub location_uuid: String, + pub library_uuid: String, } static DOTFILE_NAME: &str = ".spacedrive"; @@ -61,164 +61,154 @@ static DOTFILE_NAME: &str = ".spacedrive"; // - accessible on from the local filesystem // - already exists in the database pub async fn check_location(path: &str) -> Result { - let dotfile: DotSpacedrive = - match fs::File::open(format!("{}/{}", path.clone(), DOTFILE_NAME)) { - Ok(file) => serde_json::from_reader(file).unwrap_or(DotSpacedrive::default()), - Err(e) => return Err(LocationError::DotfileReadFailure(e)), - }; + let dotfile: DotSpacedrive = match fs::File::open(format!("{}/{}", path.clone(), DOTFILE_NAME)) { + Ok(file) => serde_json::from_reader(file).unwrap_or(DotSpacedrive::default()), + Err(e) => return Err(LocationError::DotfileReadFailure(e)), + }; - Ok(dotfile) + Ok(dotfile) } pub async fn get_location( - ctx: &CoreContext, - location_id: i32, + ctx: &CoreContext, + location_id: i32, ) -> Result { - let db = &ctx.database; + let db = &ctx.database; - // get location by location_id from db and include location_paths - let location = match db - .location() - .find_first(vec![ - Location::file_paths().some(vec![FilePath::id().equals(location_id.into())]) - ]) - .exec() - .await? - { - Some(location) => location, - None => Err(LocationError::NotFound(location_id.to_string()))?, - }; + // get location by location_id from db and include location_paths + let location = match db + .location() + .find_first(vec![ + Location::file_paths().some(vec![FilePath::id().equals(location_id.into())]) + ]) + .exec() + .await? + { + Some(location) => location, + None => Err(LocationError::NotFound(location_id.to_string()))?, + }; - info!("Retrieved location: {:?}", location); + info!("Retrieved location: {:?}", location); - Ok(location.into()) + Ok(location.into()) } pub async fn new_location_and_scan( - ctx: &CoreContext, - path: &str, + ctx: &CoreContext, + path: &str, ) -> Result { - let location = create_location(&ctx, path).await?; + let location = create_location(&ctx, path).await?; - ctx.spawn_job(Box::new(IndexerJob { - path: path.to_string(), - })); + ctx.spawn_job(Box::new(IndexerJob { + path: path.to_string(), + })); - Ok(location) + Ok(location) } -pub async fn create_location( - ctx: &CoreContext, - path: &str, -) -> Result { - let db = &ctx.database; - let config = client::get(); +pub async fn create_location(ctx: &CoreContext, path: &str) -> Result { + let db = &ctx.database; + let config = client::get(); - // check if we have access to this location - match fs::File::open(&path) { - Ok(_) => info!("Path is valid, creating location for '{}'", &path), - Err(e) => Err(LocationError::FileReadError(e))?, - } - // check if location already exists - let location = match db - .location() - .find_first(vec![Location::local_path().equals(path.to_string())]) - .exec() - .await? - { - Some(location) => location, - None => { - info!( - "Location does not exist, creating new location for '{}'", - &path - ); - let uuid = uuid::Uuid::new_v4(); - // create new location - let create_location_params = { - let volumes = match volumes::get_volumes() { - Ok(volumes) => volumes, - Err(e) => Err(LocationError::VolumeReadError(e.to_string()))?, - }; - info!("Loaded mounted volumes: {:?}", volumes); - // find mount with matching path - let volume = volumes - .into_iter() - .find(|mount| path.starts_with(&mount.mount_point)); + // check if we have access to this location + match fs::File::open(&path) { + Ok(_) => info!("Path is valid, creating location for '{}'", &path), + Err(e) => Err(LocationError::FileReadError(e))?, + } + // check if location already exists + let location = match db + .location() + .find_first(vec![Location::local_path().equals(path.to_string())]) + .exec() + .await? + { + Some(location) => location, + None => { + info!( + "Location does not exist, creating new location for '{}'", + &path + ); + let uuid = uuid::Uuid::new_v4(); + // create new location + let create_location_params = { + let volumes = match volumes::get_volumes() { + Ok(volumes) => volumes, + Err(e) => Err(LocationError::VolumeReadError(e.to_string()))?, + }; + info!("Loaded mounted volumes: {:?}", volumes); + // find mount with matching path + let volume = volumes + .into_iter() + .find(|mount| path.starts_with(&mount.mount_point)); - let volume_data = match volume { - Some(mount) => mount, - None => Volume::default(), - }; + let volume_data = match volume { + Some(mount) => mount, + None => Volume::default(), + }; - vec![ - Location::name().set(volume_data.name.to_string()), - Location::total_capacity().set(volume_data.total_capacity as i32), - Location::available_capacity() - .set(volume_data.available_capacity as i32), - Location::is_ejectable().set(false), // remove this - Location::is_removable().set(volume_data.is_removable), - Location::is_root_filesystem().set(false), // remove this - Location::is_online().set(true), - Location::local_path().set(path.to_string()), - ] - }; + vec![ + Location::name().set(volume_data.name.to_string()), + Location::total_capacity().set(volume_data.total_capacity as i32), + Location::available_capacity().set(volume_data.available_capacity as i32), + Location::is_ejectable().set(false), // remove this + Location::is_removable().set(volume_data.is_removable), + Location::is_root_filesystem().set(false), // remove this + Location::is_online().set(true), + Location::local_path().set(path.to_string()), + ] + }; - let location = db - .location() - .create_one(create_location_params) - .exec() - .await?; + let location = db + .location() + .create_one(create_location_params) + .exec() + .await?; - info!("Created location: {:?}", location); + info!("Created location: {:?}", location); - // write a file called .spacedrive to path containing the location id in JSON format - let mut dotfile = - match fs::File::create(format!("{}/{}", path.clone(), DOTFILE_NAME)) { - Ok(file) => file, - Err(e) => { - Err(LocationError::DotfileWriteFailure(e, path.to_string()))? - }, - }; + // write a file called .spacedrive to path containing the location id in JSON format + let mut dotfile = match fs::File::create(format!("{}/{}", path.clone(), DOTFILE_NAME)) { + Ok(file) => file, + Err(e) => Err(LocationError::DotfileWriteFailure(e, path.to_string()))?, + }; - let data = DotSpacedrive { - location_uuid: uuid.to_string(), - library_uuid: config.current_library_uuid, - }; + let data = DotSpacedrive { + location_uuid: uuid.to_string(), + library_uuid: config.current_library_uuid, + }; - let json = match serde_json::to_string(&data) { - Ok(json) => json, - Err(e) => { - Err(LocationError::DotfileSerializeFailure(e, path.to_string()))? - }, - }; + let json = match serde_json::to_string(&data) { + Ok(json) => json, + Err(e) => Err(LocationError::DotfileSerializeFailure(e, path.to_string()))?, + }; - match dotfile.write_all(json.as_bytes()) { - Ok(_) => (), - Err(e) => Err(LocationError::DotfileWriteFailure(e, path.to_string()))?, - } - location - }, - }; + match dotfile.write_all(json.as_bytes()) { + Ok(_) => (), + Err(e) => Err(LocationError::DotfileWriteFailure(e, path.to_string()))?, + } + location + } + }; - Ok(location.into()) + Ok(location.into()) } #[derive(Error, Debug)] pub enum LocationError { - #[error("Failed to create location (uuid {uuid:?})")] - CreateFailure { uuid: String }, - #[error("Failed to read location dotfile")] - DotfileReadFailure(io::Error), - #[error("Failed to serialize dotfile for location (at path: {1:?})")] - DotfileSerializeFailure(serde_json::Error, String), - #[error("Location not found (uuid: {1:?})")] - DotfileWriteFailure(io::Error, String), - #[error("Location not found (uuid: {0:?})")] - NotFound(String), - #[error("Failed to open file from local os")] - FileReadError(io::Error), - #[error("Failed to read mounted volumes from local os")] - VolumeReadError(String), - #[error("Failed to connect to database (error: {0:?})")] - IOError(io::Error), + #[error("Failed to create location (uuid {uuid:?})")] + CreateFailure { uuid: String }, + #[error("Failed to read location dotfile")] + DotfileReadFailure(io::Error), + #[error("Failed to serialize dotfile for location (at path: {1:?})")] + DotfileSerializeFailure(serde_json::Error, String), + #[error("Location not found (uuid: {1:?})")] + DotfileWriteFailure(io::Error, String), + #[error("Location not found (uuid: {0:?})")] + NotFound(String), + #[error("Failed to open file from local os")] + FileReadError(io::Error), + #[error("Failed to read mounted volumes from local os")] + VolumeReadError(String), + #[error("Failed to connect to database (error: {0:?})")] + IOError(io::Error), } diff --git a/core/src/sys/mod.rs b/core/src/sys/mod.rs index f3f670499..fdd0da08a 100644 --- a/core/src/sys/mod.rs +++ b/core/src/sys/mod.rs @@ -8,12 +8,12 @@ use self::locations::LocationError; #[derive(Error, Debug)] pub enum SysError { - #[error("Location error")] - LocationError(#[from] LocationError), - #[error("Error with system volumes")] - VolumeError(String), - #[error("Error from job runner")] - JobError(#[from] job::JobError), - #[error("Database error")] - DatabaseError(#[from] prisma::QueryError), + #[error("Location error")] + LocationError(#[from] LocationError), + #[error("Error with system volumes")] + VolumeError(String), + #[error("Error from job runner")] + JobError(#[from] job::JobError), + #[error("Database error")] + DatabaseError(#[from] prisma::QueryError), } diff --git a/core/src/sys/volumes.rs b/core/src/sys/volumes.rs index 2167d47e2..4d5a16818 100644 --- a/core/src/sys/volumes.rs +++ b/core/src/sys/volumes.rs @@ -13,87 +13,85 @@ use super::SysError; #[repr(C)] #[ts(export)] pub struct Volume { - pub name: String, - pub mount_point: String, - pub total_capacity: u64, - pub available_capacity: u64, - pub is_removable: bool, - pub disk_type: Option, - pub file_system: Option, - pub is_root_filesystem: bool, + pub name: String, + pub mount_point: String, + pub total_capacity: u64, + pub available_capacity: u64, + pub is_removable: bool, + pub disk_type: Option, + pub file_system: Option, + pub is_root_filesystem: bool, } pub fn get_volumes() -> Result, SysError> { - let all_volumes: Vec = System::new_all() - .disks() - .iter() - .map(|disk| { - let mut total_space = disk.total_space(); - let mount_point = disk.mount_point().to_str().unwrap_or("/").to_string(); - let available_space = disk.available_space(); - let mut name = disk.name().to_str().unwrap_or("Volume").to_string(); - let is_removable = disk.is_removable(); + let all_volumes: Vec = System::new_all() + .disks() + .iter() + .map(|disk| { + let mut total_space = disk.total_space(); + let mount_point = disk.mount_point().to_str().unwrap_or("/").to_string(); + let available_space = disk.available_space(); + let mut name = disk.name().to_str().unwrap_or("Volume").to_string(); + let is_removable = disk.is_removable(); - let file_system = String::from_utf8(disk.file_system().to_vec()) - .unwrap_or_else(|_| "Err".to_string()); + let file_system = + String::from_utf8(disk.file_system().to_vec()).unwrap_or_else(|_| "Err".to_string()); - let disk_type = match disk.type_() { - sysinfo::DiskType::SSD => "SSD".to_string(), - sysinfo::DiskType::HDD => "HDD".to_string(), - _ => "Removable Disk".to_string(), - }; + let disk_type = match disk.type_() { + sysinfo::DiskType::SSD => "SSD".to_string(), + sysinfo::DiskType::HDD => "HDD".to_string(), + _ => "Removable Disk".to_string(), + }; - if cfg!(target_os = "macos") && mount_point == "/" { - name = "Macintosh HD".to_string(); - } + if cfg!(target_os = "macos") && mount_point == "/" { + name = "Macintosh HD".to_string(); + } - if total_space < available_space && cfg!(target_os = "windows") { - let mut caption = mount_point.clone(); - caption.pop(); - let wmic_process = Command::new("cmd") - .args([ - "/C", - &format!("wmic logical disk where Caption='{caption}' get Size"), - ]) - .output() - .expect("failed to execute process"); - let wmic_process_output = String::from_utf8(wmic_process.stdout).unwrap(); - let parsed_size = - wmic_process_output.split("\r\r\n").collect::>()[1] - .to_string(); + if total_space < available_space && cfg!(target_os = "windows") { + let mut caption = mount_point.clone(); + caption.pop(); + let wmic_process = Command::new("cmd") + .args([ + "/C", + &format!("wmic logical disk where Caption='{caption}' get Size"), + ]) + .output() + .expect("failed to execute process"); + let wmic_process_output = String::from_utf8(wmic_process.stdout).unwrap(); + let parsed_size = wmic_process_output.split("\r\r\n").collect::>()[1].to_string(); - if let Ok(n) = parsed_size.trim().parse::() { - total_space = n; - } - } + if let Ok(n) = parsed_size.trim().parse::() { + total_space = n; + } + } - Volume { - name, - mount_point: mount_point.clone(), - total_capacity: total_space, - available_capacity: available_space, - is_removable, - disk_type: Some(disk_type), - file_system: Some(file_system), - is_root_filesystem: mount_point == "/", - } - }) - .collect(); + Volume { + name, + mount_point: mount_point.clone(), + total_capacity: total_space, + available_capacity: available_space, + is_removable, + disk_type: Some(disk_type), + file_system: Some(file_system), + is_root_filesystem: mount_point == "/", + } + }) + .collect(); - let volumes = all_volumes - .clone() - .into_iter() - .filter(|volume| !volume.mount_point.starts_with("/System")) - .collect(); + let volumes = all_volumes + .clone() + .into_iter() + .filter(|volume| !volume.mount_point.starts_with("/System")) + .collect(); - Ok(volumes) + Ok(volumes) } #[test] fn test_get_volumes() { - let volumes = get_volumes().unwrap(); - dbg!(&volumes); - assert!(volumes.len() > 0); + let volumes = get_volumes().unwrap(); + dbg!(&volumes); + assert!(volumes.len() > 0); } // Adapted from: https://github.com/kimlimjustin/xplorer/blob/f4f3590d06783d64949766cc2975205a3b689a56/src-tauri/src/drives.rs