feat(realtime): replace polling with a Rust-owned WebSocket connection

MessageList polled every three seconds and ChatPage bumped a refreshKey after
each send. Both are gone; messages now arrive over a socket.

The connection lives in the Tauri Rust layer rather than in the WebView. Two
reasons: the handshake can carry a real Authorization header, which is the only
channel that nginx, Apache and Caddy do not write to their access logs by
default, and tungstenite surfaces the full HTTP response on a rejected
handshake. The browser API is required by spec to collapse every failure into
close code 1006, so "token expired" and "server down" would be
indistinguishable — which matters once the client holds connections to several
servers and has to show a meaningful state for each.

The socket only carries {conversation, seq}. Rust does the backfill over REST
and hands resolved entities to the frontend, so the UI keeps no cursor logic:
it replaces by id and sorts by seq.

The client now supplies the message id. Without it an optimistically rendered
message cannot be matched against the copy that arrives over the realtime
channel and shows up twice; it also makes a retry after an unclear network
error a no-op instead of a duplicate.

icons/icon.ico is generated from the existing icon.png — tauri-build requires
it on Windows, and without it the crate does not build there at all.

Not yet done, marked as TODO in the code: token refresh (the connection ends
at expiry with 4001), the server list replacing BASE_URL, optimistic rendering,
and the DM-versus-channel rule for background fetching.

src-tauri/Cargo.toml

@@ -16,3 +16,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 reqwest = { version = "0.12", features = ["json"] }
 tokio = { version = "1", features = ["full"] }
+uuid = { version = "1.24.0", features = ["v4"] }
+futures-util = "0.3.34"
+tokio-tungstenite = { version = "0.30.0", features = ["rustls-tls-webpki-roots"] }
+rand = "0.10.2"

src-tauri/src/commands/message.rs

@@ -1,19 +1,33 @@
 use crate::models::Message;
 use crate::state::{AppState, BASE_URL};
 
+#[derive(serde::Deserialize)]
+struct History {
+    cursor: i64,
+    messages: Vec<Message>,
+}
+
 #[tauri::command]
 pub async fn list_messages(
     state: tauri::State<'_, AppState>,
     conversation_id: String,
 ) -> Result<Vec<Message>, String> {
-    state
+    let history: History = state
-        .authed_get(&format!("{}/conversations/{}/messages", BASE_URL, conversation_id))
+        .authed_get(&format!(
+            "{}/conversations/{}/messages",
+            BASE_URL, conversation_id
+        ))
         .send()
         .await
         .map_err(|e| e.to_string())?
         .json()
         .await
-        .map_err(|e| e.to_string())
+        .map_err(|e| e.to_string())?;
+
+    // Loading the history *is* a resync: adopt the cursor it came with, so the
+    // next realtime signal asks for a delta from the right place.
+    state.cursors.set(&conversation_id, history.cursor);
+    Ok(history.messages)
 }
 
 #[tauri::command]
@@ -29,7 +43,12 @@ pub async fn send_message(
         .clone()
         .ok_or_else(|| "Not logged in".to_string())?;
 
+    // The id comes from here, not from the server. Without it the optimistically
+    // rendered message cannot be matched against the copy that arrives over the
+    // realtime channel, and the same message shows up twice. It also makes a
+    // retry after an unclear network error a no-op instead of a duplicate.
     let body = serde_json::json!({
+        "id": uuid::Uuid::new_v4().to_string(),
         "conversation_id": conversation_id,
         "user_id": user_id,
         "content": content,

src-tauri/src/commands/user.rs

@@ -2,7 +2,11 @@ use crate::models::{AuthTokenResponse, User};
 use crate::state::{AppState, BASE_URL};
 
 #[tauri::command]
-pub async fn login(state: tauri::State<'_, AppState>, email: String) -> Result<User, String> {
+pub async fn login(
+    app: tauri::AppHandle,
+    state: tauri::State<'_, AppState>,
+    email: String,
+) -> Result<User, String> {
     let users: Vec<User> = state
         .client
         .get(format!("{}/users", BASE_URL))
@@ -38,5 +42,7 @@ pub async fn login(state: tauri::State<'_, AppState>, email: String) -> Result<U
     *state.token.lock().unwrap() = Some(token_resp.token);
     *state.user_id.lock().unwrap() = Some(user.id.clone());
 
+    crate::realtime::spawn(app);
+
     Ok(user)
 }

src-tauri/src/lib.rs

@@ -1,5 +1,6 @@
 mod commands;
 mod models;
+mod realtime;
 mod state;
 
 use state::AppState;

src-tauri/src/models.rs

@@ -28,12 +28,16 @@ pub struct Conversation {
     pub created_by: String,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct Message {
     pub id: String,
     pub conversation_id: String,
     pub user_id: String,
     pub content: String,
+    /// Sort key within the conversation. Display order follows this, never
+    /// `created_at` — that is transaction start time on the server, so it is
+    /// neither unique nor commit-ordered.
+    pub seq: i64,
     pub created_at: String,
 }
 

src-tauri/src/realtime.rs

@@ -0,0 +1,219 @@
+//! The realtime connection, owned by Rust rather than by the WebView.
+//!
+//! Two reasons it lives here. The handshake can carry a real
+//! `Authorization: Bearer` header — the only channel that proxy defaults do not
+//! write to their access logs — and `tungstenite` surfaces the full HTTP
+//! response on a rejected handshake, whereas the browser API is required by spec
+//! to collapse every failure into close code 1006. "Token expired" and "server
+//! down" would be indistinguishable in the WebView.
+//!
+//! The socket only ever says "conversation X is at seq N". This module then
+//! pulls the actual entities over REST and hands them to the frontend, so the
+//! frontend keeps no cursor logic at all.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use serde::{Deserialize, Serialize};
+use tauri::{AppHandle, Emitter, Manager};
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tokio_tungstenite::tungstenite::Message;
+
+use crate::models::Message as ChatMessage;
+use crate::state::{AppState, BASE_URL};
+
+pub const EVENT_CHANGED: &str = "realtime://changed";
+pub const EVENT_RESYNC: &str = "realtime://resync";
+// TODO: a connection-state event per server once the UI shows it — ticket 10
+// wants "connected / reconnecting / signed out" visible per server. Left out
+// until something listens, rather than emitting into the void.
+
+#[derive(Debug, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+enum Inbound {
+    Changed { conversation: String, seq: i64 },
+    Resync,
+}
+
+#[derive(Debug, Serialize, Clone)]
+pub struct Changed {
+    pub conversation_id: String,
+    pub messages: Vec<ChatMessage>,
+}
+
+#[derive(Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+enum Changes {
+    Delta {
+        cursor: i64,
+        messages: Vec<ChatMessage>,
+    },
+    Resync {
+        cursor: i64,
+    },
+}
+
+/// Cursor per conversation. Lives next to the connection and the token, so the
+/// frontend never has to know it exists.
+#[derive(Default)]
+pub struct Cursors(Mutex<HashMap<String, i64>>);
+
+impl Cursors {
+    fn get(&self, conversation: &str) -> i64 {
+        *self.0.lock().unwrap().get(conversation).unwrap_or(&0)
+    }
+    pub fn set(&self, conversation: &str, seq: i64) {
+        self.0.lock().unwrap().insert(conversation.to_string(), seq);
+    }
+    fn clear(&self) {
+        self.0.lock().unwrap().clear();
+    }
+}
+
+/// Close codes the server may send. 4002 and 4003 are terminal: reconnecting
+/// after being displaced would just displace somebody else, and two devices
+/// would push each other around in circles.
+const CLOSE_TOKEN_EXPIRED: u16 = 4001;
+const CLOSE_DISPLACED: u16 = 4002;
+const CLOSE_UNAUTHORIZED: u16 = 4003;
+
+pub fn spawn(app: AppHandle) {
+    tauri::async_runtime::spawn(async move {
+        // TODO: one connection per configured server once the server list
+        // replaces BASE_URL. The structure below is per-connection already; what
+        // is missing is the list and the per-server auth UI.
+        let mut attempt: u32 = 0;
+        loop {
+            match connect_once(&app).await {
+                Ok(terminal) => {
+                    if terminal {
+                        return;
+                    }
+                    attempt = 0;
+                }
+                Err(e) => eprintln!("realtime: {e}"),
+            }
+
+            // Full jitter. The random part is not optional: without it every
+            // client comes back at the same instant after a server restart.
+            let ceiling = (500u64 << attempt.min(6)).min(30_000);
+            let wait = rand::random_range(0..=ceiling);
+            tokio::time::sleep(Duration::from_millis(wait)).await;
+            attempt = attempt.saturating_add(1);
+        }
+    });
+}
+
+/// Returns Ok(true) when the server told us to stay down.
+async fn connect_once(app: &AppHandle) -> Result<bool, String> {
+    let state = app.state::<AppState>();
+    let token = state
+        .token
+        .lock()
+        .unwrap()
+        .clone()
+        .ok_or_else(|| "not signed in".to_string())?;
+
+    let url = BASE_URL.replacen("http", "ws", 1) + "/ws";
+    let mut request = url.into_client_request().map_err(|e| e.to_string())?;
+    request.headers_mut().insert(
+        "Authorization",
+        format!("Bearer {token}")
+            .parse()
+            .map_err(|_| "bad token".to_string())?,
+    );
+
+    let (socket, _) = tokio_tungstenite::connect_async(request)
+        .await
+        .map_err(|e| e.to_string())?;
+    let (mut sink, mut stream) = socket.split();
+
+    // The client drives the heartbeat: it sits behind the NAT whose path needs
+    // keeping open, and it can slow down on battery without the server caring.
+    let mut ping = tokio::time::interval(Duration::from_secs(15));
+
+    loop {
+        tokio::select! {
+            _ = ping.tick() => {
+                if sink.send(Message::Ping(Vec::new().into())).await.is_err() {
+                    return Ok(false);
+                }
+            }
+            received = stream.next() => {
+                let Some(received) = received else { return Ok(false) };
+                let message = received.map_err(|e| e.to_string())?;
+                match message {
+                    Message::Text(text) => match serde_json::from_str::<Inbound>(&text) {
+                        Ok(Inbound::Changed { conversation, seq }) => {
+                            // The server already collapses bursts per connection,
+                            // so there is nothing left to debounce here.
+                            if seq > state.cursors.get(&conversation) {
+                                if let Err(e) = backfill(app, &conversation).await {
+                                    eprintln!("realtime: backfill failed: {e}");
+                                }
+                            }
+                        }
+                        Ok(Inbound::Resync) => {
+                            state.cursors.clear();
+                            let _ = app.emit(EVENT_RESYNC, ());
+                        }
+                        Err(e) => eprintln!("realtime: undecodable frame: {e}"),
+                    },
+                    Message::Close(frame) => {
+                        let code = frame.map(|f| u16::from(f.code)).unwrap_or(0);
+                        // TOKEN_EXPIRED counts as terminal only because there is
+                        // no refresh flow yet — reconnecting with the same dead
+                        // token would just spin against 401.
+                        // TODO: refresh over REST and push the new token back
+                        // through the socket as a `reauth` frame (the server
+                        // already accepts one), then treat 4001 as recoverable.
+                        return Ok(matches!(
+                            code,
+                            CLOSE_DISPLACED | CLOSE_UNAUTHORIZED | CLOSE_TOKEN_EXPIRED
+                        ));
+                    }
+                    _ => {}
+                }
+            }
+        }
+    }
+}
+
+async fn backfill(app: &AppHandle, conversation: &str) -> Result<(), String> {
+    let state = app.state::<AppState>();
+    let since = state.cursors.get(conversation);
+
+    let response: Changes = state
+        .authed_get(&format!(
+            "{BASE_URL}/conversations/{conversation}/changes?since={since}"
+        ))
+        .send()
+        .await
+        .map_err(|e| e.to_string())?
+        .json()
+        .await
+        .map_err(|e| e.to_string())?;
+
+    match response {
+        Changes::Delta { cursor, messages } => {
+            state.cursors.set(conversation, cursor);
+            app.emit(
+                EVENT_CHANGED,
+                Changed {
+                    conversation_id: conversation.to_string(),
+                    messages,
+                },
+            )
+            .map_err(|e| e.to_string())?;
+        }
+        Changes::Resync { cursor } => {
+            // Not an error — this is the normal answer on a first connection or
+            // after a long absence. The frontend drops its cache and reloads.
+            state.cursors.set(conversation, cursor);
+            app.emit(EVENT_RESYNC, ()).map_err(|e| e.to_string())?;
+        }
+    }
+    Ok(())
+}

src-tauri/src/state.rs

@@ -6,6 +6,9 @@ pub struct AppState {
     pub client: reqwest::Client,
     pub token: Mutex<Option<String>>,
     pub user_id: Mutex<Option<String>>,
+    /// Cursor per conversation, kept here rather than in the frontend so the UI
+    /// has no realtime bookkeeping at all.
+    pub cursors: crate::realtime::Cursors,
 }
 
 impl AppState {
@@ -14,6 +17,7 @@ impl AppState {
             client: reqwest::Client::new(),
             token: Mutex::new(None),
             user_id: Mutex::new(None),
+            cursors: Default::default(),
         }
     }
 

src/components/MessageInput.tsx

@@ -3,10 +3,9 @@ import { invoke } from "@tauri-apps/api/core";
 
 interface Props {
   conversationId: string;
-  onSent: () => void;
 }
 
-export default function MessageInput({ conversationId, onSent }: Props) {
+export default function MessageInput({ conversationId }: Props) {
   const [content, setContent] = useState("");
   const [error, setError] = useState<string | null>(null);
 
@@ -16,7 +15,10 @@ export default function MessageInput({ conversationId, onSent }: Props) {
       await invoke("send_message", { conversationId, content });
       setContent("");
       setError(null);
-      onSent();
+      // Nothing to trigger: the message comes back over the realtime channel.
+      // TODO: render it optimistically instead of waiting for the round trip —
+      // the client already supplies the id, which is what makes reconciling it
+      // possible without duplicates.
     } catch (e) {
       setError(String(e));
     }

src/components/MessageList.tsx

@@ -1,42 +1,66 @@
 import { useState, useEffect, useRef } from "react";
 import { invoke } from "@tauri-apps/api/core";
+import { listen } from "@tauri-apps/api/event";
 
 interface Message {
   id: string;
   user_id: string;
   content: string;
+  /** Sort key. Never sort by created_at — see the Rust model for why. */
+  seq: number;
   created_at: string;
 }
 
+interface Changed {
+  conversation_id: string;
+  messages: Message[];
+}
+
 interface Props {
   conversationId: string;
-  refreshKey: number;
 }
 
-export default function MessageList({ conversationId, refreshKey }: Props) {
+export default function MessageList({ conversationId }: Props) {
   const [messages, setMessages] = useState<Message[]>([]);
   const [error, setError] = useState<string | null>(null);
   const bottomRef = useRef<HTMLDivElement>(null);
   const prevLengthRef = useRef(0);
 
-  // Fix 3: cancellation flag prevents stale responses from overwriting new conversation
+  // Full load. Also the resync path: the Rust side adopts the cursor that comes
+  // with the history, so the next realtime signal asks for a delta from the
+  // right place.
+  const load = () =>
+    invoke<Message[]>("list_messages", { conversationId })
+      .then(setMessages)
+      .catch((e) => setError(String(e)));
+
   useEffect(() => {
-    let cancelled = false;
+    setMessages([]);
-    const load = () =>
-      invoke<Message[]>("list_messages", { conversationId })
-        .then((msgs) => {
-          if (!cancelled) setMessages(msgs);
-        })
-        .catch((e) => { if (!cancelled) setError(String(e)); });
-
     load();
-    // ponytail: polling every 3s — replace with WebSocket when server supports it
+    return () => { prevLengthRef.current = 0; };
-    const id = setInterval(load, 3000);
+  }, [conversationId]);
-    return () => { cancelled = true; clearInterval(id); prevLengthRef.current = 0; };
+
-  // Fix 2: refreshKey in deps triggers immediate reload after send
+  // Realtime. Rust owns the connection and the cursor and hands over resolved
-  }, [conversationId, refreshKey]);
+  // entities, so there is nothing to compute here — replace by id, sort by seq.
+  useEffect(() => {
+    const changed = listen<Changed>("realtime://changed", (event) => {
+      if (event.payload.conversation_id !== conversationId) return;
+      setMessages((current) => {
+        const byId = new Map(current.map((m) => [m.id, m]));
+        for (const message of event.payload.messages) byId.set(message.id, message);
+        return [...byId.values()].sort((a, b) => a.seq - b.seq);
+      });
+    });
+
+    // The server could not build a delta — throw the cache away and reload.
+    const resync = listen("realtime://resync", load);
+
+    return () => {
+      changed.then((un) => un());
+      resync.then((un) => un());
+    };
+  }, [conversationId]);
 
-  // Fix 4: scroll only when message count increases, not on every poll
   useEffect(() => {
     if (messages.length > prevLengthRef.current) {
       bottomRef.current?.scrollIntoView({ behavior: "smooth" });

src/pages/ChatPage.tsx

@@ -7,7 +7,6 @@ import MessageInput from "../components/MessageInput";
 export default function ChatPage() {
   const { workspaceId } = useParams<{ workspaceId: string }>();
   const [conversationId, setConversationId] = useState<string | null>(null);
-  const [refreshKey, setRefreshKey] = useState(0);
 
   return (
     <div style={{ display: "flex", height: "100vh" }}>
@@ -19,11 +18,10 @@ export default function ChatPage() {
       <div style={{ flex: 1, display: "flex", flexDirection: "column" }}>
         {conversationId ? (
           <>
-            <MessageList conversationId={conversationId} refreshKey={refreshKey} />
+            <MessageList conversationId={conversationId} />
-            <MessageInput
+            {/* No onSent callback: the sent message arrives back through the
-              conversationId={conversationId}
+                realtime channel like any other. */}
-              onSent={() => setRefreshKey((k) => k + 1)}
+            <MessageInput conversationId={conversationId} />
-            />
           </>
         ) : (
           <div style={{ flex: 1, display: "flex", alignItems: "center", justifyContent: "center", color: "#888" }}>