From d44a70562bca484c084afbbd99c219e192c36964 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Tue, 4 Apr 2023 13:45:00 +0800 Subject: [PATCH] some more sync impls (#669) a lil more sync impls --- core/src/location/file_path_helper.rs | 77 ++++++++++--- core/src/location/manager/watcher/utils.rs | 123 +++++++++++++++------ core/src/sync/manager.rs | 3 +- 3 files changed, 151 insertions(+), 52 deletions(-) diff --git a/core/src/location/file_path_helper.rs b/core/src/location/file_path_helper.rs index 1fd357930..a7a766c20 100644 --- a/core/src/location/file_path_helper.rs +++ b/core/src/location/file_path_helper.rs @@ -1,3 +1,4 @@ +use crate::location::Library; use crate::prisma::{file_path, location, PrismaClient}; use std::{ @@ -11,6 +12,7 @@ use dashmap::{mapref::entry::Entry, DashMap}; use futures::future::try_join_all; use prisma_client_rust::{Direction, QueryError}; use serde::{Deserialize, Serialize}; +use serde_json::json; use thiserror::Error; use tokio::{fs, io}; use tracing::error; @@ -306,7 +308,7 @@ impl LastFilePathIdManager { #[cfg(feature = "location-watcher")] pub async fn create_file_path( &self, - db: &PrismaClient, + library @ Library { db, sync, .. }: &Library, MaterializedPath { materialized_path, is_dir, @@ -319,6 +321,9 @@ impl LastFilePathIdManager { device: u64, ) -> Result { // Keeping a reference in that map for the entire duration of the function, so we keep it locked + + use crate::sync; + let mut last_id_ref = match self.last_id_by_location.entry(location_id) { Entry::Occupied(ocupied) => ocupied.into_ref(), Entry::Vacant(vacant) => { @@ -327,24 +332,64 @@ impl LastFilePathIdManager { } }; + let location = db + .location() + .find_unique(location::id::equals(location_id)) + .select(location::select!({ id pub_id })) + .exec() + .await? + .unwrap(); + let next_id = *last_id_ref + 1; - let created_path = db - .file_path() - .create( - next_id, - location::id::equals(location_id), - materialized_path.into_owned(), - name.into_owned(), - extension.into_owned(), - inode.to_le_bytes().into(), - device.to_le_bytes().into(), - vec![ - file_path::parent_id::set(parent_id), - file_path::is_dir::set(is_dir), - ], + let created_path = library + .sync + .write_op( + &library.db, + library.sync.unique_shared_create( + sync::file_path::SyncId { + location: sync::location::SyncId { + pub_id: location.pub_id.clone(), + }, + id: next_id, + }, + [ + ("materialized_path", json!(materialized_path)), + ("name", json!(name)), + ("extension", json!(extension)), + ("inode", json!(inode.to_le_bytes())), + ("device", json!(device.to_le_bytes())), + ("is_dir", json!(is_dir)), + ] + .into_iter() + .map(Some) + .chain([parent_id.map(|parent_id| { + ( + "parent_id", + json!(sync::file_path::SyncId { + location: sync::location::SyncId { + pub_id: location.pub_id.clone() + }, + id: parent_id + }), + ) + })]) + .flatten(), + ), + library.db.file_path().create( + next_id, + location::id::equals(location_id), + materialized_path.into_owned(), + name.into_owned(), + extension.into_owned(), + inode.to_le_bytes().into(), + device.to_le_bytes().into(), + vec![ + file_path::parent_id::set(parent_id), + file_path::is_dir::set(is_dir), + ], + ), ) - .exec() .await?; *last_id_ref = next_id; diff --git a/core/src/location/manager/watcher/utils.rs b/core/src/location/manager/watcher/utils.rs index d3235cdc3..43cc8a531 100644 --- a/core/src/location/manager/watcher/utils.rs +++ b/core/src/location/manager/watcher/utils.rs @@ -21,6 +21,7 @@ use crate::{ validation::hash::file_checksum, }, prisma::{file_path, location, object}, + sync, }; #[cfg(target_family = "unix")] @@ -42,6 +43,7 @@ use chrono::{DateTime, Local}; use int_enum::IntEnum; use notify::{Event, EventKind}; use prisma_client_rust::{raw, PrismaValue}; +use serde_json::json; use tokio::{fs, io::ErrorKind}; use tracing::{error, info, trace, warn}; use uuid::Uuid; @@ -107,7 +109,7 @@ pub(super) async fn create_dir( let created_path = library .last_file_path_id_manager .create_file_path( - &library.db, + &library, materialized_path, Some(parent_directory.id), inode, @@ -168,7 +170,7 @@ pub(super) async fn create_file( let created_file = library .last_file_path_id_manager .create_file_path( - db, + library, materialized_path, Some(parent_directory.id), inode, @@ -317,10 +319,18 @@ async fn inner_update_file( location_id: LocationId, file_path: &file_path_with_object::Data, full_path: impl AsRef, - library: &Library, + library @ Library { db, sync, .. }: &Library, ) -> Result<(), LocationManagerError> { let full_path = full_path.as_ref(); - let location_path = extract_location_path(location_id, library).await?; + let location = db + .location() + .find_unique(location::id::equals(location_id)) + .exec() + .await? + .ok_or_else(|| LocationManagerError::MissingLocation(location_id))?; + + let location_path = PathBuf::from(location.path); + trace!( "Location: updating file: {}", location_path.display(), @@ -339,30 +349,67 @@ async fn inner_update_file( if let Some(old_cas_id) = &file_path.cas_id { if old_cas_id != &cas_id { + let (sync_params, db_params): (Vec<_>, Vec<_>) = [ + ( + ("cas_id", json!(old_cas_id)), + file_path::cas_id::set(Some(old_cas_id.clone())), + ), + ( + ("size_in_bytes", json!(fs_metadata.len().to_string())), + file_path::size_in_bytes::set(fs_metadata.len().to_string()), + ), + { + let date = DateTime::::from(fs_metadata.created().unwrap()).into(); + + ( + ("date_modified", json!(date)), + file_path::date_modified::set(date), + ) + }, + { + // TODO: Should this be a skip rather than a null-set? + let checksum = if file_path.integrity_checksum.is_some() { + // If a checksum was already computed, we need to recompute it + Some(file_checksum(full_path).await?) + } else { + None + }; + + ( + ("integrity_checksum", json!(checksum)), + file_path::integrity_checksum::set(checksum), + ) + }, + ] + .into_iter() + .unzip(); + // file content changed - library - .db - .file_path() - .update( - file_path::location_id_id(location_id, file_path.id), - vec![ - file_path::cas_id::set(Some(old_cas_id.clone())), - file_path::size_in_bytes::set(fs_metadata.len().to_string()), - file_path::date_modified::set( - DateTime::::from(fs_metadata.created().unwrap()).into(), - ), - file_path::integrity_checksum::set( - if file_path.integrity_checksum.is_some() { - // If a checksum was already computed, we need to recompute it - Some(file_checksum(full_path).await?) - } else { - None - }, - ), - ], - ) - .exec() - .await?; + sync.write_ops( + db, + ( + sync_params + .into_iter() + .map(|(field, value)| { + sync.shared_update( + sync::file_path::SyncId { + location: sync::location::SyncId { + pub_id: location.pub_id.clone(), + }, + id: file_path.id, + }, + field, + value, + ) + }) + .collect(), + db.file_path().update( + file_path::location_id_id(location_id, file_path.id), + db_params, + ), + ), + ) + .await?; if let Some(ref object) = file_path.object { // if this file had a thumbnail previously, we update it to match the new content @@ -371,18 +418,26 @@ async fn inner_update_file( } let int_kind = kind.int_value(); + if object.kind != int_kind { - library - .db - .object() - .update( + sync.write_op( + db, + sync.shared_update( + sync::object::SyncId { + pub_id: object.pub_id.clone(), + }, + "kind", + json!(int_kind), + ), + db.object().update( object::id::equals(object.id), vec![object::kind::set(int_kind)], - ) - .exec() - .await?; + ), + ) + .await?; } } + invalidate_query!(library, "locations.getExplorerData"); } } diff --git a/core/src/sync/manager.rs b/core/src/sync/manager.rs index 15667e8bd..483301271 100644 --- a/core/src/sync/manager.rs +++ b/core/src/sync/manager.rs @@ -455,13 +455,12 @@ impl SyncManager { })) } pub fn unique_shared_create< - const SIZE: usize, TSyncId: SyncId, TModel: SyncType, >( &self, id: TSyncId, - values: [(&'static str, Value); SIZE], + values: impl IntoIterator, ) -> CRDTOperation { self.new_op(CRDTOperationType::Shared(SharedOperation { model: TModel::MODEL.to_string(),