From 9619a892d50896692f4078f8f3ef279dc8bb149d Mon Sep 17 00:00:00 2001 From: SirBlobby Date: Tue, 21 Jul 2026 12:44:19 -0400 Subject: [PATCH] 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 | 35 +++++ server/src/devices.rs | 256 +++++++++++++++++++++++++++++++ server/src/main.rs | 16 +- server/src/projects.rs | 21 ++- src/routes/settings/+page.svelte | 133 ++++++++++++++++ 5 files changed, 454 insertions(+), 7 deletions(-) create mode 100644 server/src/devices.rs diff --git a/server/src/desktop.rs b/server/src/desktop.rs index 7c48bad..1aa4811 100644 --- a/server/src/desktop.rs +++ b/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) } diff --git a/server/src/devices.rs b/server/src/devices.rs new file mode 100644 index 0000000..6a8a9c2 --- /dev/null +++ b/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 { + 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 { + 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, + pub document_id: Option, +} + +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, +} + +pub async fn ws_handler( + State(state): State, + headers: HeaderMap, + ws: WebSocketUpgrade, +) -> Result { + 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, + pub connected: bool, + pub connected_since: Option, +} + +pub async fn list_devices( + State(state): State, + jar: SignedCookieJar, +) -> Result>, (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)>( + "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, + jar: SignedCookieJar, + Path(id): Path, +) -> Result { + 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) +} diff --git a/server/src/main.rs b/server/src/main.rs index bbefcd5..88f6fe9 100644 --- a/server/src/main.rs +++ b/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>>, + pub device_events: Arc>>>, } impl axum::extract::FromRef 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)); diff --git a/server/src/projects.rs b/server/src/projects.rs index e695c73..1ad0f94 100644 --- a/server/src/projects.rs +++ b/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, ) -> Result, (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, (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, ) -> Result { 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 { 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) } diff --git a/src/routes/settings/+page.svelte b/src/routes/settings/+page.svelte index 6e5dcf3..9a55ed1 100644 --- a/src/routes/settings/+page.svelte +++ b/src/routes/settings/+page.svelte @@ -25,6 +25,15 @@ rate_limit: number; }; + type Device = { + id: string; + name: string; + created_at: string; + last_used_at: string | null; + connected: boolean; + connected_since: string | null; + }; + let activeSection = $state('account'); let username = $state(''); @@ -62,6 +71,12 @@ let confirmRegenerateId = $state(null); let regeneratingKeyId = $state(null); + let devices = $state([]); + let devicesLoading = $state(false); + let devicesError = $state(''); + let confirmRevokeDeviceId = $state(null); + let revokingDeviceId = $state(null); + type UsagePoint = { date: string; count: number }; type UsagePeriod = '1hr' | '1day' | '1week'; let usageData = $state([]); @@ -161,6 +176,34 @@ confirmDeleteKeyId = null; } + async function loadDevices() { + devicesLoading = true; + devicesError = ''; + try { + const res = await fetch('/api/devices'); + if (res.ok) { + devices = await res.json(); + } else { + devicesError = 'Failed to load devices.'; + } + } catch { + devicesError = 'Network error.'; + } + devicesLoading = false; + } + + async function revokeDevice(id: string) { + revokingDeviceId = id; + try { + const res = await fetch(`/api/devices/${id}`, { method: 'DELETE' }); + if (res.ok) { + devices = devices.filter(d => d.id !== id); + } + } catch {} + revokingDeviceId = null; + confirmRevokeDeviceId = null; + } + async function copyKey(key: string) { await navigator.clipboard.writeText(key); copiedKey = true; @@ -300,6 +343,12 @@ } }); + $effect(() => { + if (activeSection === 'devices') { + loadDevices(); + } + }); + async function toggleAdmin(user: AdminUser) { const res = await fetch(`/api/admin/users/${user.id}`, { method: 'PATCH', @@ -408,6 +457,7 @@ { id: 'theme', label: 'Theme', icon: 'mdi:palette-outline' }, { id: 'storage', label: 'Storage', icon: 'mdi:harddisk' }, { id: 'api-keys', label: 'API Keys', icon: 'mdi:key-outline' }, + { id: 'devices', label: 'Devices', icon: 'mdi:devices' }, ...($userStore?.is_admin ? [{ id: 'admin', label: 'Admin', icon: 'mdi:shield-crown-outline' }] : []) ]); @@ -811,6 +861,89 @@ {/if} + {#if activeSection === 'devices'} +
+
+

+ + Devices +

+

+ Typst Desktop apps signed in to your account. A device stays connected while it is running with live sync; revoke a device to sign it out immediately. +

+ + {#if devicesError} +
{devicesError}
+ {/if} + + {#if devicesLoading} +
+ + Loading devices... +
+ {:else if devices.length === 0} +
+ +

No devices signed in yet.

+
+ {:else} +
+ {#each devices as device (device.id)} +
+
+ +
+
+

+ {device.name} + {#if device.connected} + + + Connected + + {/if} +

+

{device.last_used_at ? `Last used ${formatDate(device.last_used_at)}` : 'Never used'}

+
+ +
+ {#if confirmRevokeDeviceId === device.id} +
+ Revoke? + + +
+ {:else} + + {/if} +
+
+ {/each} +
+ {/if} +
+
+ {/if} + {#if activeSection === 'admin' && $userStore?.is_admin}