From 1619c51ad9f4c423d03304ab9cca534617e8fd3e Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Wed, 16 Sep 2026 10:36:20 +0100 Subject: [PATCH] Migrate server to CAS only --- sync-server/README.md | 33 ++ sync-server/build.rs | 2 +- sync-server/src/app_state/database.rs | 324 +------------- .../database/get_document_version.rs | 41 ++ .../app_state/database/get_latest_document.rs | 43 ++ .../database/get_latest_document_by_path.rs | 46 ++ .../database/get_latest_documents.rs | 56 +++ .../database/get_latest_documents_since.rs | 62 +++ .../database/get_max_update_id_in_vault.rs | 29 ++ .../database/get_push_acknowledgement.rs | 38 ++ .../database/insert_document_version.rs | 79 ++++ .../migrations/20241207143519_bootstrap.sql | 11 +- .../20250522192949_add_provenance_columns.sql | 2 - .../20251019103154_add_has_been_merged.sql | 13 - sync-server/src/app_state/database/models.rs | 2 - sync-server/src/consts.rs | 2 +- sync-server/src/server.rs | 51 +-- sync-server/src/server/create_document.rs | 109 ----- sync-server/src/server/delete_document.rs | 96 ---- sync-server/src/server/endpoints.rs | 7 + .../fetch_document_version_content.rs | 0 .../fetch_latest_document_version.rs | 0 .../endpoints/fetch_latest_documents.rs | 36 ++ .../src/server/{ => endpoints}/index.rs | 2 +- .../src/server/{ => endpoints}/ping.rs | 2 +- .../src/server/endpoints/push_document.rs | 210 +++++++++ .../server/endpoints/push_document/tests.rs | 423 ++++++++++++++++++ .../src/server/{ => endpoints}/websocket.rs | 0 .../src/server/fetch_document_version.rs | 62 --- .../src/server/fetch_latest_documents.rs | 59 --- sync-server/src/server/requests.rs | 56 +-- sync-server/src/server/responses.rs | 26 +- sync-server/src/server/update_document.rs | 271 ----------- sync-server/src/utils.rs | 2 - sync-server/src/utils/is_binary.rs | 26 -- .../src/utils/is_file_type_mergable.rs | 34 -- 36 files changed, 1163 insertions(+), 1092 deletions(-) create mode 100644 sync-server/src/app_state/database/get_document_version.rs create mode 100644 sync-server/src/app_state/database/get_latest_document.rs create mode 100644 sync-server/src/app_state/database/get_latest_document_by_path.rs create mode 100644 sync-server/src/app_state/database/get_latest_documents.rs create mode 100644 sync-server/src/app_state/database/get_latest_documents_since.rs create mode 100644 sync-server/src/app_state/database/get_max_update_id_in_vault.rs create mode 100644 sync-server/src/app_state/database/get_push_acknowledgement.rs create mode 100644 sync-server/src/app_state/database/insert_document_version.rs delete mode 100644 sync-server/src/app_state/database/migrations/20250522192949_add_provenance_columns.sql delete mode 100644 sync-server/src/app_state/database/migrations/20251019103154_add_has_been_merged.sql delete mode 100644 sync-server/src/server/create_document.rs delete mode 100644 sync-server/src/server/delete_document.rs create mode 100644 sync-server/src/server/endpoints.rs rename sync-server/src/server/{ => endpoints}/fetch_document_version_content.rs (100%) rename sync-server/src/server/{ => endpoints}/fetch_latest_document_version.rs (100%) create mode 100644 sync-server/src/server/endpoints/fetch_latest_documents.rs rename sync-server/src/server/{ => endpoints}/index.rs (68%) rename sync-server/src/server/{ => endpoints}/ping.rs (95%) create mode 100644 sync-server/src/server/endpoints/push_document.rs create mode 100644 sync-server/src/server/endpoints/push_document/tests.rs rename sync-server/src/server/{ => endpoints}/websocket.rs (100%) delete mode 100644 sync-server/src/server/fetch_document_version.rs delete mode 100644 sync-server/src/server/fetch_latest_documents.rs delete mode 100644 sync-server/src/server/update_document.rs delete mode 100644 sync-server/src/utils/is_binary.rs delete mode 100644 sync-server/src/utils/is_file_type_mergable.rs diff --git a/sync-server/README.md b/sync-server/README.md index 16509fb7..9b11a061 100644 --- a/sync-server/README.md +++ b/sync-server/README.md @@ -1,5 +1,38 @@ # Sync server +## Document pushes (API v3) + +The server stores immutable document versions. One compare-and-swap endpoint, +`PUT /vaults/:vault_id/documents/:document_id`, handles every mutation. The client +generates the document UUID; the JSON body contains: + +- `requestId`: a UUID identifying this exact push. +- `parentVersionId`: `null` to create an absent document, otherwise its base version. +- `relativePath`: the desired filename. +- `content`: `{ "type": "Snapshot", "value": "" }`, + `{ "type": "Diff", "value": [3, "inserted text", -2] }`, or `{ "type": "Delete" }`. + +All pushes return one of: + +- `Accepted`: version metadata for the exact snapshot committed. A retry with the + same request ID returns the original acknowledgement, even after a restart or + subsequent updates. Clients must persist the ID with the unchanged request. + A payload fingerprint rejects reusing that ID for a different mutation. +- `StaleBase`: the current version, including `contentBase64`. Nothing was written; + the client merges locally and submits a new push with a new request ID. + +The base check, path allocation, new version, and durable acknowledgement share +one SQLite write transaction. A deleted document stays deleted; recreating a file +uses a new document ID. Occupied paths receive an available numbered filename. +Text updates accept reconcile-text transport diffs against the base version; +the server only reconstructs snapshots and never merges concurrent edits. + +Snapshot requests use base64, adding roughly one third to the raw file size. +Server and reverse-proxy request limits apply to the encoded JSON body. + +API v3 uses a fresh schema. Update clients together with the server and initialise +new vault databases; compatibility with previous releases is not supported. + ## Creating/resetting the Database for development ```sh diff --git a/sync-server/build.rs b/sync-server/build.rs index d5068697..ee539ea3 100644 --- a/sync-server/build.rs +++ b/sync-server/build.rs @@ -1,5 +1,5 @@ // generated by `sqlx migrate build-script` fn main() { // trigger recompilation when a new migration is added - println!("cargo:rerun-if-changed=migrations"); + println!("cargo:rerun-if-changed=src/app_state/database/migrations"); } diff --git a/sync-server/src/app_state/database.rs b/sync-server/src/app_state/database.rs index 75ce6df4..216a537d 100644 --- a/sync-server/src/app_state/database.rs +++ b/sync-server/src/app_state/database.rs @@ -3,21 +3,24 @@ use std::{collections::HashMap, sync::Arc}; use anyhow::{Context as _, Result}; use log::info; -use models::{ - DocumentId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId, VaultUpdateId, -}; -use sqlx::{ConnectOptions, sqlite::SqliteConnectOptions, types::chrono::Utc}; - -pub mod models; +use models::VaultId; +use sqlx::{ConnectOptions, sqlite::SqliteConnectOptions}; use sqlx::{Pool, Sqlite, sqlite::SqlitePoolOptions}; use tokio::sync::Mutex; use tokio::time::Instant; -use uuid::fmt::Hyphenated; -use super::websocket::{ - broadcasts::Broadcasts, - models::{WebSocketServerMessage, WebSocketServerMessageWithOrigin, WebSocketVaultUpdate}, -}; +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; + +use super::websocket::broadcasts::Broadcasts; use crate::config::database_config::DatabaseConfig; #[derive(Clone)] @@ -176,305 +179,6 @@ impl Database { Ok(transaction) } - /// Return the latest state of all documents in the vault - pub async fn get_latest_documents( - &self, - vault: &VaultId, - transaction: Option<&mut Transaction<'_>>, - ) -> Result> { - let query = sqlx::query!( - r#" - select - vault_update_id, - document_id as "document_id: Hyphenated", - relative_path, - updated_date as "updated_date: chrono::DateTime", - is_deleted, - user_id, - device_id, - length(content) as "content_size: u64" - from latest_document_versions - order by vault_update_id - "#, - ); - - if let Some(transaction) = transaction { - query.fetch_all(&mut **transaction).await - } else { - query - .fetch_all(&self.get_connection_pool(vault).await?) - .await - } - .context("Cannot fetch latest documents") - .map(|rows| { - rows.into_iter() - .map(|row| DocumentVersionWithoutContent { - vault_update_id: row.vault_update_id, - document_id: row.document_id.into(), - relative_path: row.relative_path, - updated_date: row.updated_date, - is_deleted: row.is_deleted, - user_id: row.user_id, - device_id: row.device_id, - content_size: row - .content_size - .expect("Content size can't be null but sqlx can't infer it"), - }) - .collect() - }) - } - - /// Return the latest state of all documents (including deleted) in the - /// vault which have changed since the given update id - pub async fn get_latest_documents_since( - &self, - vault: &VaultId, - vault_update_id: VaultUpdateId, - transaction: Option<&mut Transaction<'_>>, - ) -> Result> { - let query = sqlx::query!( - r#" - select - vault_update_id, - document_id as "document_id: Hyphenated", - relative_path, - updated_date as "updated_date: chrono::DateTime", - is_deleted, - user_id, - device_id, - length(content) as "content_size: u64" - from latest_document_versions - where vault_update_id > ? - order by vault_update_id - "#, - vault_update_id - ); - - if let Some(transaction) = transaction { - query.fetch_all(&mut **transaction).await - } else { - query - .fetch_all(&self.get_connection_pool(vault).await?) - .await - } - .with_context(|| { - format!("Cannot fetch latest documents since vault_update_id `{vault_update_id}`") - }) - .map(|rows| { - rows.into_iter() - .map(|row| DocumentVersionWithoutContent { - vault_update_id: row.vault_update_id, - document_id: row.document_id.into(), - relative_path: row.relative_path, - updated_date: row.updated_date, - is_deleted: row.is_deleted, - user_id: row.user_id, - device_id: row.device_id, - content_size: row - .content_size - .expect("Content size can't be null but sqlx can't infer it"), - }) - .collect() - }) - } - - pub async fn get_max_update_id_in_vault( - &self, - vault: &VaultId, - transaction: Option<&mut Transaction<'_>>, - ) -> Result { - let query = sqlx::query!( - r#" - select coalesce(max(vault_update_id), 0) as max_vault_update_id - from documents - "#, - ); - - if let Some(transaction) = transaction { - query.fetch_one(&mut **transaction).await - } else { - query - .fetch_one(&self.get_connection_pool(vault).await?) - .await - } - .map(|row| row.max_vault_update_id) - .context("Cannot fetch max update id in vault") - } - - pub async fn get_latest_document_by_path( - &self, - vault: &VaultId, - relative_path: &str, - transaction: Option<&mut Transaction<'_>>, - ) -> Result> { - let query = sqlx::query_as!( - StoredDocumentVersion, - r#" - select - vault_update_id, - document_id as "document_id: Hyphenated", - relative_path, - updated_date as "updated_date: chrono::DateTime", - content, - is_deleted, - user_id, - device_id, - has_been_merged - from latest_document_versions - where relative_path = ? and is_deleted = false - order by vault_update_id desc -- `latest_document_versions` only contains a single latest version of each document, however, - -- multiple documents can have the same `relative_path`, if they have been deleted. That's - -- why we only care about the latest version of the document with the given relative path. - limit 1 - "#, - relative_path - ); - - if let Some(transaction) = transaction { - query.fetch_optional(&mut **transaction).await - } else { - query - .fetch_optional(&self.get_connection_pool(vault).await?) - .await - } - .context("Cannot fetch latest document version") - } - - pub async fn get_latest_document( - &self, - vault: &VaultId, - document_id: &DocumentId, - transaction: Option<&mut Transaction<'_>>, - ) -> Result> { - let document_id = document_id.as_hyphenated(); - let query = sqlx::query_as!( - StoredDocumentVersion, - r#" - select - vault_update_id, - document_id as "document_id: Hyphenated", - relative_path, - updated_date as "updated_date: chrono::DateTime", - content, - is_deleted, - user_id, - device_id, - has_been_merged - from latest_document_versions - where document_id = ? - "#, - document_id - ); - - if let Some(transaction) = transaction { - query.fetch_optional(&mut **transaction).await - } else { - query - .fetch_optional(&self.get_connection_pool(vault).await?) - .await - } - .context("Cannot fetch latest document version") - } - - pub async fn get_document_version( - &self, - vault: &VaultId, - vault_update_id: VaultUpdateId, - transaction: Option<&mut Transaction<'_>>, - ) -> Result> { - let query = sqlx::query_as!( - StoredDocumentVersion, - r#" - select - vault_update_id, - document_id as "document_id: Hyphenated", - relative_path, - updated_date as "updated_date: chrono::DateTime", - content, - is_deleted, - user_id, - device_id, - has_been_merged - from documents - where vault_update_id = ?"#, - vault_update_id - ); - - if let Some(transaction) = transaction { - query.fetch_optional(&mut **transaction).await - } else { - query - .fetch_optional(&self.get_connection_pool(vault).await?) - .await - } - .context("Cannot fetch document version") - } - - // inserting the document must be the last step of the transaction if there's one - pub async fn insert_document_version( - &self, - vault_id: &VaultId, - version: &StoredDocumentVersion, - transaction: Option>, - ) -> Result<()> { - let document_id = version.document_id.as_hyphenated(); - let query = sqlx::query!( - r#" - insert into documents ( - vault_update_id, - document_id, - relative_path, - updated_date, - content, - is_deleted, - user_id, - device_id - ) - values (?, ?, ?, ?, ?, ?, ?, ?) - "#, - version.vault_update_id, - document_id, - version.relative_path, - version.updated_date, - version.content, - version.is_deleted, - version.user_id, - version.device_id - ); - - if let Some(mut transaction) = transaction { - query - .execute(&mut *transaction) - .await - .context("Cannot insert document version")?; - - transaction - .commit() - .await - .context("Failed to commit transaction")?; - } else { - query - .execute(&self.get_connection_pool(vault_id).await?) - .await - .context("Cannot insert document version")?; - } - - self.broadcasts - .send_document_update( - vault_id.clone(), - WebSocketServerMessageWithOrigin::with_origin( - version.device_id.clone(), - WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate { - documents: vec![version.clone().into()], - is_initial_sync: false, - }), - ), - ) - .await; - - Ok(()) - } - /// Cleanup idle connection pools that haven't been accessed in more than 5 minutes async fn cleanup_idle_pools(&self) { let mut pools = self.connection_pools.lock().await; diff --git a/sync-server/src/app_state/database/get_document_version.rs b/sync-server/src/app_state/database/get_document_version.rs new file mode 100644 index 00000000..c4113d8b --- /dev/null +++ b/sync-server/src/app_state/database/get_document_version.rs @@ -0,0 +1,41 @@ +use anyhow::{Context as _, Result}; +use sqlx::types::chrono::Utc; +use uuid::fmt::Hyphenated; + +use super::models::{StoredDocumentVersion, VaultId, VaultUpdateId}; +use super::{Database, Transaction}; + +impl Database { + pub async fn get_document_version( + &self, + vault: &VaultId, + vault_update_id: VaultUpdateId, + transaction: Option<&mut Transaction<'_>>, + ) -> Result> { + let query = sqlx::query_as!( + StoredDocumentVersion, + r#" + select + vault_update_id, + document_id as "document_id: Hyphenated", + relative_path, + updated_date as "updated_date: chrono::DateTime", + content, + is_deleted, + user_id, + device_id + from documents + where vault_update_id = ?"#, + vault_update_id + ); + + if let Some(transaction) = transaction { + query.fetch_optional(&mut **transaction).await + } else { + query + .fetch_optional(&self.get_connection_pool(vault).await?) + .await + } + .context("Cannot fetch document version") + } +} diff --git a/sync-server/src/app_state/database/get_latest_document.rs b/sync-server/src/app_state/database/get_latest_document.rs new file mode 100644 index 00000000..0bb6ecb6 --- /dev/null +++ b/sync-server/src/app_state/database/get_latest_document.rs @@ -0,0 +1,43 @@ +use anyhow::{Context as _, Result}; +use sqlx::types::chrono::Utc; +use uuid::fmt::Hyphenated; + +use super::models::{DocumentId, StoredDocumentVersion, VaultId}; +use super::{Database, Transaction}; + +impl Database { + pub async fn get_latest_document( + &self, + vault: &VaultId, + document_id: &DocumentId, + transaction: Option<&mut Transaction<'_>>, + ) -> Result> { + let document_id = document_id.as_hyphenated(); + let query = sqlx::query_as!( + StoredDocumentVersion, + r#" + select + vault_update_id, + document_id as "document_id: Hyphenated", + relative_path, + updated_date as "updated_date: chrono::DateTime", + content, + is_deleted, + user_id, + device_id + from latest_document_versions + where document_id = ? + "#, + document_id + ); + + if let Some(transaction) = transaction { + query.fetch_optional(&mut **transaction).await + } else { + query + .fetch_optional(&self.get_connection_pool(vault).await?) + .await + } + .context("Cannot fetch latest document version") + } +} diff --git a/sync-server/src/app_state/database/get_latest_document_by_path.rs b/sync-server/src/app_state/database/get_latest_document_by_path.rs new file mode 100644 index 00000000..95e49875 --- /dev/null +++ b/sync-server/src/app_state/database/get_latest_document_by_path.rs @@ -0,0 +1,46 @@ +use anyhow::{Context as _, Result}; +use sqlx::types::chrono::Utc; +use uuid::fmt::Hyphenated; + +use super::models::{StoredDocumentVersion, VaultId}; +use super::{Database, Transaction}; + +impl Database { + pub async fn get_latest_document_by_path( + &self, + vault: &VaultId, + relative_path: &str, + transaction: Option<&mut Transaction<'_>>, + ) -> Result> { + let query = sqlx::query_as!( + StoredDocumentVersion, + r#" + select + vault_update_id, + document_id as "document_id: Hyphenated", + relative_path, + updated_date as "updated_date: chrono::DateTime", + content, + is_deleted, + user_id, + device_id + from latest_document_versions + where relative_path = ? and is_deleted = false + order by vault_update_id desc -- `latest_document_versions` only contains a single latest version of each document, however, + -- multiple documents can have the same `relative_path`, if they have been deleted. That's + -- why we only care about the latest version of the document with the given relative path. + limit 1 + "#, + relative_path + ); + + if let Some(transaction) = transaction { + query.fetch_optional(&mut **transaction).await + } else { + query + .fetch_optional(&self.get_connection_pool(vault).await?) + .await + } + .context("Cannot fetch latest document version") + } +} diff --git a/sync-server/src/app_state/database/get_latest_documents.rs b/sync-server/src/app_state/database/get_latest_documents.rs new file mode 100644 index 00000000..700a505c --- /dev/null +++ b/sync-server/src/app_state/database/get_latest_documents.rs @@ -0,0 +1,56 @@ +use anyhow::{Context as _, Result}; +use sqlx::types::chrono::Utc; +use uuid::fmt::Hyphenated; + +use super::models::{DocumentVersionWithoutContent, VaultId}; +use super::{Database, Transaction}; + +impl Database { + /// Return the latest state of all documents in the vault + pub async fn get_latest_documents( + &self, + vault: &VaultId, + transaction: Option<&mut Transaction<'_>>, + ) -> Result> { + let query = sqlx::query!( + r#" + select + vault_update_id, + document_id as "document_id: Hyphenated", + relative_path, + updated_date as "updated_date: chrono::DateTime", + is_deleted, + user_id, + device_id, + length(content) as "content_size: u64" + from latest_document_versions + order by vault_update_id + "#, + ); + + if let Some(transaction) = transaction { + query.fetch_all(&mut **transaction).await + } else { + query + .fetch_all(&self.get_connection_pool(vault).await?) + .await + } + .context("Cannot fetch latest documents") + .map(|rows| { + rows.into_iter() + .map(|row| DocumentVersionWithoutContent { + vault_update_id: row.vault_update_id, + document_id: row.document_id.into(), + relative_path: row.relative_path, + updated_date: row.updated_date, + is_deleted: row.is_deleted, + user_id: row.user_id, + device_id: row.device_id, + content_size: row + .content_size + .expect("Content size can't be null but sqlx can't infer it"), + }) + .collect() + }) + } +} diff --git a/sync-server/src/app_state/database/get_latest_documents_since.rs b/sync-server/src/app_state/database/get_latest_documents_since.rs new file mode 100644 index 00000000..48c2f0e9 --- /dev/null +++ b/sync-server/src/app_state/database/get_latest_documents_since.rs @@ -0,0 +1,62 @@ +use anyhow::{Context as _, Result}; +use sqlx::types::chrono::Utc; +use uuid::fmt::Hyphenated; + +use super::models::{DocumentVersionWithoutContent, VaultId, VaultUpdateId}; +use super::{Database, Transaction}; + +impl Database { + /// Return the latest state of all documents (including deleted) in the + /// vault which have changed since the given update id + pub async fn get_latest_documents_since( + &self, + vault: &VaultId, + vault_update_id: VaultUpdateId, + transaction: Option<&mut Transaction<'_>>, + ) -> Result> { + let query = sqlx::query!( + r#" + select + vault_update_id, + document_id as "document_id: Hyphenated", + relative_path, + updated_date as "updated_date: chrono::DateTime", + is_deleted, + user_id, + device_id, + length(content) as "content_size: u64" + from latest_document_versions + where vault_update_id > ? + order by vault_update_id + "#, + vault_update_id + ); + + if let Some(transaction) = transaction { + query.fetch_all(&mut **transaction).await + } else { + query + .fetch_all(&self.get_connection_pool(vault).await?) + .await + } + .with_context(|| { + format!("Cannot fetch latest documents since vault_update_id `{vault_update_id}`") + }) + .map(|rows| { + rows.into_iter() + .map(|row| DocumentVersionWithoutContent { + vault_update_id: row.vault_update_id, + document_id: row.document_id.into(), + relative_path: row.relative_path, + updated_date: row.updated_date, + is_deleted: row.is_deleted, + user_id: row.user_id, + device_id: row.device_id, + content_size: row + .content_size + .expect("Content size can't be null but sqlx can't infer it"), + }) + .collect() + }) + } +} diff --git a/sync-server/src/app_state/database/get_max_update_id_in_vault.rs b/sync-server/src/app_state/database/get_max_update_id_in_vault.rs new file mode 100644 index 00000000..cc4a43c7 --- /dev/null +++ b/sync-server/src/app_state/database/get_max_update_id_in_vault.rs @@ -0,0 +1,29 @@ +use anyhow::{Context as _, Result}; + +use super::models::VaultId; +use super::{Database, Transaction}; + +impl Database { + pub async fn get_max_update_id_in_vault( + &self, + vault: &VaultId, + transaction: Option<&mut Transaction<'_>>, + ) -> Result { + let query = sqlx::query!( + r#" + select coalesce(max(vault_update_id), 0) as max_vault_update_id + from documents + "#, + ); + + if let Some(transaction) = transaction { + query.fetch_one(&mut **transaction).await + } else { + query + .fetch_one(&self.get_connection_pool(vault).await?) + .await + } + .map(|row| row.max_vault_update_id) + .context("Cannot fetch max update id in vault") + } +} diff --git a/sync-server/src/app_state/database/get_push_acknowledgement.rs b/sync-server/src/app_state/database/get_push_acknowledgement.rs new file mode 100644 index 00000000..cd8cd593 --- /dev/null +++ b/sync-server/src/app_state/database/get_push_acknowledgement.rs @@ -0,0 +1,38 @@ +use anyhow::{Context as _, Result}; + +use super::models::{StoredDocumentVersion, VaultId}; +use super::{Database, Transaction}; + +impl Database { + /// Look up the immutable version acknowledged by a previous push. + pub async fn get_push_acknowledgement( + &self, + vault_id: &VaultId, + request_id: uuid::Uuid, + transaction: &mut Transaction<'_>, + ) -> Result, StoredDocumentVersion)>> { + let request_id = request_id.hyphenated().to_string(); + let acknowledgement = sqlx::query!( + "select vault_update_id, request_fingerprint from push_acknowledgements where request_id = ?", + request_id + ) + .fetch_optional(&mut **transaction) + .await + .context("Cannot fetch push acknowledgement")?; + + match acknowledgement { + Some(acknowledgement) => { + let version = self + .get_document_version( + vault_id, + acknowledgement.vault_update_id, + Some(transaction), + ) + .await? + .context("Acknowledged document version is missing")?; + Ok(Some((acknowledgement.request_fingerprint, version))) + } + None => Ok(None), + } + } +} diff --git a/sync-server/src/app_state/database/insert_document_version.rs b/sync-server/src/app_state/database/insert_document_version.rs new file mode 100644 index 00000000..16752471 --- /dev/null +++ b/sync-server/src/app_state/database/insert_document_version.rs @@ -0,0 +1,79 @@ +use anyhow::{Context as _, Result}; + +use super::models::{StoredDocumentVersion, VaultId}; +use super::{Database, Transaction}; +use crate::app_state::websocket::models::{ + WebSocketServerMessage, WebSocketServerMessageWithOrigin, WebSocketVaultUpdate, +}; + +impl Database { + /// Commit the snapshot and acknowledgement atomically, then notify subscribers. + pub async fn insert_document_version( + &self, + vault_id: &VaultId, + version: &StoredDocumentVersion, + request_id: uuid::Uuid, + request_fingerprint: &[u8], + mut transaction: Transaction<'_>, + ) -> Result<()> { + let document_id = version.document_id.as_hyphenated(); + sqlx::query!( + r#" + insert into documents ( + vault_update_id, + document_id, + relative_path, + updated_date, + content, + is_deleted, + user_id, + device_id + ) + values (?, ?, ?, ?, ?, ?, ?, ?) + "#, + version.vault_update_id, + document_id, + version.relative_path, + version.updated_date, + version.content, + version.is_deleted, + version.user_id, + version.device_id + ) + .execute(&mut *transaction) + .await + .context("Cannot insert document version")?; + + let request_id = request_id.hyphenated().to_string(); + sqlx::query!( + "insert into push_acknowledgements (request_id, request_fingerprint, vault_update_id) values (?, ?, ?)", + request_id, + request_fingerprint, + version.vault_update_id + ) + .execute(&mut *transaction) + .await + .context("Cannot save push acknowledgement")?; + + transaction + .commit() + .await + .context("Failed to commit push")?; + + // notifying the listeners here means that no update can circumvent this step + self.broadcasts + .send_document_update( + vault_id.clone(), + WebSocketServerMessageWithOrigin::with_origin( + version.device_id.clone(), + WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate { + documents: vec![version.clone().into()], + is_initial_sync: false, + }), + ), + ) + .await; + + Ok(()) + } +} diff --git a/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql b/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql index 4a9f31ba..0b3076c6 100644 --- a/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql +++ b/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql @@ -4,7 +4,9 @@ CREATE TABLE IF NOT EXISTS documents ( relative_path TEXT NOT NULL, updated_date TIMESTAMP NOT NULL, content BLOB NOT NULL, - is_deleted BOOLEAN NOT NULL + is_deleted BOOLEAN NOT NULL, + user_id TEXT NOT NULL, + device_id TEXT NOT NULL ); CREATE VIEW IF NOT EXISTS latest_document_versions AS @@ -19,3 +21,10 @@ ON d.vault_update_id = max_versions.max_version_id; CREATE INDEX IF NOT EXISTS idx_documents_vault_id_relative_path ON documents (relative_path); + +-- The immutable version and its acknowledgement commit in one transaction. +CREATE TABLE push_acknowledgements ( + request_id TEXT NOT NULL PRIMARY KEY, + request_fingerprint BLOB NOT NULL, + vault_update_id INTEGER NOT NULL REFERENCES documents(vault_update_id) +); diff --git a/sync-server/src/app_state/database/migrations/20250522192949_add_provenance_columns.sql b/sync-server/src/app_state/database/migrations/20250522192949_add_provenance_columns.sql deleted file mode 100644 index 06860174..00000000 --- a/sync-server/src/app_state/database/migrations/20250522192949_add_provenance_columns.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE documents ADD COLUMN user_id TEXT NOT NULL DEFAULT ""; -ALTER TABLE documents ADD COLUMN device_id TEXT NOT NULL DEFAULT ""; diff --git a/sync-server/src/app_state/database/migrations/20251019103154_add_has_been_merged.sql b/sync-server/src/app_state/database/migrations/20251019103154_add_has_been_merged.sql deleted file mode 100644 index 5259c175..00000000 --- a/sync-server/src/app_state/database/migrations/20251019103154_add_has_been_merged.sql +++ /dev/null @@ -1,13 +0,0 @@ -ALTER TABLE documents ADD COLUMN has_been_merged BOOLEAN NOT NULL DEFAULT False; - -DROP VIEW latest_document_versions; - -CREATE VIEW IF NOT EXISTS latest_document_versions AS --recreate view as it now includes one more field -SELECT d.* -FROM documents d -INNER JOIN ( - SELECT MAX(vault_update_id) AS max_version_id - FROM documents - GROUP BY document_id -) max_versions -ON d.vault_update_id = max_versions.max_version_id; diff --git a/sync-server/src/app_state/database/models.rs b/sync-server/src/app_state/database/models.rs index a216125a..24c0c370 100644 --- a/sync-server/src/app_state/database/models.rs +++ b/sync-server/src/app_state/database/models.rs @@ -20,8 +20,6 @@ pub struct StoredDocumentVersion { pub is_deleted: bool, pub user_id: UserId, pub device_id: DeviceId, - #[allow(dead_code)] // This is for manual analysis - pub has_been_merged: bool, } impl PartialEq for StoredDocumentVersion { diff --git a/sync-server/src/consts.rs b/sync-server/src/consts.rs index 98ed1c1f..9e9890c0 100644 --- a/sync-server/src/consts.rs +++ b/sync-server/src/consts.rs @@ -20,4 +20,4 @@ pub const DEFAULT_LOG_LEVEL: LogLevel = LogLevel::Info; pub const DEFAULT_MERGEABLE_FILE_EXTENSIONS: &[&str] = &["md", "txt"]; -pub const SUPPORTED_API_VERSION: u32 = 2; +pub const SUPPORTED_API_VERSION: u32 = 3; diff --git a/sync-server/src/server.rs b/sync-server/src/server.rs index 01b09cf6..b8304a6c 100644 --- a/sync-server/src/server.rs +++ b/sync-server/src/server.rs @@ -1,17 +1,8 @@ pub mod auth; -mod create_document; -mod delete_document; mod device_id_header; -mod fetch_document_version; -mod fetch_document_version_content; -mod fetch_latest_document_version; -mod fetch_latest_documents; -mod index; -mod ping; +mod endpoints; mod requests; mod responses; -mod update_document; -mod websocket; use anyhow::{Context as _, Result, anyhow}; use auth::auth_middleware; @@ -21,7 +12,7 @@ use axum::{ http::{self, HeaderValue, Method}, middleware, response::IntoResponse, - routing::{IntoMakeService, delete, get, post, put}, + routing::{IntoMakeService, get}, }; use device_id_header::DEVICE_ID_HEADER_NAME; use log::info; @@ -53,9 +44,12 @@ pub async fn create_server(config: Config) -> Result<()> { let app = Router::new() .nest("/", get_authed_routes(app_state.clone())) - .route("/", get(index::index)) - .route("/vaults/:vault_id/ping", get(ping::ping)) - .route("/vaults/:vault_id/ws", get(websocket::websocket_handler)) + .route("/", get(endpoints::index::index)) + .route("/vaults/:vault_id/ping", get(endpoints::ping::ping)) + .route( + "/vaults/:vault_id/ws", + get(endpoints::websocket::websocket_handler), + ) .layer(DefaultBodyLimit::disable()) .layer(RequestBodyLimitLayer::new( app_state.config.server.max_body_size_mb * 1024 * 1024, @@ -69,7 +63,7 @@ pub async fn create_server(config: Config) -> Result<()> { http::header::AUTHORIZATION, DEVICE_ID_HEADER_NAME.clone(), ]) - .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE]), + .allow_methods([Method::GET, Method::PUT]), ) .layer( TraceLayer::new_for_http() @@ -102,35 +96,16 @@ fn get_authed_routes(app_state: AppState) -> Router { Router::new() .route( "/vaults/:vault_id/documents", - get(fetch_latest_documents::fetch_latest_documents), - ) - .route( - "/vaults/:vault_id/documents", - post(create_document::create_document), + get(endpoints::fetch_latest_documents::fetch_latest_documents), ) .route( "/vaults/:vault_id/documents/:document_id", - get(fetch_latest_document_version::fetch_latest_document_version), - ) - .route( - "/vaults/:vault_id/documents/:document_id/binary", - put(update_document::update_binary), - ) - .route( - "/vaults/:vault_id/documents/:document_id/text", - put(update_document::update_text), - ) - .route( - "/vaults/:vault_id/documents/:document_id/versions/:vault_update_id", - get(fetch_document_version::fetch_document_version), + get(endpoints::fetch_latest_document_version::fetch_latest_document_version) + .put(endpoints::push_document::push_document), ) .route( "/vaults/:vault_id/documents/:document_id/versions/:vault_update_id/content", - get(fetch_document_version_content::fetch_document_version_content), - ) - .route( - "/vaults/:vault_id/documents/:document_id", - delete(delete_document::delete_document), + get(endpoints::fetch_document_version_content::fetch_document_version_content), ) .layer(middleware::from_fn_with_state(app_state, auth_middleware)) } diff --git a/sync-server/src/server/create_document.rs b/sync-server/src/server/create_document.rs deleted file mode 100644 index 859c0db4..00000000 --- a/sync-server/src/server/create_document.rs +++ /dev/null @@ -1,109 +0,0 @@ -use axum::{ - Extension, Json, - extract::{Path, State}, -}; -use axum_extra::TypedHeader; -use axum_typed_multipart::TypedMultipart; -use log::{debug, info}; -use serde::Deserialize; - -use super::{device_id_header::DeviceIdHeader, requests::CreateDocumentVersion}; -use crate::{ - app_state::{ - AppState, - database::models::{DocumentVersionWithoutContent, StoredDocumentVersion, VaultId}, - }, - config::user_config::User, - errors::{SyncServerError, client_error, server_error}, - utils::{ - find_first_available_path::find_first_available_path, normalize::normalize, - sanitize_path::sanitize_path, - }, -}; - -#[derive(Deserialize)] -pub struct CreateDocumentPathParams { - #[serde(deserialize_with = "normalize")] - vault_id: VaultId, -} - -/// Create a new document in case a document with the same doesn't exist -/// already. If a document with the same path exists, a new version is created -/// with their content merged. -#[axum::debug_handler] -pub async fn create_document( - Path(CreateDocumentPathParams { vault_id }): Path, - Extension(user): Extension, - TypedHeader(device_id): TypedHeader, - State(state): State, - TypedMultipart(request): TypedMultipart, -) -> Result, SyncServerError> { - debug!("Creating document in vault `{vault_id}`"); - - let mut transaction = state - .database - .create_write_transaction(&vault_id) - .await - .map_err(server_error)?; - - let document_id = match request.document_id { - Some(document_id) => { - let existing_version = state - .database - .get_latest_document(&vault_id, &document_id, Some(&mut transaction)) - .await - .map_err(server_error)?; - - if existing_version.is_some() { - return Err(client_error(anyhow::anyhow!( - "Document with the same ID `{document_id}` already exists" - ))); - } - - document_id - } - None => uuid::Uuid::new_v4(), - }; - - let last_update_id = state - .database - .get_max_update_id_in_vault(&vault_id, Some(&mut transaction)) - .await - .map_err(server_error)?; - - let sanitized_relative_path = sanitize_path(&request.relative_path); - let deduped_path = find_first_available_path( - &vault_id, - &sanitized_relative_path, - &state.database, - &mut transaction, - ) - .await - .map_err(server_error)?; - - if deduped_path != sanitized_relative_path { - info!( - "Document already exists at new location: `{sanitized_relative_path}` when trying to create it in vault `{vault_id}`, deconflicting by creating at `{deduped_path}`" - ); - } - - let new_version = StoredDocumentVersion { - vault_update_id: last_update_id + 1, - document_id, - relative_path: deduped_path, - content: request.content.contents.to_vec(), - updated_date: chrono::Utc::now(), - is_deleted: false, - user_id: user.name, - device_id: device_id.0, - has_been_merged: false, - }; - - state - .database - .insert_document_version(&vault_id, &new_version, Some(transaction)) - .await - .map_err(server_error)?; - - Ok(Json(new_version.into())) -} diff --git a/sync-server/src/server/delete_document.rs b/sync-server/src/server/delete_document.rs deleted file mode 100644 index e126d6b5..00000000 --- a/sync-server/src/server/delete_document.rs +++ /dev/null @@ -1,96 +0,0 @@ -use anyhow::Context; -use axum::{ - Extension, Json, - extract::{Path, State}, -}; -use axum_extra::TypedHeader; -use log::{debug, info}; -use serde::Deserialize; - -use super::{device_id_header::DeviceIdHeader, requests::DeleteDocumentVersion}; -use crate::{ - app_state::{ - AppState, - database::models::{ - DocumentId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId, - }, - }, - config::user_config::User, - errors::{SyncServerError, server_error}, - utils::{normalize::normalize, sanitize_path::sanitize_path}, -}; - -#[derive(Deserialize)] -pub struct DeleteDocumentPathParams { - #[serde(deserialize_with = "normalize")] - vault_id: VaultId, - - document_id: DocumentId, -} - -#[axum::debug_handler] -pub async fn delete_document( - Path(DeleteDocumentPathParams { - vault_id, - document_id, - }): Path, - Extension(user): Extension, - TypedHeader(device_id): TypedHeader, - State(state): State, - Json(request): Json, -) -> Result, SyncServerError> { - debug!("Deleting document `{document_id}` in vault `{vault_id}`"); - - let mut transaction = state - .database - .create_write_transaction(&vault_id) - .await - .map_err(server_error)?; - - let last_update_id = state - .database - .get_max_update_id_in_vault(&vault_id, Some(&mut transaction)) - .await - .map_err(server_error)?; - - let latest_version = state - .database - .get_latest_document(&vault_id, &document_id, Some(&mut transaction)) - .await - .map_err(server_error)?; - - if let Some(latest_version) = &latest_version - && latest_version.is_deleted - { - transaction - .rollback() - .await - .context("Failed to roll back transaction") - .map_err(server_error)?; - - info!("Document `{document_id}` has already been deleted",); - return Ok(Json(latest_version.clone().into())); - } - - let latest_content = latest_version.map_or_else(Vec::new, |version| version.content); // in case the document has never existed before deleting it - - let new_version = StoredDocumentVersion { - vault_update_id: last_update_id + 1, - document_id, - relative_path: sanitize_path(&request.relative_path), - content: latest_content, // copy the content from the latest version - updated_date: chrono::Utc::now(), - is_deleted: true, - user_id: user.name, - device_id: device_id.0, - has_been_merged: false, - }; - - state - .database - .insert_document_version(&vault_id, &new_version, Some(transaction)) - .await - .map_err(server_error)?; - - Ok(Json(new_version.into())) -} diff --git a/sync-server/src/server/endpoints.rs b/sync-server/src/server/endpoints.rs new file mode 100644 index 00000000..243cf7fa --- /dev/null +++ b/sync-server/src/server/endpoints.rs @@ -0,0 +1,7 @@ +pub mod fetch_document_version_content; +pub mod fetch_latest_document_version; +pub mod fetch_latest_documents; +pub mod index; +pub mod ping; +pub mod push_document; +pub mod websocket; diff --git a/sync-server/src/server/fetch_document_version_content.rs b/sync-server/src/server/endpoints/fetch_document_version_content.rs similarity index 100% rename from sync-server/src/server/fetch_document_version_content.rs rename to sync-server/src/server/endpoints/fetch_document_version_content.rs diff --git a/sync-server/src/server/fetch_latest_document_version.rs b/sync-server/src/server/endpoints/fetch_latest_document_version.rs similarity index 100% rename from sync-server/src/server/fetch_latest_document_version.rs rename to sync-server/src/server/endpoints/fetch_latest_document_version.rs diff --git a/sync-server/src/server/endpoints/fetch_latest_documents.rs b/sync-server/src/server/endpoints/fetch_latest_documents.rs new file mode 100644 index 00000000..eee16441 --- /dev/null +++ b/sync-server/src/server/endpoints/fetch_latest_documents.rs @@ -0,0 +1,36 @@ +use axum::{ + Json, + extract::{Path, State}, +}; +use log::debug; +use serde::Deserialize; + +use crate::{ + app_state::{ + AppState, + database::models::{DocumentVersionWithoutContent, VaultId}, + }, + errors::{SyncServerError, server_error}, + utils::normalize::normalize, +}; + +#[derive(Deserialize)] +pub struct FetchLatestDocumentsPathParams { + #[serde(deserialize_with = "normalize")] + vault_id: VaultId, +} + +#[axum::debug_handler] +pub async fn fetch_latest_documents( + Path(FetchLatestDocumentsPathParams { vault_id }): Path, + State(state): State, +) -> Result>, SyncServerError> { + debug!("Fetching latest documents in {vault_id}"); + + state + .database + .get_latest_documents(&vault_id, None) + .await + .map(Json) + .map_err(server_error) +} diff --git a/sync-server/src/server/index.rs b/sync-server/src/server/endpoints/index.rs similarity index 68% rename from sync-server/src/server/index.rs rename to sync-server/src/server/endpoints/index.rs index 64b053f7..982b0b88 100644 --- a/sync-server/src/server/index.rs +++ b/sync-server/src/server/endpoints/index.rs @@ -1,7 +1,7 @@ use axum::response::{Html, IntoResponse}; pub async fn index() -> impl IntoResponse { - const HTML_CONTENT: &str = include_str!("./assets/index.html"); + const HTML_CONTENT: &str = include_str!("../assets/index.html"); let html_content = HTML_CONTENT; Html(html_content) } diff --git a/sync-server/src/server/ping.rs b/sync-server/src/server/endpoints/ping.rs similarity index 95% rename from sync-server/src/server/ping.rs rename to sync-server/src/server/endpoints/ping.rs index 31aa8acd..c42219be 100644 --- a/sync-server/src/server/ping.rs +++ b/sync-server/src/server/endpoints/ping.rs @@ -9,7 +9,7 @@ use axum_extra::{ use log::debug; use serde::Deserialize; -use super::{auth::auth, responses::PingResponse}; +use crate::server::{auth::auth, responses::PingResponse}; use crate::{ app_state::{AppState, database::models::VaultId}, consts::SUPPORTED_API_VERSION, diff --git a/sync-server/src/server/endpoints/push_document.rs b/sync-server/src/server/endpoints/push_document.rs new file mode 100644 index 00000000..676e8baa --- /dev/null +++ b/sync-server/src/server/endpoints/push_document.rs @@ -0,0 +1,210 @@ +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/push_document/tests.rs b/sync-server/src/server/endpoints/push_document/tests.rs new file mode 100644 index 00000000..291c2def --- /dev/null +++ b/sync-server/src/server/endpoints/push_document/tests.rs @@ -0,0 +1,423 @@ +use std::path::PathBuf; + +use super::*; +use crate::{ + app_state::{ + database::models::{DocumentVersionWithoutContent, VaultUpdateId}, + websocket::broadcasts::Broadcasts, + }, + config::{database_config::DatabaseConfig, server_config::ServerConfig}, +}; + +struct Push { + document_id: DocumentId, + request_id: uuid::Uuid, + parent_version_id: Option, + relative_path: String, + content: PushContent, + user_id: String, + device_id: String, +} + +impl Push { + async fn send( + self, + database: &Database, + vault: &VaultId, + ) -> Result { + accept_push( + database, + vault, + self.document_id, + self.user_id, + self.device_id, + PushDocument { + request_id: self.request_id, + parent_version_id: self.parent_version_id, + relative_path: self.relative_path, + content: self.content, + }, + ) + .await + } +} + +struct TestVault { + directory: PathBuf, + database: Database, + vault: VaultId, +} + +impl TestVault { + async fn new() -> Self { + let directory = + std::env::temp_dir().join(format!("vaultlink-cas-{}", uuid::Uuid::new_v4())); + let database = Self::open(&directory).await; + Self { + directory, + database, + vault: "test".to_owned(), + } + } + + async fn open(directory: &std::path::Path) -> Database { + Database::try_new( + &DatabaseConfig { + databases_directory_path: directory.to_owned(), + ..Default::default() + }, + &Broadcasts::new(&ServerConfig { + max_clients_per_vault: 16, + ..Default::default() + }), + ) + .await + .unwrap() + } + + async fn send(&self, push: Push) -> DocumentUpdateResponse { + push.send(&self.database, &self.vault).await.unwrap() + } + + async fn latest(&self, document: DocumentId) -> StoredDocumentVersion { + self.database + .get_latest_document(&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, + relative_path: "note.md".to_owned(), + 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: PushDocument = serde_json::from_value(serde_json::json!({ + "requestId": uuid::Uuid::new_v4(), + "parentVersionId": null, + "relativePath": "snapshot.bin", + "content": { "type": "Snapshot", "value": STANDARD.encode(content) }, + })) + .unwrap(); + let base = accepted( + accept_push( + &vault.database, + &vault.vault, + document, + "user".to_owned(), + "device".to_owned(), + request, + ) + .await + .unwrap(), + ); + assert_eq!(vault.latest(document).await.content, content); + + let invalid: PushDocument = serde_json::from_value(serde_json::json!({ + "requestId": uuid::Uuid::new_v4(), + "parentVersionId": base.vault_update_id, + "relativePath": "snapshot.bin", + "content": { "type": "Snapshot", "value": "not base64!" }, + })) + .unwrap(); + assert!(matches!( + accept_push( + &vault.database, + &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.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 + .database + .get_document_version(&vault.vault, base.vault_update_id, None) + .await + .unwrap() + .unwrap(); + assert_eq!(original.content, b"base"); + assert_eq!( + vault + .database + .get_max_update_id_in_vault(&vault.vault, None) + .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_path = snapshot(document, None, "created"); + changed_path.relative_path = "other.md".to_owned(); + let mut changed_operation = snapshot(document, None, "created"); + changed_operation.content = PushContent::Delete; + for mut changed_payload in [ + snapshot(document, None, "different content"), + snapshot(document, Some(1), "created"), + changed_path, + changed_operation, + ] { + changed_payload.request_id = create_id; + assert!(matches!( + changed_payload.send(&reopened, &vault.vault).await, + Err(SyncServerError::ClientError(_)) + )); + } +} + +#[tokio::test] +async fn deletes_compare_the_base_and_replay_without_resurrecting_documents() { + let vault = TestVault::new().await; + let document = uuid::Uuid::new_v4(); + let base = accepted(vault.send(snapshot(document, None, "base")).await); + let updated = accepted( + vault + .send(snapshot(document, Some(base.vault_update_id), "edited")) + .await, + ); + let mut delete = snapshot(document, Some(base.vault_update_id), ""); + delete.content = PushContent::Delete; + assert!(matches!( + vault.send(delete).await, + DocumentUpdateResponse::StaleBase(_) + )); + assert!(!vault.latest(document).await.is_deleted); + + let mut delete = snapshot(document, Some(updated.vault_update_id), ""); + delete.content = PushContent::Delete; + let request_id = delete.request_id; + let deleted = accepted(vault.send(delete).await); + assert!(deleted.is_deleted); + assert_eq!(vault.latest(document).await.content, b"edited"); + let mut retry = snapshot(document, Some(updated.vault_update_id), ""); + retry.content = PushContent::Delete; + retry.request_id = request_id; + assert_eq!( + accepted(vault.send(retry).await).vault_update_id, + deleted.vault_update_id + ); + + let terminal = vault + .send(snapshot( + document, + Some(deleted.vault_update_id), + "resurrect", + )) + .await; + assert!(matches!(terminal, DocumentUpdateResponse::StaleBase(version) if version.is_deleted)); + let recreated = accepted( + vault + .send(snapshot(uuid::Uuid::new_v4(), None, "new identity")) + .await, + ); + assert_eq!(recreated.relative_path, "note.md"); +} + +#[tokio::test] +async fn deleting_a_missing_document_without_a_parent_is_a_client_error() { + let vault = TestVault::new().await; + let document = uuid::Uuid::new_v4(); + let mut delete = snapshot(document, None, ""); + delete.content = PushContent::Delete; + assert!(matches!( + delete.send(&vault.database, &vault.vault).await, + Err(SyncServerError::ClientError(_)) + )); + assert_eq!( + vault + .database + .get_latest_document(&vault.vault, &document, None) + .await + .unwrap(), + None + ); +} + +#[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.database, &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.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.database, &vault.vault).await, + Err(SyncServerError::ClientError(_)) + )); + assert_eq!( + vault.latest(binary_document).await.vault_update_id, + binary_base.vault_update_id + ); +} + +#[tokio::test] +async fn renames_allocate_paths_atomically_without_rewriting_history() { + let vault = TestVault::new().await; + let first = uuid::Uuid::new_v4(); + let second = uuid::Uuid::new_v4(); + let base = accepted(vault.send(snapshot(first, None, "first")).await); + let collision = accepted(vault.send(snapshot(second, None, "second")).await); + assert_ne!(base.relative_path, collision.relative_path); + + let mut rename = snapshot(first, Some(base.vault_update_id), "first"); + rename.relative_path = "renamed.md".to_owned(); + let renamed = accepted(vault.send(rename).await); + let mut stale_rename = snapshot(first, Some(base.vault_update_id), "first"); + stale_rename.relative_path = "stale.md".to_owned(); + assert!( + matches!(vault.send(stale_rename).await, DocumentUpdateResponse::StaleBase(version) if version.relative_path == renamed.relative_path) + ); + + let mut move_into_occupied = snapshot(second, Some(collision.vault_update_id), "second"); + move_into_occupied.relative_path = renamed.relative_path.clone(); + let allocated = accepted(vault.send(move_into_occupied).await); + assert_ne!(allocated.relative_path, renamed.relative_path); + assert_eq!(vault.latest(first).await.relative_path, "renamed.md"); + assert_eq!( + vault + .database + .get_document_version(&vault.vault, base.vault_update_id, None) + .await + .unwrap() + .unwrap() + .relative_path, + "note.md" + ); +} diff --git a/sync-server/src/server/websocket.rs b/sync-server/src/server/endpoints/websocket.rs similarity index 100% rename from sync-server/src/server/websocket.rs rename to sync-server/src/server/endpoints/websocket.rs diff --git a/sync-server/src/server/fetch_document_version.rs b/sync-server/src/server/fetch_document_version.rs deleted file mode 100644 index c30f1d76..00000000 --- a/sync-server/src/server/fetch_document_version.rs +++ /dev/null @@ -1,62 +0,0 @@ -use anyhow::anyhow; -use axum::{ - Json, - extract::{Path, State}, -}; -use log::debug; -use serde::Deserialize; - -use crate::{ - app_state::{ - AppState, - database::models::{DocumentId, DocumentVersion, VaultId, VaultUpdateId}, - }, - errors::{SyncServerError, not_found_error, server_error}, - utils::normalize::normalize, -}; - -#[derive(Deserialize)] -pub struct FetchDocumentVersionPathParams { - #[serde(deserialize_with = "normalize")] - vault_id: VaultId, - - document_id: DocumentId, - vault_update_id: VaultUpdateId, -} - -#[axum::debug_handler] -pub async fn fetch_document_version( - Path(FetchDocumentVersionPathParams { - vault_id, - document_id, - vault_update_id, - }): Path, - State(state): State, -) -> Result, SyncServerError> { - debug!( - "Fetching document version `{vault_update_id}` for document `{document_id}` in vault `{vault_id}`" - ); - - let result = state - .database - .get_document_version(&vault_id, vault_update_id, None) - .await - .map_err(server_error)? - .map_or_else( - || { - Err(not_found_error(anyhow!( - "Document with vault update id `{vault_update_id}` not found", - ))) - }, - Ok, - )?; - - if result.document_id != document_id { - return Err(not_found_error(anyhow!( - "Document with document id `{document_id}` does not have a version with id \ - `{vault_update_id}`", - ))); - } - - Ok(Json(result.into())) -} diff --git a/sync-server/src/server/fetch_latest_documents.rs b/sync-server/src/server/fetch_latest_documents.rs deleted file mode 100644 index 209374ce..00000000 --- a/sync-server/src/server/fetch_latest_documents.rs +++ /dev/null @@ -1,59 +0,0 @@ -use axum::{ - Json, - extract::{Path, Query, State}, -}; -use log::debug; -use serde::Deserialize; - -use super::responses::FetchLatestDocumentsResponse; -use crate::{ - app_state::{ - AppState, - database::models::{VaultId, VaultUpdateId}, - }, - errors::{SyncServerError, server_error}, - utils::normalize::normalize, -}; - -#[derive(Deserialize)] -pub struct FetchLatestDocumentsPathParams { - #[serde(deserialize_with = "normalize")] - vault_id: VaultId, -} - -#[derive(Deserialize)] -pub struct QueryParams { - since_update_id: Option, -} - -#[axum::debug_handler] -pub async fn fetch_latest_documents( - Path(FetchLatestDocumentsPathParams { vault_id }): Path, - Query(QueryParams { since_update_id }): Query, - State(state): State, -) -> Result, SyncServerError> { - debug!("Fetching latest documents in vault `{vault_id}` since update ID `{since_update_id:?}`"); - - let documents = if let Some(since_update_id) = since_update_id { - state - .database - .get_latest_documents_since(&vault_id, since_update_id, None) - .await - .map_err(server_error) - } else { - state - .database - .get_latest_documents(&vault_id, None) - .await - .map_err(server_error) - }?; - - Ok(Json(FetchLatestDocumentsResponse { - last_update_id: documents - .iter() - .map(|doc| doc.vault_update_id) - .max() - .unwrap_or(since_update_id.unwrap_or(0)), - latest_documents: documents, - })) -} diff --git a/sync-server/src/server/requests.rs b/sync-server/src/server/requests.rs index 119ad467..7b814559 100644 --- a/sync-server/src/server/requests.rs +++ b/sync-server/src/server/requests.rs @@ -1,51 +1,23 @@ -use axum::body::Bytes; -use axum_typed_multipart::{FieldData, TryFromMultipart}; +use crate::app_state::database::models::VaultUpdateId; use reconcile_text::NumberOrText; -use serde::{self, Deserialize}; +use serde::{Deserialize, Serialize}; use ts_rs::TS; -use crate::app_state::database::models::{DocumentId, VaultUpdateId}; - -#[derive(TS, Debug, TryFromMultipart)] -#[ts(export)] -pub struct CreateDocumentVersion { - /// The client can decide the document id (if it wishes to) in order - /// to help with syncing. If the client does not provide a document id, - /// the server will generate one. If the client provides a document id - /// it must not already exist in the database. - pub document_id: Option, - pub relative_path: String, - - #[ts(as = "Vec")] - #[form_data(limit = "unlimited")] - pub content: FieldData, +#[derive(TS, Debug, Deserialize, Serialize)] +#[serde(tag = "type", content = "value")] +pub enum PushContent { + Snapshot(String), + Diff(#[ts(type = "Array")] Vec), + Delete, } -#[derive(Debug, TryFromMultipart)] -pub struct UpdateBinaryDocumentVersion { - pub parent_version_id: VaultUpdateId, - pub relative_path: String, - - #[form_data(limit = "unlimited")] - pub content: FieldData, -} - -#[derive(TS, Debug, Deserialize)] +#[derive(TS, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] #[ts(export)] -pub struct UpdateTextDocumentVersion { - #[ts(as = "i32")] - pub parent_version_id: VaultUpdateId, - - pub relative_path: String, - - #[ts(type = "Array")] - pub content: Vec, -} - -#[derive(TS, Debug, Deserialize)] -#[serde(rename_all = "camelCase")] -#[ts(export)] -pub struct DeleteDocumentVersion { +pub struct PushDocument { + pub request_id: uuid::Uuid, + #[ts(as = "Option")] + pub parent_version_id: Option, pub relative_path: String, + pub content: PushContent, } diff --git a/sync-server/src/server/responses.rs b/sync-server/src/server/responses.rs index a8b3fcd7..ed469949 100644 --- a/sync-server/src/server/responses.rs +++ b/sync-server/src/server/responses.rs @@ -1,9 +1,7 @@ use serde::{self, Serialize}; use ts_rs::TS; -use crate::app_state::database::models::{ - DocumentVersion, DocumentVersionWithoutContent, VaultUpdateId, -}; +use crate::app_state::database::models::{DocumentVersion, DocumentVersionWithoutContent}; /// Response to a ping request. #[derive(TS, Debug, Clone, Serialize)] @@ -25,28 +23,14 @@ pub struct PingResponse { pub supported_api_version: u32, } -/// Response to a fetch latest documents request. -#[derive(TS, Debug, Clone, Serialize)] -#[serde(rename_all = "camelCase")] -#[ts(export)] -pub struct FetchLatestDocumentsResponse { - pub latest_documents: Vec, - - /// The update ID of the latest document in the response. - pub last_update_id: VaultUpdateId, -} - /// Response to an update document request. #[derive(TS, Debug, Clone, Serialize)] #[serde(tag = "type")] #[ts(export)] pub enum DocumentUpdateResponse { - /// Returned when the created/updated document's content is the same as was - /// sent in the create/update request and thus the response doesn't contain - /// the content because the client must already have it. - FastForwardUpdate(DocumentVersionWithoutContent), + /// The exact snapshot was committed, or this is its original acknowledgement. + Accepted(DocumentVersionWithoutContent), - /// Returned when the created/updated document's content is different from - /// what was sent in the create/update request. - MergingUpdate(DocumentVersion), + /// Nothing was written. The client must incorporate this version before retrying. + StaleBase(DocumentVersion), } diff --git a/sync-server/src/server/update_document.rs b/sync-server/src/server/update_document.rs deleted file mode 100644 index 00fbd008..00000000 --- a/sync-server/src/server/update_document.rs +++ /dev/null @@ -1,271 +0,0 @@ -use anyhow::{Context as _, anyhow}; -use axum::{ - Extension, Json, - extract::{Path, State}, -}; -use axum_extra::TypedHeader; -use axum_typed_multipart::TypedMultipart; -use log::{debug, info}; -use reconcile_text::{BuiltinTokenizer, EditedText, reconcile}; -use serde::Deserialize; - -use super::{ - device_id_header::DeviceIdHeader, requests::UpdateTextDocumentVersion, - responses::DocumentUpdateResponse, -}; -use crate::{ - app_state::{ - AppState, - database::models::{DocumentId, StoredDocumentVersion, VaultId, VaultUpdateId}, - }, - config::user_config::User, - errors::{SyncServerError, client_error, not_found_error, server_error}, - server::requests::UpdateBinaryDocumentVersion, - utils::{ - find_first_available_path::find_first_available_path, is_binary::is_binary, - is_file_type_mergable::is_file_type_mergable, normalize::normalize, - sanitize_path::sanitize_path, - }, -}; - -#[derive(Deserialize)] -pub struct UpdateDocumentPathParams { - #[serde(deserialize_with = "normalize")] - vault_id: VaultId, - document_id: DocumentId, -} - -#[axum::debug_handler] -pub async fn update_binary( - Path(UpdateDocumentPathParams { - vault_id, - document_id, - }): Path, - Extension(user): Extension, - TypedHeader(device_id): TypedHeader, - State(state): State, - TypedMultipart(request): TypedMultipart, -) -> Result, SyncServerError> { - let parent_document = get_parent_document(&state, &vault_id, request.parent_version_id).await?; - let content = request.content.contents.to_vec(); - - update_document( - parent_document, - vault_id, - document_id, - user, - device_id, - state, - &request.relative_path, - content, - ) - .await -} - -#[axum::debug_handler] -#[allow(clippy::too_many_lines)] -pub async fn update_text( - Path(UpdateDocumentPathParams { - vault_id, - document_id, - }): Path, - Extension(user): Extension, - TypedHeader(device_id): TypedHeader, - State(state): State, - Json(request): Json, -) -> Result, SyncServerError> { - let parent_document = get_parent_document(&state, &vault_id, request.parent_version_id).await?; - - let edited_text = EditedText::from_diff( - str::from_utf8(&parent_document.content) - .expect("parent must be valid UTF-8 because it's a text document"), - request.content, - &*BuiltinTokenizer::Word, - ) - .context("Failed to apply given diff to parent document") - .map_err(client_error)?; - - let content = edited_text.apply().text().into_bytes(); - - update_document( - parent_document, - vault_id, - document_id, - user, - device_id, - state, - &request.relative_path, - content, - ) - .await -} - -async fn get_parent_document( - state: &AppState, - vault_id: &VaultId, - parent_version_id: VaultUpdateId, -) -> Result { - state - .database - .get_document_version(vault_id, parent_version_id, None) - .await - .map_err(server_error)? - .map_or_else( - || { - Err(not_found_error(anyhow!( - "Parent version with id `{parent_version_id}` not found" - ))) - }, - Ok, - ) -} - -#[allow(clippy::too_many_lines, clippy::too_many_arguments)] -async fn update_document( - parent_document: StoredDocumentVersion, - vault_id: VaultId, - document_id: DocumentId, - user: User, - device_id: DeviceIdHeader, - state: AppState, - relative_path: &str, - content: Vec, -) -> Result, SyncServerError> { - debug!("Updating document `{document_id}` in vault `{vault_id}`"); - - let sanitized_relative_path = sanitize_path(relative_path); - - let mut transaction = state - .database - .create_write_transaction(&vault_id) - .await - .map_err(server_error)?; - - let last_update_id = state - .database - .get_max_update_id_in_vault(&vault_id, Some(&mut transaction)) - .await - .map_err(server_error)?; - - let latest_version = state - .database - .get_latest_document(&vault_id, &document_id, Some(&mut transaction)) - .await - .map_err(server_error)? - .map_or_else( - || { - Err(not_found_error(anyhow!( - "Document with id `{document_id}` not found", - ))) - }, - Ok, - )?; - - if latest_version.is_deleted { - transaction - .rollback() - .await - .context("Failed to roll back transaction") - .map_err(server_error)?; - - info!("Document `{document_id}` has been deleted, ignoring update to it",); - return Ok(Json(DocumentUpdateResponse::FastForwardUpdate( - latest_version.into(), - ))); - } - - // Return the latest version if the content and path are the same as the latest - // version - if content == latest_version.content && sanitized_relative_path == latest_version.relative_path - { - info!( - "Document content is the same as the latest version for `{document_id}`, skipping update" - ); - transaction - .rollback() - .await - .context("Failed to roll back transaction") - .map_err(server_error)?; - - return Ok(Json(DocumentUpdateResponse::FastForwardUpdate( - latest_version.into(), - ))); - } - - let are_all_participants_mergable = is_file_type_mergable( - &sanitized_relative_path, - &state.config.server.mergeable_file_extensions, - ) && !is_binary(&parent_document.content) - && !is_binary(&latest_version.content) - && !is_binary(&content); - - let merged_content = if are_all_participants_mergable { - info!("Merging changes for document `{document_id}` in vault `{vault_id}`"); - reconcile( - str::from_utf8(&parent_document.content) - .expect("parent must be valid UTF-8 because it's not binary"), - &str::from_utf8(&latest_version.content) - .expect("latest_version must be valid UTF-8 because it's not binary") - .into(), - &str::from_utf8(&content) - .expect("content must be valid UTF-8 because it's not binary") - .into(), - &*BuiltinTokenizer::Word, - ) - .apply() - .text() - .into_bytes() - } else { - content.clone() - }; - - let is_different_from_request_content = merged_content != content; - - // We can only update the relative path if we're the first one to do so - let new_relative_path = if parent_document.relative_path == latest_version.relative_path - && latest_version.relative_path != sanitized_relative_path - { - let new_path = find_first_available_path( - &vault_id, - &sanitized_relative_path, - &state.database, - &mut transaction, - ) - .await - .map_err(server_error)?; - - if new_path != sanitized_relative_path { - info!( - "Document already exists at new location: `{sanitized_relative_path}` when trying to update it in vault `{vault_id}`, deconflicting by creating at `{new_path}`" - ); - } - - new_path - } else { - latest_version.relative_path.clone() - }; - - let new_version = StoredDocumentVersion { - document_id, - vault_update_id: last_update_id + 1, - relative_path: new_relative_path, - content: merged_content, - updated_date: chrono::Utc::now(), - is_deleted: false, - user_id: user.name, - device_id: device_id.0, - has_been_merged: are_all_participants_mergable && is_different_from_request_content, - }; - - state - .database - .insert_document_version(&vault_id, &new_version, Some(transaction)) - .await - .map_err(server_error)?; - - Ok(Json(if is_different_from_request_content { - DocumentUpdateResponse::MergingUpdate(new_version.into()) - } else { - DocumentUpdateResponse::FastForwardUpdate(new_version.into()) - })) -} diff --git a/sync-server/src/utils.rs b/sync-server/src/utils.rs index b501ecb2..35d6a340 100644 --- a/sync-server/src/utils.rs +++ b/sync-server/src/utils.rs @@ -1,7 +1,5 @@ pub mod dedup_paths; pub mod find_first_available_path; -pub mod is_binary; -pub mod is_file_type_mergable; pub mod log_level; pub mod normalize; pub mod rotating_file_writer; diff --git a/sync-server/src/utils/is_binary.rs b/sync-server/src/utils/is_binary.rs deleted file mode 100644 index 09bfcf94..00000000 --- a/sync-server/src/utils/is_binary.rs +++ /dev/null @@ -1,26 +0,0 @@ -/// Heuristically determine if the given data is a binary or a text file's -/// content. -/// -/// Only text inputs can be reconciled using the crate's functions. -#[must_use] -pub fn is_binary(data: &[u8]) -> bool { - if data.contains(&0) { - // Even though the NUL character is valid in UTF-8, it's highly suspicious in - // human-readable text. - return true; - } - - std::str::from_utf8(data).is_err() -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_is_binary() { - assert!(is_binary(&[0, 159, 146, 150])); - assert!(is_binary(&[0, 12])); - assert!(!is_binary(b"hello")); - } -} diff --git a/sync-server/src/utils/is_file_type_mergable.rs b/sync-server/src/utils/is_file_type_mergable.rs deleted file mode 100644 index 7aabb393..00000000 --- a/sync-server/src/utils/is_file_type_mergable.rs +++ /dev/null @@ -1,34 +0,0 @@ -pub fn is_file_type_mergable(path_or_file_name: &str, mergeable_extensions: &[String]) -> bool { - let file_extension = path_or_file_name.split('.').next_back().unwrap_or_default(); - let file_extension_lower = file_extension.to_lowercase(); - - mergeable_extensions - .iter() - .any(|ext| ext.to_lowercase() == file_extension_lower) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_is_file_type_mergable() { - let mergeable = vec!["md".to_owned(), "txt".to_owned()]; - - assert!(is_file_type_mergable(".md", &mergeable)); - assert!(is_file_type_mergable("hi.md", &mergeable)); - assert!(is_file_type_mergable( - "my/path/to/my/document.md", - &mergeable - )); - assert!(is_file_type_mergable("hi.MD", &mergeable)); - assert!(is_file_type_mergable( - "my/path/to/my/DOCUMENT.MD", - &mergeable - )); - - assert!(!is_file_type_mergable(".json", &mergeable)); - assert!(!is_file_type_mergable("HELLO.JSON", &mergeable)); - assert!(!is_file_type_mergable("my/config.yml", &mergeable)); - } -}