Working Sync & Thumbnail Sync (#2875)

* First draft on new p2p design

* Test sync

* Sync works

* Update version

* Fix merge conflicts

* More stuff for thumbnail sync

* Get the thumbnail

* Update mod.rs

* Working thumbnail sync

* Format

* Update thumbnails.rs

* Update runner.rs

* Update runner.rs

---------

Co-authored-by: Ericson Soares <[email protected]>

Author: Rocky43007

Cargo.lock

@@ -20,50 +20,53 @@ rust-version = "1.81"
 
 [workspace.dependencies]
 # First party dependencies
-sd-cloud-schema = { git = "https://github.com/spacedriveapp/cloud-services-schema", rev = "4e4565bee4" }
+sd-cloud-schema = { git = "https://github.com/spacedriveapp/cloud-services-schema", rev = "515ba740ea" }
 
 # Third party dependencies used by one or more of our crates
+anyhow              = "1.0.94"
 async-channel       = "2.3"
 async-stream        = "0.3.6"
 async-trait         = "0.1.83"
-axum                = "0.7.7"
-axum-extra          = "0.9.4"
+axum                = "0.7.9"
+axum-extra          = "0.9.6"
 base64              = "0.22.1"
-blake3              = "1.5.4"
-bytes               = "1.7.1"                                          # Update blocked by hyper
-chrono              = "0.4.38"
+blake3              = "1.5.5"
+bytes               = "1.9.0"
+chrono              = "0.4.39"
 ed25519-dalek       = "2.1"
 flume               = "0.11.0"
 futures             = "0.3.31"
-futures-concurrency = "7.6"
+futures-concurrency = "7.6.2"
 globset             = "0.4.15"
-http                = "1.1"
-hyper               = "1.5"
-image               = "0.25.4"
+http                = "1.2.0"
+hyper               = "1.5.2"
+image               = "0.25.5"
+iroh                = "0.29.0"
 itertools           = "0.13.0"
 lending-stream      = "1.0"
-libc                = "0.2.159"
+libc                = "0.2.169"
 mimalloc            = "0.1.43"
 normpath            = "1.3"
-pin-project-lite    = "0.2.14"
+pin-project-lite    = "0.2.15"
+quic-rpc            = "0.17.3"
 rand                = "0.9.0-alpha.2"
-regex               = "1.11"
-reqwest             = { version = "0.12.8", default-features = false }
+regex               = "1.11.1"
+reqwest             = { version = "0.12.9", default-features = false }
 rmp                 = "0.8.14"
 rmp-serde           = "1.3"
 rmpv                = { version = "1.3", features = ["with-serde"] }
-serde               = "1.0"
-serde_json          = "1.0"
+serde               = "1.0.216"
+serde_json          = "1.0.133"
 specta              = "=2.0.0-rc.20"
 strum               = "0.26"
 strum_macros        = "0.26"
-tempfile            = "3.13"
-thiserror           = "1.0"
-tokio               = "1.40"
-tokio-stream        = "0.1.16"
-tokio-util          = "0.7.12"
-tracing             = "0.1.40"
-tracing-subscriber  = "0.3.18"
+tempfile            = "3.14.0"
+thiserror           = "2.0.8"
+tokio               = "1.42.0"
+tokio-stream        = "0.1.17"
+tokio-util          = "0.7.13"
+tracing             = "0.1.41"
+tracing-subscriber  = "0.3.19"
 tracing-test        = "0.2.5"
 uhlc                = "0.8.0"                                          # Must follow version used by specta
 uuid                = "1.10"                                           # Must follow version used by specta

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name    = "sd-desktop"
-version = "0.4.3"
+version = "0.5.0"
 
 authors              = ["Spacedrive Technology Inc <[email protected]>"]
 default-run          = "sd-desktop"
@@ -69,9 +69,9 @@ sd-desktop-linux = { path = "../crates/linux" }
 # WARNING: dbus must NOT be vendored, as that breaks the app on Linux,X11,Nvidia
 dbus = { version = "0.9.7", features = ["stdfd"] }
 # https://github.com/tauri-apps/tauri/blob/tauri-v2.0.0/crates/tauri/Cargo.toml#L101
+gtk        = { version = "0.18", features = ["v3_24"] }
+tao        = { version = "0.31.1", features = ["serde"] }
 webkit2gtk = { version = "=2.0.1", features = ["v2_40"] }
-gtk = { version = "0.18", features = ["v3_24"] }
-tao = { version = "0.31.1", features = ["serde"] }
 
 
 [target.'cfg(target_os = "macos")'.dependencies]

apps/desktop/src-tauri/Cargo.toml

@@ -8,6 +8,7 @@ import {
 	libraryClient,
 	RspcProvider,
 	useBridgeMutation,
+	useLibraryMutation,
 	useSelector
 } from '@sd/client';
 import {
@@ -61,21 +62,21 @@ declare global {
 }
 
 // Disabling until sync is ready.
-// SuperTokens.init({
-// 	appInfo: {
-// 		apiDomain: AUTH_SERVER_URL,
-// 		apiBasePath: '/api/auth',
-// 		appName: 'Spacedrive Auth Service'
-// 	},
-// 	cookieHandler: getCookieHandler,
-// 	windowHandler: getWindowHandler,
-// 	recipeList: [
-// 		Session.init({ tokenTransferMethod: 'header' }),
-// 		EmailPassword.init(),
-// 		ThirdParty.init(),
-// 		Passwordless.init()
-// 	]
-// });
+SuperTokens.init({
+	appInfo: {
+		apiDomain: AUTH_SERVER_URL,
+		apiBasePath: '/api/auth',
+		appName: 'Spacedrive Auth Service'
+	},
+	cookieHandler: getCookieHandler,
+	windowHandler: getWindowHandler,
+	recipeList: [
+		Session.init({ tokenTransferMethod: 'header' }),
+		EmailPassword.init()
+		// ThirdParty.init(),
+		// Passwordless.init()
+	]
+});
 
 const startupError = (window as any).__SD_ERROR__ as string | undefined;
 
@@ -232,7 +233,7 @@ type RedirectPath = { pathname: string; search: string | undefined };
 function AppInner() {
 	const [tabs, setTabs] = useState(() => [createTab()]);
 	const [selectedTabIndex, setSelectedTabIndex] = useState(0);
-	const cloudBootstrap = useBridgeMutation('cloud.bootstrap');
+	const cloudBootstrap = useLibraryMutation('cloud.bootstrap');
 
 	useEffect(() => {
 		(async () => {

apps/desktop/src/App.tsx

@@ -1,6 +1,6 @@
 [package]
 name    = "sd-core"
-version = "0.4.3"
+version = "0.5.0"
 
 authors                = ["Spacedrive Technology Inc <[email protected]>"]
 description            = "Virtual distributed filesystem engine that powers Spacedrive."
@@ -35,10 +35,10 @@ sd-ffmpeg         = { path = "../crates/ffmpeg", optional = true }
 sd-file-ext       = { path = "../crates/file-ext" }
 sd-images         = { path = "../crates/images", features = ["rspc", "serde", "specta"] }
 sd-media-metadata = { path = "../crates/media-metadata" }
-sd-p2p            = { path = "../crates/p2p", features = ["specta"] }
-sd-p2p-block      = { path = "../crates/p2p/crates/block" }
-sd-p2p-proto      = { path = "../crates/p2p/crates/proto" }
-sd-p2p-tunnel     = { path = "../crates/p2p/crates/tunnel" }
+sd-old-p2p        = { path = "../crates/old-p2p", features = ["specta"] }
+sd-old-p2p-block  = { path = "../crates/old-p2p/crates/block" }
+sd-old-p2p-proto  = { path = "../crates/old-p2p/crates/proto" }
+sd-old-p2p-tunnel = { path = "../crates/old-p2p/crates/tunnel" }
 sd-prisma         = { path = "../crates/prisma" }
 sd-sync           = { path = "../crates/sync" }
 sd-task-system    = { path = "../crates/task-system" }

apps/desktop/src/commands.ts

@@ -6,7 +6,9 @@ edition = "2021"
 
 [dependencies]
 # Core Spacedrive Sub-crates
-sd-core-sync = { path = "../sync" }
+sd-core-heavy-lifting  = { path = "../heavy-lifting" }
+sd-core-prisma-helpers = { path = "../prisma-helpers" }
+sd-core-sync           = { path = "../sync" }
 
 # Spacedrive Sub-crates
 sd-actors       = { path = "../../../crates/actors" }
@@ -16,13 +18,16 @@ sd-prisma       = { path = "../../../crates/prisma" }
 sd-utils        = { path = "../../../crates/utils" }
 
 # Workspace dependencies
+anyhow              = { workspace = true }
 async-stream        = { workspace = true }
 base64              = { workspace = true }
 blake3              = { workspace = true }
 chrono              = { workspace = true, features = ["serde"] }
 flume               = { workspace = true }
 futures             = { workspace = true }
 futures-concurrency = { workspace = true }
+iroh                = { workspace = true, features = ["discovery-local-network"] }
+quic-rpc            = { workspace = true, features = ["iroh-transport", "quinn-transport"] }
 rmp-serde           = { workspace = true }
 rspc                = { workspace = true }
 serde               = { workspace = true, features = ["derive"] }
@@ -37,12 +42,9 @@ uuid                = { workspace = true, features = ["serde"] }
 zeroize             = { workspace = true }
 
 # External dependencies
-anyhow   = "1.0.86"
-dashmap  = "6.1.0"
-iroh     = { version = "0.29.0", features = ["discovery-local-network"] }
-paste    = "=1.0.15"
-quic-rpc = { version = "0.17.1", features = ["iroh-transport", "quinn-transport"] }
-quinn    = { package = "iroh-quinn", version = "0.12" }
+dashmap = "6.1.0"
+paste   = "=1.0.15"
+quinn   = { package = "iroh-quinn", version = "0.12" }
 # Using whatever version of reqwest that reqwest-middleware uses, just putting here to enable some features
 reqwest                  = { version = "0.12", features = ["json", "native-tls-vendored", "stream"] }
 reqwest-middleware       = { version = "0.4", features = ["json"] }

core/Cargo.toml

@@ -1,27 +1,30 @@
 use crate::p2p::{NotifyUser, UserResponse};
 
-use sd_cloud_schema::{Client, Request, Response, ServicesALPN};
+use sd_cloud_schema::{Client, Service, ServicesALPN};
 
 use std::{net::SocketAddr, sync::Arc, time::Duration};
 
 use futures::Stream;
 use iroh::relay::RelayUrl;
-use quic_rpc::{transport::quinn::QuinnConnector, RpcClient, RpcMessage};
+use quic_rpc::{client::QuinnConnector, RpcClient};
 use quinn::{crypto::rustls::QuicClientConfig, ClientConfig, Endpoint};
 use reqwest::{IntoUrl, Url};
 use reqwest_middleware::{reqwest, ClientBuilder, ClientWithMiddleware};
+// use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware};
 use tokio::sync::{Mutex, RwLock};
 use tracing::warn;
 
 use super::{
 	error::Error, key_manager::KeyManager, p2p::CloudP2P, token_refresher::TokenRefresher,
 };
 
+pub type CloudServicesClient = Client<QuinnConnector<Service>>;
+
 #[derive(Debug, Default, Clone)]
-enum ClientState<In: RpcMessage, Out: RpcMessage> {
+enum ClientState {
 	#[default]
 	NotConnected,
-	Connected(Client<QuinnConnector<In, Out>>),
+	Connected(CloudServicesClient),
 }
 
 /// Cloud services are a optional feature that allows you to interact with the cloud services
@@ -34,7 +37,7 @@ enum ClientState<In: RpcMessage, Out: RpcMessage> {
 /// that core can always operate without the cloud services.
 #[derive(Debug)]
 pub struct CloudServices {
-	client_state: Arc<RwLock<ClientState<Response, Request>>>,
+	client_state: Arc<RwLock<ClientState>>,
 	get_cloud_api_address: Url,
 	http_client: ClientWithMiddleware,
 	domain_name: String,
@@ -157,7 +160,7 @@ impl CloudServices {
 		http_client: &ClientWithMiddleware,
 		get_cloud_api_address: Url,
 		domain_name: String,
-	) -> Result<Client<QuinnConnector<Response, Request>>, Error> {
+	) -> Result<CloudServicesClient, Error> {
 		let cloud_api_address = http_client
 			.get(get_cloud_api_address)
 			.send()
@@ -256,7 +259,7 @@ impl CloudServices {
 			.map_err(Error::FailedToCreateEndpoint)?;
 		endpoint.set_default_client_config(client_config);
 
-		Ok(Client::new(RpcClient::new(QuinnConnector::new(
+		Ok(Client::new(RpcClient::new(QuinnConnector::<Service>::new(
 			endpoint,
 			cloud_api_address,
 			domain_name,
@@ -268,7 +271,7 @@ impl CloudServices {
 	/// If the client is not connected, it will try to connect to the cloud services.
 	/// Available routes documented in
 	/// [`sd_cloud_schema::Service`](https://github.com/spacedriveapp/cloud-services-schema).
-	pub async fn client(&self) -> Result<Client<QuinnConnector<Response, Request>>, Error> {
+	pub async fn client(&self) -> Result<CloudServicesClient, Error> {
 		if let ClientState::Connected(client) = { &*self.client_state.read().await } {
 			return Ok(client.clone());
 		}

core/crates/cloud-services/Cargo.toml

@@ -177,10 +177,22 @@ pub enum Error {
 	ReadNonceStreamDecryption(io::Error),
 	#[error("Incomplete download bytes sync messages")]
 	IncompleteDownloadBytesSyncMessages,
+	#[error("Timed out while waiting to recive thumbnail data")]
+	ThumbnailRequestTimeout,
 
 	// Temporary errors
 	#[error("Device missing secret key for decrypting sync messages")]
 	MissingKeyHash,
+	#[error("Not Implemented yet")]
+	NotImplemented,
+	#[error("Device not found")]
+	DeviceNotFound,
+	#[error("Invalid CAS ID")]
+	InvalidCasId,
+	#[error("Internal Error")]
+	InternalError,
+	#[error("Remote Device Error")]
+	RemoteDeviceError,
 }
 
 #[derive(thiserror::Error, Debug)]

core/crates/cloud-services/src/client.rs

@@ -8,8 +8,9 @@ use sd_cloud_schema::{
 	SecretKey as IrohSecretKey,
 };
 use sd_crypto::{CryptoRng, SeedableRng};
+use sd_prisma::prisma::file_path::cas_id;
 
-use std::{sync::Arc, time::Duration};
+use std::{path::PathBuf, sync::Arc, time::Duration};
 
 use iroh::{
 	discovery::{
@@ -35,6 +36,12 @@ pub struct JoinedLibraryCreateArgs {
 	pub description: Option<String>,
 }
 
+#[derive(Debug)]
+pub struct RecivedGetThumbnailArgs {
+	pub cas_id: cas_id::Type,
+	pub error: Option<Error>,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, specta::Type)]
 #[serde(transparent)]
 #[repr(transparent)]
@@ -108,6 +115,7 @@ impl CloudP2P {
 		dns_origin_domain: String,
 		dns_pkarr_url: Url,
 		relay_url: RelayUrl,
+		data_directory: PathBuf,
 	) -> Result<Self, Error> {
 		let dht_discovery = DhtDiscovery::builder()
 			.secret_key(iroh_secret_key.clone())
@@ -156,6 +164,7 @@ impl CloudP2P {
 				cloud_services,
 				msgs_tx.clone(),
 				endpoint,
+				data_directory,
 			)
 			.await?;
 			let user_response_rx = cloud_services.user_response_rx.clone();
@@ -232,6 +241,28 @@ impl CloudP2P {
 			.await
 			.expect("Channel closed");
 	}
+
+	/// Requests the binary of a thumbnail from a specific device endpoint
+	///
+	/// # Panics
+	/// Will panic if the actor channel is closed, which should never happen
+	pub async fn request_thumbnail_data(
+		&self,
+		device_pub_id: devices::PubId,
+		cas_id: cas_id::Type,
+		library_pub_id: libraries::PubId,
+		tx: oneshot::Sender<RecivedGetThumbnailArgs>,
+	) {
+		self.msgs_tx
+			.send_async(runner::Message::Request(runner::Request::GetThumbnail {
+				device_pub_id,
+				cas_id,
+				library_pub_id,
+				tx,
+			}))
+			.await
+			.expect("Channel closed");
+	}
 }
 
 impl Drop for CloudP2P {