Support device-token auth and file ids for realtime sync

Claude-Session: https://claude.ai/code/session_01PoLmSR1pFVyHf8i5b1rNWk

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<String> {
+    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<Vec<(String, String, Vec<u8>, String)>, (StatusCode, String)> {
+) -> Result<Vec<(String, String, String, Vec<u8>, String)>, (StatusCode, String)> {
-    let rows = sqlx::query_as::<_, (String, String, Option<Vec<u8>>, Option<String>)>(
+    let rows = sqlx::query_as::<_, (String, String, String, Option<Vec<u8>>, Option<String>)>(
-        "SELECT path, kind, content, updated_at FROM project_files WHERE project_id = ? ORDER BY path ASC",
+        "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 {

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<usize>,
 }
 
+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<String>,
+    Query(params): Query<HashMap<String, String>>,
     State(state): State<AppState>,
     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<Vec<u8>> = None;
-    // (table, row_id) the autosave task persists into; None means no persistence.
+    let mut save_target: Option<YjsSaveTarget> = None;
-    let mut save_target: Option<(&'static str, String)> = 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<Vec<u8>>,)>(
                     "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<Vec<u8>> = 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)
+                    let result = sqlx::query(query)
-                        .bind(content)
+                        .bind(&content)
-                        .bind(&row_id)
+                        .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;
+                    }
                 }
             });
         }