diff --git a/Cargo.lock b/Cargo.lock index 9271162..2c2e758 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2581,7 +2581,7 @@ dependencies = [ [[package]] name = "rustpad" -version = "0.2.30" +version = "0.2.31" dependencies = [ "argon2", "aws-config", diff --git a/Cargo.toml b/Cargo.toml index b2a13ef..3377814 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rustpad" -version = "0.2.30" +version = "0.2.31" edition = "2024" rust-version = "1.94" description = "Collaborative Markdown notepad built with Axum, WebSockets and SQLite, PostgreSQL and MySQL" diff --git a/migrations/mysql/0028_collaboration_updates.sql b/migrations/mysql/0028_collaboration_updates.sql new file mode 100644 index 0000000..fcc831c --- /dev/null +++ b/migrations/mysql/0028_collaboration_updates.sql @@ -0,0 +1,11 @@ +ALTER TABLE note_revisions + ADD COLUMN collaboration_client_id VARCHAR(64) NULL, + ADD COLUMN collaboration_update_id BIGINT NULL, + ADD UNIQUE INDEX idx_note_revisions_collaboration_update + (note_id, collaboration_client_id, collaboration_update_id); + +ALTER TABLE revisions + ADD COLUMN collaboration_client_id VARCHAR(64) NULL, + ADD COLUMN collaboration_update_id BIGINT NULL, + ADD UNIQUE INDEX idx_revisions_collaboration_update + (pad_id, collaboration_client_id, collaboration_update_id); diff --git a/migrations/postgres/0028_collaboration_updates.sql b/migrations/postgres/0028_collaboration_updates.sql new file mode 100644 index 0000000..3bae1ad --- /dev/null +++ b/migrations/postgres/0028_collaboration_updates.sql @@ -0,0 +1,11 @@ +ALTER TABLE note_revisions ADD COLUMN collaboration_client_id TEXT; +ALTER TABLE note_revisions ADD COLUMN collaboration_update_id BIGINT; +CREATE UNIQUE INDEX idx_note_revisions_collaboration_update + ON note_revisions(note_id, collaboration_client_id, collaboration_update_id) + WHERE collaboration_client_id IS NOT NULL AND collaboration_update_id IS NOT NULL; + +ALTER TABLE revisions ADD COLUMN collaboration_client_id TEXT; +ALTER TABLE revisions ADD COLUMN collaboration_update_id BIGINT; +CREATE UNIQUE INDEX idx_revisions_collaboration_update + ON revisions(pad_id, collaboration_client_id, collaboration_update_id) + WHERE collaboration_client_id IS NOT NULL AND collaboration_update_id IS NOT NULL; diff --git a/migrations/sqlite/0028_collaboration_updates.sql b/migrations/sqlite/0028_collaboration_updates.sql new file mode 100644 index 0000000..798847a --- /dev/null +++ b/migrations/sqlite/0028_collaboration_updates.sql @@ -0,0 +1,11 @@ +ALTER TABLE note_revisions ADD COLUMN collaboration_client_id TEXT; +ALTER TABLE note_revisions ADD COLUMN collaboration_update_id INTEGER; +CREATE UNIQUE INDEX IF NOT EXISTS idx_note_revisions_collaboration_update + ON note_revisions(note_id, collaboration_client_id, collaboration_update_id) + WHERE collaboration_client_id IS NOT NULL AND collaboration_update_id IS NOT NULL; + +ALTER TABLE revisions ADD COLUMN collaboration_client_id TEXT; +ALTER TABLE revisions ADD COLUMN collaboration_update_id INTEGER; +CREATE UNIQUE INDEX IF NOT EXISTS idx_revisions_collaboration_update + ON revisions(pad_id, collaboration_client_id, collaboration_update_id) + WHERE collaboration_client_id IS NOT NULL AND collaboration_update_id IS NOT NULL; diff --git a/src/api/mod.rs b/src/api/mod.rs index 3956abe..996b5c8 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -30,6 +30,7 @@ use sha2::{Digest, Sha256}; use slug::slugify; use crate::{ + collab::{self, AppliedOperation}, db, queries, state::{NoteUpdate, RoomEvent, SharedState}, }; @@ -301,10 +302,17 @@ pub struct ListPaginationMeta { total_pages: usize, } -fn default_list_page() -> usize { 1 } -fn default_list_per_page() -> usize { 25 } +fn default_list_page() -> usize { + 1 +} +fn default_list_per_page() -> usize { + 25 +} fn normalize_list_per_page(value: usize) -> usize { - match value { 25 | 50 | 100 => value, _ => 25 } + match value { + 25 | 50 | 100 => value, + _ => 25, + } } #[derive(Debug, Serialize)] @@ -614,7 +622,12 @@ pub async fn open_workspace( search.is_empty() || note.title.to_lowercase().contains(&search) || note.slug.to_lowercase().contains(&search) - || note.created_by.as_deref().unwrap_or_default().to_lowercase().contains(&search) + || note + .created_by + .as_deref() + .unwrap_or_default() + .to_lowercase() + .contains(&search) }) .map(|note| { let stats = stats.get(¬e.id); @@ -658,7 +671,12 @@ pub async fn open_workspace( Ok(Json(WorkspaceOpenResponse { workspace: workspace_info_from(&workspace, access_level), notes, - pagination: ListPaginationMeta { page, per_page, total, total_pages }, + pagination: ListPaginationMeta { + page, + per_page, + total, + total_pages, + }, })) } @@ -1053,26 +1071,62 @@ pub async fn restore( .fetch_optional(state.db.pool()) .await?; let content = content.ok_or_else(ApiError::not_found_revision)?; + let room_key = crate::state::AppState::note_room_key(&workspace_slug, ¬e_slug); + let channel = state.note_channel(&workspace_slug, ¬e_slug).await; + let collaboration_snapshot = db::note_collaboration_snapshot(&state.db, note.id).await?; + let collaborative_document = state + .collaborative_document( + &room_key, + collaboration_snapshot.content, + collaboration_snapshot.owner_map, + collaboration_snapshot.revision_id, + ) + .await; + let mut document = collaborative_document.lock().await; + let base_revision_id = document.revision_id; + let operation = + collab::replace_operation(document.content.encode_utf16().count(), content, Vec::new()); + let (content, owner_map) = collab::apply_operation_to_document( + &document.content, + &document.owner_map, + &operation, + &[], + ) + .map_err(|_| ApiError::bad_request("The selected revision could not be restored"))?; let (revision_id, updated_at) = db::save_revision( &state.db, note.id, workspace.id, &content, Some("restore"), - "[]", + &owner_map, ) .await?; + let update_id = u64::try_from(revision_id).unwrap_or_default().max(1); + let applied = AppliedOperation { + base_revision_id, + revision_id, + client_id: "server_restore".into(), + update_id, + operation: operation.clone(), + owner_replacements: Vec::new(), + }; + document.content.clone_from(&content); + document.owner_map.clone_from(&owner_map); + document.revision_id = revision_id; + document.record(applied); let update = NoteUpdate { - content, + base_revision_id, revision_id, updated_at, author: Some("restore".into()), - owner_map: "[]".into(), + client_id: "server_restore".into(), + update_id, + operation, + owner_replacements: Vec::new(), }; - let _ = state - .note_channel(&workspace_slug, ¬e_slug) - .await - .send(RoomEvent::Document(update)); + let _ = channel.send(RoomEvent::Document(update)); + drop(document); Ok(Json(serde_json::json!({"ok": true}))) } @@ -1217,15 +1271,8 @@ async fn effective_header_access_level( is_private: i64, password_protected: bool, ) -> Result { - let mut level = request_access_level( - state, - headers, - kind, - slug, - None, - bearer_token(headers), - ) - .await?; + let mut level = + request_access_level(state, headers, kind, slug, None, bearer_token(headers)).await?; if is_private == 0 && !password_protected { level = std::cmp::max(level, AccessLevel::Write); } diff --git a/src/api/pads_public.rs b/src/api/pads_public.rs index 5b994e5..9218fb1 100644 --- a/src/api/pads_public.rs +++ b/src/api/pads_public.rs @@ -465,6 +465,144 @@ pub async fn public_page( })) } +async fn commit_public_pad_task_update( + state: &SharedState, + page: &db::PublishedPage, + source_line: usize, + checked: bool, +) -> Result<(), ApiError> { + let pad_id = page.pad_id.ok_or_else(ApiError::not_found_note)?; + let room_key = crate::state::AppState::pad_room_key(&page.resource_slug); + let channel = state.pad_channel(&page.resource_slug).await; + let collaboration_snapshot = db::pad_collaboration_snapshot(&state.db, pad_id).await?; + let collaborative_document = state + .collaborative_document( + &room_key, + collaboration_snapshot.content, + collaboration_snapshot.owner_map, + collaboration_snapshot.revision_id, + ) + .await; + let mut document = collaborative_document.lock().await; + let Some(next_content) = + db::updated_public_task_content(&document.content, source_line, checked) + else { + return Ok(()); + }; + let base_revision_id = document.revision_id; + let operation = + collab::operation_from_edit(&document.content, &next_content, &document.owner_map); + let (content, owner_map) = collab::apply_operation_to_document( + &document.content, + &document.owner_map, + &operation, + &[], + ) + .map_err(|_| ApiError::bad_request("The task could not be updated"))?; + let (revision_id, updated_at) = + db::save_pad_revision(&state.db, pad_id, &content, Some("public"), &owner_map).await?; + let update_id = u64::try_from(revision_id).unwrap_or_default().max(1); + document.content.clone_from(&content); + document.owner_map.clone_from(&owner_map); + document.revision_id = revision_id; + document.record(AppliedOperation { + base_revision_id, + revision_id, + client_id: "public_task".into(), + update_id, + operation: operation.clone(), + owner_replacements: Vec::new(), + }); + let _ = channel.send(RoomEvent::Document(NoteUpdate { + base_revision_id, + revision_id, + updated_at, + author: Some("public".into()), + client_id: "public_task".into(), + update_id, + operation, + owner_replacements: Vec::new(), + })); + drop(document); + Ok(()) +} + +async fn commit_public_note_task_update( + state: &SharedState, + page: &db::PublishedPage, + source_line: usize, + checked: bool, +) -> Result<(), ApiError> { + let note_id = page.note_id.ok_or_else(ApiError::not_found_note)?; + let workspace_id = page.workspace_id.ok_or_else(ApiError::not_found_note)?; + let workspace_slug = page + .workspace_slug + .as_deref() + .ok_or_else(ApiError::not_found_note)?; + let room_key = crate::state::AppState::note_room_key(workspace_slug, &page.resource_slug); + let channel = state + .note_channel(workspace_slug, &page.resource_slug) + .await; + let collaboration_snapshot = db::note_collaboration_snapshot(&state.db, note_id).await?; + let collaborative_document = state + .collaborative_document( + &room_key, + collaboration_snapshot.content, + collaboration_snapshot.owner_map, + collaboration_snapshot.revision_id, + ) + .await; + let mut document = collaborative_document.lock().await; + let Some(next_content) = + db::updated_public_task_content(&document.content, source_line, checked) + else { + return Ok(()); + }; + let base_revision_id = document.revision_id; + let operation = + collab::operation_from_edit(&document.content, &next_content, &document.owner_map); + let (content, owner_map) = collab::apply_operation_to_document( + &document.content, + &document.owner_map, + &operation, + &[], + ) + .map_err(|_| ApiError::bad_request("The task could not be updated"))?; + let (revision_id, updated_at) = db::save_revision( + &state.db, + note_id, + workspace_id, + &content, + Some("public"), + &owner_map, + ) + .await?; + let update_id = u64::try_from(revision_id).unwrap_or_default().max(1); + document.content.clone_from(&content); + document.owner_map.clone_from(&owner_map); + document.revision_id = revision_id; + document.record(AppliedOperation { + base_revision_id, + revision_id, + client_id: "public_task".into(), + update_id, + operation: operation.clone(), + owner_replacements: Vec::new(), + }); + let _ = channel.send(RoomEvent::Document(NoteUpdate { + base_revision_id, + revision_id, + updated_at, + author: Some("public".into()), + client_id: "public_task".into(), + update_id, + operation, + owner_replacements: Vec::new(), + })); + drop(document); + Ok(()) +} + pub async fn update_public_task( State(state): State, headers: HeaderMap, @@ -480,7 +618,14 @@ pub async fn update_public_task( "Task updates are disabled for this page", )); } - let page = db::update_public_task(&state.db, &token, payload.source_line, payload.checked) + if current.pad_id.is_some() { + commit_public_pad_task_update(&state, ¤t, payload.source_line, payload.checked) + .await?; + } else { + commit_public_note_task_update(&state, ¤t, payload.source_line, payload.checked) + .await?; + } + let page = db::find_published_page(&state.db, &token) .await? .ok_or_else(ApiError::not_found_note)?; let files = @@ -564,19 +709,59 @@ pub async fn pad_restore( .fetch_optional(state.db.pool()) .await?; let owner_map = owner_map.unwrap_or_else(|| "[]".into()); + let room_key = crate::state::AppState::pad_room_key(&slug); + let channel = state.pad_channel(&slug).await; + let collaboration_snapshot = db::pad_collaboration_snapshot(&state.db, pad.id).await?; + let collaborative_document = state + .collaborative_document( + &room_key, + collaboration_snapshot.content, + collaboration_snapshot.owner_map, + collaboration_snapshot.revision_id, + ) + .await; + let mut document = collaborative_document.lock().await; + let base_revision_id = document.revision_id; + let restored_owners = collab::owner_spans_from_map(&content, &owner_map); + let operation = collab::replace_operation( + document.content.encode_utf16().count(), + content, + restored_owners, + ); + let (content, owner_map) = collab::apply_operation_to_document( + &document.content, + &document.owner_map, + &operation, + &[], + ) + .map_err(|_| ApiError::bad_request("The selected revision could not be restored"))?; let (revision_id, updated_at) = db::save_pad_revision(&state.db, pad.id, &content, Some("restore"), &owner_map).await?; + let update_id = u64::try_from(revision_id).unwrap_or_default().max(1); + let applied = AppliedOperation { + base_revision_id, + revision_id, + client_id: "server_restore".into(), + update_id, + operation: operation.clone(), + owner_replacements: Vec::new(), + }; + document.content.clone_from(&content); + document.owner_map.clone_from(&owner_map); + document.revision_id = revision_id; + document.record(applied); let update = NoteUpdate { - content, + base_revision_id, revision_id, updated_at, author: Some("restore".into()), - owner_map, + client_id: "server_restore".into(), + update_id, + operation, + owner_replacements: Vec::new(), }; - let _ = state - .pad_channel(&slug) - .await - .send(RoomEvent::Document(update)); + let _ = channel.send(RoomEvent::Document(update)); + drop(document); Ok(Json(serde_json::json!({"ok": true}))) } @@ -591,7 +776,8 @@ pub(super) async fn authorized_pad( let pad = db::find_pad(&state.db, slug) .await? .ok_or_else(ApiError::not_found_note)?; - let token_level = request_access_level(state, headers, "pad", slug, access_token, bearer).await?; + let token_level = + request_access_level(state, headers, "pad", slug, access_token, bearer).await?; if pad.is_private != 0 && token_level == AccessLevel::None { return Err(ApiError::not_found_note()); } diff --git a/src/assets.rs b/src/assets.rs index 39de823..b9e295f 100644 --- a/src/assets.rs +++ b/src/assets.rs @@ -17,6 +17,8 @@ const MODULES: &[&str] = &[ "authorship", "auth-ui", "clipboard", + "collaboration", + "collaboration-session", "editor-format", "emoji-data", "emoji-picker", diff --git a/src/auth/mod.rs b/src/auth/mod.rs index 8cf9059..5806632 100644 --- a/src/auth/mod.rs +++ b/src/auth/mod.rs @@ -342,10 +342,17 @@ pub struct PaginationMeta { total_pages: usize, } -fn default_page() -> usize { 1 } -fn default_per_page() -> usize { 25 } +fn default_page() -> usize { + 1 +} +fn default_per_page() -> usize { + 25 +} fn normalized_per_page(value: usize) -> usize { - match value { 25 | 50 | 100 => value, _ => 25 } + match value { + 25 | 50 | 100 => value, + _ => 25, + } } #[derive(Serialize)] pub struct SessionResponse { @@ -1237,14 +1244,13 @@ pub async fn resources( .fetch_all(state.db.pool()) .await .map_err(AuthError::database)?; - let pads = sqlx::query_as::<_, ResourceItem>( - queries::get(state.db.kind(), queries::USER_LIST_PADS), - ) - .bind(user.id) - .bind(user.id) - .fetch_all(state.db.pool()) - .await - .map_err(AuthError::database)?; + let pads = + sqlx::query_as::<_, ResourceItem>(queries::get(state.db.kind(), queries::USER_LIST_PADS)) + .bind(user.id) + .bind(user.id) + .fetch_all(state.db.pool()) + .await + .map_err(AuthError::database)?; let search = query.q.trim().to_lowercase(); let mut items = workspaces @@ -1278,7 +1284,12 @@ pub async fn resources( Ok(Json(ResourceList { items, - pagination: PaginationMeta { page, per_page, total, total_pages }, + pagination: PaginationMeta { + page, + per_page, + total, + total_pages, + }, })) } @@ -1817,10 +1828,9 @@ pub async fn create_share_link( header::CACHE_CONTROL, "no-store, max-age=0".parse().expect("valid cache-control"), ); - response.headers_mut().insert( - header::PRAGMA, - "no-cache".parse().expect("valid pragma"), - ); + response + .headers_mut() + .insert(header::PRAGMA, "no-cache".parse().expect("valid pragma")); Ok(response) } @@ -2059,18 +2069,15 @@ pub async fn create_share_session( .map_err(AuthError::database)?; let token = random_token(); - sqlx::query(queries::get( - state.db.kind(), - queries::SHARE_SESSION_INSERT, - )) - .bind(hash_token(&token)) - .bind(source.token_hash) - .bind(kind) - .bind(slug) - .bind(expires_at.to_rfc3339()) - .execute(state.db.pool()) - .await - .map_err(AuthError::database)?; + sqlx::query(queries::get(state.db.kind(), queries::SHARE_SESSION_INSERT)) + .bind(hash_token(&token)) + .bind(source.token_hash) + .bind(kind) + .bind(slug) + .bind(expires_at.to_rfc3339()) + .execute(state.db.pool()) + .await + .map_err(AuthError::database)?; Ok(Some(ShareSession { token, diff --git a/src/collab.rs b/src/collab.rs new file mode 100644 index 0000000..8149d1c --- /dev/null +++ b/src/collab.rs @@ -0,0 +1,909 @@ +/* + * Copyright (C) 2026 Mateusz Gruszczynski @linuxiarz.pl + * Source-Available Code / Dual-Licensed. + * + * Free for non-commercial and evaluation use under terms of BSL/GPLv3. + * Commercial or production use requires a valid paid license. + * See LICENSE file in repository root for details. + */ + +use serde::{Deserialize, Serialize}; +use std::{ + collections::{HashMap, VecDeque}, + error::Error, + fmt, +}; + +pub const MAX_OPERATION_COMPONENTS: usize = 4096; +const MAX_OPERATION_OWNER_SPANS: usize = 8192; +const MAX_OPERATION_INSERT_BYTES: usize = 2_000_000; +const MAX_OWNER_LENGTH: usize = 120; +const MAX_OPERATION_HISTORY: usize = 512; +const MAX_OPERATION_HISTORY_BYTES: usize = 8 * 1024 * 1024; +const AUTHORSHIP_VERSION: u8 = 2; +const OWNER_COLOR_SEPARATOR: char = '\u{001f}'; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct OwnerSpan { + pub start: usize, + pub end: usize, + pub owner: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum OperationComponent { + Retain { + count: usize, + }, + Delete { + count: usize, + }, + Insert { + text: String, + #[serde(default)] + owners: Vec, + }, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct TextOperation { + #[serde(default)] + pub components: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct OwnerReplacement { + pub owner: String, + pub replacement: String, +} + +#[derive(Debug, Clone, Serialize)] +pub struct AppliedOperation { + pub base_revision_id: i64, + pub revision_id: i64, + pub client_id: String, + pub update_id: u64, + pub operation: TextOperation, + pub owner_replacements: Vec, +} + +#[derive(Debug)] +pub struct CollaborativeDocument { + pub content: String, + pub owner_map: String, + pub revision_id: i64, + history: VecDeque, + history_bytes: usize, + acknowledged_updates: HashMap, +} + +impl CollaborativeDocument { + pub fn new(content: String, owner_map: String, revision_id: i64) -> Self { + Self { + content, + owner_map, + revision_id, + history: VecDeque::new(), + history_bytes: 0, + acknowledged_updates: HashMap::new(), + } + } + + pub fn transform_from( + &self, + base_revision_id: i64, + operation: &TextOperation, + client_id: &str, + update_id: u64, + ) -> Result { + let mut transformed = normalize_operation(operation)?; + if base_revision_id == self.revision_id { + return Ok(transformed); + } + + let Some(start) = self + .history + .iter() + .position(|entry| entry.base_revision_id == base_revision_id) + else { + return Err(OperationError::RevisionUnavailable); + }; + + let mut expected_revision = base_revision_id; + for applied in self.history.iter().skip(start) { + if applied.base_revision_id != expected_revision { + return Err(OperationError::RevisionUnavailable); + } + let incoming_has_priority = + operation_key_before(client_id, update_id, &applied.client_id, applied.update_id); + transformed = + transform_operation(&transformed, &applied.operation, incoming_has_priority)?; + expected_revision = applied.revision_id; + if expected_revision == self.revision_id { + return Ok(transformed); + } + } + Err(OperationError::RevisionUnavailable) + } + + pub fn acknowledge(&mut self, client_id: &str, update_id: u64) { + self.acknowledged_updates + .entry(client_id.to_owned()) + .and_modify(|acknowledged| *acknowledged = (*acknowledged).max(update_id)) + .or_insert(update_id); + } + + pub fn has_applied_update(&self, client_id: &str, update_id: u64) -> bool { + self.acknowledged_updates + .get(client_id) + .is_some_and(|acknowledged| update_id <= *acknowledged) + } + + pub fn acknowledged_updates(&self, client_id: &str) -> Vec { + self.acknowledged_updates + .get(client_id) + .copied() + .into_iter() + .collect() + } + + pub fn operations_after(&self, revision_id: i64) -> Option> { + if revision_id == self.revision_id { + return Some(Vec::new()); + } + let start = self + .history + .iter() + .position(|entry| entry.base_revision_id == revision_id)?; + let mut expected_revision = revision_id; + let mut operations = Vec::new(); + for applied in self.history.iter().skip(start) { + if applied.base_revision_id != expected_revision { + return None; + } + operations.push(applied.clone()); + expected_revision = applied.revision_id; + if expected_revision == self.revision_id { + return Some(operations); + } + } + None + } + + pub fn record(&mut self, operation: AppliedOperation) { + self.acknowledge(&operation.client_id, operation.update_id); + self.history_bytes = self + .history_bytes + .saturating_add(applied_operation_size(&operation)); + self.history.push_back(operation); + while self.history.len() > MAX_OPERATION_HISTORY + || self.history_bytes > MAX_OPERATION_HISTORY_BYTES + { + let Some(removed) = self.history.pop_front() else { + break; + }; + self.history_bytes = self + .history_bytes + .saturating_sub(applied_operation_size(&removed)); + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ComponentKind { + Retain, + Delete, + Insert, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum OperationError { + InvalidComponent, + InvalidUtf16Boundary, + LengthMismatch, + TooManyComponents, + RevisionUnavailable, + Serialization, +} + +impl fmt::Display for OperationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + let message = match self { + Self::InvalidComponent => "invalid text operation component", + Self::InvalidUtf16Boundary => "text operation splits a UTF-16 character", + Self::LengthMismatch => "text operation length does not match the document", + Self::TooManyComponents => "text operation contains too many components", + Self::RevisionUnavailable => "the base revision is no longer available", + Self::Serialization => "invalid authorship metadata", + }; + formatter.write_str(message) + } +} + +impl Error for OperationError {} + +fn applied_operation_size(operation: &AppliedOperation) -> usize { + let components = operation + .operation + .components + .iter() + .map(|component| match component { + OperationComponent::Retain { .. } | OperationComponent::Delete { .. } => 24, + OperationComponent::Insert { text, owners } => { + 32usize.saturating_add(text.len()).saturating_add( + owners + .iter() + .map(|span| 24usize.saturating_add(span.owner.len())) + .sum::(), + ) + } + }) + .sum::(); + components + .saturating_add(operation.client_id.len()) + .saturating_add( + operation + .owner_replacements + .iter() + .map(|replacement| replacement.owner.len() + replacement.replacement.len() + 16) + .sum::(), + ) +} + +fn operation_key_before( + left_client_id: &str, + left_update_id: u64, + right_client_id: &str, + right_update_id: u64, +) -> bool { + left_client_id < right_client_id + || (left_client_id == right_client_id && left_update_id < right_update_id) +} + +fn component_kind(component: &OperationComponent) -> ComponentKind { + match component { + OperationComponent::Retain { .. } => ComponentKind::Retain, + OperationComponent::Delete { .. } => ComponentKind::Delete, + OperationComponent::Insert { .. } => ComponentKind::Insert, + } +} + +fn component_length(component: &OperationComponent) -> usize { + match component { + OperationComponent::Retain { count } | OperationComponent::Delete { count } => *count, + OperationComponent::Insert { text, .. } => text.encode_utf16().count(), + } +} + +fn normalize_owner_spans(spans: &[OwnerSpan], length: usize) -> Vec { + let mut sorted = spans + .iter() + .filter_map(|span| { + let start = span.start.min(length); + let end = span.end.min(length).max(start); + if span.owner.is_empty() || end <= start { + None + } else { + Some(OwnerSpan { + start, + end, + owner: span.owner.clone(), + }) + } + }) + .collect::>(); + sorted.sort_by_key(|span| (span.start, span.end)); + let mut result: Vec = Vec::new(); + for mut span in sorted { + if let Some(previous) = result.last_mut() { + if previous.owner == span.owner && span.start <= previous.end { + previous.end = previous.end.max(span.end); + continue; + } + if span.start < previous.end { + span.start = previous.end; + } + } + if span.end > span.start { + result.push(span); + } + } + result +} + +fn slice_owner_spans(spans: &[OwnerSpan], start: usize, length: usize) -> Vec { + let end = start.saturating_add(length); + let sliced = spans + .iter() + .filter_map(|span| { + let overlap_start = start.max(span.start); + let overlap_end = end.min(span.end); + (overlap_end > overlap_start).then(|| OwnerSpan { + start: overlap_start - start, + end: overlap_end - start, + owner: span.owner.clone(), + }) + }) + .collect::>(); + normalize_owner_spans(&sliced, length) +} + +fn shift_owner_spans(spans: &[OwnerSpan], offset: usize) -> Vec { + spans + .iter() + .map(|span| OwnerSpan { + start: span.start + offset, + end: span.end + offset, + owner: span.owner.clone(), + }) + .collect() +} + +fn append_component( + components: &mut Vec, + component: OperationComponent, +) -> Result<(), OperationError> { + match component { + OperationComponent::Retain { count } => { + if count == 0 { + return Ok(()); + } + if let Some(OperationComponent::Retain { count: previous }) = components.last_mut() { + *previous = previous + .checked_add(count) + .ok_or(OperationError::InvalidComponent)?; + } else { + components.push(OperationComponent::Retain { count }); + } + } + OperationComponent::Delete { count } => { + if count == 0 { + return Ok(()); + } + if let Some(OperationComponent::Delete { count: previous }) = components.last_mut() { + *previous = previous + .checked_add(count) + .ok_or(OperationError::InvalidComponent)?; + } else { + components.push(OperationComponent::Delete { count }); + } + } + OperationComponent::Insert { text, owners } => { + let length = text.encode_utf16().count(); + if length == 0 { + return Ok(()); + } + let owners = normalize_owner_spans(&owners, length); + if let Some(OperationComponent::Insert { + text: previous_text, + owners: previous_owners, + }) = components.last_mut() + { + let offset = previous_text.encode_utf16().count(); + previous_text.push_str(&text); + previous_owners.extend(shift_owner_spans(&owners, offset)); + *previous_owners = + normalize_owner_spans(previous_owners, previous_text.encode_utf16().count()); + } else { + components.push(OperationComponent::Insert { text, owners }); + } + } + } + if components.len() > MAX_OPERATION_COMPONENTS { + return Err(OperationError::TooManyComponents); + } + Ok(()) +} + +pub fn normalize_operation(operation: &TextOperation) -> Result { + if operation.components.len() > MAX_OPERATION_COMPONENTS { + return Err(OperationError::TooManyComponents); + } + let mut inserted_bytes = 0usize; + let mut owner_span_count = 0usize; + let mut components = Vec::with_capacity(operation.components.len()); + for component in &operation.components { + if let OperationComponent::Insert { text, owners } = component { + inserted_bytes = inserted_bytes + .checked_add(text.len()) + .ok_or(OperationError::InvalidComponent)?; + owner_span_count = owner_span_count + .checked_add(owners.len()) + .ok_or(OperationError::InvalidComponent)?; + let text_length = text.encode_utf16().count(); + if inserted_bytes > MAX_OPERATION_INSERT_BYTES + || owner_span_count > MAX_OPERATION_OWNER_SPANS + || owners.iter().any(|span| { + span.start > span.end + || span.end > text_length + || utf16_byte_index(text, span.start).is_err() + || utf16_byte_index(text, span.end).is_err() + || span.owner.chars().count() > MAX_OWNER_LENGTH + || span.owner.chars().any(|character| { + character.is_control() && character != OWNER_COLOR_SEPARATOR + }) + }) + { + return Err(OperationError::InvalidComponent); + } + } + append_component(&mut components, component.clone())?; + } + Ok(TextOperation { components }) +} + +pub fn operation_base_length(operation: &TextOperation) -> Result { + normalize_operation(operation)? + .components + .iter() + .try_fold(0usize, |length, component| { + let component_length = match component { + OperationComponent::Retain { count } | OperationComponent::Delete { count } => { + *count + } + OperationComponent::Insert { .. } => 0, + }; + length + .checked_add(component_length) + .ok_or(OperationError::InvalidComponent) + }) +} + +pub fn operation_from_edit( + previous_content: &str, + next_content: &str, + next_owner_map: &str, +) -> TextOperation { + let mut previous_prefix_bytes = 0usize; + let mut next_prefix_bytes = 0usize; + for (previous, next) in previous_content.chars().zip(next_content.chars()) { + if previous != next { + break; + } + previous_prefix_bytes += previous.len_utf8(); + next_prefix_bytes += next.len_utf8(); + } + + let previous_remainder = &previous_content[previous_prefix_bytes..]; + let next_remainder = &next_content[next_prefix_bytes..]; + let mut previous_suffix_bytes = 0usize; + let mut next_suffix_bytes = 0usize; + for (previous, next) in previous_remainder + .chars() + .rev() + .zip(next_remainder.chars().rev()) + { + if previous != next { + break; + } + previous_suffix_bytes += previous.len_utf8(); + next_suffix_bytes += next.len_utf8(); + } + + let previous_middle_end = previous_content.len() - previous_suffix_bytes; + let next_middle_end = next_content.len() - next_suffix_bytes; + let previous_prefix = &previous_content[..previous_prefix_bytes]; + let previous_middle = &previous_content[previous_prefix_bytes..previous_middle_end]; + let next_middle = &next_content[next_prefix_bytes..next_middle_end]; + let suffix = &previous_content[previous_middle_end..]; + + let prefix_length = previous_prefix.encode_utf16().count(); + let deleted_length = previous_middle.encode_utf16().count(); + let inserted_length = next_middle.encode_utf16().count(); + let suffix_length = suffix.encode_utf16().count(); + let next_authorship = parse_authorship(next_content, next_owner_map); + let inserted_owners = slice_owner_spans(&next_authorship.spans, prefix_length, inserted_length); + + let mut components = Vec::new(); + if prefix_length > 0 { + components.push(OperationComponent::Retain { + count: prefix_length, + }); + } + if deleted_length > 0 { + components.push(OperationComponent::Delete { + count: deleted_length, + }); + } + if !next_middle.is_empty() { + components.push(OperationComponent::Insert { + text: next_middle.to_owned(), + owners: inserted_owners, + }); + } + if suffix_length > 0 { + components.push(OperationComponent::Retain { + count: suffix_length, + }); + } + TextOperation { components } +} + +pub fn replace_operation( + base_length: usize, + content: String, + owners: Vec, +) -> TextOperation { + let mut components = Vec::new(); + if base_length > 0 { + components.push(OperationComponent::Delete { count: base_length }); + } + if !content.is_empty() { + components.push(OperationComponent::Insert { + text: content, + owners, + }); + } + TextOperation { components } +} + +struct OperationCursor { + components: Vec, + index: usize, + offset: usize, +} + +impl OperationCursor { + fn new(operation: &TextOperation) -> Result { + Ok(Self { + components: normalize_operation(operation)?.components, + index: 0, + offset: 0, + }) + } + + fn current(&self) -> Option<&OperationComponent> { + self.components.get(self.index) + } + + fn kind(&self) -> Option { + self.current().map(component_kind) + } + + fn remaining(&self) -> usize { + self.current() + .map(|component| component_length(component).saturating_sub(self.offset)) + .unwrap_or(0) + } + + fn take(&mut self, count: usize) -> Result { + let component = self + .current() + .cloned() + .ok_or(OperationError::InvalidComponent)?; + if count == 0 || count > self.remaining() { + return Err(OperationError::InvalidComponent); + } + let component_length = component_length(&component); + let part = match component { + OperationComponent::Retain { .. } => OperationComponent::Retain { count }, + OperationComponent::Delete { .. } => OperationComponent::Delete { count }, + OperationComponent::Insert { text, owners } => OperationComponent::Insert { + text: slice_utf16(&text, self.offset, count)?.to_owned(), + owners: slice_owner_spans(&owners, self.offset, count), + }, + }; + self.offset += count; + if self.offset == component_length { + self.index += 1; + self.offset = 0; + } + Ok(part) + } + + fn take_remaining(&mut self) -> Result { + let count = self.remaining(); + self.take(count) + } +} + +pub fn transform_operation( + left_operation: &TextOperation, + right_operation: &TextOperation, + left_before_right: bool, +) -> Result { + if operation_base_length(left_operation)? != operation_base_length(right_operation)? { + return Err(OperationError::LengthMismatch); + } + let mut left = OperationCursor::new(left_operation)?; + let mut right = OperationCursor::new(right_operation)?; + let mut left_prime = Vec::new(); + + while left.current().is_some() || right.current().is_some() { + if left.kind() == Some(ComponentKind::Insert) + && (right.kind() != Some(ComponentKind::Insert) || left_before_right) + { + append_component(&mut left_prime, left.take_remaining()?)?; + continue; + } + if right.kind() == Some(ComponentKind::Insert) { + let count = right.remaining(); + right.take_remaining()?; + append_component(&mut left_prime, OperationComponent::Retain { count })?; + continue; + } + let (Some(left_kind), Some(right_kind)) = (left.kind(), right.kind()) else { + return Err(OperationError::InvalidComponent); + }; + let count = left.remaining().min(right.remaining()); + match (left_kind, right_kind) { + (ComponentKind::Retain, ComponentKind::Retain) => { + append_component(&mut left_prime, OperationComponent::Retain { count })?; + } + (ComponentKind::Delete, ComponentKind::Retain) => { + append_component(&mut left_prime, OperationComponent::Delete { count })?; + } + (ComponentKind::Retain, ComponentKind::Delete) + | (ComponentKind::Delete, ComponentKind::Delete) => {} + _ => return Err(OperationError::InvalidComponent), + } + left.take(count)?; + right.take(count)?; + } + Ok(TextOperation { + components: left_prime, + }) +} + +fn utf16_byte_index(value: &str, offset: usize) -> Result { + if offset == 0 { + return Ok(0); + } + let mut current = 0usize; + for (byte_index, character) in value.char_indices() { + if current == offset { + return Ok(byte_index); + } + current += character.len_utf16(); + if current > offset { + return Err(OperationError::InvalidUtf16Boundary); + } + } + if current == offset { + Ok(value.len()) + } else { + Err(OperationError::LengthMismatch) + } +} + +fn slice_utf16(value: &str, start: usize, length: usize) -> Result<&str, OperationError> { + let start_byte = utf16_byte_index(value, start)?; + let end_byte = utf16_byte_index(value, start.saturating_add(length))?; + value + .get(start_byte..end_byte) + .ok_or(OperationError::InvalidUtf16Boundary) +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct AuthorshipModel { + #[serde(default = "authorship_version")] + version: u8, + #[serde(default)] + spans: Vec, +} + +fn authorship_version() -> u8 { + AUTHORSHIP_VERSION +} + +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum RawAuthorship { + Model(AuthorshipModel), + LineOwners(Vec), +} + +fn parse_authorship(content: &str, raw: &str) -> AuthorshipModel { + let length = content.encode_utf16().count(); + match serde_json::from_str::(raw) { + Ok(RawAuthorship::Model(model)) if model.version == AUTHORSHIP_VERSION => AuthorshipModel { + version: AUTHORSHIP_VERSION, + spans: normalize_owner_spans(&model.spans, length), + }, + Ok(RawAuthorship::LineOwners(owners)) => { + let lines = content.split('\n').collect::>(); + let mut offset = 0usize; + let mut spans = Vec::new(); + for (index, line) in lines.iter().enumerate() { + let line_length = + line.encode_utf16().count() + usize::from(index + 1 < lines.len()); + let owner = owners.get(index).cloned().unwrap_or_default(); + if !owner.is_empty() && line_length > 0 { + spans.push(OwnerSpan { + start: offset, + end: offset + line_length, + owner, + }); + } + offset += line_length; + } + AuthorshipModel { + version: AUTHORSHIP_VERSION, + spans: normalize_owner_spans(&spans, length), + } + } + _ => AuthorshipModel { + version: AUTHORSHIP_VERSION, + spans: Vec::new(), + }, + } +} + +fn copy_retained_spans( + target: &mut Vec, + spans: &[OwnerSpan], + source_start: usize, + length: usize, + output_start: usize, +) { + let source_end = source_start + length; + for span in spans { + let start = source_start.max(span.start); + let end = source_end.min(span.end); + if end > start { + target.push(OwnerSpan { + start: output_start + start - source_start, + end: output_start + end - source_start, + owner: span.owner.clone(), + }); + } + } +} + +pub fn apply_operation_to_document( + content: &str, + owner_map: &str, + operation: &TextOperation, + owner_replacements: &[OwnerReplacement], +) -> Result<(String, String), OperationError> { + let operation = normalize_operation(operation)?; + let content_length = content.encode_utf16().count(); + if operation_base_length(&operation)? != content_length { + return Err(OperationError::LengthMismatch); + } + + let source_model = parse_authorship(content, owner_map); + let mut output_spans = Vec::new(); + let mut source_offset = 0usize; + let mut output_offset = 0usize; + let mut output_content = String::new(); + + for component in &operation.components { + match component { + OperationComponent::Retain { count } => { + output_content.push_str(slice_utf16(content, source_offset, *count)?); + copy_retained_spans( + &mut output_spans, + &source_model.spans, + source_offset, + *count, + output_offset, + ); + source_offset += *count; + output_offset += *count; + } + OperationComponent::Delete { count } => { + source_offset += *count; + } + OperationComponent::Insert { text, owners } => { + output_content.push_str(text); + output_spans.extend(shift_owner_spans(owners, output_offset)); + output_offset += text.encode_utf16().count(); + } + } + } + if source_offset != content_length { + return Err(OperationError::LengthMismatch); + } + + for span in &mut output_spans { + let identity = span + .owner + .split(OWNER_COLOR_SEPARATOR) + .next() + .unwrap_or_default(); + if let Some(replacement) = owner_replacements.iter().find(|replacement| { + replacement.owner == identity && !replacement.replacement.is_empty() + }) { + span.owner.clone_from(&replacement.replacement); + } + } + let model = AuthorshipModel { + version: AUTHORSHIP_VERSION, + spans: normalize_owner_spans(&output_spans, output_offset), + }; + let owner_map = serde_json::to_string(&model).map_err(|_| OperationError::Serialization)?; + Ok((output_content, owner_map)) +} + +pub fn owner_spans_from_map(content: &str, owner_map: &str) -> Vec { + parse_authorship(content, owner_map).spans +} + +#[cfg(test)] +mod tests { + use super::*; + + fn operation(components: Vec) -> TextOperation { + TextOperation { components } + } + + #[test] + fn concurrent_insertions_have_stable_order() { + let left = operation(vec![ + OperationComponent::Retain { count: 1 }, + OperationComponent::Insert { + text: "X".into(), + owners: Vec::new(), + }, + OperationComponent::Retain { count: 1 }, + ]); + let right = operation(vec![ + OperationComponent::Retain { count: 1 }, + OperationComponent::Insert { + text: "Y".into(), + owners: Vec::new(), + }, + OperationComponent::Retain { count: 1 }, + ]); + let left_prime = transform_operation(&left, &right, true).unwrap(); + let right_prime = transform_operation(&right, &left, false).unwrap(); + let after_right = apply_operation_to_document("aYb", "[]", &left_prime, &[]) + .unwrap() + .0; + let after_left = apply_operation_to_document("aXb", "[]", &right_prime, &[]) + .unwrap() + .0; + assert_eq!(after_right, "aXYb"); + assert_eq!(after_left, "aXYb"); + } + + #[test] + fn utf16_offsets_support_emoji() { + let operation = operation(vec![ + OperationComponent::Retain { count: 3 }, + OperationComponent::Insert { + text: "x".into(), + owners: Vec::new(), + }, + OperationComponent::Retain { count: 1 }, + ]); + let result = apply_operation_to_document("A😀B", "[]", &operation, &[]) + .unwrap() + .0; + assert_eq!(result, "A😀xB"); + } + + #[test] + fn operation_from_edit_preserves_utf16_boundaries() { + let operation = operation_from_edit("A😀B", "A😀xB", "[]"); + let result = apply_operation_to_document("A😀B", "[]", &operation, &[]) + .unwrap() + .0; + assert_eq!(result, "A😀xB"); + } + + #[test] + fn acknowledgements_survive_history_compaction() { + let mut document = CollaborativeDocument::new(String::new(), "[]".into(), 0); + for update_id in 1..=MAX_OPERATION_HISTORY as u64 + 8 { + let base_revision_id = document.revision_id; + let revision_id = base_revision_id + 1; + document.revision_id = revision_id; + document.record(AppliedOperation { + base_revision_id, + revision_id, + client_id: "client-123".into(), + update_id, + operation: TextOperation::default(), + owner_replacements: Vec::new(), + }); + } + assert!(document.has_applied_update("client-123", 1)); + assert_eq!( + document.acknowledged_updates("client-123"), + vec![MAX_OPERATION_HISTORY as u64 + 8] + ); + } +} diff --git a/src/db/editor_preferences.rs b/src/db/editor_preferences.rs index b8b270c..b5e6a20 100644 --- a/src/db/editor_preferences.rs +++ b/src/db/editor_preferences.rs @@ -78,14 +78,11 @@ pub async fn load_editor_preferences( user_id: i64, resource: EditorPreferenceResource, ) -> Result, sqlx::Error> { - let Some(row) = sqlx::query(queries::get( - pool.kind(), - preference_select_query(resource), - )) - .bind(user_id) - .bind(resource_id(resource)) - .fetch_optional(pool.pool()) - .await? + let Some(row) = sqlx::query(queries::get(pool.kind(), preference_select_query(resource))) + .bind(user_id) + .bind(resource_id(resource)) + .fetch_optional(pool.pool()) + .await? else { return Ok(None); }; @@ -113,20 +110,17 @@ pub async fn save_editor_configuration( let user_id = user_id.ok_or_else(|| { sqlx::Error::Protocol("user id is required for personal editor preferences".into()) })?; - sqlx::query(queries::get( - pool.kind(), - preference_upsert_query(resource), - )) - .bind(user_id) - .bind(resource_id(resource)) - .bind(preferences.compact_view) - .bind(preferences.editor_line_numbers) - .bind(preferences.preview_line_numbers) - .bind(preferences.line_links) - .bind(&preferences.font_family) - .bind(preferences.font_size) - .execute(&mut *tx) - .await?; + sqlx::query(queries::get(pool.kind(), preference_upsert_query(resource))) + .bind(user_id) + .bind(resource_id(resource)) + .bind(preferences.compact_view) + .bind(preferences.editor_line_numbers) + .bind(preferences.preview_line_numbers) + .bind(preferences.line_links) + .bind(&preferences.font_family) + .bind(preferences.font_size) + .execute(&mut *tx) + .await?; } if let Some((resource_kind, resource_slug, settings)) = resource_settings { diff --git a/src/db/mod.rs b/src/db/mod.rs index 89703e6..90d1499 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -72,6 +72,13 @@ pub struct Note { pub created_by_guest_id: Option, } +#[derive(Debug, Clone)] +pub struct CollaborationSnapshot { + pub content: String, + pub owner_map: String, + pub revision_id: i64, +} + #[derive(Debug, Clone, FromRow)] struct SqliteNote { id: i64, @@ -316,6 +323,68 @@ pub async fn save_revision( Ok((revision_id, updated_at)) } +pub async fn save_collaborative_revision( + pool: &Database, + note_id: i64, + workspace_id: i64, + content: &str, + author: Option<&str>, + owner_map: &str, + collaboration_client_id: &str, + collaboration_update_id: i64, +) -> Result<(i64, String), sqlx::Error> { + let mut tx = pool.pool().begin().await?; + sqlx::query(queries::get(pool.kind(), queries::Q006)) + .bind(content) + .bind(owner_map) + .bind(note_id) + .execute(&mut *tx) + .await?; + sqlx::query(queries::get(pool.kind(), queries::Q007)) + .bind(workspace_id) + .execute(&mut *tx) + .await?; + sqlx::query(queries::get(pool.kind(), queries::Q054)) + .bind(note_id) + .bind(content) + .bind(author) + .bind(owner_map) + .bind(collaboration_client_id) + .bind(collaboration_update_id) + .execute(&mut *tx) + .await?; + let revision_id = inserted_id(pool.kind(), &mut tx, "note_revisions").await?; + let updated_at: String = sqlx::query_scalar(queries::get(pool.kind(), queries::Q009)) + .bind(note_id) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok((revision_id, updated_at)) +} + +pub async fn latest_note_collaboration_update_id( + pool: &Database, + note_id: i64, + collaboration_client_id: &str, +) -> Result, sqlx::Error> { + let update_id = sqlx::query_scalar::<_, Option>(queries::get(pool.kind(), queries::Q056)) + .bind(note_id) + .bind(collaboration_client_id) + .fetch_one(pool.pool()) + .await?; + Ok(update_id.and_then(|value| u64::try_from(value).ok())) +} + +pub async fn note_collaboration_snapshot( + pool: &Database, + note_id: i64, +) -> Result { + sqlx::query_as::<_, CollaborationSnapshot>(queries::get(pool.kind(), queries::Q058)) + .bind(note_id) + .fetch_one(pool.pool()) + .await +} + pub async fn list_revisions(pool: &Database, note_id: i64) -> Result, sqlx::Error> { sqlx::query_as::<_, Revision>(queries::get(pool.kind(), queries::Q010)) .bind(note_id) @@ -392,7 +461,6 @@ pub struct Pad { pub password_hash: Option, pub created_at: String, pub updated_at: String, - pub owner_map: String, pub is_private: i64, pub created_by_guest_id: Option, } @@ -476,6 +544,63 @@ pub async fn save_pad_revision( Ok((revision_id, updated_at)) } +pub async fn save_collaborative_pad_revision( + pool: &Database, + pad_id: i64, + content: &str, + author: Option<&str>, + owner_map: &str, + collaboration_client_id: &str, + collaboration_update_id: i64, +) -> Result<(i64, String), sqlx::Error> { + let mut tx = pool.pool().begin().await?; + sqlx::query(queries::get(pool.kind(), queries::Q013)) + .bind(content) + .bind(owner_map) + .bind(pad_id) + .execute(&mut *tx) + .await?; + sqlx::query(queries::get(pool.kind(), queries::Q055)) + .bind(pad_id) + .bind(content) + .bind(author) + .bind(owner_map) + .bind(collaboration_client_id) + .bind(collaboration_update_id) + .execute(&mut *tx) + .await?; + let revision_id = inserted_id(pool.kind(), &mut tx, "revisions").await?; + let updated_at: String = sqlx::query_scalar(queries::get(pool.kind(), queries::Q015)) + .bind(pad_id) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok((revision_id, updated_at)) +} + +pub async fn latest_pad_collaboration_update_id( + pool: &Database, + pad_id: i64, + collaboration_client_id: &str, +) -> Result, sqlx::Error> { + let update_id = sqlx::query_scalar::<_, Option>(queries::get(pool.kind(), queries::Q057)) + .bind(pad_id) + .bind(collaboration_client_id) + .fetch_one(pool.pool()) + .await?; + Ok(update_id.and_then(|value| u64::try_from(value).ok())) +} + +pub async fn pad_collaboration_snapshot( + pool: &Database, + pad_id: i64, +) -> Result { + sqlx::query_as::<_, CollaborationSnapshot>(queries::get(pool.kind(), queries::Q059)) + .bind(pad_id) + .fetch_one(pool.pool()) + .await +} + pub async fn list_pad_revisions( pool: &Database, pad_id: i64, @@ -486,6 +611,16 @@ pub async fn list_pad_revisions( .await } +impl<'r> sqlx::FromRow<'r, AnyRow> for CollaborationSnapshot { + fn from_row(row: &'r AnyRow) -> Result { + Ok(Self { + content: crate::row_decode::text(row, "content")?, + owner_map: crate::row_decode::text(row, "owner_map")?, + revision_id: row.try_get("revision_id")?, + }) + } +} + impl<'r> sqlx::FromRow<'r, AnyRow> for Workspace { fn from_row(row: &'r AnyRow) -> Result { Ok(Self { @@ -538,7 +673,6 @@ impl<'r> sqlx::FromRow<'r, AnyRow> for Pad { password_hash: crate::row_decode::optional_text(row, "password_hash")?, created_at: crate::row_decode::text(row, "created_at")?, updated_at: crate::row_decode::text(row, "updated_at")?, - owner_map: crate::row_decode::text(row, "owner_map")?, is_private: row.try_get("is_private")?, created_by_guest_id: crate::row_decode::optional_text(row, "created_by_guest_id")?, }) @@ -570,7 +704,6 @@ mod password_verification_tests { password_hash, created_at: String::new(), updated_at: String::new(), - owner_map: "[]".into(), is_private: 1, created_by_guest_id: None, } @@ -579,7 +712,10 @@ mod password_verification_tests { #[test] fn missing_password_does_not_grant_password_access() { assert!(!verify_workspace_password(&workspace(None), None)); - assert!(!verify_workspace_password(&workspace(None), Some("anything"))); + assert!(!verify_workspace_password( + &workspace(None), + Some("anything") + )); assert!(!verify_pad_password(&pad(None), None)); assert!(!verify_pad_password(&pad(None), Some("anything"))); } diff --git a/src/db/public_pages.rs b/src/db/public_pages.rs index 38161de..006bd68 100644 --- a/src/db/public_pages.rs +++ b/src/db/public_pages.rs @@ -15,6 +15,10 @@ pub struct PublishedPage { pub pad_id: Option, pub note_id: Option, pub allow_task_updates: bool, + pub resource_slug: String, + pub workspace_id: Option, + pub workspace_slug: Option, + pub owner_map: String, pub title: String, pub content: String, pub updated_at: String, @@ -26,6 +30,10 @@ struct PublishedPageRow { pad_id: Option, note_id: Option, allow_task_updates: i64, + resource_slug: String, + workspace_id: Option, + workspace_slug: Option, + owner_map: String, title: String, content: String, updated_at: String, @@ -37,6 +45,10 @@ struct PostgresPublishedPageRow { pad_id: Option, note_id: Option, allow_task_updates: bool, + resource_slug: String, + workspace_id: Option, + workspace_slug: Option, + owner_map: String, title: String, content: String, updated_at: String, @@ -49,6 +61,10 @@ impl From for PublishedPage { pad_id: value.pad_id, note_id: value.note_id, allow_task_updates: value.allow_task_updates, + resource_slug: value.resource_slug, + workspace_id: value.workspace_id, + workspace_slug: value.workspace_slug, + owner_map: value.owner_map, title: value.title, content: value.content, updated_at: value.updated_at, @@ -62,6 +78,10 @@ impl From for PublishedPage { pad_id: value.pad_id, note_id: value.note_id, allow_task_updates: value.allow_task_updates != 0, + resource_slug: value.resource_slug, + workspace_id: value.workspace_id, + workspace_slug: value.workspace_slug, + owner_map: value.owner_map, title: value.title, content: value.content, updated_at: value.updated_at, @@ -310,29 +330,23 @@ pub async fn set_note_public_page_unprotected( Ok(()) } -pub async fn update_public_task( - pool: &Database, - token: &str, +pub fn updated_public_task_content( + content: &str, source_line: usize, checked: bool, -) -> Result, sqlx::Error> { - let Some(mut page) = find_published_page(pool, token).await? else { - return Ok(None); - }; - if !page.allow_task_updates || source_line == 0 { - return Ok(Some(page)); +) -> Option { + if source_line == 0 { + return None; } - let mut lines: Vec = page.content.split('\n').map(str::to_owned).collect(); - let Some(line) = lines.get_mut(source_line - 1) else { - return Ok(Some(page)); - }; + let mut lines: Vec = content.split('\n').map(str::to_owned).collect(); + let line = lines.get_mut(source_line - 1)?; let bytes = line.as_bytes(); let mut i = 0usize; while i < bytes.len() && bytes[i].is_ascii_whitespace() { i += 1; } if i >= bytes.len() || !matches!(bytes[i], b'-' | b'*' | b'+') { - return Ok(Some(page)); + return None; } i += 1; while i < bytes.len() && bytes[i].is_ascii_whitespace() { @@ -343,24 +357,10 @@ pub async fn update_public_task( || !matches!(bytes[i + 1], b' ' | b'x' | b'X') || bytes[i + 2] != b']' { - return Ok(Some(page)); + return None; } line.replace_range(i + 1..i + 2, if checked { "x" } else { " " }); - page.content = lines.join("\n"); - if let Some(id) = page.pad_id { - sqlx::query(queries::get(pool.kind(), queries::Q042)) - .bind(&page.content) - .bind(id) - .execute(pool.pool()) - .await?; - } else if let Some(id) = page.note_id { - sqlx::query(queries::get(pool.kind(), queries::Q043)) - .bind(&page.content) - .bind(id) - .execute(pool.pool()) - .await?; - } - find_published_page(pool, token).await + Some(lines.join("\n")) } pub async fn pad_file_token(pool: &Database, pad_id: i64) -> Result { @@ -416,6 +416,10 @@ impl<'r> sqlx::FromRow<'r, AnyRow> for PublishedPageRow { pad_id: row.try_get("pad_id")?, note_id: row.try_get("note_id")?, allow_task_updates: row.try_get("allow_task_updates")?, + resource_slug: crate::row_decode::text(row, "resource_slug")?, + workspace_id: row.try_get("workspace_id")?, + workspace_slug: crate::row_decode::optional_text(row, "workspace_slug")?, + owner_map: crate::row_decode::text(row, "owner_map")?, title: crate::row_decode::text(row, "title")?, content: crate::row_decode::text(row, "content")?, updated_at: crate::row_decode::text(row, "updated_at")?, @@ -429,6 +433,10 @@ impl<'r> sqlx::FromRow<'r, AnyRow> for PostgresPublishedPageRow { pad_id: row.try_get("pad_id")?, note_id: row.try_get("note_id")?, allow_task_updates: row.try_get("allow_task_updates")?, + resource_slug: crate::row_decode::text(row, "resource_slug")?, + workspace_id: row.try_get("workspace_id")?, + workspace_slug: crate::row_decode::optional_text(row, "workspace_slug")?, + owner_map: crate::row_decode::text(row, "owner_map")?, title: crate::row_decode::text(row, "title")?, content: crate::row_decode::text(row, "content")?, updated_at: crate::row_decode::text(row, "updated_at")?, @@ -437,17 +445,14 @@ impl<'r> sqlx::FromRow<'r, AnyRow> for PostgresPublishedPageRow { } pub async fn pad_public_page_disabled(pool: &Database, pad_id: i64) -> Result { - let sql = match pool.kind() { - DatabaseKind::Postgres => "SELECT public_page_disabled FROM pads WHERE id = $1", - _ => "SELECT public_page_disabled FROM pads WHERE id = ?", - }; + let query = queries::get(pool.kind(), queries::PAD_PUBLIC_PAGE_DISABLED); if pool.kind() == DatabaseKind::Postgres { - return Ok(sqlx::query_scalar::<_, bool>(sql) + return Ok(sqlx::query_scalar::<_, bool>(query) .bind(pad_id) .fetch_one(pool.pool()) .await?); } - Ok(sqlx::query_scalar::<_, i64>(sql) + Ok(sqlx::query_scalar::<_, i64>(query) .bind(pad_id) .fetch_one(pool.pool()) .await? @@ -455,17 +460,14 @@ pub async fn pad_public_page_disabled(pool: &Database, pad_id: i64) -> Result Result { - let sql = match pool.kind() { - DatabaseKind::Postgres => "SELECT public_page_disabled FROM notes WHERE id = $1", - _ => "SELECT public_page_disabled FROM notes WHERE id = ?", - }; + let query = queries::get(pool.kind(), queries::NOTE_PUBLIC_PAGE_DISABLED); if pool.kind() == DatabaseKind::Postgres { - return Ok(sqlx::query_scalar::<_, bool>(sql) + return Ok(sqlx::query_scalar::<_, bool>(query) .bind(note_id) .fetch_one(pool.pool()) .await?); } - Ok(sqlx::query_scalar::<_, i64>(sql) + Ok(sqlx::query_scalar::<_, i64>(query) .bind(note_id) .fetch_one(pool.pool()) .await? diff --git a/src/main.rs b/src/main.rs index 3fc50a6..d7e7d7e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -12,6 +12,7 @@ mod app; mod assets; mod auth; mod cache; +mod collab; mod config; mod database; mod db; @@ -304,9 +305,10 @@ mod startup_tests { let credential = startup_credential(); assert!(credential.contains(&format!("RustPad {}", env!("CARGO_PKG_VERSION")))); assert!(credential.contains("Mateusz GruszczyÅ„ski @linuxiarz.pl")); - assert!(credential.contains( - "https://git.linuxiarz.pl/gru/rustpad/src/branch/master/LICENSE.md" - )); + assert!( + credential + .contains("https://git.linuxiarz.pl/gru/rustpad/src/branch/master/LICENSE.md") + ); } } diff --git a/src/queries/mod.rs b/src/queries/mod.rs index 3628a95..ad1c997 100644 --- a/src/queries/mod.rs +++ b/src/queries/mod.rs @@ -110,6 +110,8 @@ pub enum Query { SHARE_SESSION_PERMISSION, SHARE_SESSIONS_DELETE_BY_LINK, SHARE_SESSIONS_DELETE_EXPIRED, + PAD_PUBLIC_PAGE_DISABLED, + NOTE_PUBLIC_PAGE_DISABLED, Q001, Q002, Q003, @@ -153,14 +155,18 @@ pub enum Query { Q047, Q040, Q041, - Q042, - Q043, Q044, Q045, Q048, Q049, Q050, Q051, + Q054, + Q055, + Q056, + Q057, + Q058, + Q059, } pub fn get(kind: DatabaseKind, query: Query) -> &'static str { @@ -269,6 +275,8 @@ pub const SHARE_SESSION_INSERT: Query = Query::SHARE_SESSION_INSERT; pub const SHARE_SESSION_PERMISSION: Query = Query::SHARE_SESSION_PERMISSION; pub const SHARE_SESSIONS_DELETE_BY_LINK: Query = Query::SHARE_SESSIONS_DELETE_BY_LINK; pub const SHARE_SESSIONS_DELETE_EXPIRED: Query = Query::SHARE_SESSIONS_DELETE_EXPIRED; +pub const PAD_PUBLIC_PAGE_DISABLED: Query = Query::PAD_PUBLIC_PAGE_DISABLED; +pub const NOTE_PUBLIC_PAGE_DISABLED: Query = Query::NOTE_PUBLIC_PAGE_DISABLED; pub const Q001: Query = Query::Q001; pub const Q002: Query = Query::Q002; pub const Q003: Query = Query::Q003; @@ -312,14 +320,18 @@ pub const Q046: Query = Query::Q046; pub const Q047: Query = Query::Q047; pub const Q040: Query = Query::Q040; pub const Q041: Query = Query::Q041; -pub const Q042: Query = Query::Q042; -pub const Q043: Query = Query::Q043; pub const Q044: Query = Query::Q044; pub const Q045: Query = Query::Q045; pub const Q048: Query = Query::Q048; pub const Q049: Query = Query::Q049; pub const Q050: Query = Query::Q050; pub const Q051: Query = Query::Q051; +pub const Q054: Query = Query::Q054; +pub const Q055: Query = Query::Q055; +pub const Q056: Query = Query::Q056; +pub const Q057: Query = Query::Q057; +pub const Q058: Query = Query::Q058; +pub const Q059: Query = Query::Q059; #[cfg(test)] mod tests { @@ -340,6 +352,8 @@ mod tests { SHARE_SESSION_PERMISSION, SHARE_SESSIONS_DELETE_BY_LINK, SHARE_SESSIONS_DELETE_EXPIRED, + PAD_PUBLIC_PAGE_DISABLED, + NOTE_PUBLIC_PAGE_DISABLED, ] { assert!(!get(DatabaseKind::Sqlite, query).is_empty()); assert!(!get(DatabaseKind::Postgres, query).is_empty()); diff --git a/src/queries/mysql.rs b/src/queries/mysql.rs index 1af5d08..de95815 100644 --- a/src/queries/mysql.rs +++ b/src/queries/mysql.rs @@ -31,9 +31,7 @@ pub fn get(query: Query) -> &'static str { Query::AUTH_UPDATE_EDITOR_COLOR => { r#"UPDATE users SET editor_color = ?, updated_at = ? WHERE id = ?"# } - Query::AUTH_UPDATE_THEME => { - r#"UPDATE users SET theme = ?, updated_at = ? WHERE id = ?"# - } + Query::AUTH_UPDATE_THEME => r#"UPDATE users SET theme = ?, updated_at = ? WHERE id = ?"#, Query::AUTH_EDITOR_COLOR_BY_USER => r#"SELECT editor_color FROM users WHERE id = ?"#, Query::RESOURCE_COLOR_BY_USER => { r#"SELECT color FROM user_resource_colors WHERE user_id = ? AND resource_kind = ? AND resource_slug = ?"# @@ -174,10 +172,10 @@ pub fn get(query: Query) -> &'static str { r#"INSERT INTO user_pads (user_id, pad_id) SELECT ?, id FROM pads WHERE slug = ?"# } Query::USER_LIST_WORKSPACES => { - r#"SELECT w.slug, CAST(w.title AS CHAR CHARACTER SET utf8mb4) AS title, CASE WHEN w.password_hash IS NULL THEN 0 ELSE 1 END AS protected, w.updated_at, CASE WHEN w.is_private THEN 1 ELSE 0 END AS private, 1 AS owned, 'rw' AS permission, '' AS shared_by FROM user_workspaces uw JOIN workspaces w ON w.id = uw.workspace_id WHERE uw.user_id = ? UNION SELECT w.slug, CAST(w.title AS CHAR CHARACTER SET utf8mb4) AS title, CASE WHEN w.password_hash IS NULL THEN 0 ELSE 1 END, w.updated_at, CASE WHEN w.is_private THEN 1 ELSE 0 END, 0, rp.permission, COALESCE((SELECT u.nickname FROM user_workspaces owner_uw JOIN users u ON u.id = owner_uw.user_id WHERE owner_uw.workspace_id = w.id LIMIT 1), 'Unknown user') AS shared_by FROM resource_permissions rp JOIN workspaces w ON w.slug = rp.resource_slug WHERE rp.resource_kind = 'workspace' AND rp.user_id = ? ORDER BY updated_at DESC"# + r#"SELECT w.slug, CAST(w.title AS CHAR CHARACTER SET utf8mb4) AS title, CAST(CASE WHEN w.password_hash IS NULL THEN 0 ELSE 1 END AS SIGNED) AS protected, w.updated_at, CAST(CASE WHEN w.is_private THEN 1 ELSE 0 END AS SIGNED) AS private, CAST(1 AS SIGNED) AS owned, 'rw' AS permission, '' AS shared_by FROM user_workspaces uw JOIN workspaces w ON w.id = uw.workspace_id WHERE uw.user_id = ? UNION SELECT w.slug, CAST(w.title AS CHAR CHARACTER SET utf8mb4) AS title, CAST(CASE WHEN w.password_hash IS NULL THEN 0 ELSE 1 END AS SIGNED), w.updated_at, CAST(CASE WHEN w.is_private THEN 1 ELSE 0 END AS SIGNED), CAST(0 AS SIGNED), rp.permission, COALESCE((SELECT u.nickname FROM user_workspaces owner_uw JOIN users u ON u.id = owner_uw.user_id WHERE owner_uw.workspace_id = w.id LIMIT 1), 'Unknown user') AS shared_by FROM resource_permissions rp JOIN workspaces w ON w.slug = rp.resource_slug WHERE rp.resource_kind = 'workspace' AND rp.user_id = ? ORDER BY updated_at DESC"# } Query::USER_LIST_PADS => { - r#"SELECT p.slug, CAST(p.title AS CHAR CHARACTER SET utf8mb4) AS title, CASE WHEN p.password_hash IS NULL THEN 0 ELSE 1 END AS protected, p.updated_at, CASE WHEN p.is_private THEN 1 ELSE 0 END AS private, 1 AS owned, 'rw' AS permission, '' AS shared_by FROM user_pads up JOIN pads p ON p.id = up.pad_id WHERE up.user_id = ? UNION SELECT p.slug, CAST(p.title AS CHAR CHARACTER SET utf8mb4) AS title, CASE WHEN p.password_hash IS NULL THEN 0 ELSE 1 END, p.updated_at, CASE WHEN p.is_private THEN 1 ELSE 0 END, 0, rp.permission, COALESCE((SELECT u.nickname FROM user_pads owner_up JOIN users u ON u.id = owner_up.user_id WHERE owner_up.pad_id = p.id LIMIT 1), 'Unknown user') AS shared_by FROM resource_permissions rp JOIN pads p ON p.slug = rp.resource_slug WHERE rp.resource_kind = 'pad' AND rp.user_id = ? ORDER BY updated_at DESC"# + r#"SELECT p.slug, CAST(p.title AS CHAR CHARACTER SET utf8mb4) AS title, CAST(CASE WHEN p.password_hash IS NULL THEN 0 ELSE 1 END AS SIGNED) AS protected, p.updated_at, CAST(CASE WHEN p.is_private THEN 1 ELSE 0 END AS SIGNED) AS private, CAST(1 AS SIGNED) AS owned, 'rw' AS permission, '' AS shared_by FROM user_pads up JOIN pads p ON p.id = up.pad_id WHERE up.user_id = ? UNION SELECT p.slug, CAST(p.title AS CHAR CHARACTER SET utf8mb4) AS title, CAST(CASE WHEN p.password_hash IS NULL THEN 0 ELSE 1 END AS SIGNED), p.updated_at, CAST(CASE WHEN p.is_private THEN 1 ELSE 0 END AS SIGNED), CAST(0 AS SIGNED), rp.permission, COALESCE((SELECT u.nickname FROM user_pads owner_up JOIN users u ON u.id = owner_up.user_id WHERE owner_up.pad_id = p.id LIMIT 1), 'Unknown user') AS shared_by FROM resource_permissions rp JOIN pads p ON p.slug = rp.resource_slug WHERE rp.resource_kind = 'pad' AND rp.user_id = ? ORDER BY updated_at DESC"# } Query::USER_OWNS_WORKSPACE => { r#"SELECT COUNT(*) FROM user_workspaces uw JOIN workspaces w ON w.id = uw.workspace_id WHERE uw.user_id = ? AND w.slug = ?"# @@ -268,6 +266,13 @@ pub fn get(query: Query) -> &'static str { Query::SHARE_SESSIONS_DELETE_EXPIRED => { r#"DELETE FROM resource_share_sessions WHERE expires_at <= ?"# } + + Query::PAD_PUBLIC_PAGE_DISABLED => { + r#"SELECT CAST(CASE WHEN public_page_disabled THEN 1 ELSE 0 END AS SIGNED) FROM pads WHERE id = ?"# + } + Query::NOTE_PUBLIC_PAGE_DISABLED => { + r#"SELECT CAST(CASE WHEN public_page_disabled THEN 1 ELSE 0 END AS SIGNED) FROM notes WHERE id = ?"# + } Query::Q001 => { r#"SELECT id, slug, CAST(title AS CHAR CHARACTER SET utf8mb4) AS title, CAST(password_hash AS CHAR CHARACTER SET utf8mb4) AS password_hash, created_at, updated_at, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS SIGNED) AS is_private FROM workspaces WHERE slug = ?"# } @@ -293,9 +298,11 @@ pub fn get(query: Query) -> &'static str { r#"SELECT id, CAST(content AS CHAR CHARACTER SET utf8mb4) AS content, created_at, CAST(author AS CHAR CHARACTER SET utf8mb4) AS author, CAST(owner_map AS CHAR CHARACTER SET utf8mb4) AS owner_map FROM note_revisions WHERE note_id = ? ORDER BY id DESC LIMIT 100"# } Query::Q011 => { - r#"SELECT id, slug, CAST(title AS CHAR CHARACTER SET utf8mb4) AS title, CAST(content AS CHAR CHARACTER SET utf8mb4) AS content, CAST(password_hash AS CHAR CHARACTER SET utf8mb4) AS password_hash, created_at, updated_at, CAST(owner_map AS CHAR CHARACTER SET utf8mb4) AS owner_map, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS SIGNED) AS is_private, CAST(created_by_guest_id AS CHAR CHARACTER SET utf8mb4) AS created_by_guest_id FROM pads WHERE slug = ?"# + r#"SELECT id, slug, CAST(title AS CHAR CHARACTER SET utf8mb4) AS title, CAST(content AS CHAR CHARACTER SET utf8mb4) AS content, CAST(password_hash AS CHAR CHARACTER SET utf8mb4) AS password_hash, created_at, updated_at, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS SIGNED) AS is_private, CAST(created_by_guest_id AS CHAR CHARACTER SET utf8mb4) AS created_by_guest_id FROM pads WHERE slug = ?"# + } + Query::Q012 => { + r#"INSERT INTO pads (slug, title, password_hash, created_by_guest_id) VALUES (?, ?, ?, ?)"# } - Query::Q012 => r#"INSERT INTO pads (slug, title, password_hash, created_by_guest_id) VALUES (?, ?, ?, ?)"#, Query::Q013 => { r#"UPDATE pads SET content = ?, owner_map = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"# } @@ -311,7 +318,7 @@ pub fn get(query: Query) -> &'static str { Query::Q019 => r#"SELECT token FROM published_pages WHERE note_id = ?"#, Query::Q020 => r#"INSERT INTO published_pages (token, note_id) VALUES (?, ?)"#, Query::Q021 => { - r#"SELECT pp.token, pp.pad_id, pp.note_id, CAST(CASE WHEN pp.allow_task_updates THEN 1 ELSE 0 END AS SIGNED) AS allow_task_updates, CAST(COALESCE(p.title, n.title) AS CHAR CHARACTER SET utf8mb4) AS title, CAST(COALESCE(p.content, n.content) AS CHAR CHARACTER SET utf8mb4) AS content, COALESCE(p.updated_at, n.updated_at) AS updated_at FROM published_pages pp LEFT JOIN pads p ON p.id = pp.pad_id LEFT JOIN notes n ON n.id = pp.note_id WHERE pp.token = ?"# + r#"SELECT pp.token, pp.pad_id, pp.note_id, CAST(CASE WHEN pp.allow_task_updates THEN 1 ELSE 0 END AS SIGNED) AS allow_task_updates, CAST(COALESCE(p.slug, n.slug) AS CHAR CHARACTER SET utf8mb4) AS resource_slug, n.workspace_id AS workspace_id, CAST(w.slug AS CHAR CHARACTER SET utf8mb4) AS workspace_slug, CAST(COALESCE(p.owner_map, n.owner_map, '[]') AS CHAR CHARACTER SET utf8mb4) AS owner_map, CAST(COALESCE(p.title, n.title) AS CHAR CHARACTER SET utf8mb4) AS title, CAST(COALESCE(p.content, n.content) AS CHAR CHARACTER SET utf8mb4) AS content, COALESCE(p.updated_at, n.updated_at) AS updated_at FROM published_pages pp LEFT JOIN pads p ON p.id = pp.pad_id LEFT JOIN notes n ON n.id = pp.note_id LEFT JOIN workspaces w ON w.id = n.workspace_id WHERE pp.token = ?"# } Query::Q022 => r#"SELECT file_token FROM pads WHERE id = ?"#, Query::Q023 => r#"UPDATE pads SET file_token = ? WHERE id = ? AND file_token IS NULL"#, @@ -353,12 +360,6 @@ pub fn get(query: Query) -> &'static str { Query::Q047 => r#"DELETE FROM pad_files WHERE id = ? AND pad_id = ?"#, Query::Q040 => r#"UPDATE published_pages SET allow_task_updates = ? WHERE pad_id = ?"#, Query::Q041 => r#"UPDATE published_pages SET allow_task_updates = ? WHERE note_id = ?"#, - Query::Q042 => { - r#"UPDATE pads SET content = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"# - } - Query::Q043 => { - r#"UPDATE notes SET content = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"# - } Query::Q044 => { r#"SELECT CAST(CASE WHEN allow_task_updates THEN 1 ELSE 0 END AS SIGNED) FROM published_pages WHERE pad_id = ?"# } @@ -373,6 +374,68 @@ pub fn get(query: Query) -> &'static str { r#"SELECT CAST(CASE WHEN unprotected THEN 1 ELSE 0 END AS SIGNED) FROM published_pages WHERE note_id = ?"# } Query::Q050 => r#"UPDATE published_pages SET unprotected = ? WHERE pad_id = ?"#, + Query::Q054 => { + r#"INSERT INTO note_revisions (note_id, content, author, owner_map, collaboration_client_id, collaboration_update_id) VALUES (?, ?, ?, ?, ?, ?)"# + } + Query::Q055 => { + r#"INSERT INTO revisions (pad_id, content, author, owner_map, collaboration_client_id, collaboration_update_id) VALUES (?, ?, ?, ?, ?, ?)"# + } + Query::Q056 => { + r#"SELECT MAX(collaboration_update_id) FROM note_revisions WHERE note_id = ? AND collaboration_client_id = ?"# + } + Query::Q057 => { + r#"SELECT MAX(collaboration_update_id) FROM revisions WHERE pad_id = ? AND collaboration_client_id = ?"# + } + Query::Q058 => { + r#"SELECT n.content, n.owner_map, COALESCE((SELECT MAX(r.id) FROM note_revisions r WHERE r.note_id = n.id), 0) AS revision_id FROM notes n WHERE n.id = ?"# + } + Query::Q059 => { + r#"SELECT p.content, p.owner_map, COALESCE((SELECT MAX(r.id) FROM revisions r WHERE r.pad_id = p.id), 0) AS revision_id FROM pads p WHERE p.id = ?"# + } Query::Q051 => r#"UPDATE published_pages SET unprotected = ? WHERE note_id = ?"#, } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn boolean_projections_are_normalized_for_sqlx_any() { + // MySQL BOOLEAN is TINYINT(1), which sqlx::Any 0.8 cannot map directly. + for (query, expected_casts) in [ + (Query::RESOURCE_EDITOR_SETTINGS_SELECT, 1), + (Query::EDITOR_PREFERENCES_SELECT_PAD, 4), + (Query::EDITOR_PREFERENCES_SELECT_NOTE, 4), + (Query::AUTH_USER_BY_EXTERNAL_ID, 1), + (Query::AUTH_USER_BY_SESSION, 1), + (Query::AUTH_USER_BY_NICKNAME, 1), + (Query::AUTH_USER_BY_EMAIL, 1), + (Query::AUTH_USER_BY_SHARE_IDENTIFIER, 1), + (Query::USER_LIST_WORKSPACES, 6), + (Query::USER_LIST_PADS, 6), + (Query::PAD_PUBLIC_PAGE_DISABLED, 1), + (Query::NOTE_PUBLIC_PAGE_DISABLED, 1), + (Query::Q001, 1), + (Query::Q003, 1), + (Query::Q004, 1), + (Query::Q011, 1), + (Query::Q021, 1), + (Query::Q033, 1), + (Query::Q036, 1), + (Query::Q038, 1), + (Query::Q046, 1), + (Query::Q044, 1), + (Query::Q045, 1), + (Query::Q048, 1), + (Query::Q049, 1), + ] { + let sql = get(query); + assert_eq!( + sql.matches("AS SIGNED").count(), + expected_casts, + "MySQL boolean projection is not normalized in {query:?}: {sql}" + ); + } + } +} diff --git a/src/queries/postgres.rs b/src/queries/postgres.rs index b71f0ad..d880244 100644 --- a/src/queries/postgres.rs +++ b/src/queries/postgres.rs @@ -31,9 +31,7 @@ pub fn get(query: Query) -> &'static str { Query::AUTH_UPDATE_EDITOR_COLOR => { r#"UPDATE users SET editor_color = $1, updated_at = $2 WHERE id = $3"# } - Query::AUTH_UPDATE_THEME => { - r#"UPDATE users SET theme = $1, updated_at = $2 WHERE id = $3"# - } + Query::AUTH_UPDATE_THEME => r#"UPDATE users SET theme = $1, updated_at = $2 WHERE id = $3"#, Query::AUTH_EDITOR_COLOR_BY_USER => r#"SELECT editor_color FROM users WHERE id = $1"#, Query::RESOURCE_COLOR_BY_USER => { r#"SELECT color FROM user_resource_colors WHERE user_id = $1 AND resource_kind = $2 AND resource_slug = $3"# @@ -270,6 +268,11 @@ pub fn get(query: Query) -> &'static str { Query::SHARE_SESSIONS_DELETE_EXPIRED => { r#"DELETE FROM resource_share_sessions WHERE expires_at <= $1"# } + + Query::PAD_PUBLIC_PAGE_DISABLED => r#"SELECT public_page_disabled FROM pads WHERE id = $1"#, + Query::NOTE_PUBLIC_PAGE_DISABLED => { + r#"SELECT public_page_disabled FROM notes WHERE id = $1"# + } Query::Q001 => { r#"SELECT id, slug, title, password_hash, created_at, updated_at, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS BIGINT) AS is_private FROM workspaces WHERE slug = $1"# } @@ -297,9 +300,11 @@ pub fn get(query: Query) -> &'static str { r#"SELECT id, content, created_at, author, owner_map FROM note_revisions WHERE note_id = $1 ORDER BY id DESC LIMIT 100"# } Query::Q011 => { - r#"SELECT id, slug, title, content, password_hash, created_at, updated_at, owner_map, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS BIGINT) AS is_private, created_by_guest_id FROM pads WHERE slug = $1"# + r#"SELECT id, slug, title, content, password_hash, created_at, updated_at, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS BIGINT) AS is_private, created_by_guest_id FROM pads WHERE slug = $1"# + } + Query::Q012 => { + r#"INSERT INTO pads (slug, title, password_hash, created_by_guest_id) VALUES ($1, $2, $3, $4)"# } - Query::Q012 => r#"INSERT INTO pads (slug, title, password_hash, created_by_guest_id) VALUES ($1, $2, $3, $4)"#, Query::Q013 => { r#"UPDATE pads SET content = $1, owner_map = $2, updated_at = (CURRENT_TIMESTAMP::text) WHERE id = $3"# } @@ -315,7 +320,7 @@ pub fn get(query: Query) -> &'static str { Query::Q019 => r#"SELECT token FROM published_pages WHERE note_id = $1"#, Query::Q020 => r#"INSERT INTO published_pages (token, note_id) VALUES ($1, $2)"#, Query::Q021 => { - r#"SELECT pp.token, pp.pad_id, pp.note_id, pp.allow_task_updates, COALESCE(p.title, n.title) AS title, COALESCE(p.content, n.content) AS content, COALESCE(p.updated_at, n.updated_at) AS updated_at FROM published_pages pp LEFT JOIN pads p ON p.id = pp.pad_id LEFT JOIN notes n ON n.id = pp.note_id WHERE pp.token = $1"# + r#"SELECT pp.token, pp.pad_id, pp.note_id, pp.allow_task_updates, COALESCE(p.slug, n.slug) AS resource_slug, n.workspace_id AS workspace_id, w.slug AS workspace_slug, COALESCE(p.owner_map, n.owner_map, '[]') AS owner_map, COALESCE(p.title, n.title) AS title, COALESCE(p.content, n.content) AS content, COALESCE(p.updated_at, n.updated_at) AS updated_at FROM published_pages pp LEFT JOIN pads p ON p.id = pp.pad_id LEFT JOIN notes n ON n.id = pp.note_id LEFT JOIN workspaces w ON w.id = n.workspace_id WHERE pp.token = $1"# } Query::Q022 => r#"SELECT file_token FROM pads WHERE id = $1"#, Query::Q023 => r#"UPDATE pads SET file_token = $1 WHERE id = $2 AND file_token IS NULL"#, @@ -351,18 +356,30 @@ pub fn get(query: Query) -> &'static str { Query::Q047 => r#"DELETE FROM pad_files WHERE id = $1 AND pad_id = $2"#, Query::Q040 => r#"UPDATE published_pages SET allow_task_updates = $1 WHERE pad_id = $2"#, Query::Q041 => r#"UPDATE published_pages SET allow_task_updates = $1 WHERE note_id = $2"#, - Query::Q042 => { - r#"UPDATE pads SET content = $1, updated_at = (CURRENT_TIMESTAMP::text) WHERE id = $2"# - } - Query::Q043 => { - r#"UPDATE notes SET content = $1, updated_at = (CURRENT_TIMESTAMP::text) WHERE id = $2"# - } Query::Q044 => r#"SELECT allow_task_updates FROM published_pages WHERE pad_id = $1"#, Query::Q045 => r#"SELECT allow_task_updates FROM published_pages WHERE note_id = $1"#, Query::Q048 => r#"SELECT unprotected FROM published_pages WHERE pad_id = $1"#, Query::Q049 => r#"SELECT unprotected FROM published_pages WHERE note_id = $1"#, Query::Q050 => r#"UPDATE published_pages SET unprotected = $1 WHERE pad_id = $2"#, + Query::Q054 => { + r#"INSERT INTO note_revisions (note_id, content, author, owner_map, collaboration_client_id, collaboration_update_id) VALUES ($1, $2, $3, $4, $5, $6)"# + } + Query::Q055 => { + r#"INSERT INTO revisions (pad_id, content, author, owner_map, collaboration_client_id, collaboration_update_id) VALUES ($1, $2, $3, $4, $5, $6)"# + } + Query::Q056 => { + r#"SELECT MAX(collaboration_update_id) FROM note_revisions WHERE note_id = $1 AND collaboration_client_id = $2"# + } + Query::Q057 => { + r#"SELECT MAX(collaboration_update_id) FROM revisions WHERE pad_id = $1 AND collaboration_client_id = $2"# + } + Query::Q058 => { + r#"SELECT n.content, n.owner_map, COALESCE((SELECT MAX(r.id) FROM note_revisions r WHERE r.note_id = n.id), 0) AS revision_id FROM notes n WHERE n.id = $1"# + } + Query::Q059 => { + r#"SELECT p.content, p.owner_map, COALESCE((SELECT MAX(r.id) FROM revisions r WHERE r.pad_id = p.id), 0) AS revision_id FROM pads p WHERE p.id = $1"# + } Query::Q051 => r#"UPDATE published_pages SET unprotected = $1 WHERE note_id = $2"#, } } diff --git a/src/queries/sqlite.rs b/src/queries/sqlite.rs index 8ab14b7..1d739b0 100644 --- a/src/queries/sqlite.rs +++ b/src/queries/sqlite.rs @@ -31,9 +31,7 @@ pub fn get(query: Query) -> &'static str { Query::AUTH_UPDATE_EDITOR_COLOR => { r#"UPDATE users SET editor_color = ?, updated_at = ? WHERE id = ?"# } - Query::AUTH_UPDATE_THEME => { - r#"UPDATE users SET theme = ?, updated_at = ? WHERE id = ?"# - } + Query::AUTH_UPDATE_THEME => r#"UPDATE users SET theme = ?, updated_at = ? WHERE id = ?"#, Query::AUTH_EDITOR_COLOR_BY_USER => r#"SELECT editor_color FROM users WHERE id = ?"#, Query::RESOURCE_COLOR_BY_USER => { r#"SELECT color FROM user_resource_colors WHERE user_id = ? AND resource_kind = ? AND resource_slug = ?"# @@ -268,6 +266,13 @@ pub fn get(query: Query) -> &'static str { Query::SHARE_SESSIONS_DELETE_EXPIRED => { r#"DELETE FROM resource_share_sessions WHERE expires_at <= ?"# } + + Query::PAD_PUBLIC_PAGE_DISABLED => { + r#"SELECT CASE WHEN public_page_disabled THEN 1 ELSE 0 END FROM pads WHERE id = ?"# + } + Query::NOTE_PUBLIC_PAGE_DISABLED => { + r#"SELECT CASE WHEN public_page_disabled THEN 1 ELSE 0 END FROM notes WHERE id = ?"# + } Query::Q001 => { r#"SELECT id, slug, title, password_hash, created_at, updated_at, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS INTEGER) AS is_private FROM workspaces WHERE slug = ?"# } @@ -293,9 +298,11 @@ pub fn get(query: Query) -> &'static str { r#"SELECT id, content, created_at, author, owner_map FROM note_revisions WHERE note_id = ? ORDER BY id DESC LIMIT 100"# } Query::Q011 => { - r#"SELECT id, slug, title, content, password_hash, created_at, updated_at, owner_map, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS INTEGER) AS is_private, created_by_guest_id FROM pads WHERE slug = ?"# + r#"SELECT id, slug, title, content, password_hash, created_at, updated_at, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS INTEGER) AS is_private, created_by_guest_id FROM pads WHERE slug = ?"# + } + Query::Q012 => { + r#"INSERT INTO pads (slug, title, password_hash, created_by_guest_id) VALUES (?, ?, ?, ?)"# } - Query::Q012 => r#"INSERT INTO pads (slug, title, password_hash, created_by_guest_id) VALUES (?, ?, ?, ?)"#, Query::Q013 => { r#"UPDATE pads SET content = ?, owner_map = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"# } @@ -311,7 +318,7 @@ pub fn get(query: Query) -> &'static str { Query::Q019 => r#"SELECT token FROM published_pages WHERE note_id = ?"#, Query::Q020 => r#"INSERT INTO published_pages (token, note_id) VALUES (?, ?)"#, Query::Q021 => { - r#"SELECT pp.token, pp.pad_id, pp.note_id, CASE WHEN pp.allow_task_updates THEN 1 ELSE 0 END AS allow_task_updates, COALESCE(p.title, n.title) AS title, COALESCE(p.content, n.content) AS content, COALESCE(p.updated_at, n.updated_at) AS updated_at FROM published_pages pp LEFT JOIN pads p ON p.id = pp.pad_id LEFT JOIN notes n ON n.id = pp.note_id WHERE pp.token = ?"# + r#"SELECT pp.token, pp.pad_id, pp.note_id, CASE WHEN pp.allow_task_updates THEN 1 ELSE 0 END AS allow_task_updates, COALESCE(p.slug, n.slug) AS resource_slug, n.workspace_id AS workspace_id, w.slug AS workspace_slug, COALESCE(p.owner_map, n.owner_map, '[]') AS owner_map, COALESCE(p.title, n.title) AS title, COALESCE(p.content, n.content) AS content, COALESCE(p.updated_at, n.updated_at) AS updated_at FROM published_pages pp LEFT JOIN pads p ON p.id = pp.pad_id LEFT JOIN notes n ON n.id = pp.note_id LEFT JOIN workspaces w ON w.id = n.workspace_id WHERE pp.token = ?"# } Query::Q022 => r#"SELECT file_token FROM pads WHERE id = ?"#, Query::Q023 => r#"UPDATE pads SET file_token = ? WHERE id = ? AND file_token IS NULL"#, @@ -347,12 +354,6 @@ pub fn get(query: Query) -> &'static str { Query::Q047 => r#"DELETE FROM pad_files WHERE id = ? AND pad_id = ?"#, Query::Q040 => r#"UPDATE published_pages SET allow_task_updates = ? WHERE pad_id = ?"#, Query::Q041 => r#"UPDATE published_pages SET allow_task_updates = ? WHERE note_id = ?"#, - Query::Q042 => { - r#"UPDATE pads SET content = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"# - } - Query::Q043 => { - r#"UPDATE notes SET content = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"# - } Query::Q044 => { r#"SELECT CASE WHEN allow_task_updates THEN 1 ELSE 0 END FROM published_pages WHERE pad_id = ?"# } @@ -367,6 +368,24 @@ pub fn get(query: Query) -> &'static str { r#"SELECT CASE WHEN unprotected THEN 1 ELSE 0 END FROM published_pages WHERE note_id = ?"# } Query::Q050 => r#"UPDATE published_pages SET unprotected = ? WHERE pad_id = ?"#, + Query::Q054 => { + r#"INSERT INTO note_revisions (note_id, content, author, owner_map, collaboration_client_id, collaboration_update_id) VALUES (?, ?, ?, ?, ?, ?)"# + } + Query::Q055 => { + r#"INSERT INTO revisions (pad_id, content, author, owner_map, collaboration_client_id, collaboration_update_id) VALUES (?, ?, ?, ?, ?, ?)"# + } + Query::Q056 => { + r#"SELECT MAX(collaboration_update_id) FROM note_revisions WHERE note_id = ? AND collaboration_client_id = ?"# + } + Query::Q057 => { + r#"SELECT MAX(collaboration_update_id) FROM revisions WHERE pad_id = ? AND collaboration_client_id = ?"# + } + Query::Q058 => { + r#"SELECT n.content, n.owner_map, COALESCE((SELECT MAX(r.id) FROM note_revisions r WHERE r.note_id = n.id), 0) AS revision_id FROM notes n WHERE n.id = ?"# + } + Query::Q059 => { + r#"SELECT p.content, p.owner_map, COALESCE((SELECT MAX(r.id) FROM revisions r WHERE r.pad_id = p.id), 0) AS revision_id FROM pads p WHERE p.id = ?"# + } Query::Q051 => r#"UPDATE published_pages SET unprotected = ? WHERE note_id = ?"#, } } diff --git a/src/security.rs b/src/security.rs index 825c37d..4e4dab5 100644 --- a/src/security.rs +++ b/src/security.rs @@ -50,11 +50,7 @@ pub fn resource_token<'a>(headers: &'a HeaderMap, kind: &str, slug: &str) -> Opt cookie_value(headers, &name) } -pub fn share_session_token<'a>( - headers: &'a HeaderMap, - kind: &str, - slug: &str, -) -> Option<&'a str> { +pub fn share_session_token<'a>(headers: &'a HeaderMap, kind: &str, slug: &str) -> Option<&'a str> { let name = share_session_cookie_name(kind, slug); cookie_value(headers, &name) } diff --git a/src/state.rs b/src/state.rs index 5e404cc..f3f2d34 100644 --- a/src/state.rs +++ b/src/state.rs @@ -7,12 +7,12 @@ * See LICENSE file in repository root for details. */ -use crate::database::Database; +use crate::{collab::CollaborativeDocument, database::Database}; use serde::Serialize; use std::{ collections::HashMap, sync::{ - Arc, + Arc, Weak, atomic::{AtomicU64, Ordering}, }, time::{Duration, Instant}, @@ -49,11 +49,14 @@ pub struct SmtpConfig { #[derive(Debug, Clone)] pub struct NoteUpdate { - pub content: String, + pub base_revision_id: i64, pub revision_id: i64, pub updated_at: String, pub author: Option, - pub owner_map: String, + pub client_id: String, + pub update_id: u64, + pub operation: crate::collab::TextOperation, + pub owner_replacements: Vec, } #[derive(Debug, Clone, Serialize)] @@ -108,6 +111,7 @@ pub struct AppState { pub unconfirmed_account_ttl_days: i64, pub ldap: Option, channels: RwLock>>, + collaborative_documents: RwLock>>>, presence: RwLock>>, next_connection_id: AtomicU64, rate_limits: Mutex>, @@ -148,6 +152,7 @@ impl AppState { unconfirmed_account_ttl_days, ldap, channels: RwLock::new(HashMap::new()), + collaborative_documents: RwLock::new(HashMap::new()), presence: RwLock::new(HashMap::new()), next_connection_id: AtomicU64::new(1), rate_limits: Mutex::new(HashMap::new()), @@ -197,6 +202,38 @@ impl AppState { self.rate_limits.lock().await.remove(key); } + pub async fn collaborative_document( + &self, + key: &str, + content: String, + owner_map: String, + revision_id: i64, + ) -> Arc> { + if let Some(document) = self + .collaborative_documents + .read() + .await + .get(key) + .and_then(|document| document.upgrade()) + { + return document; + } + + let mut documents = self.collaborative_documents.write().await; + if let Some(document) = documents.get(key).and_then(|document| document.upgrade()) { + return document; + } + documents.retain(|_, document| document.strong_count() > 0); + + let document = Arc::new(Mutex::new(CollaborativeDocument::new( + content, + owner_map, + revision_id, + ))); + documents.insert(key.to_owned(), Arc::downgrade(&document)); + document + } + async fn channel_for_key(&self, key: String) -> broadcast::Sender { if let Some(sender) = self.channels.read().await.get(&key) { return sender.clone(); diff --git a/src/websocket/mod.rs b/src/websocket/mod.rs index 09dd750..9d1706c 100644 --- a/src/websocket/mod.rs +++ b/src/websocket/mod.rs @@ -1,5 +1,7 @@ use crate::{ - auth, db, + auth, + collab::{self, AppliedOperation, OwnerReplacement, TextOperation}, + db, state::{AppState, NoteUpdate, PresenceUser, RoomEvent, SharedState}, }; use axum::{ @@ -13,7 +15,10 @@ use axum::{ use futures_util::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use std::time::{Duration, Instant}; +use std::{ + collections::HashSet, + time::{Duration, Instant}, +}; use tracing::{debug, info, warn}; mod pad; @@ -91,10 +96,17 @@ enum ClientMessage { color: Option, #[serde(default)] diagnostics: Option, + #[serde(default)] + client_id: Option, + #[serde(default)] + known_revision_id: Option, }, Update { - content: String, - owner_map: Option, + base_revision_id: i64, + update_id: u64, + operation: TextOperation, + #[serde(default)] + owner_replacements: Vec, }, Ping { nonce: u64, @@ -115,14 +127,27 @@ enum ServerMessage { note_title: String, content: String, owner_map: String, + revision_id: i64, access_level: String, + catchup_operations: Vec, + acknowledged_update_ids: Vec, + resync_required: bool, }, Document { - content: String, + base_revision_id: i64, revision_id: i64, updated_at: String, author: Option, + client_id: String, + update_id: u64, + operation: TextOperation, + owner_replacements: Vec, + }, + Resync { + content: String, + revision_id: i64, owner_map: String, + acknowledged_update_ids: Vec, }, Presence { users: Vec, @@ -259,8 +284,7 @@ async fn current_resource_access( let permission = resource_permission_from_tokens(state, kind, slug, access_tokens, session_token).await; let password_token_ok = password_access_from_tokens(state, kind, slug, access_tokens).await; - let write_allowed = - password_ok || password_token_ok || permission.as_deref() == Some("rw"); + let write_allowed = password_ok || password_token_ok || permission.as_deref() == Some("rw"); let read_allowed = write_allowed || permission.as_deref() == Some("ro"); (read_allowed, write_allowed) } @@ -329,31 +353,44 @@ async fn handle_socket( let _ = send_error(&mut socket, "Note not found").await; return; }; - let (password, access_token, nickname, guest_id, color, client_diagnostics) = - match socket.recv().await { - Some(Ok(Message::Text(text))) => match serde_json::from_str::(&text) { - Ok(ClientMessage::Authenticate { - password, - access_token, - nickname, - guest_id, - color, - diagnostics, - }) => ( - password, - access_token, - clean_nickname(nickname), - clean_guest_id(guest_id), - clean_color(color), - diagnostics, - ), - _ => { - let _ = send_error(&mut socket, "Wymagane uwierzytelnienie").await; - return; - } - }, - _ => return, - }; + let ( + password, + access_token, + nickname, + guest_id, + color, + client_diagnostics, + collaboration_client_id, + known_revision_id, + ) = match socket.recv().await { + Some(Ok(Message::Text(text))) => match serde_json::from_str::(&text) { + Ok(ClientMessage::Authenticate { + password, + access_token, + nickname, + guest_id, + color, + diagnostics, + client_id, + known_revision_id, + }) => ( + password, + access_token, + clean_nickname(nickname), + clean_guest_id(guest_id), + clean_color(color), + diagnostics, + clean_collaboration_client_id(client_id) + .unwrap_or_else(|| format!("legacy_{}", db::random_suffix(24))), + known_revision_id.filter(|revision_id| *revision_id >= 0), + ), + _ => { + let _ = send_error(&mut socket, "Wymagane uwierzytelnienie").await; + return; + } + }, + _ => return, + }; let session_token = cookie_session_token; let explicit_access_token = access_token .as_deref() @@ -392,13 +429,8 @@ async fn handle_socket( session_token.as_deref(), ) .await; - let anonymous_token_ok = password_access_from_tokens( - &state, - "workspace", - &workspace_slug, - &external_tokens, - ) - .await; + let anonymous_token_ok = + password_access_from_tokens(&state, "workspace", &workspace_slug, &external_tokens).await; let password_limit_key = format!("resource-password:{client_key}:workspace:{workspace_slug}"); let password_attempted = password .as_deref() @@ -464,13 +496,80 @@ async fn handle_socket( ) .await; info!(workspace_id = workspace.id, note_id = note.id, nickname = ?nickname, "note websocket authenticated"); + let room_key = AppState::note_room_key(&workspace_slug, ¬e_slug); + let collaboration_snapshot = match db::note_collaboration_snapshot(&state.db, note.id).await { + Ok(snapshot) => snapshot, + Err(error) => { + warn!(%error, workspace_id = workspace.id, note_id = note.id, "failed to load collaborative document"); + let _ = send_error(&mut socket, "Failed to load the document").await; + return; + } + }; + let collaborative_document = state + .collaborative_document( + &room_key, + collaboration_snapshot.content, + collaboration_snapshot.owner_map, + collaboration_snapshot.revision_id, + ) + .await; + // Subscribe before taking the authentication snapshot. Updates committed after + // the snapshot are then queued for this connection instead of falling into a gap. + let channel = state.note_channel(&workspace_slug, ¬e_slug).await; + let mut updates = channel.subscribe(); + let persisted_acknowledged_update_id = match db::latest_note_collaboration_update_id( + &state.db, + note.id, + &collaboration_client_id, + ) + .await + { + Ok(update_id) => update_id, + Err(error) => { + warn!(%error, workspace_id = workspace.id, note_id = note.id, "failed to load collaborative acknowledgement"); + let _ = send_error(&mut socket, "Failed to load the document").await; + return; + } + }; + let ( + authenticated_content, + authenticated_owner_map, + authenticated_revision_id, + catchup_operations, + acknowledged_update_ids, + resync_required, + ) = { + let mut document = collaborative_document.lock().await; + if let Some(update_id) = persisted_acknowledged_update_id { + document.acknowledge(&collaboration_client_id, update_id); + } + let (catchup_operations, resync_required) = match known_revision_id { + Some(revision_id) => match document.operations_after(revision_id) { + Some(operations) => (operations, false), + None => (Vec::new(), revision_id != document.revision_id), + }, + None => (Vec::new(), false), + }; + ( + document.content.clone(), + document.owner_map.clone(), + document.revision_id, + catchup_operations, + document.acknowledged_updates(&collaboration_client_id), + resync_required, + ) + }; if send( &mut socket, &ServerMessage::Authenticated { workspace_title: workspace.title.clone(), note_title: note.title.clone(), - content: note.content.clone(), - owner_map: note.owner_map.clone(), + content: authenticated_content, + owner_map: authenticated_owner_map, + revision_id: authenticated_revision_id, + catchup_operations, + acknowledged_update_ids, + resync_required, access_level: if write_allowed { "full".into() } else { @@ -483,9 +582,6 @@ async fn handle_socket( { return; } - let room_key = AppState::note_room_key(&workspace_slug, ¬e_slug); - let channel = state.note_channel(&workspace_slug, ¬e_slug).await; - let mut updates = channel.subscribe(); let display_name = nickname.clone().unwrap_or_else(|| "Guest".into()); let (connection_id, users) = state .join_room(&room_key, display_name.clone(), color, presence_identity) @@ -513,7 +609,7 @@ async fn handle_socket( tokio::select! { incoming=receiver.next()=>match incoming { Some(Ok(Message::Text(text)))=>match serde_json::from_str::(&text) { - Ok(ClientMessage::Update{content,owner_map})=>{ + Ok(ClientMessage::Update{base_revision_id,update_id,operation,owner_replacements})=>{ let (read_allowed, current_write_allowed) = current_resource_access( &state, "workspace", @@ -524,11 +620,119 @@ async fn handle_socket( ).await; if !read_allowed { let _=send_split(&mut sender,&ServerMessage::Error{message:"Access expired or revoked".into()}).await; break; } if !current_write_allowed { let _=send_split(&mut sender,&ServerMessage::Error{message:"Read-only access".into()}).await; continue; } - if content.len()>2_000_000 { let _=send_split(&mut sender,&ServerMessage::Error{message:"The document is too large".into()}).await; continue; } - let owner_map=owner_map.unwrap_or_else(||"[]".into()); - match db::save_revision(&state.db,note.id,workspace.id,&content,nickname.as_deref(),&owner_map).await { - Ok((revision_id,updated_at))=>{let _=channel.send(RoomEvent::Document(NoteUpdate{content,revision_id,updated_at,author:nickname.clone(),owner_map}));} - Err(error)=>warn!(%error, workspace_id = workspace.id, note_id = note.id, "failed to save revision"), + if update_id == 0 + || update_id > i64::MAX as u64 + || !valid_owner_replacements(&owner_replacements) + { + let _=send_split(&mut sender,&ServerMessage::Error{message:"Invalid collaborative update".into()}).await; + continue; + } + let mut document = collaborative_document.lock().await; + if document.has_applied_update(&collaboration_client_id, update_id) { + let snapshot = ( + document.content.clone(), + document.revision_id, + document.owner_map.clone(), + document.acknowledged_updates(&collaboration_client_id), + ); + drop(document); + let _ = send_split(&mut sender, &ServerMessage::Resync { + content: snapshot.0, + revision_id: snapshot.1, + owner_map: snapshot.2, + acknowledged_update_ids: snapshot.3, + }).await; + continue; + } + let transformed = match document.transform_from( + base_revision_id, + &operation, + &collaboration_client_id, + update_id, + ) { + Ok(operation) => operation, + Err(collab::OperationError::RevisionUnavailable) => { + let snapshot = ( + document.content.clone(), + document.revision_id, + document.owner_map.clone(), + document.acknowledged_updates(&collaboration_client_id), + ); + drop(document); + let _=send_split(&mut sender,&ServerMessage::Resync{ + content:snapshot.0, + revision_id:snapshot.1, + owner_map:snapshot.2, + acknowledged_update_ids:snapshot.3, + }).await; + continue; + } + Err(error) => { + drop(document); + warn!(%error, workspace_id = workspace.id, note_id = note.id, "invalid collaborative operation"); + let _=send_split(&mut sender,&ServerMessage::Error{message:"Invalid collaborative update".into()}).await; + continue; + } + }; + let applied_base_revision_id = document.revision_id; + let (content,owner_map)=match collab::apply_operation_to_document( + &document.content, + &document.owner_map, + &transformed, + &owner_replacements, + ) { + Ok(document) => document, + Err(error) => { + drop(document); + warn!(%error, workspace_id = workspace.id, note_id = note.id, "failed to apply collaborative operation"); + let _=send_split(&mut sender,&ServerMessage::Error{message:"Invalid collaborative update".into()}).await; + continue; + } + }; + if content.len()>2_000_000 { + drop(document); + let _=send_split(&mut sender,&ServerMessage::Error{message:"The document is too large".into()}).await; + continue; + } + match db::save_collaborative_revision( + &state.db, + note.id, + workspace.id, + &content, + nickname.as_deref(), + &owner_map, + &collaboration_client_id, + update_id as i64, + ).await { + Ok((revision_id,updated_at))=>{ + document.content=content.clone(); + document.owner_map=owner_map.clone(); + document.revision_id=revision_id; + document.record(AppliedOperation{ + base_revision_id:applied_base_revision_id, + revision_id, + client_id:collaboration_client_id.clone(), + update_id, + operation:transformed.clone(), + owner_replacements:owner_replacements.clone(), + }); + let _=channel.send(RoomEvent::Document(NoteUpdate{ + base_revision_id:applied_base_revision_id, + revision_id, + updated_at, + author:nickname.clone(), + client_id:collaboration_client_id.clone(), + update_id, + operation:transformed, + owner_replacements, + })); + drop(document); + } + Err(error)=>{ + drop(document); + warn!(%error, workspace_id = workspace.id, note_id = note.id, "failed to save revision"); + let _=send_split(&mut sender,&ServerMessage::Error{message:"Failed to save the document".into()}).await; + } } } Ok(ClientMessage::Ping{nonce})=>{ let _=send_split(&mut sender,&ServerMessage::Pong{nonce}).await; }, @@ -569,10 +773,15 @@ async fn handle_socket( break; } match update { - Ok(RoomEvent::Document(update))=>if send_split(&mut sender,&ServerMessage::Document{content:update.content,revision_id:update.revision_id,updated_at:update.updated_at,author:update.author,owner_map:update.owner_map}).await.is_err(){break;}, + Ok(RoomEvent::Document(update))=>if send_split(&mut sender,&ServerMessage::Document{base_revision_id:update.base_revision_id,revision_id:update.revision_id,updated_at:update.updated_at,author:update.author,client_id:update.client_id,update_id:update.update_id,operation:update.operation,owner_replacements:update.owner_replacements}).await.is_err(){break;}, Ok(RoomEvent::Presence(users))=>if send_split(&mut sender,&ServerMessage::Presence{users}).await.is_err(){break;}, Ok(RoomEvent::Chat{sender:chat_sender,text})=>if send_split(&mut sender,&ServerMessage::Chat{sender:chat_sender,text}).await.is_err(){break;}, - Err(tokio::sync::broadcast::error::RecvError::Lagged(_))=>if let Ok(Some(current))=db::find_note(&state.db,workspace.id,¬e_slug).await { if send_split(&mut sender,&ServerMessage::Document{content:current.content,revision_id:0,updated_at:current.updated_at,author:None,owner_map:current.owner_map}).await.is_err(){break;} }, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_))=>{ + let document=collaborative_document.lock().await; + let snapshot=(document.content.clone(),document.revision_id,document.owner_map.clone(),document.acknowledged_updates(&collaboration_client_id)); + drop(document); + if send_split(&mut sender,&ServerMessage::Resync{content:snapshot.0,revision_id:snapshot.1,owner_map:snapshot.2,acknowledged_update_ids:snapshot.3}).await.is_err(){break;} + }, Err(tokio::sync::broadcast::error::RecvError::Closed)=>break, } } @@ -586,6 +795,36 @@ async fn handle_socket( "note websocket disconnected" ); } +fn clean_collaboration_client_id(value: Option) -> Option { + value + .map(|value| value.trim().chars().take(64).collect::()) + .filter(|value| { + value.len() >= 8 + && value.chars().all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '-' | '_') + }) + }) +} + +fn valid_owner_replacements(replacements: &[OwnerReplacement]) -> bool { + if replacements.len() > 64 { + return false; + } + let mut owners = HashSet::with_capacity(replacements.len()); + replacements.iter().all(|replacement| { + !replacement.owner.is_empty() + && replacement.owner.chars().count() <= 80 + && !replacement.replacement.is_empty() + && replacement.replacement.chars().count() <= 120 + && !replacement.owner.chars().any(char::is_control) + && !replacement + .replacement + .chars() + .any(|character| character.is_control() && character != '\u{001f}') + && owners.insert(replacement.owner.as_str()) + }) +} + fn clean_nickname(value: Option) -> Option { value .map(|v| v.trim().chars().take(40).collect::()) diff --git a/src/websocket/pad.rs b/src/websocket/pad.rs index 4c67f8a..f9666c5 100644 --- a/src/websocket/pad.rs +++ b/src/websocket/pad.rs @@ -7,14 +7,27 @@ enum PadServerMessage { title: String, content: String, owner_map: String, + revision_id: i64, access_level: String, + catchup_operations: Vec, + acknowledged_update_ids: Vec, + resync_required: bool, }, Document { - content: String, + base_revision_id: i64, revision_id: i64, updated_at: String, author: Option, + client_id: String, + update_id: u64, + operation: TextOperation, + owner_replacements: Vec, + }, + Resync { + content: String, + revision_id: i64, owner_map: String, + acknowledged_update_ids: Vec, }, Presence { users: Vec, @@ -84,37 +97,50 @@ async fn handle_pad_socket( .await; return; }; - let (password, access_token, nickname, guest_id, color, client_diagnostics) = - match socket.recv().await { - Some(Ok(Message::Text(text))) => match serde_json::from_str::(&text) { - Ok(ClientMessage::Authenticate { - password, - access_token, - nickname, - guest_id, - color, - diagnostics, - }) => ( - password, - access_token, - clean_nickname(nickname), - clean_guest_id(guest_id), - clean_color(color), - diagnostics, - ), - _ => { - let _ = send_pad( - &mut socket, - &PadServerMessage::Error { - message: "Wymagane uwierzytelnienie".into(), - }, - ) - .await; - return; - } - }, - _ => return, - }; + let ( + password, + access_token, + nickname, + guest_id, + color, + client_diagnostics, + collaboration_client_id, + known_revision_id, + ) = match socket.recv().await { + Some(Ok(Message::Text(text))) => match serde_json::from_str::(&text) { + Ok(ClientMessage::Authenticate { + password, + access_token, + nickname, + guest_id, + color, + diagnostics, + client_id, + known_revision_id, + }) => ( + password, + access_token, + clean_nickname(nickname), + clean_guest_id(guest_id), + clean_color(color), + diagnostics, + clean_collaboration_client_id(client_id) + .unwrap_or_else(|| format!("legacy_{}", db::random_suffix(24))), + known_revision_id.filter(|revision_id| *revision_id >= 0), + ), + _ => { + let _ = send_pad( + &mut socket, + &PadServerMessage::Error { + message: "Wymagane uwierzytelnienie".into(), + }, + ) + .await; + return; + } + }, + _ => return, + }; let session_token = cookie_session_token; let explicit_access_token = access_token .as_deref() @@ -228,12 +254,88 @@ async fn handle_pad_socket( ) .await; info!(pad_id = pad.id, nickname = ?nickname, "pad websocket authenticated"); + let room_key = AppState::pad_room_key(&slug); + let collaboration_snapshot = match db::pad_collaboration_snapshot(&state.db, pad.id).await { + Ok(snapshot) => snapshot, + Err(error) => { + warn!(%error, pad_id = pad.id, "failed to load collaborative document"); + let _ = send_pad( + &mut socket, + &PadServerMessage::Error { + message: "Failed to load the document".into(), + }, + ) + .await; + return; + } + }; + let collaborative_document = state + .collaborative_document( + &room_key, + collaboration_snapshot.content, + collaboration_snapshot.owner_map, + collaboration_snapshot.revision_id, + ) + .await; + // Subscribe before taking the authentication snapshot. Updates committed after + // the snapshot are then queued for this connection instead of falling into a gap. + let channel = state.pad_channel(&slug).await; + let mut updates = channel.subscribe(); + let persisted_acknowledged_update_id = + match db::latest_pad_collaboration_update_id(&state.db, pad.id, &collaboration_client_id) + .await + { + Ok(update_id) => update_id, + Err(error) => { + warn!(%error, pad_id = pad.id, "failed to load collaborative acknowledgement"); + let _ = send_pad( + &mut socket, + &PadServerMessage::Error { + message: "Failed to load the document".into(), + }, + ) + .await; + return; + } + }; + let ( + authenticated_content, + authenticated_owner_map, + authenticated_revision_id, + catchup_operations, + acknowledged_update_ids, + resync_required, + ) = { + let mut document = collaborative_document.lock().await; + if let Some(update_id) = persisted_acknowledged_update_id { + document.acknowledge(&collaboration_client_id, update_id); + } + let (catchup_operations, resync_required) = match known_revision_id { + Some(revision_id) => match document.operations_after(revision_id) { + Some(operations) => (operations, false), + None => (Vec::new(), revision_id != document.revision_id), + }, + None => (Vec::new(), false), + }; + ( + document.content.clone(), + document.owner_map.clone(), + document.revision_id, + catchup_operations, + document.acknowledged_updates(&collaboration_client_id), + resync_required, + ) + }; if send_pad( &mut socket, &PadServerMessage::Authenticated { title: pad.title.clone(), - content: pad.content.clone(), - owner_map: pad.owner_map.clone(), + content: authenticated_content, + owner_map: authenticated_owner_map, + revision_id: authenticated_revision_id, + catchup_operations, + acknowledged_update_ids, + resync_required, access_level: if write_allowed { "full".into() } else { @@ -246,9 +348,6 @@ async fn handle_pad_socket( { return; } - let room_key = AppState::pad_room_key(&slug); - let channel = state.pad_channel(&slug).await; - let mut updates = channel.subscribe(); let display_name = nickname.clone().unwrap_or_else(|| "Guest".into()); let (connection_id, users) = state .join_room(&room_key, display_name.clone(), color, presence_identity) @@ -276,7 +375,7 @@ async fn handle_pad_socket( tokio::select! { incoming=receiver.next()=>match incoming{ Some(Ok(Message::Text(text)))=>match serde_json::from_str::(&text){ - Ok(ClientMessage::Update{content,owner_map})=>{ + Ok(ClientMessage::Update{base_revision_id,update_id,operation,owner_replacements})=>{ let (read_allowed, current_write_allowed) = current_resource_access( &state, "pad", @@ -287,10 +386,118 @@ async fn handle_pad_socket( ).await; if !read_allowed { let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Access expired or revoked".into()}).await;break; } if !current_write_allowed{let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Read-only access".into()}).await;continue;} - if content.len()>2_000_000 { let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"The document is too large".into()}).await; continue; } - let owner_map=owner_map.unwrap_or_else(||"[]".into()); - if let Ok((revision_id,updated_at))=db::save_pad_revision(&state.db,pad.id,&content,nickname.as_deref(),&owner_map).await{ - let _=channel.send(RoomEvent::Document(NoteUpdate{content,revision_id,updated_at,author:nickname.clone(),owner_map})); + if update_id == 0 + || update_id > i64::MAX as u64 + || !valid_owner_replacements(&owner_replacements) + { + let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Invalid collaborative update".into()}).await; + continue; + } + let mut document = collaborative_document.lock().await; + if document.has_applied_update(&collaboration_client_id, update_id) { + let snapshot = ( + document.content.clone(), + document.revision_id, + document.owner_map.clone(), + document.acknowledged_updates(&collaboration_client_id), + ); + drop(document); + let _ = send_pad_split(&mut sender, &PadServerMessage::Resync { + content: snapshot.0, + revision_id: snapshot.1, + owner_map: snapshot.2, + acknowledged_update_ids: snapshot.3, + }).await; + continue; + } + let transformed = match document.transform_from( + base_revision_id, + &operation, + &collaboration_client_id, + update_id, + ) { + Ok(operation) => operation, + Err(collab::OperationError::RevisionUnavailable) => { + let snapshot = ( + document.content.clone(), + document.revision_id, + document.owner_map.clone(), + document.acknowledged_updates(&collaboration_client_id), + ); + drop(document); + let _=send_pad_split(&mut sender,&PadServerMessage::Resync{ + content:snapshot.0, + revision_id:snapshot.1, + owner_map:snapshot.2, + acknowledged_update_ids:snapshot.3, + }).await; + continue; + } + Err(error) => { + drop(document); + warn!(%error, pad_id = pad.id, "invalid collaborative operation"); + let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Invalid collaborative update".into()}).await; + continue; + } + }; + let applied_base_revision_id = document.revision_id; + let (content,owner_map)=match collab::apply_operation_to_document( + &document.content, + &document.owner_map, + &transformed, + &owner_replacements, + ) { + Ok(document) => document, + Err(error) => { + drop(document); + warn!(%error, pad_id = pad.id, "failed to apply collaborative operation"); + let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Invalid collaborative update".into()}).await; + continue; + } + }; + if content.len()>2_000_000 { + drop(document); + let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"The document is too large".into()}).await; + continue; + } + match db::save_collaborative_pad_revision( + &state.db, + pad.id, + &content, + nickname.as_deref(), + &owner_map, + &collaboration_client_id, + update_id as i64, + ).await{ + Ok((revision_id,updated_at))=>{ + document.content=content.clone(); + document.owner_map=owner_map.clone(); + document.revision_id=revision_id; + document.record(AppliedOperation{ + base_revision_id:applied_base_revision_id, + revision_id, + client_id:collaboration_client_id.clone(), + update_id, + operation:transformed.clone(), + owner_replacements:owner_replacements.clone(), + }); + let _=channel.send(RoomEvent::Document(NoteUpdate{ + base_revision_id:applied_base_revision_id, + revision_id, + updated_at, + author:nickname.clone(), + client_id:collaboration_client_id.clone(), + update_id, + operation:transformed, + owner_replacements, + })); + drop(document); + } + Err(error)=>{ + drop(document); + warn!(%error, pad_id = pad.id, "failed to save revision"); + let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Failed to save the document".into()}).await; + } } } Ok(ClientMessage::Ping{nonce})=>{ let _=send_pad_split(&mut sender,&PadServerMessage::Pong{nonce}).await; }, @@ -334,10 +541,15 @@ async fn handle_pad_socket( break; } match update { - Ok(RoomEvent::Document(u))=>if send_pad_split(&mut sender,&PadServerMessage::Document{content:u.content,revision_id:u.revision_id,updated_at:u.updated_at,author:u.author,owner_map:u.owner_map}).await.is_err(){break;}, + Ok(RoomEvent::Document(u))=>if send_pad_split(&mut sender,&PadServerMessage::Document{base_revision_id:u.base_revision_id,revision_id:u.revision_id,updated_at:u.updated_at,author:u.author,client_id:u.client_id,update_id:u.update_id,operation:u.operation,owner_replacements:u.owner_replacements}).await.is_err(){break;}, Ok(RoomEvent::Presence(users))=>if send_pad_split(&mut sender,&PadServerMessage::Presence{users}).await.is_err(){break;}, Ok(RoomEvent::Chat{sender:chat_sender,text})=>if send_pad_split(&mut sender,&PadServerMessage::Chat{sender:chat_sender,text}).await.is_err(){break;}, - Err(tokio::sync::broadcast::error::RecvError::Lagged(_))=>if let Ok(Some(current))=db::find_pad(&state.db,&slug).await { if send_pad_split(&mut sender,&PadServerMessage::Document{content:current.content,revision_id:0,updated_at:current.updated_at,author:None,owner_map:current.owner_map}).await.is_err(){break;} }, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_))=>{ + let document=collaborative_document.lock().await; + let snapshot=(document.content.clone(),document.revision_id,document.owner_map.clone(),document.acknowledged_updates(&collaboration_client_id)); + drop(document); + if send_pad_split(&mut sender,&PadServerMessage::Resync{content:snapshot.0,revision_id:snapshot.1,owner_map:snapshot.2,acknowledged_update_ids:snapshot.3}).await.is_err(){break;} + }, Err(tokio::sync::broadcast::error::RecvError::Closed)=>break, } } diff --git a/static/js/collaboration-session.js b/static/js/collaboration-session.js new file mode 100644 index 0000000..78f37a8 --- /dev/null +++ b/static/js/collaboration-session.js @@ -0,0 +1,277 @@ +/* + * Copyright (C) 2026 Mateusz Gruszczynski @linuxiarz.pl + * Source-Available Code / Dual-Licensed. + * + * Free for non-commercial and evaluation use under terms of BSL/GPLv3. + * Commercial or production use requires a valid paid license. + * See LICENSE file in repository root for details. + */ + +import { + applyOperationToDocument, + compareOperationKeys, + composeOperations, + documentAfterPending, + identityOperation, + normalizeOperation, + operationBaseLength, + operationFromEdit, + operationTargetLength, + transformOperations, +} from "@rustpad/collaboration"; +import { parseAuthorship } from "@rustpad/authorship"; + +function hasTextEffect(operation) { + return normalizeOperation(operation).components.some(component => component.kind === "insert" || component.kind === "delete"); +} + +function mergeOwnerReplacements(previous = [], next = []) { + const replacements = new Map(); + for (const item of [...previous, ...next]) { + const owner = String(item?.owner || ""); + const replacement = String(item?.replacement || ""); + if (owner && replacement) replacements.set(owner, { owner, replacement }); + } + return [...replacements.values()]; +} + +function pendingEnvelope(clientId, updateId, operation, ownerReplacements = []) { + return { + clientId, + updateId, + operation: normalizeOperation(operation), + ownerReplacements: mergeOwnerReplacements([], ownerReplacements), + }; +} + +function serverEnvelope(message) { + return { + baseRevisionId: Number(message?.base_revision_id ?? message?.baseRevisionId), + revisionId: Number(message?.revision_id ?? message?.revisionId), + clientId: String(message?.client_id ?? message?.clientId ?? ""), + updateId: Number(message?.update_id ?? message?.updateId ?? 0), + operation: normalizeOperation(message?.operation), + ownerReplacements: mergeOwnerReplacements([], message?.owner_replacements ?? message?.ownerReplacements ?? []), + }; +} + +export class CollaborationRevisionGapError extends Error { + constructor(expected, actual) { + super(`Collaborative revision gap: expected ${expected}, received ${actual}`); + this.name = "CollaborationRevisionGapError"; + this.expected = expected; + this.actual = actual; + } +} + +export class CollaborationSession { + constructor(clientId) { + this.clientId = String(clientId || ""); + this.ready = false; + this.serverContent = ""; + this.serverOwnerMap = "[]"; + this.revisionId = 0; + this.outstanding = null; + this.buffer = null; + this.nextUpdateId = 1; + } + + initialize(content, ownerMap, revisionId, { clearPending = true } = {}) { + this.serverContent = String(content || ""); + this.serverOwnerMap = ownerMap == null ? "[]" : String(ownerMap); + this.revisionId = Number(revisionId) || 0; + if (clearPending) { + this.outstanding = null; + this.buffer = null; + } + this.ready = true; + } + + localDocument() { + return documentAfterPending( + this.serverContent, + this.serverOwnerMap, + this.outstanding, + this.buffer, + ); + } + + hasPending() { + return Boolean(this.outstanding || this.buffer); + } + + queue(operation, ownerReplacements = []) { + operation = normalizeOperation(operation); + const replacements = mergeOwnerReplacements([], ownerReplacements); + if (!hasTextEffect(operation) && !replacements.length) return null; + + const localLength = this.localDocument().content.length; + if (operationBaseLength(operation) !== localLength) { + throw new Error("Local operation base length does not match the collaborative document"); + } + + if (!this.buffer) { + this.buffer = pendingEnvelope(this.clientId, this.nextUpdateId++, operation, replacements); + } else { + this.buffer.operation = composeOperations(this.buffer.operation, operation); + this.buffer.ownerReplacements = mergeOwnerReplacements(this.buffer.ownerReplacements, replacements); + } + + if (operationTargetLength(this.buffer.operation) !== this.localDocument().content.length) { + throw new Error("Buffered operation target length does not match the collaborative document"); + } + return this.buffer; + } + + sendable() { + if (!this.ready || this.outstanding || !this.buffer) return null; + return { + baseRevisionId: this.revisionId, + updateId: this.buffer.updateId, + operation: this.buffer.operation, + ownerReplacements: this.buffer.ownerReplacements, + }; + } + + markSent(updateId) { + if (this.outstanding || !this.buffer || this.buffer.updateId !== Number(updateId)) return false; + this.outstanding = this.buffer; + this.buffer = null; + return true; + } + + integrate(message) { + const remote = serverEnvelope(message); + if (!Number.isFinite(remote.baseRevisionId) || !Number.isFinite(remote.revisionId)) { + throw new Error("Collaborative update is missing revision metadata"); + } + if (remote.revisionId <= this.revisionId) return { duplicate: true, ownAck: false, remote: false }; + if (remote.baseRevisionId !== this.revisionId) { + throw new CollaborationRevisionGapError(this.revisionId, remote.baseRevisionId); + } + + const nextServer = applyOperationToDocument( + this.serverContent, + this.serverOwnerMap, + remote.operation, + remote.ownerReplacements, + ); + const ownAck = Boolean( + this.outstanding + && remote.clientId === this.clientId + && remote.updateId === this.outstanding.updateId + ); + + if (ownAck) { + this.outstanding = null; + } else { + let remoteForPending = remote.operation; + if (this.outstanding) { + const outstandingBeforeRemote = compareOperationKeys(this.outstanding, remote) < 0; + const [outstandingPrime, remotePrime] = transformOperations( + this.outstanding.operation, + remoteForPending, + outstandingBeforeRemote, + ); + this.outstanding.operation = outstandingPrime; + remoteForPending = remotePrime; + } + if (this.buffer) { + const bufferBeforeRemote = compareOperationKeys(this.buffer, remote) < 0; + const [bufferPrime] = transformOperations( + this.buffer.operation, + remoteForPending, + bufferBeforeRemote, + ); + this.buffer.operation = bufferPrime; + } + } + + this.serverContent = nextServer.content; + this.serverOwnerMap = nextServer.ownerMap; + this.revisionId = remote.revisionId; + return { duplicate: false, ownAck, remote: !ownAck }; + } + + resynchronize(message) { + const canonicalContent = String(message?.content || ""); + const canonicalOwnerMap = message?.owner_map == null ? "[]" : String(message.owner_map); + const canonicalRevisionId = Number(message?.revision_id) || 0; + const outstanding = this.outstanding; + const buffer = this.buffer; + const local = this.localDocument(); + const acknowledgedIds = (message?.acknowledged_update_ids || []) + .map(Number) + .filter(Number.isFinite); + const acknowledgedThrough = acknowledgedIds.length ? Math.max(...acknowledgedIds) : 0; + const messageClientId = String(message?.client_id ?? message?.clientId ?? ""); + const messageUpdateId = Number(message?.update_id ?? message?.updateId ?? 0); + const outstandingAcknowledged = Boolean( + outstanding + && ( + acknowledgedThrough >= outstanding.updateId + || (messageClientId === this.clientId && messageUpdateId === outstanding.updateId) + ) + ); + const pendingOwnerReplacements = mergeOwnerReplacements( + outstanding?.ownerReplacements || [], + buffer?.ownerReplacements || [], + ); + let replayOperation = null; + let replayOwnerReplacements = []; + + if (this.ready && (outstanding || buffer)) { + const canonicalAuthorship = parseAuthorship(canonicalContent, canonicalOwnerMap); + if (outstandingAcknowledged) { + if (buffer) { + const afterOutstanding = applyOperationToDocument( + this.serverContent, + this.serverOwnerMap, + outstanding.operation, + outstanding.ownerReplacements, + ); + const missedRemote = operationFromEdit( + afterOutstanding.content, + canonicalContent, + canonicalAuthorship, + ); + [replayOperation] = transformOperations(buffer.operation, missedRemote, false); + replayOwnerReplacements = buffer.ownerReplacements; + } + } else { + const localAuthorship = parseAuthorship(local.content, local.ownerMap); + const localOperation = operationFromEdit(this.serverContent, local.content, localAuthorship); + const missedRemote = operationFromEdit( + this.serverContent, + canonicalContent, + canonicalAuthorship, + ); + [replayOperation] = transformOperations(localOperation, missedRemote, false); + replayOwnerReplacements = pendingOwnerReplacements; + } + } + + this.initialize(canonicalContent, canonicalOwnerMap, canonicalRevisionId); + if (replayOperation || replayOwnerReplacements.length) { + this.queue( + replayOperation || identityOperation(canonicalContent.length), + replayOwnerReplacements, + ); + } + return { + replayed: Boolean(this.buffer), + outstandingAcknowledged, + }; + } + + adoptCanonicalSnapshot(content, ownerMap, revisionId) { + revisionId = Number(revisionId) || 0; + if (revisionId !== this.revisionId) { + throw new CollaborationRevisionGapError(this.revisionId, revisionId); + } + if (String(content || "") !== this.serverContent) { + throw new Error("Canonical collaborative content does not match applied operations"); + } + this.serverOwnerMap = ownerMap == null ? "[]" : String(ownerMap); + } +} diff --git a/static/js/collaboration.js b/static/js/collaboration.js new file mode 100644 index 0000000..0fd0657 --- /dev/null +++ b/static/js/collaboration.js @@ -0,0 +1,331 @@ +/* + * Copyright (C) 2026 Mateusz Gruszczynski @linuxiarz.pl + * Source-Available Code / Dual-Licensed. + * + * Free for non-commercial and evaluation use under terms of BSL/GPLv3. + * Commercial or production use requires a valid paid license. + * See LICENSE file in repository root for details. + */ + +import { parseAuthorship, serializeAuthorship } from "@rustpad/authorship"; + +function utf16Length(value) { + return String(value || "").length; +} + +function normalizeOwnerSpans(spans, length) { + const result = []; + const sorted = [...(Array.isArray(spans) ? spans : [])].sort((left, right) => (Number(left?.start) || 0) - (Number(right?.start) || 0) || (Number(left?.end) || 0) - (Number(right?.end) || 0)); + for (const source of sorted) { + const start = Math.max(0, Math.min(length, Number(source?.start) || 0)); + const end = Math.max(start, Math.min(length, Number(source?.end) || 0)); + const owner = String(source?.owner || ""); + if (!owner || end <= start) continue; + const previous = result.at(-1); + if (previous && previous.owner === owner && start <= previous.end) { + previous.end = Math.max(previous.end, end); + continue; + } + const clippedStart = previous && start < previous.end ? previous.end : start; + if (end > clippedStart) result.push({ start: clippedStart, end, owner }); + } + return result; +} + +function sliceOwnerSpans(spans, start, length) { + const end = start + length; + return normalizeOwnerSpans((spans || []).flatMap(span => { + const overlapStart = Math.max(start, span.start); + const overlapEnd = Math.min(end, span.end); + return overlapEnd > overlapStart + ? [{ start: overlapStart - start, end: overlapEnd - start, owner: span.owner }] + : []; + }), length); +} + +function shiftOwnerSpans(spans, offset) { + return (spans || []).map(span => ({ start: span.start + offset, end: span.end + offset, owner: span.owner })); +} + +function appendComponent(components, component) { + if (!component) return; + if (component.kind === "retain" || component.kind === "delete") { + const count = Number(component.count) || 0; + if (count <= 0) return; + const previous = components.at(-1); + if (previous?.kind === component.kind) previous.count += count; + else components.push({ kind: component.kind, count }); + return; + } + if (component.kind !== "insert") throw new Error("Unknown operation component"); + const text = String(component.text || ""); + if (!text) return; + const owners = normalizeOwnerSpans(component.owners, text.length); + const previous = components.at(-1); + if (previous?.kind === "insert") { + const offset = previous.text.length; + previous.text += text; + previous.owners = normalizeOwnerSpans([ + ...(previous.owners || []), + ...shiftOwnerSpans(owners, offset), + ], previous.text.length); + } else components.push({ kind: "insert", text, owners }); +} + +export function normalizeOperation(operation) { + const components = []; + for (const component of operation?.components || []) appendComponent(components, component); + return { components }; +} + +export function operationBaseLength(operation) { + return normalizeOperation(operation).components.reduce((length, component) => + length + (component.kind === "retain" || component.kind === "delete" ? component.count : 0), 0); +} + +export function operationTargetLength(operation) { + return normalizeOperation(operation).components.reduce((length, component) => + length + (component.kind === "retain" ? component.count : component.kind === "insert" ? component.text.length : 0), 0); +} + +export function identityOperation(length) { + return normalizeOperation({ components: length > 0 ? [{ kind: "retain", count: length }] : [] }); +} + +function isUtf16Boundary(value, offset) { + if (offset <= 0 || offset >= value.length) return true; + const previous = value.charCodeAt(offset - 1); + const next = value.charCodeAt(offset); + return !(previous >= 0xd800 && previous <= 0xdbff && next >= 0xdc00 && next <= 0xdfff); +} + +export function operationFromEdit(previousText, nextText, nextAuthorship) { + previousText = String(previousText || ""); + nextText = String(nextText || ""); + let prefix = 0; + const shared = Math.min(previousText.length, nextText.length); + while (prefix < shared && previousText.charCodeAt(prefix) === nextText.charCodeAt(prefix)) prefix++; + while (prefix > 0 && (!isUtf16Boundary(previousText, prefix) || !isUtf16Boundary(nextText, prefix))) prefix--; + + let oldSuffix = previousText.length; + let newSuffix = nextText.length; + while (oldSuffix > prefix && newSuffix > prefix && previousText.charCodeAt(oldSuffix - 1) === nextText.charCodeAt(newSuffix - 1)) { + oldSuffix--; + newSuffix--; + } + while (!isUtf16Boundary(previousText, oldSuffix) || !isUtf16Boundary(nextText, newSuffix)) { + oldSuffix++; + newSuffix++; + } + + const components = []; + appendComponent(components, { kind: "retain", count: prefix }); + appendComponent(components, { kind: "delete", count: oldSuffix - prefix }); + const insertedText = nextText.slice(prefix, newSuffix); + appendComponent(components, { + kind: "insert", + text: insertedText, + owners: sliceOwnerSpans(nextAuthorship?.spans || [], prefix, insertedText.length), + }); + appendComponent(components, { kind: "retain", count: previousText.length - oldSuffix }); + return { components }; +} + +class OperationCursor { + constructor(operation) { + this.components = normalizeOperation(operation).components; + this.index = 0; + this.offset = 0; + } + + get current() { return this.components[this.index] || null; } + get kind() { return this.current?.kind || null; } + get remaining() { + const component = this.current; + if (!component) return 0; + return (component.kind === "insert" ? component.text.length : component.count) - this.offset; + } + + take(count = this.remaining) { + const component = this.current; + if (!component || count <= 0 || count > this.remaining) throw new Error("Invalid operation cursor read"); + let part; + if (component.kind === "insert") { + part = { + kind: "insert", + text: component.text.slice(this.offset, this.offset + count), + owners: sliceOwnerSpans(component.owners, this.offset, count), + }; + } else part = { kind: component.kind, count }; + this.offset += count; + if (this.offset === (component.kind === "insert" ? component.text.length : component.count)) { + this.index++; + this.offset = 0; + } + return part; + } +} + +export function composeOperations(first, second) { + first = normalizeOperation(first); + second = normalizeOperation(second); + if (operationTargetLength(first) !== operationBaseLength(second)) throw new Error("Cannot compose operations with different lengths"); + const left = new OperationCursor(first); + const right = new OperationCursor(second); + const components = []; + + while (left.current || right.current) { + if (right.kind === "insert") { + appendComponent(components, right.take()); + continue; + } + if (left.kind === "delete") { + appendComponent(components, left.take()); + continue; + } + if (!left.current || !right.current) throw new Error("Incomplete operation composition"); + const count = Math.min(left.remaining, right.remaining); + const leftKind = left.kind; + const rightKind = right.kind; + if (leftKind === "retain" && rightKind === "retain") { + appendComponent(components, { kind: "retain", count }); + left.take(count); + right.take(count); + } else if (leftKind === "retain" && rightKind === "delete") { + appendComponent(components, { kind: "delete", count }); + left.take(count); + right.take(count); + } else if (leftKind === "insert" && rightKind === "retain") { + appendComponent(components, left.take(count)); + right.take(count); + } else if (leftKind === "insert" && rightKind === "delete") { + left.take(count); + right.take(count); + } else throw new Error("Unsupported operation composition"); + } + return { components }; +} + +export function transformOperations(leftOperation, rightOperation, leftBeforeRight = true) { + leftOperation = normalizeOperation(leftOperation); + rightOperation = normalizeOperation(rightOperation); + if (operationBaseLength(leftOperation) !== operationBaseLength(rightOperation)) throw new Error("Cannot transform operations with different base lengths"); + const left = new OperationCursor(leftOperation); + const right = new OperationCursor(rightOperation); + const leftPrime = []; + const rightPrime = []; + + while (left.current || right.current) { + if (left.kind === "insert" && (right.kind !== "insert" || leftBeforeRight)) { + const part = left.take(); + appendComponent(leftPrime, part); + appendComponent(rightPrime, { kind: "retain", count: part.text.length }); + continue; + } + if (right.kind === "insert") { + const part = right.take(); + appendComponent(leftPrime, { kind: "retain", count: part.text.length }); + appendComponent(rightPrime, part); + continue; + } + if (!left.current || !right.current) throw new Error("Incomplete operation transform"); + const count = Math.min(left.remaining, right.remaining); + if (left.kind === "retain" && right.kind === "retain") { + appendComponent(leftPrime, { kind: "retain", count }); + appendComponent(rightPrime, { kind: "retain", count }); + } else if (left.kind === "delete" && right.kind === "retain") { + appendComponent(leftPrime, { kind: "delete", count }); + } else if (left.kind === "retain" && right.kind === "delete") { + appendComponent(rightPrime, { kind: "delete", count }); + } else if (left.kind !== "delete" || right.kind !== "delete") throw new Error("Unsupported operation transform"); + left.take(count); + right.take(count); + } + return [{ components: leftPrime }, { components: rightPrime }]; +} + +export function applyOperation(text, operation) { + text = String(text || ""); + operation = normalizeOperation(operation); + if (operationBaseLength(operation) !== text.length) throw new Error("Operation base length does not match document"); + let cursor = 0; + let result = ""; + for (const component of operation.components) { + if (component.kind === "retain") { + result += text.slice(cursor, cursor + component.count); + cursor += component.count; + } else if (component.kind === "delete") cursor += component.count; + else result += component.text; + } + if (cursor !== text.length) throw new Error("Operation did not consume the whole document"); + return result; +} + +function copyRetainedSpans(target, spans, sourceStart, length, outputStart) { + const sourceEnd = sourceStart + length; + for (const span of spans || []) { + const start = Math.max(sourceStart, span.start); + const end = Math.min(sourceEnd, span.end); + if (end > start) target.push({ start: outputStart + start - sourceStart, end: outputStart + end - sourceStart, owner: span.owner }); + } +} + +export function applyOperationToDocument(content, ownerMap, operation, ownerReplacements = []) { + content = String(content || ""); + operation = normalizeOperation(operation); + if (operationBaseLength(operation) !== content.length) throw new Error("Operation base length does not match document"); + const sourceModel = parseAuthorship(content, ownerMap); + const outputSpans = []; + let sourceOffset = 0; + let outputOffset = 0; + let nextContent = ""; + for (const component of operation.components) { + if (component.kind === "retain") { + nextContent += content.slice(sourceOffset, sourceOffset + component.count); + copyRetainedSpans(outputSpans, sourceModel.spans, sourceOffset, component.count, outputOffset); + sourceOffset += component.count; + outputOffset += component.count; + } else if (component.kind === "delete") sourceOffset += component.count; + else { + nextContent += component.text; + outputSpans.push(...shiftOwnerSpans(component.owners, outputOffset)); + outputOffset += component.text.length; + } + } + if (sourceOffset !== content.length) throw new Error("Operation did not consume the whole document"); + const replacementMap = new Map((ownerReplacements || []).map(item => [String(item?.owner || ""), String(item?.replacement || "")])); + const replacedSpans = outputSpans.map(span => { + const identity = String(span.owner || "").split("\u001f", 1)[0]; + const replacement = replacementMap.get(identity); + return replacement ? { ...span, owner: replacement } : span; + }); + const model = { version: 2, spans: normalizeOwnerSpans(replacedSpans, nextContent.length) }; + return { content: nextContent, ownerMap: serializeAuthorship(model, nextContent.length), authorship: model }; +} + +export function operationEquals(left, right) { + return JSON.stringify(normalizeOperation(left)) === JSON.stringify(normalizeOperation(right)); +} + +export function compareOperationKeys(left, right) { + const leftClient = String(left?.clientId || left?.client_id || ""); + const rightClient = String(right?.clientId || right?.client_id || ""); + if (leftClient !== rightClient) return leftClient < rightClient ? -1 : 1; + const leftUpdate = Number(left?.updateId ?? left?.update_id ?? 0); + const rightUpdate = Number(right?.updateId ?? right?.update_id ?? 0); + return leftUpdate === rightUpdate ? 0 : leftUpdate < rightUpdate ? -1 : 1; +} + +export function documentAfterPending(serverContent, serverOwnerMap, outstanding, buffer) { + let documentState = { content: serverContent, ownerMap: serverOwnerMap }; + for (const pending of [outstanding, buffer]) { + if (!pending) continue; + documentState = applyOperationToDocument( + documentState.content, + documentState.ownerMap, + pending.operation, + pending.ownerReplacements, + ); + } + return documentState; +} diff --git a/static/js/note-editor.js b/static/js/note-editor.js index ad0ecbe..63ed7c7 100644 --- a/static/js/note-editor.js +++ b/static/js/note-editor.js @@ -11,6 +11,8 @@ import { installGlobalDiagnostics, logInfo } from "@rustpad/logger"; installGlobalDiagnostics(); import { applyAuthorshipEdit, authorshipOwners, lineAuthors, mapSelectionThroughEdit, parseAuthorship, renderAuthorshipLayer, replaceAuthorshipOwner, serializeAuthorship, syncAuthorshipLayer } from "@rustpad/authorship"; +import { identityOperation, operationFromEdit } from "@rustpad/collaboration"; +import { CollaborationRevisionGapError, CollaborationSession } from "@rustpad/collaboration-session"; import { copyText } from "@rustpad/clipboard"; import { lineFromHash, lineLink, lineStartOffset } from "@rustpad/line-links"; import { applyFormat, bindFormatShortcuts, bindIndentationShortcuts } from "@rustpad/editor-format"; @@ -37,7 +39,11 @@ export function startNoteEditor(adapter) { const mobileFontFamily = document.querySelector("#mobile-font-family"), mobileFontSize = document.querySelector("#mobile-font-size"), mobileLineToggle = document.querySelector("#mobile-line-numbers-toggle"), mobilePreviewLineToggle = document.querySelector("#mobile-preview-line-numbers-toggle"), mobileCompactToggle = document.querySelector("#mobile-compact-toggle"), mobileLineLinksToggle = document.querySelector("#mobile-line-links-toggle"); const shareToken = new URLSearchParams(location.search).get("share"); const notePreferenceKey = name => `rustpad:${name}:${location.pathname}`; - let accessToken = shareToken || getAccessToken(adapter.access.kind, adapter.access.key), password = "", nickname = getNickname(), info, socket, saveTimer, applyingRemote = false, applyingHistory = false, resourceUnlocked = false, uiState = readEditorState(), authorship = parseAuthorship("", "[]"), previousContent = "", lastServerContent = "", lastServerOwnerMap = "[]", globalColor = "", noteColor = "", presenceUsers = [], authorshipMode = "simple", authorshipColorsEnabled = true, lastRevealedLineHash = ""; + const collaborationClientId = typeof crypto.randomUUID === "function" + ? crypto.randomUUID().replaceAll("-", "") + : `${Date.now().toString(36)}${Math.random().toString(36).slice(2)}`; + const collaboration = new CollaborationSession(collaborationClientId); + let accessToken = shareToken || getAccessToken(adapter.access.kind, adapter.access.key), password = "", nickname = getNickname(), info, socket, saveTimer, applyingRemote = false, applyingHistory = false, resourceUnlocked = false, uiState = readEditorState(), authorship = parseAuthorship("", "[]"), previousContent = "", globalColor = "", noteColor = "", presenceUsers = [], authorshipMode = "simple", authorshipColorsEnabled = true, lastRevealedLineHash = "", flushRequested = false; let editorSettingsSaveTimer, editorSettingsSaveInFlight = false, pendingPersonalSettingsSave = false, pendingAuthorshipSettingsSave = false, connectionNoticeTimer = 0, connectionWasInterrupted = false; const editHistory = { entries: [], @@ -971,6 +977,53 @@ export function startNoteEditor(adapter) { } function renderNow() { if (uiState.mode === "markdown") { preview.classList.remove("preview--raw"); preview.innerHTML = renderMarkdown(editor.value); scheduleAliasFileRefresh(editor.value); document.querySelector("#preview-label").textContent = "Preview (mermaid / markdown)"; renderMermaid(); renderCodeHighlight(); } else { preview.classList.add("preview--raw"); preview.innerHTML = editor.value.split("\n").map((line, index) => `
${escapeHtml(line) || "
"}
`).join(""); document.querySelector("#preview-label").textContent = "Text preview"; } alignPreviewLineNumbers(preview); document.querySelector("#characters").textContent = `${editor.value.length} characters`; document.querySelector("#words").textContent = `${editor.value.trim() ? editor.value.trim().split(/\s+/).length : 0} words`; renderGutter(); requestAnimationFrame(syncPreviewScroll); } const render = createRenderQueue(renderNow); + function queueCollaborativeOperation(operation, ownerReplacements = []) { + if (!collaboration.ready || !canEditDocument()) return false; + try { + collaboration.queue(operation, ownerReplacements); + return true; + } catch (error) { + console.error("Failed to queue collaborative operation", error); + saveState.textContent = "Synchronization error"; + socket?.stop(); + queueMicrotask(connect); + return false; + } + } + + function flushCollaborativeUpdate() { + clearTimeout(saveTimer); + if (!canEditDocument()) { + saveState.textContent = "Read only"; + return; + } + const pending = collaboration.sendable(); + if (!pending) { + if (collaboration.outstanding) { + flushRequested = Boolean(collaboration.buffer); + saveState.textContent = "Saving…"; + } else if (!collaboration.buffer) { + flushRequested = false; + saveState.textContent = "Changes are saved automatically"; + } + return; + } + const sent = socket?.update( + pending.baseRevisionId, + pending.updateId, + pending.operation, + pending.ownerReplacements, + ); + if (!sent) { + flushRequested = true; + saveState.textContent = "Waiting for connection…"; + return; + } + collaboration.markSent(pending.updateId); + flushRequested = false; + saveState.textContent = "Saving…"; + } + function scheduleDocumentSave() { clearTimeout(saveTimer); if (!canEditDocument()) { @@ -978,10 +1031,19 @@ export function startNoteEditor(adapter) { return; } saveState.textContent = "Saving…"; - saveTimer = setTimeout(() => socket?.update(editor.value, serializeAuthorship(authorship, editor.value.length)), 250); + saveTimer = setTimeout(flushCollaborativeUpdate, 250); } + + function queueOwnerReplacement(owner, replacement) { + if (!owner || !replacement) return; + if (queueCollaborativeOperation(identityOperation(editor.value.length), [{ owner, replacement }])) { + scheduleDocumentSave(); + } + } + function restoreHistorySnapshot(snapshot) { if (!snapshot || !canEditDocument()) return; + const previous = editor.value; const maxOffset = snapshot.content.length; const selectionStart = Math.min(snapshot.selectionStart ?? maxOffset, maxOffset); const selectionEnd = Math.min(snapshot.selectionEnd ?? selectionStart, maxOffset); @@ -996,6 +1058,7 @@ export function startNoteEditor(adapter) { syncEditorLayers(); applyingHistory = false; editor.focus({ preventScroll: true }); + queueCollaborativeOperation(operationFromEdit(previous, snapshot.content, authorship)); scheduleDocumentSave(); } function activeView() { @@ -1044,9 +1107,7 @@ export function startNoteEditor(adapter) { if (snapshot.raw) editRawPreviewLine(target); } - function applyRemote(content, ownerMap) { - lastServerContent = content; - if (ownerMap != null) lastServerOwnerMap = ownerMap; + function applyEditorDocument(content, ownerMap, { resetHistory = true } = {}) { if (content === editor.value) { if (ownerMap != null) authorship = adoptCurrentOwnerAliases(parseAuthorship(content, ownerMap), content.length); previousContent = content; @@ -1066,7 +1127,8 @@ export function startNoteEditor(adapter) { editor.scrollTop = scrollTop; editor.scrollLeft = scrollLeft; applyingRemote = false; - editHistory.reset(); + if (resetHistory) editHistory.reset(); + else editHistory.syncCurrent(); render(); editor.scrollTop = scrollTop; editor.scrollLeft = scrollLeft; @@ -1075,6 +1137,108 @@ export function startNoteEditor(adapter) { requestAnimationFrame(revealLinkedLine); } + function applyCollaborativeView({ resetHistory = false } = {}) { + const local = collaboration.localDocument(); + applyEditorDocument(local.content, local.ownerMap, { resetHistory }); + } + + function recoverCollaborativeSnapshot(message, reason = "resync") { + try { + const result = collaboration.resynchronize(message); + if (result.replayed) { + flushRequested = true; + toast(reason === "resync" ? "Connection state was resynchronized; pending edits were merged." : "A missed update was merged with your local edits."); + } + } catch (error) { + console.error("Failed to transform pending changes during resynchronization", error); + const localContent = editor.value; + const serverContent = String(message?.content || ""); + const serverOwnerMap = message?.owner_map == null ? "[]" : String(message.owner_map); + const revisionId = Number(message?.revision_id) || 0; + const separator = serverContent && !serverContent.endsWith("\n") ? "\n\n" : ""; + const recoveryContent = `${serverContent}${separator}\n${localContent}`; + collaboration.initialize(serverContent, serverOwnerMap, revisionId); + if (new TextEncoder().encode(recoveryContent).length <= 2_000_000) { + collaboration.queue( + operationFromEdit(serverContent, recoveryContent, parseAuthorship(recoveryContent, "[]")), + ); + flushRequested = true; + toast("A synchronization conflict was preserved as a local recovery block."); + } else { + toast("Synchronization failed because the recoverable document exceeds the size limit."); + } + } + applyCollaborativeView({ resetHistory: true }); + if (collaboration.buffer) scheduleDocumentSave(); + } + + + function integrateCollaborativeEnvelope(message, { renderView = true } = {}) { + const result = collaboration.integrate(message); + if (result.duplicate) return result; + if (message.content != null && String(message.content) !== collaboration.serverContent) { + throw new Error("Server snapshot does not match the collaborative operation"); + } + if (message.owner_map != null) collaboration.serverOwnerMap = String(message.owner_map); + if (renderView) applyCollaborativeView({ resetHistory: result.remote }); + return result; + } + + function handleCollaborativeAuthentication(message) { + const revisionId = Number(message.revision_id) || 0; + if (!collaboration.ready) { + collaboration.initialize(message.content, message.owner_map, revisionId); + applyCollaborativeView({ resetHistory: true }); + return; + } + if (message.resync_required) { + recoverCollaborativeSnapshot(message, "resync"); + return; + } + let resetHistory = false; + try { + for (const operation of message.catchup_operations || []) { + const result = integrateCollaborativeEnvelope(operation, { renderView: false }); + resetHistory ||= result.remote; + } + collaboration.adoptCanonicalSnapshot(message.content, message.owner_map, revisionId); + applyCollaborativeView({ resetHistory }); + } catch (error) { + console.error("Failed to apply collaborative catch-up", error); + recoverCollaborativeSnapshot(message, error instanceof CollaborationRevisionGapError ? "gap" : "resync"); + } + } + + function resendOutstandingUpdate() { + const pending = collaboration.outstanding; + if (!pending || !canEditDocument()) return; + const sent = socket?.update( + collaboration.revisionId, + pending.updateId, + pending.operation, + pending.ownerReplacements, + ); + if (sent) saveState.textContent = "Saving…"; + else flushRequested = true; + } + + function handleCollaborativeDocument(message) { + try { + const result = integrateCollaborativeEnvelope(message); + if (result.duplicate) return; + const timestamp = new Date(message.updated_at).toLocaleTimeString("en-US", { hour: "2-digit", minute: "2-digit" }); + if (collaboration.hasPending()) saveState.textContent = "Saving…"; + else saveState.textContent = editor.readOnly ? "Read only" : `${message.author ? `${message.author} · ` : ""}${timestamp}`; + if (result.ownAck && collaboration.buffer && flushRequested) flushCollaborativeUpdate(); + } catch (error) { + console.error("Failed to integrate collaborative update", error); + saveState.textContent = "Resynchronizing…"; + socket?.stop(); + queueMicrotask(connect); + } + } + + const { loadFiles } = bindNoteFiles({ editor, toast, getAccessToken: () => accessToken, canDelete: () => Boolean(info?.can_delete_files), @@ -1139,6 +1303,8 @@ export function startNoteEditor(adapter) { color: currentUserColor() || null, sessionToken: null, guestId: getGuestId(), + clientId: collaborationClientId, + getKnownRevision: () => collaboration.ready ? collaboration.revisionId : null, onStatus: handleSocketStatus, onAuthenticated: message => { resourceUnlocked = true; @@ -1146,14 +1312,21 @@ export function startNoteEditor(adapter) { const readOnly = message.access_level === "read_only"; setDocumentReadOnly(readOnly); accessLevel.textContent = readOnly ? "Access: read only" : "Access: full"; - applyRemote(message.content, message.owner_map); - saveState.textContent = readOnly ? "Read only" : "Changes are saved automatically"; - if (!readOnly) editor.focus(); - }, - onDocument: message => { - applyRemote(message.content, message.owner_map); - saveState.textContent = editor.readOnly ? "Read only" : `${message.author ? `${message.author} · ` : ""}${new Date(message.updated_at).toLocaleTimeString("en-US", { hour: "2-digit", minute: "2-digit" })}`; + if (readOnly) { + collaboration.initialize(message.content, message.owner_map, message.revision_id); + applyCollaborativeView({ resetHistory: true }); + flushRequested = false; + saveState.textContent = "Read only"; + return; + } + handleCollaborativeAuthentication(message); + if (collaboration.outstanding) resendOutstandingUpdate(); + else if (collaboration.buffer) flushCollaborativeUpdate(); + else saveState.textContent = "Changes are saved automatically"; + editor.focus(); }, + onDocument: handleCollaborativeDocument, + onResync: message => recoverCollaborativeSnapshot(message, "resync"), onPresence: updatePresence, onLatency: updateLatency, onDiagnostics: renderConnectionDiagnostics, @@ -1166,7 +1339,9 @@ export function startNoteEditor(adapter) { toast(friendly); accessLevel.textContent = "Access: read only"; setDocumentReadOnly(true, "Read only — changes not saved"); - applyRemote(lastServerContent, lastServerOwnerMap); + collaboration.initialize(collaboration.serverContent, collaboration.serverOwnerMap, collaboration.revisionId); + applyCollaborativeView({ resetHistory: true }); + flushRequested = false; queueMicrotask(connect); return; } @@ -1725,7 +1900,7 @@ export function startNoteEditor(adapter) { authorship = replaceAuthorshipOwner(authorship, owner => ownerName(owner) === nickname, replacement, editor.value.length); updateCurrentUser(); render(); socket?.setColor(noteColor); - if (socket && canEditDocument()) socket.update(editor.value, serializeAuthorship(authorship, editor.value.length)); + if (socket && canEditDocument()) queueOwnerReplacement(nickname, replacement); } userColorPicker.addEventListener("change", () => saveUserColor(userColorPicker.value)); mobileColorPicker?.addEventListener("change", () => saveUserColor(mobileColorPicker.value)); @@ -1739,7 +1914,7 @@ export function startNoteEditor(adapter) { authorship = replaceAuthorshipOwner(authorship, owner => ownerName(owner) === nickname, replacement, editor.value.length); updateCurrentUser(); render(); socket?.setColor(currentUserColor() || null); - if (socket && canEditDocument()) socket.update(editor.value, serializeAuthorship(authorship, editor.value.length)); + if (socket && canEditDocument()) queueOwnerReplacement(nickname, replacement); toast("Global profile color restored"); }); editor.addEventListener("keydown", continueIndentation); @@ -1749,17 +1924,19 @@ export function startNoteEditor(adapter) { }); editor.addEventListener("input", event => { if (!canEditDocument() && !applyingRemote) { - applyRemote(lastServerContent, lastServerOwnerMap); + applyCollaborativeView({ resetHistory: true }); saveState.textContent = "Read only"; return; } + const previous = previousContent; const nextContent = editor.value; - authorship = adoptCurrentOwnerAliases(authorship, previousContent.length); - authorship = replaceAuthorshipOwner(authorship, owner => ownerName(owner) === nickname, currentOwner(), previousContent.length); - authorship = applyAuthorshipEdit(authorship, previousContent, nextContent, currentOwner()); + authorship = adoptCurrentOwnerAliases(authorship, previous.length); + authorship = replaceAuthorshipOwner(authorship, owner => ownerName(owner) === nickname, currentOwner(), previous.length); + authorship = applyAuthorshipEdit(authorship, previous, nextContent, currentOwner()); previousContent = nextContent; render(); if (applyingRemote || applyingHistory) return; + queueCollaborativeOperation(operationFromEdit(previous, nextContent, authorship)); editHistory.record(event.inputType || ""); scheduleDocumentSave(); }); diff --git a/static/js/socket.js b/static/js/socket.js index ea68a5f..3d2b4c1 100644 --- a/static/js/socket.js +++ b/static/js/socket.js @@ -97,6 +97,8 @@ class RoomSocket { guest_id: this.guestId || null, color: this.color || null, diagnostics: this.clientDiagnostics(), + client_id: this.clientId || null, + known_revision_id: this.getKnownRevision?.() ?? null, }); this.emitDiagnostics(); }); @@ -130,6 +132,7 @@ class RoomSocket { return; } if (message.type === "document") this.onDocument?.(message); + if (message.type === "resync") this.onResync?.(message); if (message.type === "presence") this.onPresence?.(message.users || []); if (message.type === "chat") this.onChat?.(message); if (message.type === "pong") { @@ -302,7 +305,15 @@ class RoomSocket { return true; } - update(content, ownerMap = "[]") { this.send({ type: "update", content, owner_map: ownerMap }); } + update(baseRevisionId, updateId, operation, ownerReplacements = []) { + return this.send({ + type: "update", + base_revision_id: baseRevisionId, + update_id: updateId, + operation, + owner_replacements: ownerReplacements, + }); + } chat(text) { this.send({ type: "chat", text }); } setColor(color) { this.color = color || null; this.send({ type: "set_color", color: this.color }); }