gru/rustpad
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix in db and collaborate

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-08-04 14:22:26 +02:00
1 parent 1a77ccd1bf
commit 8d58549d11
26 files changed
+3009 -298
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/websocket/mod.rs
+291 -52
View File
@@ -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<String>,
#[serde(default)]
diagnostics: Option<ClientDiagnostics>,
#[serde(default)]
client_id: Option<String>,
#[serde(default)]
known_revision_id: Option<i64>,
},
Update {
content: String,
owner_map: Option<String>,
base_revision_id: i64,
update_id: u64,
operation: TextOperation,
#[serde(default)]
owner_replacements: Vec<OwnerReplacement>,
},
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<AppliedOperation>,
acknowledged_update_ids: Vec<u64>,
resync_required: bool,
},
Document {
content: String,
base_revision_id: i64,
revision_id: i64,
updated_at: String,
author: Option<String>,
client_id: String,
update_id: u64,
operation: TextOperation,
owner_replacements: Vec<OwnerReplacement>,
},
Resync {
content: String,
revision_id: i64,
owner_map: String,
acknowledged_update_ids: Vec<u64>,
},
Presence {
users: Vec<PresenceUser>,
@@ -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::<ClientMessage>(&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::<ClientMessage>(&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, &note_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, &note_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, &note_slug);
let channel = state.note_channel(&workspace_slug, &note_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::<ClientMessage>(&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,&note_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<String>) -> Option<String> {
value
.map(|value| value.trim().chars().take(64).collect::<String>())
.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<String>) -> Option<String> {
value
.map(|v| v.trim().chars().take(40).collect::<String>())
src/websocket/pad.rs
+256 -44
View File
@@ -7,14 +7,27 @@ enum PadServerMessage {
title: String,
content: String,
owner_map: String,
revision_id: i64,
access_level: String,
catchup_operations: Vec<AppliedOperation>,
acknowledged_update_ids: Vec<u64>,
resync_required: bool,
},
Document {
content: String,
base_revision_id: i64,
revision_id: i64,
updated_at: String,
author: Option<String>,
client_id: String,
update_id: u64,
operation: TextOperation,
owner_replacements: Vec<OwnerReplacement>,
},
Resync {
content: String,
revision_id: i64,
owner_map: String,
acknowledged_update_ids: Vec<u64>,
},
Presence {
users: Vec<PresenceUser>,
@@ -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::<ClientMessage>(&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::<ClientMessage>(&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::<ClientMessage>(&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,
}
}