From 69a1e91d4ba8a723da1f7c68a95505e5672a3f21 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Sat, 19 Sep 2026 17:06:31 +0100 Subject: [PATCH] Add new endpoints --- sync-server/src/app_state/cursors.rs | 17 +- sync-server/src/app_state/database.rs | 20 +- .../src/app_state/websocket/broadcasts.rs | 84 +-- sync-server/src/app_state/websocket/models.rs | 35 +- sync-server/src/app_state/websocket/utils.rs | 25 +- sync-server/src/server/endpoints.rs | 10 +- sync-server/src/server/endpoints/events.rs | 49 ++ .../fetch_latest_document_version.rs | 2 +- .../src/server/endpoints/get_file_manifest.rs | 43 ++ sync-server/src/server/endpoints/index.rs | 1 + .../src/server/endpoints/push_document.rs | 210 ------ .../src/server/endpoints/put_file_content.rs | 624 ++++++++++++++++++ .../src/server/endpoints/put_file_manifest.rs | 122 ++++ sync-server/src/server/endpoints/utils.rs | 32 + .../src/server/endpoints/vault_snapshot.rs | 36 + sync-server/src/server/endpoints/websocket.rs | 238 +++---- sync-server/src/server/requests.rs | 142 +++- sync-server/src/server/responses.rs | 20 +- 18 files changed, 1199 insertions(+), 511 deletions(-) create mode 100644 sync-server/src/server/endpoints/events.rs create mode 100644 sync-server/src/server/endpoints/get_file_manifest.rs delete mode 100644 sync-server/src/server/endpoints/push_document.rs create mode 100644 sync-server/src/server/endpoints/put_file_content.rs create mode 100644 sync-server/src/server/endpoints/put_file_manifest.rs create mode 100644 sync-server/src/server/endpoints/utils.rs create mode 100644 sync-server/src/server/endpoints/vault_snapshot.rs diff --git a/sync-server/src/app_state/cursors.rs b/sync-server/src/app_state/cursors.rs index d083e1ac..2c57a49d 100644 --- a/sync-server/src/app_state/cursors.rs +++ b/sync-server/src/app_state/cursors.rs @@ -6,11 +6,8 @@ use tokio::sync::Mutex; use super::{ database::models::{DeviceId, VaultId}, websocket::{ - broadcasts::Broadcasts, - models::{ - ClientCursors, CursorPositionFromServer, WebSocketServerMessage, - WebSocketServerMessageWithOrigin, - }, + broadcasts::{Broadcasts, Notification}, + models::{ClientCursors, CursorPositionFromServer}, }, }; use crate::{ @@ -91,13 +88,11 @@ impl Cursors { for (vault_id, cursors) in vault_to_cursors.iter() { self.broadcasts - .send_document_update( + .send( vault_id.clone(), - WebSocketServerMessageWithOrigin::new(WebSocketServerMessage::CursorPositions( - CursorPositionFromServer { - clients: cursors.iter().map(|c| c.client_cursors.clone()).collect(), - }, - )), + Notification::Cursors(CursorPositionFromServer { + clients: cursors.iter().map(|c| c.client_cursors.clone()).collect(), + }), ) .await; } diff --git a/sync-server/src/app_state/database.rs b/sync-server/src/app_state/database.rs index 216a537d..3bd5e9c0 100644 --- a/sync-server/src/app_state/database.rs +++ b/sync-server/src/app_state/database.rs @@ -11,16 +11,9 @@ use tokio::time::Instant; pub mod models; -mod get_document_version; -mod get_latest_document; -mod get_latest_document_by_path; -mod get_latest_documents; -mod get_latest_documents_since; -mod get_max_update_id_in_vault; -mod get_push_acknowledgement; -mod insert_document_version; +mod mutations; +mod queries; -use super::websocket::broadcasts::Broadcasts; use crate::config::database_config::DatabaseConfig; #[derive(Clone)] @@ -41,14 +34,13 @@ impl std::fmt::Debug for PoolWithTimestamp { #[derive(Clone, Debug)] pub struct Database { config: DatabaseConfig, - broadcasts: Broadcasts, connection_pools: Arc>>, } pub type Transaction<'a> = sqlx::Transaction<'a, Sqlite>; impl Database { - pub async fn try_new(config: &DatabaseConfig, broadcasts: &Broadcasts) -> Result { + pub async fn try_new(config: &DatabaseConfig) -> Result { tokio::fs::create_dir_all(&config.databases_directory_path) .await .with_context(|| { @@ -86,7 +78,6 @@ impl Database { let database = Self { config: config.clone(), connection_pools: Arc::new(Mutex::new(connection_pools)), - broadcasts: broadcasts.clone(), }; // Start background task to cleanup idle connection pools @@ -109,6 +100,9 @@ impl Database { .auto_vacuum(sqlx::sqlite::SqliteAutoVacuum::Full) .busy_timeout(Duration::from_secs(3600)) .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .synchronous(sqlx::sqlite::SqliteSynchronous::Full) + .pragma("fullfsync", "ON") + .foreign_keys(true) .log_slow_statements(log::LevelFilter::Warn, Duration::from_secs(30)); let pool = SqlitePoolOptions::new() @@ -172,7 +166,7 @@ impl Database { let mut transaction = self.create_readonly_transaction(vault).await?; // sqlx doesn't support immediate transactions for sqlite: https://github.com/launchbadge/sqlx/issues/481 - sqlx::query!("END; BEGIN IMMEDIATE;") + sqlx::query("END; BEGIN IMMEDIATE;") .execute(&mut *transaction) .await?; diff --git a/sync-server/src/app_state/websocket/broadcasts.rs b/sync-server/src/app_state/websocket/broadcasts.rs index 60ae0219..1b157b13 100644 --- a/sync-server/src/app_state/websocket/broadcasts.rs +++ b/sync-server/src/app_state/websocket/broadcasts.rs @@ -1,69 +1,47 @@ +use super::models::CursorPositionFromServer; +use crate::{app_state::database::models::VaultId, config::server_config::ServerConfig}; use std::{collections::HashMap, sync::Arc}; - -use anyhow::Context; -use log::{debug, warn}; use tokio::sync::{Mutex, broadcast}; -use super::models::WebSocketServerMessageWithOrigin; -use crate::{ - app_state::database::models::VaultId, config::server_config::ServerConfig, errors::server_error, -}; +#[derive(Debug, Clone)] +pub enum Notification { + VaultUpdate, + Cursors(CursorPositionFromServer), +} #[derive(Debug, Clone)] pub struct Broadcasts { - max_clients_per_vault: usize, - tx: Arc>>>, + capacity: usize, + senders: Arc>>>, } impl Broadcasts { - pub fn new(server_config: &ServerConfig) -> Self { + pub fn new(config: &ServerConfig) -> Self { Self { - max_clients_per_vault: server_config.max_clients_per_vault, - tx: Arc::new(Mutex::new(HashMap::new())), + capacity: config.max_clients_per_vault.max(1), + senders: Arc::default(), } } - pub async fn get_receiver( - &self, - vault: VaultId, - ) -> broadcast::Receiver { - let tx = self.get_or_create(vault).await; - - tx.subscribe() - } - - /// Notify all clients (who are subscribed to the vault) about an update. - /// We only log failures and don't propagate them. - pub async fn send_document_update( - &self, - vault: VaultId, - document: WebSocketServerMessageWithOrigin, - ) { - let tx = self.get_or_create(vault.clone()).await; - - if tx.receiver_count() == 0 { - debug!("Skipping broadcast, no clients connected for vault `{vault}`"); - return; - } - - let result = tx - .send(document) - .context("Cannot broadcast server message to websocket listeners") - .map_err(server_error); - - if result.is_err() { - warn!("Failed to send message: {result:?}"); - } - } - - async fn get_or_create( - &self, - vault: VaultId, - ) -> broadcast::Sender { - let mut tx = self.tx.lock().await; - - tx.entry(vault) - .or_insert_with(|| broadcast::channel(self.max_clients_per_vault).0.clone()) + async fn sender(&self, vault: VaultId) -> broadcast::Sender { + self.senders + .lock() + .await + .entry(vault) + .or_insert_with(|| broadcast::channel(self.capacity).0) .clone() } + + pub async fn get_receiver(&self, vault: VaultId) -> broadcast::Receiver { + self.sender(vault).await.subscribe() + } + + pub async fn send(&self, vault: VaultId, notification: Notification) { + // No listeners (or lagging listeners) is normal; they replay the log. + let _ = self.sender(vault).await.send(notification); + } + + pub async fn notify_about_vault_update(&self, vault: VaultId) { + self.send(vault, Notification::VaultUpdate).await; + } } diff --git a/sync-server/src/app_state/websocket/models.rs b/sync-server/src/app_state/websocket/models.rs index e037fb7e..32d97966 100644 --- a/sync-server/src/app_state/websocket/models.rs +++ b/sync-server/src/app_state/websocket/models.rs @@ -1,9 +1,7 @@ use serde::{Deserialize, Serialize}; use ts_rs::TS; -use crate::app_state::database::models::{ - DeviceId, DocumentId, DocumentVersionWithoutContent, VaultUpdateId, -}; +use crate::app_state::database::models::{DeviceId, DocumentId, EventBatch, VaultUpdateId}; #[derive(TS, Deserialize, Clone, Debug)] #[serde(rename_all = "camelCase")] @@ -57,13 +55,6 @@ pub struct CursorPositionFromServer { pub clients: Vec, } -#[derive(TS, Serialize, Clone, Debug)] -#[serde(rename_all = "camelCase")] -pub struct WebSocketVaultUpdate { - pub documents: Vec, - pub is_initial_sync: bool, -} - #[derive(TS, Deserialize, Clone, Debug)] #[serde(rename_all = "camelCase", tag = "type")] #[ts(export)] @@ -76,28 +67,6 @@ pub enum WebSocketClientMessage { #[serde(rename_all = "camelCase", tag = "type")] #[ts(export)] pub enum WebSocketServerMessage { - VaultUpdate(WebSocketVaultUpdate), + VaultEvents(EventBatch), CursorPositions(CursorPositionFromServer), } - -#[derive(Clone, Debug)] -pub struct WebSocketServerMessageWithOrigin { - pub origin_device_id: Option, - pub message: WebSocketServerMessage, -} - -impl WebSocketServerMessageWithOrigin { - pub fn new(message: WebSocketServerMessage) -> Self { - Self { - origin_device_id: None, - message, - } - } - - pub fn with_origin(origin_device_id: DeviceId, message: WebSocketServerMessage) -> Self { - Self { - origin_device_id: Some(origin_device_id), - message, - } - } -} diff --git a/sync-server/src/app_state/websocket/utils.rs b/sync-server/src/app_state/websocket/utils.rs index 1e0dd243..e72a3775 100644 --- a/sync-server/src/app_state/websocket/utils.rs +++ b/sync-server/src/app_state/websocket/utils.rs @@ -4,10 +4,7 @@ use futures::{sink::SinkExt, stream::SplitSink}; use super::models::{WebSocketClientMessage, WebSocketHandshake, WebSocketServerMessage}; use crate::{ - app_state::{ - AppState, - database::models::{DocumentVersionWithoutContent, VaultId, VaultUpdateId}, - }, + app_state::{AppState, database::models::VaultId}, config::user_config::User, errors::{SyncServerError, server_error, unauthenticated_error}, server::auth::auth, @@ -44,26 +41,6 @@ pub fn get_authenticated_handshake( } } -pub async fn get_unseen_documents( - state: &AppState, - vault_id: &VaultId, - last_seen_vault_update_id: Option, -) -> Result, SyncServerError> { - if let Some(update_id) = last_seen_vault_update_id { - state - .database - .get_latest_documents_since(vault_id, update_id, None) - .await - .map_err(server_error) - } else { - state - .database - .get_latest_documents(vault_id, None) - .await - .map_err(server_error) - } -} - pub async fn send_update_over_websocket( update: &WebSocketServerMessage, sender: &mut SplitSink, diff --git a/sync-server/src/server/endpoints.rs b/sync-server/src/server/endpoints.rs index 243cf7fa..ed873473 100644 --- a/sync-server/src/server/endpoints.rs +++ b/sync-server/src/server/endpoints.rs @@ -1,7 +1,13 @@ +pub mod events; pub mod fetch_document_version_content; pub mod fetch_latest_document_version; -pub mod fetch_latest_documents; +pub mod get_file_manifest; pub mod index; +pub mod method_not_allowed; +pub mod not_found; pub mod ping; -pub mod push_document; +pub mod put_file_content; +pub mod put_file_manifest; +mod utils; +pub mod vault_snapshot; pub mod websocket; diff --git a/sync-server/src/server/endpoints/events.rs b/sync-server/src/server/endpoints/events.rs new file mode 100644 index 00000000..ef113864 --- /dev/null +++ b/sync-server/src/server/endpoints/events.rs @@ -0,0 +1,49 @@ +use axum::{ + Json, + extract::{Path, Query, State}, +}; +use log::debug; +use serde::Deserialize; + +use crate::{ + app_state::{ + AppState, + database::models::{EventBatch, VaultId}, + }, + errors::{SyncServerError, client_error, server_error}, + utils::normalize::normalize, +}; + +#[derive(Deserialize)] +pub struct EventsPath { + #[serde(deserialize_with = "normalize")] + vault_id: VaultId, +} + +#[derive(Deserialize)] +pub struct EventsQuery { + pub after: i64, +} + +#[axum::debug_handler] +pub async fn events( + Path(path): Path, + Query(query): Query, + State(state): State, +) -> Result, SyncServerError> { + debug!( + "Fetching events after `{}` for vault `{}`", + query.after, path.vault_id + ); + + if query.after < 0 { + return Err(client_error(anyhow::anyhow!("Invalid event cursor"))); + } + + state + .database + .events_after(&path.vault_id, query.after) + .await + .map(Json) + .map_err(server_error) +} diff --git a/sync-server/src/server/endpoints/fetch_latest_document_version.rs b/sync-server/src/server/endpoints/fetch_latest_document_version.rs index a9973606..a6b662e9 100644 --- a/sync-server/src/server/endpoints/fetch_latest_document_version.rs +++ b/sync-server/src/server/endpoints/fetch_latest_document_version.rs @@ -35,7 +35,7 @@ pub async fn fetch_latest_document_version( let latest_version = state .database - .get_latest_document(&vault_id, &document_id, None) + .get_latest_document_version(&vault_id, &document_id, None) .await .map_err(server_error)? .map_or_else( diff --git a/sync-server/src/server/endpoints/get_file_manifest.rs b/sync-server/src/server/endpoints/get_file_manifest.rs new file mode 100644 index 00000000..cfafda02 --- /dev/null +++ b/sync-server/src/server/endpoints/get_file_manifest.rs @@ -0,0 +1,43 @@ +use axum::{ + Json, + extract::{Path, State}, +}; +use log::debug; +use serde::Deserialize; + +use crate::{ + app_state::{ + AppState, + database::{ + Database, + models::{FileManifest, VaultId}, + }, + }, + errors::{SyncServerError, server_error}, + utils::normalize::normalize, +}; + +#[derive(Deserialize)] +pub struct GetFileManifestPath { + #[serde(deserialize_with = "normalize")] + vault_id: VaultId, +} + +#[axum::debug_handler] +pub async fn get_file_manifest( + Path(path): Path, + State(state): State, +) -> Result, SyncServerError> { + debug!("Fetching file manifest for vault `{}`", path.vault_id); + + let mut transaction = state + .database + .create_readonly_transaction(&path.vault_id) + .await + .map_err(server_error)?; + + Database::current_file_manifest(&mut transaction) + .await + .map(Json) + .map_err(server_error) +} diff --git a/sync-server/src/server/endpoints/index.rs b/sync-server/src/server/endpoints/index.rs index 982b0b88..f6ea022d 100644 --- a/sync-server/src/server/endpoints/index.rs +++ b/sync-server/src/server/endpoints/index.rs @@ -1,5 +1,6 @@ use axum::response::{Html, IntoResponse}; +#[axum::debug_handler] pub async fn index() -> impl IntoResponse { const HTML_CONTENT: &str = include_str!("../assets/index.html"); let html_content = HTML_CONTENT; diff --git a/sync-server/src/server/endpoints/push_document.rs b/sync-server/src/server/endpoints/push_document.rs deleted file mode 100644 index 676e8baa..00000000 --- a/sync-server/src/server/endpoints/push_document.rs +++ /dev/null @@ -1,210 +0,0 @@ -use anyhow::{Context as _, anyhow}; -use axum::{ - Extension, Json, - extract::{Path, State}, -}; -use axum_extra::TypedHeader; -use base64::{Engine as _, engine::general_purpose::STANDARD}; -use reconcile_text::{BuiltinTokenizer, EditedText, NumberOrText}; -use serde::Deserialize; -use sha2::{Digest, Sha256}; - -use crate::server::{ - device_id_header::DeviceIdHeader, - requests::{PushContent, PushDocument}, - responses::DocumentUpdateResponse, -}; -use crate::{ - app_state::{ - AppState, - database::{ - Database, - models::{DocumentId, StoredDocumentVersion, VaultId}, - }, - }, - config::user_config::User, - errors::{SyncServerError, client_error, not_found_error, server_error}, - utils::{ - find_first_available_path::find_first_available_path, normalize::normalize, - sanitize_path::sanitize_path, - }, -}; - -#[derive(Deserialize)] -pub struct PushDocumentPath { - #[serde(deserialize_with = "normalize")] - vault_id: VaultId, - - document_id: DocumentId, -} - -#[axum::debug_handler] -pub async fn push_document( - Path(PushDocumentPath { - vault_id, - document_id, - }): Path, - Extension(user): Extension, - TypedHeader(device_id): TypedHeader, - State(state): State, - Json(push): Json, -) -> Result, SyncServerError> { - accept_push( - &state.database, - &vault_id, - document_id, - user.name, - device_id.0, - push, - ) - .await - .map(Json) -} - -/// A push is a compare-and-swap of one document's immutable head. The write lock -/// covers idempotency lookup, base validation, path allocation, and the commit. -async fn accept_push( - database: &Database, - vault_id: &VaultId, - document_id: DocumentId, - user_id: String, - device_id: String, - push: PushDocument, -) -> Result { - let fingerprint = fingerprint(document_id, &push); - - let mut transaction = database - .create_write_transaction(vault_id) - .await - .map_err(server_error)?; - - if let Some((stored_fingerprint, acknowledged)) = database - .get_push_acknowledgement(vault_id, push.request_id, &mut transaction) - .await - .map_err(server_error)? - { - if stored_fingerprint != fingerprint { - return Err(client_error(anyhow!( - "Request ID was reused with a different payload" - ))); - } - return Ok(DocumentUpdateResponse::Accepted(acknowledged.into())); - } - - let latest = database - .get_latest_document(vault_id, &document_id, Some(&mut transaction)) - .await - .map_err(server_error)?; - - match &latest { - Some(latest) - if latest.is_deleted || Some(latest.vault_update_id) != push.parent_version_id => - { - return Ok(DocumentUpdateResponse::StaleBase(latest.clone().into())); - } - - None if push.parent_version_id.is_some() => { - return Err(not_found_error(anyhow!( - "Document with id `{document_id}` not found" - ))); - } - - _ => {} - } - - let is_deleted = matches!(&push.content, PushContent::Delete); - let (content, requested_path) = match push.content { - PushContent::Snapshot(content) => ( - STANDARD - .decode(content) - .context("Invalid base64 snapshot") - .map_err(client_error)?, - sanitize_path(&push.relative_path), - ), - PushContent::Diff(diff) => ( - reconstruct_diff(latest.as_ref(), diff)?, - sanitize_path(&push.relative_path), - ), - PushContent::Delete => { - let latest = latest - .as_ref() - .ok_or_else(|| client_error(anyhow!("Cannot delete a missing document")))?; - (latest.content.clone(), latest.relative_path.clone()) - } - }; - - let relative_path = if latest - .as_ref() - .is_some_and(|version| version.relative_path == requested_path) - { - requested_path - } else { - find_first_available_path(vault_id, &requested_path, database, &mut transaction) - .await - .map_err(server_error)? - }; - - let new_version = StoredDocumentVersion { - vault_update_id: database - .get_max_update_id_in_vault(vault_id, Some(&mut transaction)) - .await - .map_err(server_error)? - + 1, - document_id, - relative_path, - content, - updated_date: chrono::Utc::now(), - is_deleted, - user_id, - device_id, - }; - - database - .insert_document_version( - vault_id, - &new_version, - push.request_id, - &fingerprint, - transaction, - ) - .await - .map_err(server_error)?; - Ok(DocumentUpdateResponse::Accepted(new_version.into())) -} - -fn reconstruct_diff( - parent: Option<&StoredDocumentVersion>, - diff: Vec, -) -> Result, SyncServerError> { - // reconcile-text 0.8 negates deletion lengths; i64::MIN would overflow. - if diff - .iter() - .any(|item| matches!(item, NumberOrText::Number(i64::MIN))) - { - return Err(client_error(anyhow!("Invalid diff length"))); - } - let parent = parent.ok_or_else(|| client_error(anyhow!("A diff requires a parent version")))?; - let parent_text = str::from_utf8(&parent.content) - .context("Parent is not UTF-8 text") - .map_err(client_error)?; - // Transport decoding only: reconciliation belongs exclusively to clients. - Ok( - EditedText::from_diff(parent_text, diff, &*BuiltinTokenizer::Word) - .context("Failed to apply diff to parent document") - .map_err(client_error)? - .apply() - .text() - .into_bytes(), - ) -} - -/// Session/device IDs may change across retries; the requested mutation may not. -fn fingerprint(document_id: DocumentId, push: &PushDocument) -> Vec { - Sha256::digest( - serde_json::to_vec(&(document_id, push)).expect("Document pushes are serializable"), - ) - .to_vec() -} - -#[cfg(test)] -mod tests; diff --git a/sync-server/src/server/endpoints/put_file_content.rs b/sync-server/src/server/endpoints/put_file_content.rs new file mode 100644 index 00000000..9d757bb8 --- /dev/null +++ b/sync-server/src/server/endpoints/put_file_content.rs @@ -0,0 +1,624 @@ +use super::utils::find_already_processed_event; +use crate::{ + app_state::{ + AppState, + database::{ + Database, + models::{DocumentId, EventRecord, StoredDocumentVersion, VaultEvent, VaultId}, + }, + }, + config::user_config::User, + errors::{SyncServerError, client_error, not_found_error, server_error}, + server::{ + device_id_header::DeviceIdHeader, + requests::{PushContent, PutFileContent}, + responses::DocumentUpdateResponse, + }, + utils::normalize::normalize, +}; +use anyhow::anyhow; +use axum::{ + Extension, Json, + extract::{Path, State}, +}; +use axum_extra::TypedHeader; +use base64::{Engine as _, engine::general_purpose::STANDARD}; +use log::debug; +use reconcile_text::{BuiltinTokenizer, EditedText, NumberOrText}; +use serde::Deserialize; +use sha2::{Digest, Sha256}; + +#[derive(Deserialize)] +pub struct PutFileContentPath { + #[serde(deserialize_with = "normalize")] + vault_id: VaultId, + document_id: DocumentId, +} + +#[axum::debug_handler] +pub async fn put_file_content( + Path(path): Path, + Extension(user): Extension, + TypedHeader(device): TypedHeader, + State(state): State, + Json(push): Json, +) -> Result, SyncServerError> { + debug!( + "Pushing document `{}` in vault `{}`", + path.document_id, path.vault_id + ); + + let fingerprint = Sha256::digest( + serde_json::to_vec(&("content", path.document_id, &push)) + .map_err(|error| server_error(error.into()))?, + ); + + let mut tx = state + .database + .create_write_transaction(&path.vault_id) + .await + .map_err(server_error)?; + + if let Some(event) = + find_already_processed_event(&mut tx, push.request_id, &fingerprint).await? + { + return match event { + VaultEvent::Content { document } => { + Ok(Json(DocumentUpdateResponse::Accepted(document))) + } + VaultEvent::FileManifest { .. } => Err(server_error(anyhow!( + "Stored event does not match content request" + ))), + }; + } + + let parent = state + .database + .get_latest_document_version(&path.vault_id, &path.document_id, Some(&mut tx)) + .await + .map_err(server_error)?; + + if let Some(parent) = &parent { + if Some(parent.vault_update_id) != push.parent_version_id { + return Ok(Json(DocumentUpdateResponse::StaleBase( + parent.clone().into(), + ))); + } + } else if push.parent_version_id.is_some() { + return Err(not_found_error(anyhow!( + "Document {} does not exist", + path.document_id + ))); + } + + let content = match push.content { + PushContent::Snapshot(bytes) => STANDARD + .decode(bytes) + .map_err(|error| client_error(error.into()))?, + + PushContent::Diff(diff) => { + if diff + .iter() + .any(|n| matches!(n, NumberOrText::Number(i64::MIN))) + { + return Err(client_error(anyhow!("Invalid diff length"))); + } + + let parent = parent + .as_ref() + .ok_or_else(|| client_error(anyhow!("Diff requires a parent")))?; + + let text = + str::from_utf8(&parent.content).map_err(|error| client_error(error.into()))?; + + EditedText::from_diff(text, diff, &*BuiltinTokenizer::Word) + .map_err(|error| client_error(error.into()))? + .apply() + .text() + .into_bytes() + } + }; + + let version = StoredDocumentVersion { + vault_update_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint) + .await + .map_err(server_error)?, + document_id: path.document_id, + content, + updated_date: chrono::Utc::now(), + user_id: user.name, + device_id: device.0, + }; + + Database::insert_document_version(&mut tx, &version) + .await + .map_err(server_error)?; + + let response = DocumentUpdateResponse::Accepted((&version).into()); + + let event = EventRecord { + event_id: version.vault_update_id, + request_id: push.request_id, + event: VaultEvent::Content { + document: version.into(), + }, + }; + + Database::write_event(&mut tx, &event) + .await + .map_err(server_error)?; + + tx.commit() + .await + .map_err(|error| server_error(error.into()))?; + + state + .broadcasts + .notify_about_vault_update(path.vault_id) + .await; + + Ok(Json(response)) +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use super::*; + use crate::{ + app_state::database::{ + Database, + models::{DocumentVersionWithoutContent, StoredDocumentVersion, VaultUpdateId}, + }, + config::{Config, database_config::DatabaseConfig, server_config::ServerConfig}, + }; + + async fn send_document( + state: &AppState, + vault: &VaultId, + document_id: DocumentId, + user_id: String, + device_id: String, + push: PutFileContent, + ) -> Result { + put_file_content( + Path(PutFileContentPath { + vault_id: vault.clone(), + document_id, + }), + Extension(User { + name: user_id, + token: String::new(), + vault_access: Default::default(), + }), + TypedHeader(DeviceIdHeader(device_id)), + State(state.clone()), + Json(push), + ) + .await + .map(|Json(response)| response) + } + + struct Push { + document_id: DocumentId, + request_id: uuid::Uuid, + parent_version_id: Option, + content: PushContent, + user_id: String, + device_id: String, + } + + impl Push { + async fn send( + self, + state: &AppState, + vault: &VaultId, + ) -> Result { + send_document( + state, + vault, + self.document_id, + self.user_id, + self.device_id, + PutFileContent { + request_id: self.request_id, + parent_version_id: self.parent_version_id, + content: self.content, + }, + ) + .await + } + } + + struct TestVault { + directory: PathBuf, + state: AppState, + vault: VaultId, + } + + impl TestVault { + async fn new() -> Self { + let directory = + std::env::temp_dir().join(format!("vaultlink-cas-{}", uuid::Uuid::new_v4())); + let state = Self::open(&directory).await; + Self { + directory, + state, + vault: "test".to_owned(), + } + } + + async fn open(directory: &std::path::Path) -> AppState { + AppState::try_new(Config { + database: DatabaseConfig { + databases_directory_path: directory.to_owned(), + ..Default::default() + }, + server: ServerConfig { + max_clients_per_vault: 16, + ..Default::default() + }, + ..Default::default() + }) + .await + .unwrap() + } + + async fn send(&self, push: Push) -> DocumentUpdateResponse { + push.send(&self.state, &self.vault).await.unwrap() + } + + async fn latest(&self, document: DocumentId) -> StoredDocumentVersion { + self.state + .database + .get_latest_document_version(&self.vault, &document, None) + .await + .unwrap() + .unwrap() + } + } + + impl Drop for TestVault { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.directory); + } + } + + fn snapshot(document_id: DocumentId, parent: Option, text: &str) -> Push { + Push { + document_id, + request_id: uuid::Uuid::new_v4(), + parent_version_id: parent, + content: PushContent::Snapshot(STANDARD.encode(text)), + user_id: "user".to_owned(), + device_id: "device".to_owned(), + } + } + + fn accepted(response: DocumentUpdateResponse) -> DocumentVersionWithoutContent { + match response { + DocumentUpdateResponse::Accepted(version) => version, + response @ DocumentUpdateResponse::StaleBase(_) => { + panic!("Expected Accepted, got {response:?}") + } + } + } + + #[tokio::test] + async fn json_snapshot_preserves_bytes_and_rejects_invalid_base64_without_a_version() { + let vault = TestVault::new().await; + let document = uuid::Uuid::new_v4(); + let content = [0, 255, 13, 10, 42]; + let request: PutFileContent = serde_json::from_value(serde_json::json!({ + "requestId": uuid::Uuid::new_v4(), + "parentVersionId": null, + "content": { "type": "Snapshot", "value": STANDARD.encode(content) }, + })) + .unwrap(); + let base = accepted( + send_document( + &vault.state, + &vault.vault, + document, + "user".to_owned(), + "device".to_owned(), + request, + ) + .await + .unwrap(), + ); + assert_eq!(vault.latest(document).await.content, content); + let mut tx = vault + .state + .database + .create_readonly_transaction(&vault.vault) + .await + .unwrap(); + let stored_document_id: String = + sqlx::query_scalar("SELECT document_id FROM documents WHERE vault_update_id = ?") + .bind(base.vault_update_id) + .fetch_one(&mut *tx) + .await + .unwrap(); + assert_eq!(stored_document_id, document.to_string()); + tx.commit().await.unwrap(); + assert!(serde_json::to_value(&base).unwrap().get("mtime").is_none()); + + let invalid: PutFileContent = serde_json::from_value(serde_json::json!({ + "requestId": uuid::Uuid::new_v4(), + "parentVersionId": base.vault_update_id, + "content": { "type": "Snapshot", "value": "not base64!" }, + })) + .unwrap(); + assert!(matches!( + send_document( + &vault.state, + &vault.vault, + document, + "user".to_owned(), + "device".to_owned(), + invalid, + ) + .await, + Err(SyncServerError::ClientError(_)) + )); + assert_eq!( + vault.latest(document).await.vault_update_id, + base.vault_update_id + ); + } + + #[tokio::test] + async fn concurrent_writers_only_accept_one_snapshot_and_preserve_the_base() { + let vault = TestVault::new().await; + let document = uuid::Uuid::new_v4(); + let base = accepted(vault.send(snapshot(document, None, "base")).await); + let (left, right) = tokio::join!( + vault.send(snapshot(document, Some(base.vault_update_id), "left")), + vault.send(snapshot(document, Some(base.vault_update_id), "right")), + ); + let (winner, rejected) = match (left, right) { + ( + DocumentUpdateResponse::Accepted(winner), + DocumentUpdateResponse::StaleBase(rejected), + ) + | ( + DocumentUpdateResponse::StaleBase(rejected), + DocumentUpdateResponse::Accepted(winner), + ) => (winner, rejected), + outcomes => panic!("Expected one accepted push and one stale base: {outcomes:?}"), + }; + let latest = vault.latest(document).await; + assert_eq!(winner.vault_update_id, rejected.metadata.vault_update_id); + assert_eq!( + STANDARD.decode(rejected.content_base64).unwrap(), + latest.content + ); + assert!(latest.content == b"left" || latest.content == b"right"); + let original = vault + .state + .database + .get_document_version(&vault.vault, base.vault_update_id, None) + .await + .unwrap() + .unwrap(); + assert_eq!(original.content, b"base"); + let mut transaction = vault + .state + .database + .create_readonly_transaction(&vault.vault) + .await + .unwrap(); + assert_eq!( + Database::latest_event_id(&mut transaction).await.unwrap(), + 2 + ); + } + + #[tokio::test] + async fn acknowledgements_survive_reopening_and_return_the_original_version() { + let vault = TestVault::new().await; + let document = uuid::Uuid::new_v4(); + let create = snapshot(document, None, "created"); + let create_id = create.request_id; + let base = accepted(vault.send(create).await); + let update = snapshot(document, Some(base.vault_update_id), "updated"); + let update_id = update.request_id; + let updated = accepted(vault.send(update).await); + let latest = accepted( + vault + .send(snapshot(document, Some(updated.vault_update_id), "newest")) + .await, + ); + + let reopened = TestVault::open(&vault.directory).await; + for (request_id, parent, text, original) in [ + (create_id, None, "created", base), + (update_id, Some(1), "updated", updated), + ] { + let mut retry = snapshot(document, parent, text); + retry.request_id = request_id; + retry.device_id = "new-session-after-restart".to_owned(); + let acknowledged = accepted(retry.send(&reopened, &vault.vault).await.unwrap()); + assert_eq!( + serde_json::to_value(acknowledged).unwrap(), + serde_json::to_value(original).unwrap() + ); + } + assert_eq!( + vault.latest(document).await.vault_update_id, + latest.vault_update_id + ); + + let mut reused = snapshot(uuid::Uuid::new_v4(), None, "wrong document"); + reused.request_id = create_id; + assert!(matches!( + reused.send(&reopened, &vault.vault).await, + Err(SyncServerError::ClientError(_)) + )); + + let mut changed_operation = snapshot(document, None, "created"); + changed_operation.content = PushContent::Diff(vec![]); + for mut changed_payload in [ + snapshot(document, None, "different content"), + snapshot(document, Some(1), "created"), + changed_operation, + ] { + changed_payload.request_id = create_id; + assert!(matches!( + changed_payload.send(&reopened, &vault.vault).await, + Err(SyncServerError::ClientError(_)) + )); + } + let events = reopened + .database + .events_after(&vault.vault, 0) + .await + .unwrap(); + assert_eq!(events.head_event_id, latest.vault_update_id); + assert_eq!(events.events.len(), 3); + assert_eq!(events.events[0].request_id, create_id); + assert_eq!(events.events[1].request_id, update_id); + } + + #[tokio::test] + async fn concurrent_retries_return_the_same_acknowledgement_and_append_one_event() { + let vault = TestVault::new().await; + let document = uuid::Uuid::new_v4(); + let original = snapshot(document, None, "created"); + let request_id = original.request_id; + let mut retry = snapshot(document, None, "created"); + retry.request_id = request_id; + retry.device_id = "reconnected-device".to_owned(); + + let (left, right) = tokio::join!(vault.send(original), vault.send(retry)); + assert_eq!( + serde_json::to_value(accepted(left)).unwrap(), + serde_json::to_value(accepted(right)).unwrap() + ); + let events = vault + .state + .database + .events_after(&vault.vault, 0) + .await + .unwrap(); + assert_eq!(events.head_event_id, 1); + assert_eq!(events.events.len(), 1); + assert_eq!(events.events[0].request_id, request_id); + } + + #[tokio::test] + async fn failed_event_commit_rolls_back_content_and_allows_retrying_the_request() { + let vault = TestVault::new().await; + let mut tx = vault + .state + .database + .create_write_transaction(&vault.vault) + .await + .unwrap(); + sqlx::query("CREATE TRIGGER reject_event BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'simulated event write failure'); END") + .execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); + + let document = uuid::Uuid::new_v4(); + let request = snapshot(document, None, "created"); + let request_id = request.request_id; + assert!(request.send(&vault.state, &vault.vault).await.is_err()); + assert!( + vault + .state + .database + .get_latest_document_version(&vault.vault, &document, None) + .await + .unwrap() + .is_none() + ); + let events = vault + .state + .database + .events_after(&vault.vault, 0) + .await + .unwrap(); + assert_eq!(events.head_event_id, 0); + assert!(events.events.is_empty()); + + let mut tx = vault + .state + .database + .create_write_transaction(&vault.vault) + .await + .unwrap(); + sqlx::query("DROP TRIGGER reject_event") + .execute(&mut *tx) + .await + .unwrap(); + tx.commit().await.unwrap(); + let mut retry = snapshot(document, None, "created"); + retry.request_id = request_id; + let version = accepted(vault.send(retry).await); + assert_eq!(version.vault_update_id, 1); + let events = vault + .state + .database + .events_after(&vault.vault, 0) + .await + .unwrap(); + assert_eq!(events.events.len(), 1); + assert_eq!(events.events[0].request_id, request_id); + } + + #[tokio::test] + async fn transport_diffs_reconstruct_exact_snapshots_and_reject_invalid_bases() { + let vault = TestVault::new().await; + let document = uuid::Uuid::new_v4(); + let original = "Hello world 🌍"; + let desired = "Hi beautiful world 🌍"; + let base = accepted(vault.send(snapshot(document, None, original)).await); + let mut update = snapshot(document, Some(base.vault_update_id), ""); + update.content = + PushContent::Diff(EditedText::from_strings(original, &desired.into()).to_diff()); + let updated = accepted(vault.send(update).await); + assert_eq!(vault.latest(document).await.content, desired.as_bytes()); + + for invalid_length in [i64::MIN, i64::MAX] { + let mut malformed = snapshot(document, Some(updated.vault_update_id), ""); + malformed.content = PushContent::Diff(vec![NumberOrText::Number(invalid_length)]); + assert!(matches!( + malformed.send(&vault.state, &vault.vault).await, + Err(SyncServerError::ClientError(_)) + )); + } + assert_eq!( + vault.latest(document).await.vault_update_id, + updated.vault_update_id + ); + + let another = accepted( + vault + .send(snapshot(uuid::Uuid::new_v4(), None, "another document")) + .await, + ); + let mut foreign_base = snapshot(document, Some(another.vault_update_id), ""); + foreign_base.content = PushContent::Diff(vec![]); + assert!(matches!( + vault.send(foreign_base).await, + DocumentUpdateResponse::StaleBase(version) + if version.metadata.vault_update_id == updated.vault_update_id + )); + + let binary_document = uuid::Uuid::new_v4(); + let mut binary = snapshot(binary_document, None, ""); + binary.content = PushContent::Snapshot(STANDARD.encode([0xff])); + let binary_base = accepted(vault.send(binary).await); + let mut invalid_text = snapshot(binary_document, Some(binary_base.vault_update_id), ""); + invalid_text.content = PushContent::Diff(vec![]); + assert!(matches!( + invalid_text.send(&vault.state, &vault.vault).await, + Err(SyncServerError::ClientError(_)) + )); + assert_eq!( + vault.latest(binary_document).await.vault_update_id, + binary_base.vault_update_id + ); + } +} diff --git a/sync-server/src/server/endpoints/put_file_manifest.rs b/sync-server/src/server/endpoints/put_file_manifest.rs new file mode 100644 index 00000000..db99c374 --- /dev/null +++ b/sync-server/src/server/endpoints/put_file_manifest.rs @@ -0,0 +1,122 @@ +use anyhow::anyhow; +use axum::{ + Json, + extract::{Path, State}, +}; +use log::debug; +use serde::Deserialize; +use sha2::{Digest, Sha256}; + +use super::utils::find_already_processed_event; +use crate::{ + app_state::{ + AppState, + database::{ + Database, + models::{EventRecord, FileManifest, VaultEvent, VaultId}, + }, + }, + errors::{SyncServerError, client_error, server_error}, + server::{requests::PushFileManifest, responses::FileManifestUpdateResponse}, + utils::normalize::normalize, +}; + +#[derive(Deserialize)] +pub struct PutFileManifestPath { + #[serde(deserialize_with = "normalize")] + vault_id: VaultId, +} + +#[axum::debug_handler] +pub async fn put_file_manifest( + Path(path): Path, + State(state): State, + Json(push): Json, +) -> Result, SyncServerError> { + debug!("Pushing file manifest for vault `{}`", path.vault_id); + + // BTreeMap canonicalizes file manifest entries for request fingerprinting. + let fingerprint = Sha256::digest( + serde_json::to_vec(&("file_manifest", &push)) + .map_err(|error| server_error(error.into()))?, + ); + + let mut tx = state + .database + .create_write_transaction(&path.vault_id) + .await + .map_err(server_error)?; + + if let Some(event) = + find_already_processed_event(&mut tx, push.request_id, &fingerprint).await? + { + return match event { + VaultEvent::FileManifest { file_manifest } => { + Ok(Json(FileManifestUpdateResponse::Accepted { + file_manifest_id: file_manifest.file_manifest_id, + })) + } + VaultEvent::Content { .. } => Err(client_error(anyhow!( + "Stored event does not match file manifest request" + ))), + }; + } + + let latest = Database::current_file_manifest(&mut tx) + .await + .map_err(server_error)?; + + if latest.file_manifest_id != push.parent_file_manifest_id { + return Ok(Json(FileManifestUpdateResponse::StaleBase(latest))); + } + + for id in push.entries.keys() { + if state + .database + .get_latest_document_version(&path.vault_id, id, Some(&mut tx)) + .await + .map_err(server_error)? + .is_none() + { + return Err(client_error(anyhow!( + "File manifest references missing content: {id}" + ))); + } + } + + let file_manifest = FileManifest { + file_manifest_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint) + .await + .map_err(server_error)?, + entries: push.entries, + }; + + Database::insert_file_manifest(&mut tx, &file_manifest) + .await + .map_err(server_error)?; + + let response = FileManifestUpdateResponse::Accepted { + file_manifest_id: file_manifest.file_manifest_id, + }; + + let event = EventRecord { + event_id: file_manifest.file_manifest_id, + request_id: push.request_id, + event: VaultEvent::FileManifest { file_manifest }, + }; + + Database::write_event(&mut tx, &event) + .await + .map_err(server_error)?; + + tx.commit() + .await + .map_err(|error| server_error(error.into()))?; + + state + .broadcasts + .notify_about_vault_update(path.vault_id) + .await; + + Ok(Json(response)) +} diff --git a/sync-server/src/server/endpoints/utils.rs b/sync-server/src/server/endpoints/utils.rs new file mode 100644 index 00000000..75e08989 --- /dev/null +++ b/sync-server/src/server/endpoints/utils.rs @@ -0,0 +1,32 @@ +use anyhow::anyhow; + +use crate::{ + app_state::database::{ + Database, Transaction, + models::{EventRecord, VaultEvent}, + }, + errors::{SyncServerError, client_error, server_error}, +}; + +pub(super) async fn find_already_processed_event( + tx: &mut Transaction<'_>, + request_id: uuid::Uuid, + fingerprint: &[u8], +) -> Result, SyncServerError> { + let Some((stored_fingerprint, event_json)) = Database::get_request_event(tx, request_id) + .await + .map_err(server_error)? + else { + return Ok(None); + }; + + if stored_fingerprint != fingerprint { + return Err(client_error(anyhow!( + "Request ID was reused with a different payload" + ))); + } + + serde_json::from_str::(&event_json) + .map(|record| Some(record.event)) + .map_err(|error| server_error(error.into())) +} diff --git a/sync-server/src/server/endpoints/vault_snapshot.rs b/sync-server/src/server/endpoints/vault_snapshot.rs new file mode 100644 index 00000000..552517c5 --- /dev/null +++ b/sync-server/src/server/endpoints/vault_snapshot.rs @@ -0,0 +1,36 @@ +use axum::{ + Json, + extract::{Path, State}, +}; +use log::debug; +use serde::Deserialize; + +use crate::{ + app_state::{ + AppState, + database::models::{VaultId, VaultSnapshot}, + }, + errors::{SyncServerError, server_error}, + utils::normalize::normalize, +}; + +#[derive(Deserialize)] +pub struct VaultSnapshotPath { + #[serde(deserialize_with = "normalize")] + vault_id: VaultId, +} + +#[axum::debug_handler] +pub async fn vault_snapshot( + Path(path): Path, + State(state): State, +) -> Result, SyncServerError> { + debug!("Fetching vault snapshot for vault `{}`", path.vault_id); + + state + .database + .vault_snapshot(&path.vault_id) + .await + .map(Json) + .map_err(server_error) +} diff --git a/sync-server/src/server/endpoints/websocket.rs b/sync-server/src/server/endpoints/websocket.rs index bb10b49f..97257215 100644 --- a/sync-server/src/server/endpoints/websocket.rs +++ b/sync-server/src/server/endpoints/websocket.rs @@ -1,4 +1,16 @@ -use anyhow::Context; +use crate::{ + app_state::{ + AppState, + database::models::VaultId, + websocket::{ + broadcasts::Notification, + models::{CursorPositionFromServer, WebSocketClientMessage, WebSocketServerMessage}, + utils::{get_authenticated_handshake, send_update_over_websocket}, + }, + }, + errors::{SyncServerError, client_error, server_error}, + utils::normalize::normalize, +}; use axum::{ extract::{ Path, State, @@ -6,27 +18,11 @@ use axum::{ }, response::Response, }; -use futures::stream::StreamExt; +use futures::StreamExt; use log::{debug, info}; use serde::Deserialize; - -use crate::{ - app_state::{ - AppState, - database::models::VaultId, - websocket::{ - models::{ - CursorPositionFromServer, WebSocketClientMessage, WebSocketServerMessage, - WebSocketVaultUpdate, - }, - utils::{ - get_authenticated_handshake, get_unseen_documents, send_update_over_websocket, - }, - }, - }, - errors::{SyncServerError, client_error, server_error}, - utils::normalize::normalize, -}; +use std::time::Duration; +use tokio::sync::broadcast::error::RecvError; #[derive(Deserialize)] pub struct WebSocketPathParams { @@ -34,160 +30,96 @@ pub struct WebSocketPathParams { vault_id: VaultId, } +#[axum::debug_handler] pub async fn websocket_handler( ws: WebSocketUpgrade, - Path(WebSocketPathParams { vault_id }): Path, + Path(path): Path, State(state): State, ) -> Result { - Ok(ws.on_upgrade(move |socket| websocket_wrapped(state, socket, vault_id))) -} - -async fn websocket_wrapped(state: AppState, stream: WebSocket, vault_id: VaultId) { - info!("WebSocket connection opened on vault `{vault_id}`"); - - let result = websocket(state, stream, vault_id.clone()).await; - - if let Err(err) = result { - debug!("WebSocket connection error on vault `{vault_id}`: {err}"); - } -} - -#[allow(clippy::too_many_lines)] -async fn websocket( - state: AppState, - stream: WebSocket, - vault_id: VaultId, -) -> Result<(), SyncServerError> { - let (mut sender, mut websocket_receiver) = stream.split(); - - let authed_handshake = get_authenticated_handshake( - &state, - &vault_id, - websocket_receiver - .next() - .await - .transpose() - .unwrap_or_default(), - )?; - - info!( - "WebSocket handshake successful for vault `{vault_id}` for `{}`", - authed_handshake.handshake.device_id + debug!( + "Upgrading WebSocket connection for vault `{}`", + path.vault_id ); - let mut broadcast_receiver = state.broadcasts.get_receiver(vault_id.clone()).await; + Ok(ws.on_upgrade(move |socket| async move { + if let Err(error) = websocket(state, socket, path.vault_id).await { + debug!("WebSocket disconnected: {error}"); + } + })) +} - send_update_over_websocket( - &WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate { - documents: get_unseen_documents( - &state, - &vault_id, - authed_handshake.handshake.last_seen_vault_update_id, - ) - .await?, - is_initial_sync: true, - }), - &mut sender, - ) - .await?; +async fn websocket( + state: AppState, + socket: WebSocket, + vault: VaultId, +) -> Result<(), SyncServerError> { + let (mut sender, mut receiver) = socket.split(); + let authenticated = get_authenticated_handshake( + &state, + &vault, + receiver.next().await.transpose().unwrap_or_default(), + )?; - send_update_over_websocket( - &WebSocketServerMessage::CursorPositions(CursorPositionFromServer { - clients: state.cursors.get_cursors(&vault_id).await, - }), - &mut sender, - ) - .await?; + let device = authenticated.handshake.device_id; + let mut after = authenticated + .handshake + .last_seen_vault_update_id + .unwrap_or(0); + let mut notifications = state.broadcasts.get_receiver(vault.clone()).await; + let mut timer = tokio::time::interval(Duration::from_secs(2)); + timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - let device_id = authed_handshake.handshake.device_id.clone(); - let mut send_task = tokio::spawn(async move { - while let Ok(update) = broadcast_receiver.recv().await { - if Some(&device_id) == update.origin_device_id.as_ref() { - continue; + info!("WebSocket connected to vault {vault}"); + + // Exactly one sender drains durable events in commit order. In-memory + // notifications are only wakeups, so lag and commit/notify crashes are safe. + let result = async { + loop { + let batch = state.database.events_after(&vault, after).await.map_err(|error| server_error(error.into()))?; + + if !batch.events.is_empty() { + let head = batch.head_event_id; + send_update_over_websocket(&WebSocketServerMessage::VaultEvents(batch), &mut sender).await?; + after = head; } - let message = match update.message { - WebSocketServerMessage::CursorPositions(CursorPositionFromServer { clients }) => { - WebSocketServerMessage::CursorPositions(CursorPositionFromServer { - clients: clients - .into_iter() - .filter(|client| client.device_id != device_id) - .collect(), - }) - } - WebSocketServerMessage::VaultUpdate(_) => update.message, - }; + tokio::select! { + _ = timer.tick() => {}, + notification = notifications.recv() => match notification { + Ok(update) => if let Notification::Cursors(mut positions) = update { + positions.clients.retain(|c| c.device_id != device); + send_update_over_websocket(&WebSocketServerMessage::CursorPositions(positions), &mut sender).await?; + }, - send_update_over_websocket(&message, &mut sender).await?; - } + Err(RecvError::Lagged(_)) => {}, // Drain the durable log on the next iteration. + Err(RecvError::Closed) => break, + }, - Ok::<(), SyncServerError>(()) - }); - - let device_id = authed_handshake.handshake.device_id.clone(); - let vault_id_clone = vault_id.clone(); - let cursor_manager = state.cursors.clone(); - let mut receive_task = tokio::spawn(async move { - while let Some(Ok(Message::Text(message))) = websocket_receiver.next().await { - let message: WebSocketClientMessage = serde_json::from_str(&message) - .context("Failed to parse WebSocket message from client") - .map_err(server_error)?; - - match message { - WebSocketClientMessage::Handshake(_) => { - return Err(client_error(anyhow::anyhow!( - "Unexpected handshake message" - ))); - } - WebSocketClientMessage::CursorPositions(cursors) => { - cursor_manager - .update_cursors( - vault_id_clone.clone(), - authed_handshake.user.name.clone(), - &device_id, - cursors.documents_with_cursors, - ) - .await; + incoming = receiver.next() => match incoming { + Some(Ok(Message::Text(text))) => { + let message: WebSocketClientMessage = serde_json::from_str(&text).map_err(|error| client_error(error.into()))?; + match message { + WebSocketClientMessage::Handshake(_) => return Err(client_error(anyhow::anyhow!("Unexpected handshake"))), + WebSocketClientMessage::CursorPositions(positions) => { + state.cursors.update_cursors(vault.clone(), authenticated.user.name.clone(), &device, positions.documents_with_cursors).await; + let clients = state.cursors.get_cursors(&vault).await.into_iter().filter(|c| c.device_id != device).collect(); + send_update_over_websocket(&WebSocketServerMessage::CursorPositions(CursorPositionFromServer { clients }), &mut sender).await?; + } + } + }, + Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {}, + _ => break, } } } - - Ok::<(), SyncServerError>(()) - }); - - tokio::select! { - _ = &mut send_task => receive_task.abort(), - _ = &mut receive_task => send_task.abort(), - }; - - let result: Result<(), SyncServerError> = (async { - send_task - .await - .context("WebSocket send task failed") - .map_err(client_error) - .and_then(|err| err)?; - - receive_task - .await - .context("WebSocket receive task failed") - .map_err(client_error) - .and_then(|err| err)?; - + Ok(()) - }) - .await; + }.await; state .cursors - .remove_cursors_of_device(&vault_id, &authed_handshake.handshake.device_id) + .remove_cursors_of_device(&vault, &device) .await; - if result.is_err() { - info!( - "WebSocket disconnected on vault `{vault_id}` for `{}`", - authed_handshake.handshake.device_id - ); - } - result } diff --git a/sync-server/src/server/requests.rs b/sync-server/src/server/requests.rs index 7b814559..a27e62a9 100644 --- a/sync-server/src/server/requests.rs +++ b/sync-server/src/server/requests.rs @@ -1,23 +1,149 @@ -use crate::app_state::database::models::VaultUpdateId; +use crate::{ + app_state::database::models::{DocumentId, FileManifestEntries, VaultUpdateId}, + utils::portable_path::validate_file_manifest, +}; use reconcile_text::NumberOrText; -use serde::{Deserialize, Serialize}; +use serde::{ + Deserialize, Deserializer, Serialize, + de::{Error, MapAccess, Visitor}, +}; +use std::{collections::BTreeMap, fmt}; use ts_rs::TS; +#[derive(TS, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +#[serde(deny_unknown_fields)] +#[ts(export)] +pub struct PutFileContent { + pub request_id: uuid::Uuid, + #[ts(as = "Option")] + pub parent_version_id: Option, + pub content: PushContent, +} + #[derive(TS, Debug, Deserialize, Serialize)] #[serde(tag = "type", content = "value")] +#[serde(deny_unknown_fields)] pub enum PushContent { Snapshot(String), Diff(#[ts(type = "Array")] Vec), - Delete, } #[derive(TS, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] +#[serde(deny_unknown_fields)] #[ts(export)] -pub struct PushDocument { +pub struct PushFileManifest { pub request_id: uuid::Uuid, - #[ts(as = "Option")] - pub parent_version_id: Option, - pub relative_path: String, - pub content: PushContent, + #[ts(as = "f64")] + pub parent_file_manifest_id: VaultUpdateId, + #[serde(deserialize_with = "deserialize_file_manifest_entries")] + #[ts(type = "Record")] + pub entries: FileManifestEntries, +} + +// A normal JSON map decoder silently overwrites duplicate keys. UUID parsing also +// catches differently spelled keys that denote the same identity. +fn deserialize_file_manifest_entries<'de, D: Deserializer<'de>>( + d: D, +) -> Result { + struct Entries; + + impl<'de> Visitor<'de> for Entries { + type Value = FileManifestEntries; + + fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("unique document IDs mapped to paths") + } + + fn visit_map>(self, mut map: M) -> Result { + let mut entries = BTreeMap::new(); + + while let Some((id, path)) = map.next_entry::()? { + if entries.insert(id, path).is_some() { + return Err(M::Error::custom("Duplicate document ID")); + } + } + + Ok(entries) + } + } + + let entries = d.deserialize_map(Entries)?; + + validate_file_manifest(&entries).map_err(D::Error::custom)?; + + Ok(entries) +} + +#[cfg(test)] +mod tests { + use super::PushFileManifest; + + fn deserialize_manifest(entries: &str) -> serde_json::Result { + serde_json::from_str(&format!( + r#"{{ + "requestId": "00000000-0000-0000-0000-000000000001", + "parentFileManifestId": 0, + "entries": {entries} + }}"# + )) + } + + #[test] + fn deserializes_empty_and_valid_file_manifests() { + assert!(deserialize_manifest("{}").unwrap().entries.is_empty()); + + let push = deserialize_manifest( + r#"{ + "00000000-0000-0000-0000-000000000001": "Notes/Café.md", + "00000000-0000-0000-0000-000000000002": "Notes/日記.md" + }"#, + ) + .unwrap(); + + assert_eq!( + push.entries + .values() + .map(String::as_str) + .collect::>(), + ["Notes/Café.md", "Notes/日記.md"] + ); + } + + #[test] + fn deserialization_rejects_invalid_paths_and_manifest_conflicts() { + for (entries, expected_error) in [ + ( + r#"{"00000000-0000-0000-0000-000000000001": "../notes.md"}"#, + "Invalid path component", + ), + ( + r#"{ + "00000000-0000-0000-0000-000000000001": "Notes/a.md", + "00000000-0000-0000-0000-000000000002": "notes/b.md" + }"#, + "Conflicting path", + ), + ] { + let error = deserialize_manifest(entries).unwrap_err(); + + assert!(error.to_string().contains(expected_error), "{error}"); + } + } + + #[test] + fn deserialization_rejects_duplicate_document_ids_including_uuid_aliases() { + let id = "00000000-0000-0000-0000-000000000abc"; + + for duplicate in [id.to_owned(), id.to_uppercase()] { + let entries = format!(r#"{{"{id}": "a.md", "{duplicate}": "b.md"}}"#); + let error = deserialize_manifest(&entries).unwrap_err(); + + assert!( + error.to_string().contains("Duplicate document ID"), + "{error}" + ); + } + } } diff --git a/sync-server/src/server/responses.rs b/sync-server/src/server/responses.rs index ed469949..66c5b5cb 100644 --- a/sync-server/src/server/responses.rs +++ b/sync-server/src/server/responses.rs @@ -1,7 +1,9 @@ -use serde::{self, Serialize}; +use serde::{self, Deserialize, Serialize}; use ts_rs::TS; -use crate::app_state::database::models::{DocumentVersion, DocumentVersionWithoutContent}; +use crate::app_state::database::models::{ + DocumentVersion, DocumentVersionWithoutContent, FileManifest, VaultUpdateId, +}; /// Response to a ping request. #[derive(TS, Debug, Clone, Serialize)] @@ -24,7 +26,7 @@ pub struct PingResponse { } /// Response to an update document request. -#[derive(TS, Debug, Clone, Serialize)] +#[derive(TS, Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type")] #[ts(export)] pub enum DocumentUpdateResponse { @@ -34,3 +36,15 @@ pub enum DocumentUpdateResponse { /// Nothing was written. The client must incorporate this version before retrying. StaleBase(DocumentVersion), } + +#[derive(TS, Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type")] +#[ts(export)] +pub enum FileManifestUpdateResponse { + Accepted { + #[serde(rename = "fileManifestId")] + #[ts(as = "f64")] + file_manifest_id: VaultUpdateId, + }, + StaleBase(FileManifest), +}