mirror/spacedrive
1
0 Fork 0
mirror of https://github.com/spacedriveapp/spacedrive.git synced 2026-10-24 09:58:35 -04:00
Code Issues Packages Projects Releases Wiki Activity

better rust formating

Browse Source
This commit is contained in:
Jamie Pine committed 2022-04-01 07:41:42 -07:00
1 parent f6e99da024
commit 95f7bb6e52
21 files changed
+1576 -1618
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

core/.rustfmt.toml
+14 -6
View File
@@ -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
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
core/src/client/mod.rs
+48 -48
View File
@@ -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(())
}
core/src/crypto/encryption.rs
+4 -4
View File
@@ -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,
}
core/src/db/migrate.rs
+101 -106
View File
@@ -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::<serde_json::Value>(
"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::<serde_json::Value>(
"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<serde_json::Value> = client
._query_raw("SELECT name FROM sqlite_master WHERE type='table' AND name='_migrations'")
.await
.unwrap();
let value: Vec<serde_json::Value> = 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::<Vec<_>>();
let mut migration_subdirs = MIGRATIONS_DIR
.dirs()
.filter(|subdir| {
subdir
.path()
.file_name()
.map(|name| name != OsStr::new("migration_table"))
.unwrap_or(false)
})
.collect::<Vec<_>>();
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::<i64>().unwrap();
let b_time = b_name[..14].parse::<i64>().unwrap();
let a_time = a_name[..14].parse::<i64>().unwrap();
let b_time = b_name[..14].parse::<i64>().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::<Vec<&str>>();
let steps = &steps[0..steps.len() - 1];
let steps = migration_sql.split(";").collect::<Vec<&str>>();
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(())
}
core/src/db/mod.rs
+11 -11
View File
@@ -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<PrismaClient, DatabaseError> {
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)
}
core/src/file/checksum.rs
+51 -51
View File
@@ -10,67 +10,67 @@ static SAMPLE_COUNT: u64 = 4;
static SAMPLE_SIZE: u64 = 10000;
pub fn partial_checksum(path: &str, size: u64) -> Result<String> {
// 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<String> {
// 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<R: Read>(mut reader: R) -> io::Result<Digest> {
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())
}
core/src/file/explorer/mod.rs
+21 -24
View File
@@ -2,32 +2,29 @@ use crate::{file::DirectoryWithContents, prisma::FilePath, CoreContext};
use super::FileError;
pub async fn open_dir(
ctx: &CoreContext,
path: &str,
) -> Result<DirectoryWithContents, FileError> {
let db = &ctx.database;
pub async fn open_dir(ctx: &CoreContext, path: &str) -> Result<DirectoryWithContents, FileError> {
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(),
})
}
core/src/file/indexer.rs
+213 -220
View File
@@ -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<ScanProgress>) + Send + Sync + 'static,
ctx: &CoreContext,
path: &str,
on_progress: impl Fn(Vec<ScanProgress>) + 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<i32>,
}
// grab the next id so we can increment in memory for batch inserting
let first_file_id = match db
._query_raw::<QueryRes>(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<i32>,
}
// grab the next id so we can increment in memory for batch inserting
let first_file_id = match db
._query_raw::<QueryRes>(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<i32>)> = Vec::new();
// store a hashmap of directories to their file ids for fast lookup
let mut dirs: HashMap<String, i32> = 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<i32>)> = Vec::new();
// store a hashmap of directories to their file ids for fast lookup
let mut dirs: HashMap<String, i32> = 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<String> = 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<String> = 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<i32>,
file_path: &PathBuf,
id: i32,
location: &LocationResource,
parent_id: &Option<i32>,
) -> Result<String> {
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<Utc> = metadata.created().unwrap().into();
let parsed_date_created = date_created.to_rfc3339_opts(SecondsFormat::Millis, true);
let date_created: DateTime<Utc> = 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
}
core/src/file/mod.rs
+69 -69
View File
@@ -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<String>,
pub size_in_bytes: String,
pub encryption: EncryptionAlgorithm,
pub file_type: FileType,
#[ts(type = "string")]
pub date_created: chrono::DateTime<chrono::Utc>,
#[ts(type = "string")]
pub date_modified: chrono::DateTime<chrono::Utc>,
#[ts(type = "string")]
pub date_indexed: chrono::DateTime<chrono::Utc>,
pub ipfs_id: Option<String>,
pub file_paths: Vec<FilePath>,
pub id: i32,
pub partial_checksum: String,
pub checksum: Option<String>,
pub size_in_bytes: String,
pub encryption: EncryptionAlgorithm,
pub file_type: FileType,
#[ts(type = "string")]
pub date_created: chrono::DateTime<chrono::Utc>,
#[ts(type = "string")]
pub date_modified: chrono::DateTime<chrono::Utc>,
#[ts(type = "string")]
pub date_indexed: chrono::DateTime<chrono::Utc>,
pub ipfs_id: Option<String>,
pub file_paths: Vec<FilePath>,
}
// 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<i32>,
pub parent_id: Option<i32>,
#[ts(type = "string")]
pub date_indexed: chrono::DateTime<chrono::Utc>,
pub permissions: Option<String>,
pub id: i32,
pub is_dir: bool,
pub location_id: i32,
pub materialized_path: String,
pub file_id: Option<i32>,
pub parent_id: Option<i32>,
#[ts(type = "string")]
pub date_indexed: chrono::DateTime<chrono::Utc>,
pub permissions: Option<String>,
}
#[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<File> 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<FilePath> 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<FilePath>,
pub directory: FilePath,
pub contents: Vec<FilePath>,
}
#[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),
}
core/src/file/thumb.rs
+80 -90
View File
@@ -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<Vec<FilePathData>> {
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<Vec<FilePathData>> {
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)
}
core/src/file/watcher.rs
+16 -16
View File
@@ -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));
}
core/src/job/jobs.rs
+130 -128
View File
@@ -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<Box<dyn Job>>,
// workers are spawned when jobs are picked off the queue
running_workers: HashMap<String, Arc<Mutex<Worker>>>,
// job_queue: Vec<Box<dyn Job>>,
// workers are spawned when jobs are picked off the queue
running_workers: HashMap<String, Arc<Mutex<Worker>>>,
}
impl Jobs {
pub fn new() -> Self {
Self {
// job_queue: vec![],
running_workers: HashMap::new(),
}
}
pub async fn ingest(&mut self, ctx: &CoreContext, job: Box<dyn Job>) {
// 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<dyn Job>) {
// 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<JobReport> {
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<JobReport> {
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<Vec<JobReport>, 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<Vec<JobReport>, 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<chrono::Utc>,
#[ts(type = "string")]
pub date_modified: chrono::DateTime<chrono::Utc>,
pub id: String,
// client_id: i32,
#[ts(type = "string")]
pub date_created: chrono::DateTime<chrono::Utc>,
#[ts(type = "string")]
pub date_modified: chrono::DateTime<chrono::Utc>,
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<JobReport> 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,
}
core/src/job/mod.rs
+4 -4
View File
@@ -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 {
core/src/job/worker.rs
+151 -147
View File
@@ -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<JobReportUpdate>),
Completed,
Failed,
Progressed(Vec<JobReportUpdate>),
Completed,
Failed,
}
enum WorkerState {
Pending(Box<dyn Job>, UnboundedReceiver<WorkerEvent>),
Running,
Pending(Box<dyn Job>, UnboundedReceiver<WorkerEvent>),
Running,
}
#[derive(Clone)]
pub struct WorkerContext {
pub uuid: String,
pub core_ctx: CoreContext,
pub sender: UnboundedSender<WorkerEvent>,
pub uuid: String,
pub core_ctx: CoreContext,
pub sender: UnboundedSender<WorkerEvent>,
}
impl WorkerContext {
pub fn progress(&self, updates: Vec<JobReportUpdate>) {
self.sender
.send(WorkerEvent::Progressed(updates))
.unwrap_or(());
}
pub fn progress(&self, updates: Vec<JobReportUpdate>) {
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<WorkerEvent>,
pub job_report: JobReport,
state: WorkerState,
worker_sender: UnboundedSender<WorkerEvent>,
}
impl Worker {
pub fn new(job: Box<dyn Job>) -> Self {
let (worker_sender, worker_receiver) = unbounded_channel();
let uuid = uuid::Uuid::new_v4().to_string();
pub fn new(job: Box<dyn Job>) -> 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<Mutex<Self>>, 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<Mutex<Self>>, 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<Mutex<Self>>,
mut channel: UnboundedReceiver<WorkerEvent>,
ctx: CoreContext,
) {
while let Some(command) = channel.recv().await {
let mut worker = worker.lock().await;
async fn track_progress(
worker: Arc<Mutex<Self>>,
mut channel: UnboundedReceiver<WorkerEvent>,
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;
}
}
}
}
}
core/src/lib.rs
+280 -299
View File
@@ -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<D, R = Result<CoreResponse, CoreError>> {
data: D,
return_sender: oneshot::Sender<R>,
data: D,
return_sender: oneshot::Sender<R>,
}
// core controller is passed to the client to communicate with the core which runs in a dedicated thread
pub struct CoreController {
query_sender: UnboundedSender<ReturnableMessage<ClientQuery>>,
command_sender: UnboundedSender<ReturnableMessage<ClientCommand>>,
query_sender: UnboundedSender<ReturnableMessage<ClientQuery>>,
command_sender: UnboundedSender<ReturnableMessage<ClientCommand>>,
}
impl CoreController {
pub async fn query(&self, query: ClientQuery) -> Result<CoreResponse, CoreError> {
// 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<CoreResponse, CoreError> {
// 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<CoreResponse, CoreError> {
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<CoreResponse, CoreError> {
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<dyn Job>),
JobComplete(String),
JobIngest(Box<dyn Job>),
JobComplete(String),
}
#[derive(Clone)]
pub struct CoreContext {
pub database: Arc<PrismaClient>,
pub event_sender: mpsc::Sender<CoreEvent>,
pub internal_sender: UnboundedSender<InternalEvent>,
pub database: Arc<PrismaClient>,
pub event_sender: mpsc::Sender<CoreEvent>,
pub internal_sender: UnboundedSender<InternalEvent>,
}
impl CoreContext {
pub fn spawn_job(&self, job: Box<dyn Job>) {
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<dyn Job>) {
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<PrismaClient>,
// filetype_registry: library::TypeRegistry,
// extension_registry: library::ExtensionRegistry,
state: ClientState,
jobs: job::jobs::Jobs,
database: Arc<PrismaClient>,
// filetype_registry: library::TypeRegistry,
// extension_registry: library::ExtensionRegistry,
// global messaging channels
query_channel: (
UnboundedSender<ReturnableMessage<ClientQuery>>,
UnboundedReceiver<ReturnableMessage<ClientQuery>>,
),
command_channel: (
UnboundedSender<ReturnableMessage<ClientCommand>>,
UnboundedReceiver<ReturnableMessage<ClientCommand>>,
),
event_sender: mpsc::Sender<CoreEvent>,
// global messaging channels
query_channel: (
UnboundedSender<ReturnableMessage<ClientQuery>>,
UnboundedReceiver<ReturnableMessage<ClientQuery>>,
),
command_channel: (
UnboundedSender<ReturnableMessage<ClientCommand>>,
UnboundedReceiver<ReturnableMessage<ClientCommand>>,
),
event_sender: mpsc::Sender<CoreEvent>,
// a channel for child threads to send events back to the core
internal_channel: (
UnboundedSender<InternalEvent>,
UnboundedReceiver<InternalEvent>,
),
// a channel for child threads to send events back to the core
internal_channel: (
UnboundedSender<InternalEvent>,
UnboundedReceiver<InternalEvent>,
),
}
impl Core {
// create new instance of core, run startup tasks
pub async fn new(
mut data_dir: std::path::PathBuf,
) -> (Core, mpsc::Receiver<CoreEvent>) {
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<CoreEvent>) {
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::<InternalEvent>();
let internal_channel = unbounded_channel::<InternalEvent>();
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<CoreResponse, CoreError> {
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<CoreResponse, CoreError> {
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<CoreResponse, CoreError> {
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<CoreResponse, CoreError> {
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<String> },
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<String> },
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<String> },
// most all events should be once of these two
InvalidateQuery(ClientQuery),
InvalidateQueryDebounced(ClientQuery),
InvalidateResource(CoreResource),
Log { message: String },
DatabaseDisconnected { reason: Option<String> },
}
#[derive(Serialize, Deserialize, Debug, TS)]
#[serde(tag = "key", content = "data")]
#[ts(export)]
pub enum CoreResponse {
Success(()),
SysGetVolumes(Vec<sys::volumes::Volume>),
SysGetLocation(sys::locations::LocationResource),
LibGetExplorerDir(file::DirectoryWithContents),
ClientGetState(ClientState),
LocCreate(sys::locations::LocationResource),
JobGetRunning(Vec<JobReport>),
JobGetHistory(Vec<JobReport>),
Success(()),
SysGetVolumes(Vec<sys::volumes::Volume>),
SysGetLocation(sys::locations::LocationResource),
LibGetExplorerDir(file::DirectoryWithContents),
ClientGetState(ClientState),
LocCreate(sys::locations::LocationResource),
JobGetRunning(Vec<JobReport>),
JobGetHistory(Vec<JobReport>),
}
#[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,
}
core/src/library/loader.rs
+57 -57
View File
@@ -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<LibraryData, LibraryError> {
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<String>) -> 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(())
}
core/src/p2p/listener.rs
+34 -34
View File
@@ -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<u32>) -> Result<(), Box<dyn Error>> {
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),
_ => {}
}
}
}
core/src/state/client.rs
+68 -68
View File
@@ -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<LibraryState>,
// 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<LibraryState>,
// 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<Option<ClientState>> = RwLock::new(None);
static ref CONFIG: RwLock<Option<ClientState>> = 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<Self> {
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<Self> {
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)
}
}
core/src/sys/locations.rs
+152 -162
View File
@@ -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<String>,
pub path: Option<String>,
pub total_capacity: Option<i32>,
pub available_capacity: Option<i32>,
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<chrono::Utc>,
pub id: i32,
pub name: Option<String>,
pub path: Option<String>,
pub total_capacity: Option<i32>,
pub available_capacity: Option<i32>,
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<chrono::Utc>,
}
impl Into<LocationResource> 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<DotSpacedrive, LocationError> {
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<LocationResource, SysError> {
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<LocationResource, SysError> {
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<LocationResource, SysError> {
let db = &ctx.database;
let config = client::get();
pub async fn create_location(ctx: &CoreContext, path: &str) -> Result<LocationResource, SysError> {
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),
}
core/src/sys/mod.rs
+8 -8
View File
@@ -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),
}
core/src/sys/volumes.rs
+64 -66
View File
@@ -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<String>,
pub file_system: Option<String>,
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<String>,
pub file_system: Option<String>,
pub is_root_filesystem: bool,
}
pub fn get_volumes() -> Result<Vec<Volume>, SysError> {
let all_volumes: Vec<Volume> = 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<Volume> = 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::<Vec<&str>>()[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::<Vec<&str>>()[1].to_string();
if let Ok(n) = parsed_size.trim().parse::<u64>() {
total_space = n;
}
}
if let Ok(n) = parsed_size.trim().parse::<u64>() {
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