diff --git a/server/src/desktop.rs b/server/src/desktop.rs index 1aa4811..d6a51a6 100644 --- a/server/src/desktop.rs +++ b/server/src/desktop.rs @@ -61,7 +61,7 @@ fn content_hash(bytes: &[u8]) -> String { format!("{:x}", Sha256::digest(bytes)) } -fn hash_token(token: &str) -> String { +pub(crate) fn hash_token(token: &str) -> String { format!("{:x}", Sha256::digest(token.as_bytes())) } @@ -110,6 +110,26 @@ pub async fn authenticate( Ok(user_id) } +pub(crate) async fn user_id_for_token(state: &AppState, token: &str) -> Option { + let row = sqlx::query_as::<_, (String, String)>( + "SELECT id, user_id FROM device_tokens WHERE token_hash = ?", + ) + .bind(hash_token(token)) + .fetch_optional(&state.db) + .await + .ok()??; + + let (token_id, user_id) = row; + + let _ = sqlx::query("UPDATE device_tokens SET last_used_at = ? WHERE id = ?") + .bind(chrono::Utc::now().to_rfc3339()) + .bind(&token_id) + .execute(&state.db) + .await; + + Some(user_id) +} + async fn owned_project( state: &AppState, project_id: &str, @@ -409,6 +429,7 @@ pub async fn delete_project( #[derive(Serialize)] pub struct ManifestEntry { + pub id: String, pub path: String, pub kind: String, pub hash: String, @@ -428,9 +449,9 @@ pub struct ProjectManifest { async fn plain_contents( state: &AppState, project_id: &str, -) -> Result, String)>, (StatusCode, String)> { - let rows = sqlx::query_as::<_, (String, String, Option>, Option)>( - "SELECT path, kind, content, updated_at FROM project_files WHERE project_id = ? ORDER BY path ASC", +) -> Result, String)>, (StatusCode, String)> { + let rows = sqlx::query_as::<_, (String, String, String, Option>, Option)>( + "SELECT id, path, kind, content, updated_at FROM project_files WHERE project_id = ? ORDER BY path ASC", ) .bind(project_id) .fetch_all(&state.db) @@ -439,14 +460,14 @@ async fn plain_contents( Ok(rows .into_iter() - .map(|(path, kind, content, updated_at)| { + .map(|(id, path, kind, content, updated_at)| { let raw = content.unwrap_or_default(); let plain = if kind == "binary" { raw } else { decode_text_blob(&raw).into_bytes() }; - (path, kind, plain, updated_at.unwrap_or_default()) + (id, path, kind, plain, updated_at.unwrap_or_default()) }) .collect()) } @@ -462,7 +483,8 @@ pub async fn get_manifest( let files = plain_contents(&state, &project_id) .await? .into_iter() - .map(|(path, kind, plain, updated_at)| ManifestEntry { + .map(|(id, path, kind, plain, updated_at)| ManifestEntry { + id, path, kind, hash: content_hash(&plain), @@ -740,7 +762,7 @@ pub async fn pull_project( let files = plain_contents(&state, &project_id) .await? .into_iter() - .map(|(path, kind, plain, _)| { + .map(|(_, path, kind, plain, _)| { let hash = content_hash(&plain); let (encoding, content) = encode_for_transport(&kind, plain); BundleFile { diff --git a/server/src/handlers.rs b/server/src/handlers.rs index 5cbba95..53cb1bb 100644 --- a/server/src/handlers.rs +++ b/server/src/handlers.rs @@ -1,10 +1,11 @@ use axum::{ - extract::{Path, State, Multipart}, + extract::{Path, Query, State, Multipart}, http::{header, StatusCode}, response::IntoResponse, Json, }; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; use std::sync::Arc; use tokio::sync::{Mutex, RwLock}; use yrs_axum::ws::AxumSink; @@ -14,6 +15,7 @@ use yrs::{Doc, ReadTxn, Transact, Update}; use yrs::updates::decoder::Decode; use futures_util::stream::{StreamExt, Stream}; use crate::AppState; +use crate::devices::{notify_devices, DeviceEvent}; use crate::models::Document; pub struct ViewerFilterStream { @@ -89,22 +91,35 @@ pub struct Diagnostic { pub to: Option, } +struct YjsSaveTarget { + table: &'static str, + row_id: String, + owner_id: String, + event: DeviceEvent, +} + pub async fn yjs_handler( ws: axum::extract::ws::WebSocketUpgrade, Path(id): Path, + Query(params): Query>, State(state): State, jar: axum_extra::extract::cookie::SignedCookieJar, ) -> impl IntoResponse { - let user_id_opt = jar.get("session_user_id").map(|c| c.value().to_string()); + let user_id_opt = match jar.get("session_user_id").map(|c| c.value().to_string()) { + Some(uid) => Some(uid), + None => match params.get("token") { + Some(token) => crate::desktop::user_id_for_token(&state, token).await, + None => None, + }, + }; let mut is_viewer = true; let mut initial_content: Option> = None; - // (table, row_id) the autosave task persists into; None means no persistence. - let mut save_target: Option<(&'static str, String)> = None; + let mut save_target: Option = None; if let Some(rest) = id.strip_prefix("project:") { if let Some((project_id, file_id)) = rest.split_once(':') { - if let Some((_project, role)) = crate::projects::project_role(&state, project_id, &user_id_opt).await { + if let Some((project, role)) = crate::projects::project_role(&state, project_id, &user_id_opt).await { is_viewer = role == "viewer"; if let Ok(Some((content,))) = sqlx::query_as::<_, (Option>,)>( "SELECT content FROM project_files WHERE id = ? AND project_id = ?" @@ -116,7 +131,12 @@ pub async fn yjs_handler( { initial_content = content; } - save_target = Some(("project_files", file_id.to_string())); + save_target = Some(YjsSaveTarget { + table: "project_files", + row_id: file_id.to_string(), + owner_id: project.owner_id, + event: DeviceEvent::project(project_id), + }); } } } else { @@ -148,8 +168,13 @@ pub async fn yjs_handler( } } initial_content = d.content.clone(); + save_target = Some(YjsSaveTarget { + table: "documents", + row_id: id.clone(), + owner_id: d.owner_id.clone(), + event: DeviceEvent::document(&id), + }); } - save_target = Some(("documents", id.clone())); } let mut bcast_map = state.bcast_map.lock().await; @@ -168,25 +193,38 @@ pub async fn yjs_handler( let new_bcast = Arc::new(BroadcastGroup::new(awareness.clone(), 10).await); bcast_map.insert(id.clone(), new_bcast.clone()); - if let Some((table, row_id)) = save_target { + if let Some(target) = save_target { let save_db = state.db.clone(); let save_awareness = awareness.clone(); + let save_state = state.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); + let mut last_content: Option> = None; loop { interval.tick().await; let doc = save_awareness.read().await; let content = doc.doc().transact().encode_state_as_update_v1(&yrs::StateVector::default()); - let query = if table == "project_files" { + drop(doc); + + if last_content.as_ref() == Some(&content) { + continue; + } + + let query = if target.table == "project_files" { "UPDATE project_files SET content = ? WHERE id = ?" } else { "UPDATE documents SET content = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?" }; - let _ = sqlx::query(query) - .bind(content) - .bind(&row_id) + let result = sqlx::query(query) + .bind(&content) + .bind(&target.row_id) .execute(&save_db) .await; + + if result.is_ok() { + last_content = Some(content); + notify_devices(&save_state, &target.owner_id, target.event.clone()).await; + } } }); }