Connect to server over websocket for live cloud sync

Desktop app now holds a persistent websocket to the server and syncs
on push notifications instead of a fixed timer, falling back to the
old polling interval only when the socket is down. Cloud folder/file
listings are cached locally so the browser stays usable offline.

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

src-tauri/src/db.rs

@@ -19,7 +19,7 @@ pub struct DocumentLink {
     pub synced_at: Option<String>,
 }
 
-const SCHEMA: [&str; 5] = [
+const SCHEMA: [&str; 6] = [
     "CREATE TABLE IF NOT EXISTS settings (
         key TEXT PRIMARY KEY,
         value TEXT NOT NULL
@@ -51,6 +51,11 @@ const SCHEMA: [&str; 5] = [
         data TEXT NOT NULL,
         source_modified INTEGER NOT NULL
     )",
+    "CREATE TABLE IF NOT EXISTS cloud_cache (
+        key TEXT PRIMARY KEY,
+        payload TEXT NOT NULL,
+        cached_at TEXT NOT NULL
+    )",
 ];
 
 const MIGRATIONS: [&str; 2] = [
@@ -416,4 +421,30 @@ impl Store {
             Ok(())
         })
     }
+
+    pub fn cloud_cache(&self, key: &str) -> Result<Option<String>, String> {
+        self.with(|connection| {
+            connection
+                .query_row(
+                    "SELECT payload FROM cloud_cache WHERE key = ?1",
+                    params![key],
+                    |row| row.get(0),
+                )
+                .optional()
+        })
+    }
+
+    pub fn save_cloud_cache(&self, key: &str, payload: &str) -> Result<(), String> {
+        self.with(|connection| {
+            connection.execute(
+                "INSERT INTO cloud_cache (key, payload, cached_at)
+                 VALUES (?1, ?2, ?3)
+                 ON CONFLICT(key) DO UPDATE SET
+                   payload = excluded.payload,
+                   cached_at = excluded.cached_at",
+                params![key, payload, chrono::Utc::now().to_rfc3339()],
+            )?;
+            Ok(())
+        })
+    }
 }

src-tauri/src/lib.rs

@@ -6,6 +6,7 @@ mod sync;
 mod thumbnails;
 mod workspace;
 mod world;
+mod ws;
 
 use base64::{engine::general_purpose::STANDARD as BASE64, Engine as _};
 use serde::{Deserialize, Serialize};
@@ -16,6 +17,7 @@ use assets::Asset;
 use db::Store;
 use compiler::{CompileResult, Diagnostic};
 use lsp::{LspHandle, LspState};
+use ws::WsState;
 use sync::{Account, ProjectSummary, SyncReport};
 use workspace::{
     browse, is_project_dir, is_text_file, list_files, load_settings, project_file_path,
@@ -535,6 +537,16 @@ fn clear_thumbnails(store: State<'_, Store>) -> Result<(), String> {
     store.clear_thumbnails()
 }
 
+#[tauri::command]
+fn get_cloud_cache(store: State<'_, Store>, key: String) -> Result<Option<String>, String> {
+    store.cloud_cache(&key)
+}
+
+#[tauri::command]
+fn save_cloud_cache(store: State<'_, Store>, key: String, payload: String) -> Result<(), String> {
+    store.save_cloud_cache(&key, &payload)
+}
+
 #[tauri::command]
 fn list_assets(app: AppHandle, store: State<'_, Store>) -> Result<Vec<Asset>, String> {
     assets::list_assets(&app, &store)
@@ -703,6 +715,7 @@ fn cloud_check_compatibility(server_url: String) -> sync::CompatibilityStatus {
 fn cloud_login(
     app: AppHandle,
     store: State<'_, Store>,
+    ws_state: State<'_, WsState>,
     server_url: String,
     email: String,
     password: String,
@@ -712,12 +725,14 @@ fn cloud_login(
     let response = sync::login(&server_url, &email, &password, &device_name)?;
 
     let mut settings = load_settings(&app, &store)?;
-    settings.server_url = server_url;
-    settings.device_token = Some(response.token);
+    settings.server_url = server_url.clone();
+    settings.device_token = Some(response.token.clone());
     settings.account_email = Some(response.email.clone());
     settings.account_username = Some(response.username.clone());
     save_settings(&store, &settings)?;
 
+    ws_state.start(app, server_url, response.token);
+
     Ok(Account {
         user_id: response.user_id,
         username: response.username,
@@ -726,7 +741,11 @@ fn cloud_login(
 }
 
 #[tauri::command]
-fn cloud_logout(app: AppHandle, store: State<'_, Store>) -> Result<(), String> {
+fn cloud_logout(
+    app: AppHandle,
+    store: State<'_, Store>,
+    ws_state: State<'_, WsState>,
+) -> Result<(), String> {
     let mut settings = load_settings(&app, &store)?;
     if let Some(token) = &settings.device_token {
         let _ = sync::logout(&settings.server_url, token);
@@ -735,7 +754,31 @@ fn cloud_logout(app: AppHandle, store: State<'_, Store>) -> Result<(), String> {
     settings.device_token = None;
     settings.account_email = None;
     settings.account_username = None;
-    save_settings(&store, &settings)
+    save_settings(&store, &settings)?;
+
+    ws_state.stop(&app);
+
+    Ok(())
+}
+
+#[tauri::command]
+fn cloud_ws_start(app: AppHandle, store: State<'_, Store>, ws_state: State<'_, WsState>) -> Result<(), String> {
+    let settings = load_settings(&app, &store)?;
+    if let Some(token) = settings.device_token {
+        ws_state.start(app, settings.server_url, token);
+    }
+    Ok(())
+}
+
+#[tauri::command]
+fn cloud_ws_stop(app: AppHandle, ws_state: State<'_, WsState>) -> Result<(), String> {
+    ws_state.stop(&app);
+    Ok(())
+}
+
+#[tauri::command]
+fn cloud_ws_status(ws_state: State<'_, WsState>) -> Result<String, String> {
+    Ok(ws_state.status())
 }
 
 #[tauri::command]
@@ -1311,6 +1354,7 @@ pub fn run() {
             Ok(())
         })
         .manage(LspState::default())
+        .manage(WsState::default())
         .on_window_event(|window, event| {
             if matches!(event, tauri::WindowEvent::Destroyed) {
                 if let Some(state) = window.app_handle().try_state::<LspState>() {
@@ -1342,6 +1386,8 @@ pub fn run() {
             thumbnail,
             read_image,
             clear_thumbnails,
+            get_cloud_cache,
+            save_cloud_cache,
             list_assets,
             list_resources,
             list_font_families,
@@ -1357,6 +1403,9 @@ pub fn run() {
             cloud_login,
             cloud_logout,
             cloud_account,
+            cloud_ws_start,
+            cloud_ws_stop,
+            cloud_ws_status,
             cloud_list_projects,
             cloud_list_folders,
             cloud_create_folder,

src-tauri/src/ws.rs

@@ -0,0 +1,156 @@
+use serde::{Deserialize, Serialize};
+use std::net::TcpStream;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Mutex;
+use std::time::Duration;
+use tauri::{AppHandle, Emitter, Manager};
+use tungstenite::client::IntoClientRequest;
+use tungstenite::stream::MaybeTlsStream;
+use tungstenite::Message;
+
+pub const STATUS_EVENT: &str = "cloud://ws-status";
+pub const SYNC_EVENT: &str = "cloud://sync-event";
+
+const READ_TIMEOUT: Duration = Duration::from_secs(10);
+const RECONNECT_DELAY: Duration = Duration::from_secs(5);
+
+#[derive(Deserialize, Serialize, Clone)]
+pub struct DeviceEvent {
+    pub kind: String,
+    pub project_id: Option<String>,
+    pub document_id: Option<String>,
+}
+
+pub struct WsState {
+    generation: AtomicU64,
+    status: Mutex<String>,
+}
+
+impl Default for WsState {
+    fn default() -> Self {
+        Self {
+            generation: AtomicU64::new(0),
+            status: Mutex::new("offline".to_string()),
+        }
+    }
+}
+
+impl WsState {
+    pub fn status(&self) -> String {
+        self.status
+            .lock()
+            .map(|slot| slot.clone())
+            .unwrap_or_else(|_| "offline".to_string())
+    }
+
+    fn set_status(&self, app: &AppHandle, status: &str) {
+        if let Ok(mut slot) = self.status.lock() {
+            *slot = status.to_string();
+        }
+        let _ = app.emit(STATUS_EVENT, status);
+    }
+
+    pub fn start(&self, app: AppHandle, server_url: String, token: String) {
+        let generation = self.generation.fetch_add(1, Ordering::SeqCst) + 1;
+        self.set_status(&app, "connecting");
+
+        std::thread::spawn(move || run_loop(app, server_url, token, generation));
+    }
+
+    pub fn stop(&self, app: &AppHandle) {
+        self.generation.fetch_add(1, Ordering::SeqCst);
+        self.set_status(app, "offline");
+    }
+}
+
+fn still_current(app: &AppHandle, generation: u64) -> bool {
+    app.state::<WsState>().generation.load(Ordering::SeqCst) == generation
+}
+
+fn ws_url(server_url: &str) -> String {
+    let trimmed = server_url.trim_end_matches('/');
+    if let Some(rest) = trimmed.strip_prefix("https://") {
+        format!("wss://{}/api/desktop/ws", rest)
+    } else if let Some(rest) = trimmed.strip_prefix("http://") {
+        format!("ws://{}/api/desktop/ws", rest)
+    } else {
+        format!("ws://{}/api/desktop/ws", trimmed)
+    }
+}
+
+fn configure_read_timeout(stream: &MaybeTlsStream<TcpStream>) {
+    let tcp = match stream {
+        MaybeTlsStream::Plain(stream) => Some(stream),
+        MaybeTlsStream::NativeTls(stream) => Some(stream.get_ref()),
+        _ => None,
+    };
+
+    if let Some(tcp) = tcp {
+        let _ = tcp.set_read_timeout(Some(READ_TIMEOUT));
+    }
+}
+
+fn is_timeout(error: &tungstenite::Error) -> bool {
+    matches!(
+        error,
+        tungstenite::Error::Io(io_error)
+            if io_error.kind() == std::io::ErrorKind::WouldBlock
+                || io_error.kind() == std::io::ErrorKind::TimedOut
+    )
+}
+
+fn connect_and_listen(app: &AppHandle, url: &str, token: &str, generation: u64) -> Result<(), String> {
+    let mut request = url
+        .into_client_request()
+        .map_err(|e| e.to_string())?;
+    let header_value = format!("Bearer {}", token)
+        .parse()
+        .map_err(|_| "Invalid device token".to_string())?;
+    request.headers_mut().insert("Authorization", header_value);
+
+    let (mut socket, _response) = tungstenite::connect(request).map_err(|e| e.to_string())?;
+    configure_read_timeout(socket.get_ref());
+
+    app.state::<WsState>().set_status(app, "connected");
+
+    loop {
+        if !still_current(app, generation) {
+            let _ = socket.close(None);
+            return Ok(());
+        }
+
+        match socket.read() {
+            Ok(Message::Text(text)) => {
+                if let Ok(event) = serde_json::from_str::<DeviceEvent>(text.as_ref()) {
+                    let _ = app.emit(SYNC_EVENT, event);
+                }
+            }
+            Ok(Message::Ping(_)) => {
+                let _ = socket.flush();
+            }
+            Ok(Message::Close(_)) => return Ok(()),
+            Ok(_) => {}
+            Err(ref error) if is_timeout(error) => continue,
+            Err(error) => return Err(error.to_string()),
+        }
+    }
+}
+
+fn run_loop(app: AppHandle, server_url: String, token: String, generation: u64) {
+    let url = ws_url(&server_url);
+
+    loop {
+        if !still_current(&app, generation) {
+            return;
+        }
+
+        let _ = connect_and_listen(&app, &url, &token, generation);
+
+        if !still_current(&app, generation) {
+            return;
+        }
+
+        app.state::<WsState>().set_status(&app, "offline");
+        std::thread::sleep(RECONNECT_DELAY);
+    }
+}