From 33fd127cf6c6377580f9ec2faec9a30810e0ef36 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Tue, 8 Apr 2025 22:22:37 +0100 Subject: [PATCH] Take last_seen_vault_update_id as a WS message instead of query parameter --- backend/sync_server/src/server/websocket.rs | 30 +++++++-------------- 1 file changed, 9 insertions(+), 21 deletions(-) diff --git a/backend/sync_server/src/server/websocket.rs b/backend/sync_server/src/server/websocket.rs index 82a37af6..204391bd 100644 --- a/backend/sync_server/src/server/websocket.rs +++ b/backend/sync_server/src/server/websocket.rs @@ -1,7 +1,7 @@ use anyhow::Context; use axum::{ extract::{ - Path, Query, State, + Path, State, ws::{Message, WebSocket, WebSocketUpgrade}, }, response::Response, @@ -21,7 +21,7 @@ use crate::{ database::models::{DeviceId, DocumentVersionWithoutContent, VaultId, VaultUpdateId}, }, errors::{SyncServerError, server_error, unauthenticated_error}, - utils::normalize::{normalize, normalize_string}, + utils::normalize::normalize, }; // This is required for aide to infer the path parameter types and names @@ -31,30 +31,18 @@ pub struct WebsocketPathParams { vault_id: VaultId, } -// This is required for aide to infer the path parameter types and names -#[derive(Deserialize, JsonSchema)] -pub struct QueryParams { - since_update_id: Option, -} - pub async fn websocket_handler( ws: WebSocketUpgrade, Path(WebsocketPathParams { vault_id }): Path, - Query(QueryParams { since_update_id }): Query, State(state): State, ) -> Result { - Ok(ws.on_upgrade(move |socket| websocket_wrapped(state, socket, vault_id, since_update_id))) + Ok(ws.on_upgrade(move |socket| websocket_wrapped(state, socket, vault_id))) } -async fn websocket_wrapped( - state: AppState, - stream: WebSocket, - vault_id: VaultId, - since_update_id: Option, -) { +async fn websocket_wrapped(state: AppState, stream: WebSocket, vault_id: VaultId) { info!("Websocket connection opened on vault '{vault_id}'"); - let result = websocket(state, stream, vault_id.clone(), since_update_id).await; + let result = websocket(state, stream, vault_id.clone()).await; if let Err(err) = result { error!("Websocket connection error on vault '{vault_id}': {err}"); @@ -68,13 +56,13 @@ async fn websocket_wrapped( struct WebsocketHandshake { pub token: String, pub device_id: DeviceId, + pub last_seen_vault_update_id: Option, } async fn websocket( state: AppState, stream: WebSocket, vault_id: VaultId, - since_update_id: Option, ) -> Result<(), SyncServerError> { let (mut sender, mut receiver) = stream.split(); @@ -83,7 +71,7 @@ async fn websocket( .context("Failed to parse token") .map_err(server_error)?; - auth(&state, handshake.token.trim(), &normalize_string(&vault_id))?; + auth(&state, handshake.token.trim(), &vault_id)?; handshake } else { @@ -94,10 +82,10 @@ async fn websocket( let mut rx = state.broadcasts.get_receiver(vault_id.clone()).await; - let documents = if let Some(since_update_id) = since_update_id { + let documents = if let Some(update_id) = handshake.last_seen_vault_update_id { state .database - .get_latest_documents_since(&vault_id, since_update_id, None) + .get_latest_documents_since(&vault_id, update_id, None) .await .map_err(server_error) } else {