Add websocket-based device sync and connected devices UI

Replaces polling-only desktop sync with a live push channel: desktop
apps hold a websocket to the server and get notified on project,
document, and folder changes from either the app or the web editor.
Settings now lists connected devices with revoke support.

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

server/src/desktop.rs

@@ -14,6 +14,7 @@ use argon2::{
 };
 
 use crate::{
+    devices::{notify_devices, DeviceEvent},
     models::{Project, User},
     projects::{decode_text_blob, encode_text_blob},
     AppState,
@@ -366,6 +367,8 @@ pub async fn create_project(
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(ProjectSummary {
         id: project.id,
         name: project.name,
@@ -399,6 +402,8 @@ pub async fn delete_project(
         return Err((StatusCode::NOT_FOUND, "Project not found".to_string()));
     }
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -673,6 +678,8 @@ pub async fn push_file(
         .execute(&state.db)
         .await;
 
+    notify_devices(&state, &user_id, DeviceEvent::project(&project_id)).await;
+
     Ok(PushOutcome::Applied(Json(PushFileResponse {
         path: payload.path,
         hash: content_hash(&incoming),
@@ -700,6 +707,8 @@ pub async fn delete_file(
         return Err((StatusCode::NOT_FOUND, "File not found".to_string()));
     }
 
+    notify_devices(&state, &user_id, DeviceEvent::project(&project_id)).await;
+
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -838,6 +847,8 @@ pub async fn create_folder(
         .await
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(CloudFolder {
         id: folder_id,
         name,
@@ -873,6 +884,8 @@ pub async fn rename_folder(
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
     .ok_or((StatusCode::NOT_FOUND, "Folder not found".to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(CloudFolder {
         id,
         name: row.0,
@@ -955,6 +968,8 @@ pub async fn move_folder(
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
     .ok_or((StatusCode::NOT_FOUND, "Folder not found".to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(CloudFolder {
         id,
         name: row.0,
@@ -1012,6 +1027,8 @@ pub async fn delete_folder(
         return Err((StatusCode::NOT_FOUND, "Folder not found or unauthorized".to_string()));
     }
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -1044,6 +1061,8 @@ pub async fn move_project(
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
     .ok_or((StatusCode::NOT_FOUND, "Project not found".to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(ProjectSummary {
         id: row.0,
         name: row.1,
@@ -1289,6 +1308,8 @@ pub async fn push_document(
         .await
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::document(&id)).await;
+
     Ok(PushOutcome::Applied(Json(PushFileResponse {
         path: id,
         hash: incoming_hash,
@@ -1323,6 +1344,8 @@ pub async fn create_document(
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(DocumentContent {
         id: document_id,
         title: payload.title,
@@ -1361,6 +1384,8 @@ pub async fn move_document(
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
     .ok_or((StatusCode::NOT_FOUND, "Document not found".to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(CloudDocument {
         id: row.0,
         title: row.1,
@@ -1388,6 +1413,8 @@ pub async fn delete_document(
         return Err((StatusCode::NOT_FOUND, "Document not found or unauthorized".to_string()));
     }
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -1487,6 +1514,8 @@ pub async fn upload_account_file(
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(CloudFile {
         id: file_id,
         name,
@@ -1525,6 +1554,8 @@ pub async fn rename_account_file(
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
     .ok_or((StatusCode::NOT_FOUND, "File not found".to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(CloudFile {
         id,
         name: row.0,
@@ -1563,6 +1594,8 @@ pub async fn move_account_file(
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
     .ok_or((StatusCode::NOT_FOUND, "File not found".to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(CloudFile {
         id,
         name: row.0,
@@ -1625,5 +1658,7 @@ pub async fn delete_account_file(
         return Err((StatusCode::NOT_FOUND, "File not found or unauthorized".to_string()));
     }
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(StatusCode::NO_CONTENT)
 }

server/src/devices.rs

@@ -0,0 +1,256 @@
+use axum::{
+    extract::{
+        ws::{Message, WebSocket, WebSocketUpgrade},
+        Path, State,
+    },
+    http::{HeaderMap, StatusCode},
+    Json,
+};
+use axum_extra::extract::cookie::SignedCookieJar;
+use futures_util::{SinkExt, StreamExt};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::time::Duration;
+use tokio::sync::{broadcast, watch};
+use uuid::Uuid;
+
+use crate::AppState;
+
+fn hash_token(token: &str) -> String {
+    format!("{:x}", Sha256::digest(token.as_bytes()))
+}
+
+fn bearer_token(headers: &HeaderMap) -> Option<String> {
+    headers
+        .get("Authorization")
+        .and_then(|value| value.to_str().ok())
+        .filter(|value| value.starts_with("Bearer "))
+        .map(|value| value[7..].to_string())
+}
+
+fn get_user_id(jar: &SignedCookieJar) -> Option<String> {
+    jar.get("session_user_id").map(|c| c.value().to_string())
+}
+
+async fn authenticate_device(
+    state: &AppState,
+    headers: &HeaderMap,
+) -> Result<(String, String), (StatusCode, String)> {
+    let token = bearer_token(headers).ok_or((
+        StatusCode::UNAUTHORIZED,
+        "Missing Authorization header".to_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
+    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    let (device_id, user_id) = row.ok_or((
+        StatusCode::UNAUTHORIZED,
+        "Invalid device token".to_string(),
+    ))?;
+
+    Ok((user_id, device_id))
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct DeviceEvent {
+    pub kind: String,
+    pub project_id: Option<String>,
+    pub document_id: Option<String>,
+}
+
+impl DeviceEvent {
+    pub fn project(project_id: &str) -> Self {
+        Self {
+            kind: "project".to_string(),
+            project_id: Some(project_id.to_string()),
+            document_id: None,
+        }
+    }
+
+    pub fn document(document_id: &str) -> Self {
+        Self {
+            kind: "document".to_string(),
+            project_id: None,
+            document_id: Some(document_id.to_string()),
+        }
+    }
+
+    pub fn structure() -> Self {
+        Self {
+            kind: "structure".to_string(),
+            project_id: None,
+            document_id: None,
+        }
+    }
+}
+
+pub async fn notify_devices(state: &AppState, user_id: &str, event: DeviceEvent) {
+    let events = state.device_events.lock().await;
+    if let Some(sender) = events.get(user_id) {
+        let _ = sender.send(event);
+    }
+}
+
+pub struct DevicePresence {
+    pub connected_since: String,
+    pub connection_id: String,
+    pub stop: watch::Sender<bool>,
+}
+
+pub async fn ws_handler(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    ws: WebSocketUpgrade,
+) -> Result<axum::response::Response, (StatusCode, String)> {
+    let (user_id, device_id) = authenticate_device(&state, &headers).await?;
+
+    Ok(ws.on_upgrade(move |socket| handle_socket(state, socket, user_id, device_id)))
+}
+
+async fn handle_socket(state: AppState, socket: WebSocket, user_id: String, device_id: String) {
+    let mut receiver = {
+        let mut events = state.device_events.lock().await;
+        let sender = events
+            .entry(user_id)
+            .or_insert_with(|| broadcast::channel(16).0)
+            .clone();
+        sender.subscribe()
+    };
+
+    let connection_id = Uuid::new_v4().to_string();
+    let (stop_tx, mut stop_rx) = watch::channel(false);
+
+    {
+        let mut presence = state.device_presence.lock().await;
+        presence.insert(
+            device_id.clone(),
+            DevicePresence {
+                connected_since: chrono::Utc::now().to_rfc3339(),
+                connection_id: connection_id.clone(),
+                stop: stop_tx,
+            },
+        );
+    }
+
+    let (mut sink, mut stream) = socket.split();
+    let mut heartbeat = tokio::time::interval(Duration::from_secs(20));
+    heartbeat.tick().await;
+
+    loop {
+        tokio::select! {
+            event = receiver.recv() => {
+                match event {
+                    Ok(event) => {
+                        let Ok(payload) = serde_json::to_string(&event) else { continue };
+                        if sink.send(Message::Text(payload.into())).await.is_err() {
+                            break;
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(broadcast::error::RecvError::Closed) => break,
+                }
+            }
+            _ = heartbeat.tick() => {
+                if sink.send(Message::Ping(Vec::new().into())).await.is_err() {
+                    break;
+                }
+            }
+            incoming = stream.next() => {
+                match incoming {
+                    Some(Ok(Message::Close(_))) | None => break,
+                    Some(Err(_)) => break,
+                    _ => {}
+                }
+            }
+            _ = stop_rx.changed() => {
+                break;
+            }
+        }
+    }
+
+    let mut presence = state.device_presence.lock().await;
+    if presence
+        .get(&device_id)
+        .map(|entry| entry.connection_id == connection_id)
+        .unwrap_or(false)
+    {
+        presence.remove(&device_id);
+    }
+}
+
+#[derive(Serialize)]
+pub struct DeviceView {
+    pub id: String,
+    pub name: String,
+    pub created_at: String,
+    pub last_used_at: Option<String>,
+    pub connected: bool,
+    pub connected_since: Option<String>,
+}
+
+pub async fn list_devices(
+    State(state): State<AppState>,
+    jar: SignedCookieJar,
+) -> Result<Json<Vec<DeviceView>>, (StatusCode, String)> {
+    let user_id =
+        get_user_id(&jar).ok_or((StatusCode::UNAUTHORIZED, "Not authenticated".to_string()))?;
+
+    let rows = sqlx::query_as::<_, (String, String, String, Option<String>)>(
+        "SELECT id, name, created_at, last_used_at FROM device_tokens WHERE user_id = ? ORDER BY created_at DESC",
+    )
+    .bind(&user_id)
+    .fetch_all(&state.db)
+    .await
+    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    let presence = state.device_presence.lock().await;
+
+    Ok(Json(
+        rows.into_iter()
+            .map(|(id, name, created_at, last_used_at)| {
+                let entry = presence.get(&id);
+                DeviceView {
+                    connected: entry.is_some(),
+                    connected_since: entry.map(|e| e.connected_since.clone()),
+                    id,
+                    name,
+                    created_at,
+                    last_used_at,
+                }
+            })
+            .collect(),
+    ))
+}
+
+pub async fn revoke_device(
+    State(state): State<AppState>,
+    jar: SignedCookieJar,
+    Path(id): Path<String>,
+) -> Result<StatusCode, (StatusCode, String)> {
+    let user_id =
+        get_user_id(&jar).ok_or((StatusCode::UNAUTHORIZED, "Not authenticated".to_string()))?;
+
+    let result = sqlx::query("DELETE FROM device_tokens WHERE id = ? AND user_id = ?")
+        .bind(&id)
+        .bind(&user_id)
+        .execute(&state.db)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    if result.rows_affected() == 0 {
+        return Err((StatusCode::NOT_FOUND, "Device not found".to_string()));
+    }
+
+    let presence = state.device_presence.lock().await;
+    if let Some(entry) = presence.get(&id) {
+        let _ = entry.stop.send(true);
+    }
+
+    Ok(StatusCode::NO_CONTENT)
+}

server/src/main.rs

@@ -6,8 +6,10 @@ use axum_extra::extract::cookie::Key;
 use sqlx::AnyPool;
 use std::sync::Arc;
 use std::collections::HashMap;
-use tokio::sync::Mutex;
+use tokio::sync::{broadcast, Mutex};
 use yrs_axum::broadcast::BroadcastGroup;
+
+use devices::{DeviceEvent, DevicePresence};
 use tower_http::services::{ServeDir, ServeFile};
 use tower_http::trace::TraceLayer;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -18,6 +20,7 @@ mod auth;
 mod compiler;
 mod db;
 mod desktop;
+mod devices;
 mod docs;
 mod folders;
 mod files;
@@ -43,6 +46,8 @@ pub struct AppState {
     pub key: Key,
     pub registration_enabled: bool,
     pub rate_limiter: RateLimiterMap,
+    pub device_presence: Arc<Mutex<HashMap<String, DevicePresence>>>,
+    pub device_events: Arc<Mutex<HashMap<String, broadcast::Sender<DeviceEvent>>>>,
 }
 
 impl axum::extract::FromRef<AppState> for Key {
@@ -93,6 +98,8 @@ async fn main() {
         key,
         registration_enabled,
         rate_limiter: Arc::new(Mutex::new(HashMap::new())),
+        device_presence: Arc::new(Mutex::new(HashMap::new())),
+        device_events: Arc::new(Mutex::new(HashMap::new())),
     };
 
     let api_routes = Router::new()
@@ -139,7 +146,9 @@ async fn main() {
         .route("/projects/{id}/files/{fid}", get(projects::get_project_file).patch(projects::update_project_file).delete(projects::delete_project_file))
         .route("/packages", get(packages::list_packages))
         .route("/packages/publish", post(packages::publish_package))
-        .route("/packages/{name}", get(packages::list_versions).delete(packages::delete_package));
+        .route("/packages/{name}", get(packages::list_versions).delete(packages::delete_package))
+        .route("/devices", get(devices::list_devices))
+        .route("/devices/{id}", delete(devices::revoke_device));
 
     let desktop_routes = Router::new()
         .route("/version", get(desktop::version_info))
@@ -158,7 +167,8 @@ async fn main() {
         .route("/files", get(desktop::list_account_files).post(desktop::upload_account_file))
         .route("/files/{id}", get(desktop::pull_account_file).patch(desktop::rename_account_file).delete(desktop::delete_account_file))
         .route("/files/{id}/move", patch(desktop::move_account_file))
-        .route("/projects/{id}/file", get(desktop::pull_file).put(desktop::push_file).delete(desktop::delete_file));
+        .route("/projects/{id}/file", get(desktop::pull_file).put(desktop::push_file).delete(desktop::delete_file))
+        .route("/ws", get(devices::ws_handler));
 
     let v1_routes = Router::new()
         .route("/render", post(public_api::render_handler));

server/src/projects.rs

@@ -13,6 +13,7 @@ use yrs::Update;
 
 use crate::{
     compiler::ProjectInput,
+    devices::{notify_devices, DeviceEvent},
     models::{
         CreateProjectFileRequest, CreateProjectRequest, Project, ProjectFile, UpdateProjectFileRequest,
         UpdateProjectRequest,
@@ -336,6 +337,8 @@ pub async fn update_project(
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(Json(project))
 }
 
@@ -363,6 +366,8 @@ pub async fn delete_project(
         return Err((StatusCode::NOT_FOUND, "Project not found or unauthorized".to_string()));
     }
 
+    notify_devices(&state, &user_id, DeviceEvent::structure()).await;
+
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -394,7 +399,7 @@ pub async fn create_project_file(
     Json(payload): Json<CreateProjectFileRequest>,
 ) -> Result<Json<ProjectFile>, (StatusCode, String)> {
     let user_id_opt = jar.get("session_user_id").map(|c| c.value().to_string());
-    let (_, role) = project_role(&state, &id, &user_id_opt)
+    let (project, role) = project_role(&state, &id, &user_id_opt)
         .await
         .ok_or((StatusCode::UNAUTHORIZED, "Unauthorized".to_string()))?;
     if role == "viewer" {
@@ -417,6 +422,8 @@ pub async fn create_project_file(
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
+    notify_devices(&state, &project.owner_id, DeviceEvent::project(&id)).await;
+
     Ok(Json(file))
 }
 
@@ -427,7 +434,7 @@ pub async fn upload_project_file(
     mut multipart: Multipart,
 ) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
     let user_id_opt = jar.get("session_user_id").map(|c| c.value().to_string());
-    let (_, role) = project_role(&state, &id, &user_id_opt)
+    let (project, role) = project_role(&state, &id, &user_id_opt)
         .await
         .ok_or((StatusCode::UNAUTHORIZED, "Unauthorized".to_string()))?;
     if role == "viewer" {
@@ -465,6 +472,8 @@ pub async fn upload_project_file(
         uploaded.push(path);
     }
 
+    notify_devices(&state, &project.owner_id, DeviceEvent::project(&id)).await;
+
     Ok(Json(serde_json::json!({ "files": uploaded })))
 }
 
@@ -505,7 +514,7 @@ pub async fn update_project_file(
     Json(payload): Json<UpdateProjectFileRequest>,
 ) -> Result<StatusCode, (StatusCode, String)> {
     let user_id_opt = jar.get("session_user_id").map(|c| c.value().to_string());
-    let (_, role) = project_role(&state, &id, &user_id_opt)
+    let (project, role) = project_role(&state, &id, &user_id_opt)
         .await
         .ok_or((StatusCode::UNAUTHORIZED, "Unauthorized".to_string()))?;
     if role == "viewer" {
@@ -524,6 +533,8 @@ pub async fn update_project_file(
         return Err((StatusCode::NOT_FOUND, "File not found".to_string()));
     }
 
+    notify_devices(&state, &project.owner_id, DeviceEvent::project(&id)).await;
+
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -533,7 +544,7 @@ pub async fn delete_project_file(
     jar: SignedCookieJar,
 ) -> Result<StatusCode, (StatusCode, String)> {
     let user_id_opt = jar.get("session_user_id").map(|c| c.value().to_string());
-    let (_, role) = project_role(&state, &id, &user_id_opt)
+    let (project, role) = project_role(&state, &id, &user_id_opt)
         .await
         .ok_or((StatusCode::UNAUTHORIZED, "Unauthorized".to_string()))?;
     if role == "viewer" {
@@ -551,5 +562,7 @@ pub async fn delete_project_file(
         return Err((StatusCode::NOT_FOUND, "File not found".to_string()));
     }
 
+    notify_devices(&state, &project.owner_id, DeviceEvent::project(&id)).await;
+
     Ok(StatusCode::NO_CONTENT)
 }