feat(staged): enable web PR polling transport

apps/staged/src-tauri/src/web_server.rs

@@ -23,7 +23,7 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use axum::extract::ws::{Message, WebSocket};
-use axum::extract::{Path, Request, State, WebSocketUpgrade};
+use axum::extract::{Path, Query, Request, State, WebSocketUpgrade};
 use axum::http::StatusCode;
 use axum::middleware::{self, Next};
 use axum::response::{IntoResponse, Json, Response};
@@ -288,15 +288,36 @@ async fn authenticate(
 // WebSocket endpoint — /api/events
 // =============================================================================
 
-async fn ws_events(ws: WebSocketUpgrade, State(state): State<WebAppState>) -> Response {
-    ws.on_upgrade(move |socket| handle_ws(socket, state))
+#[derive(serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct EventsQuery {
+    client_id: Option<String>,
+}
+
+async fn ws_events(
+    ws: WebSocketUpgrade,
+    Query(query): Query<EventsQuery>,
+    State(state): State<WebAppState>,
+) -> Response {
+    let client_id = query
+        .client_id
+        .map(|id| id.trim().to_string())
+        .filter(|id| !id.is_empty());
+    ws.on_upgrade(move |socket| handle_ws(socket, state, client_id))
 }
 
 // clippy's suggested fix (collapsing the inner `if` into a match guard) doesn't
 // compile because `data: Bytes` can't be moved out of the pattern binding into
 // the guard expression.
 #[allow(clippy::collapsible_match)]
-async fn handle_ws(mut socket: WebSocket, state: WebAppState) {
+async fn handle_ws(mut socket: WebSocket, state: WebAppState, client_id: Option<String>) {
+    let pr_scheduler = client_id.as_ref().map(|client_id| {
+        use tauri::Manager;
+        let scheduler = Arc::clone(&state.app_handle.state::<Arc<PrPollScheduler>>());
+        scheduler.touch(client_id.clone());
+        scheduler
+    });
+
     let mut rx = state.event_tx.subscribe();
     loop {
         tokio::select! {
@@ -314,11 +335,30 @@ async fn handle_ws(mut socket: WebSocket, state: WebAppState) {
                     Err(broadcast::error::RecvError::Closed) => break,
                 }
             }
-            // Handle incoming messages (ping/pong, close)
+            // Handle incoming messages (heartbeat, ping/pong, close)
             msg = socket.recv() => {
                 let pong_data = match msg {
                     Some(Ok(Message::Close(_))) | None => break,
-                    Some(Ok(Message::Ping(data))) => data,
+                    Some(Ok(Message::Text(text))) => {
+                        if is_heartbeat_message(text.as_str()) {
+                            if let (Some(scheduler), Some(client_id)) = (&pr_scheduler, &client_id) {
+                                scheduler.touch(client_id.clone());
+                            }
+                        }
+                        continue;
+                    }
+                    Some(Ok(Message::Ping(data))) => {
+                        if let (Some(scheduler), Some(client_id)) = (&pr_scheduler, &client_id) {
+                            scheduler.touch(client_id.clone());
+                        }
+                        data
+                    }
+                    Some(Ok(Message::Pong(_))) => {
+                        if let (Some(scheduler), Some(client_id)) = (&pr_scheduler, &client_id) {
+                            scheduler.touch(client_id.clone());
+                        }
+                        continue;
+                    }
                     _ => continue,
                 };
                 if socket.send(Message::Pong(pong_data)).await.is_err() {
@@ -327,6 +367,26 @@ async fn handle_ws(mut socket: WebSocket, state: WebAppState) {
             }
         }
     }
+
+    if let (Some(scheduler), Some(client_id)) = (pr_scheduler, client_id) {
+        scheduler.disconnect_client(client_id);
+    }
+}
+
+fn is_heartbeat_message(text: &str) -> bool {
+    if text == "heartbeat" {
+        return true;
+    }
+
+    serde_json::from_str::<Value>(text)
+        .ok()
+        .and_then(|value| {
+            value
+                .get("type")
+                .and_then(Value::as_str)
+                .map(|message_type| message_type == "heartbeat")
+        })
+        .unwrap_or(false)
 }
 
 // =============================================================================
@@ -3548,20 +3608,13 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result<Val
 mod tests {
     use std::collections::BTreeSet;
 
-    const INTENTIONALLY_UNSUPPORTED_WEB_COMMANDS: &[&str] = &[];
-
     #[test]
     fn web_dispatch_covers_tauri_commands() {
         let tauri_commands = extract_generate_handler_commands(include_str!("lib.rs"));
         let dispatch_commands = extract_dispatch_commands(include_str!("web_server.rs"));
-        let intentionally_unsupported = INTENTIONALLY_UNSUPPORTED_WEB_COMMANDS
-            .iter()
-            .copied()
-            .collect::<BTreeSet<_>>();
 
         let missing = tauri_commands
             .difference(&dispatch_commands)
-            .filter(|command| !intentionally_unsupported.contains(command.as_str()))
             .cloned()
             .collect::<Vec<_>>();
 

apps/staged/src/lib/services/prPollingService.test.ts

@@ -0,0 +1,94 @@
+// @vitest-environment jsdom
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+describe('prPollingService in web mode', () => {
+  let invokeCommand: ReturnType<typeof vi.fn>;
+  let listenToEvent: ReturnType<typeof vi.fn>;
+  let unlistenRefresh: ReturnType<typeof vi.fn>;
+  let unlistenStale: ReturnType<typeof vi.fn>;
+  let randomUUID: ReturnType<typeof vi.fn>;
+
+  beforeEach(() => {
+    vi.resetModules();
+    invokeCommand = vi.fn().mockResolvedValue(undefined);
+    unlistenRefresh = vi.fn();
+    unlistenStale = vi.fn();
+    listenToEvent = vi.fn().mockReturnValueOnce(unlistenRefresh).mockReturnValueOnce(unlistenStale);
+    randomUUID = vi.fn(() => 'web-client-1');
+
+    vi.stubGlobal('crypto', { randomUUID });
+    vi.spyOn(document, 'hasFocus').mockReturnValue(true);
+    vi.doMock('../transport', () => ({
+      isTauri: false,
+      invokeCommand,
+      listenToEvent,
+    }));
+  });
+
+  afterEach(() => {
+    vi.doUnmock('../transport');
+    vi.restoreAllMocks();
+    vi.unstubAllGlobals();
+  });
+
+  it('keeps a stable browser client id and sends it with every hint', async () => {
+    const service = await import('./prPollingService');
+    const clientId = service.getPrPollClientId();
+
+    expect(clientId).toBe('web-client-1');
+    expect(service.getPrPollClientId()).toBe(clientId);
+    expect(randomUUID).toHaveBeenCalledTimes(1);
+
+    service.init();
+    service.setSelectedProject('project-1');
+    service.updateChecksStatus('branch-1', 'project-1', true);
+    service.refreshNow('project-1');
+    window.dispatchEvent(new Event('blur'));
+    window.dispatchEvent(new Event('focus'));
+    service.dispose();
+
+    expect(listenToEvent.mock.calls.map(([event]) => event)).toEqual([
+      'pr-refresh-state',
+      'pr-status-stale',
+    ]);
+    expect(invokeCommand.mock.calls).toEqual([
+      ['set_focus', { clientId, focused: true }],
+      ['set_foreground_project', { clientId, projectId: 'project-1' }],
+      [
+        'set_branch_pending',
+        { clientId, branchId: 'branch-1', projectId: 'project-1', pending: true },
+      ],
+      ['refresh_now', { clientId, projectId: 'project-1' }],
+      ['set_focus', { clientId, focused: false }],
+      ['set_focus', { clientId, focused: true }],
+      ['disconnect_client', { clientId }],
+    ]);
+    expect(unlistenRefresh).toHaveBeenCalledTimes(1);
+    expect(unlistenStale).toHaveBeenCalledTimes(1);
+  });
+
+  it('replays the current interest hints after a web socket reconnect', async () => {
+    const service = await import('./prPollingService');
+    const clientId = service.getPrPollClientId();
+
+    service.init();
+    service.setSelectedProject('project-1');
+    service.updateChecksStatus('branch-1', 'project-1', true);
+    service.updateChecksStatus('branch-2', 'project-1', true);
+    service.updateChecksStatus('branch-2', 'project-1', false);
+
+    invokeCommand.mockClear();
+    await service.replayPrPollInterestHints();
+
+    expect(invokeCommand.mock.calls).toEqual([
+      ['set_focus', { clientId, focused: true }],
+      ['set_foreground_project', { clientId, projectId: 'project-1' }],
+      [
+        'set_branch_pending',
+        { clientId, branchId: 'branch-1', projectId: 'project-1', pending: true },
+      ],
+    ]);
+
+    service.dispose();
+  });
+});

apps/staged/src/lib/services/prPollingService.ts

@@ -64,6 +64,11 @@ export function getPrPollClientId(): string {
 type StaleCallback = (projectId: string, isStale: boolean) => void;
 type RefreshingCallback = (projectId: string, isRefreshing: boolean) => void;
 
+interface PendingBranchHint {
+  branchId: string;
+  projectId: string;
+}
+
 interface PrRefreshStateEvent {
   projectId: string;
   refreshing: boolean;
@@ -87,6 +92,9 @@ const refreshingCallbacks = new Set<RefreshingCallback>();
 /** Projects the backend currently reports as refreshing. */
 const refreshingProjects = new Set<string>();
 
+let selectedProjectId: string | null = null;
+const pendingBranchHints = new Map<string, PendingBranchHint>();
+
 let initialized = false;
 let unlistenRefreshState: UnlistenFn | null = null;
 let unlistenStale: UnlistenFn | null = null;
@@ -146,10 +154,10 @@ function handleBlur() {
 /**
  * Wire up the interest/hint layer: forward window focus to the backend and
  * subscribe to its refresh/stale lifecycle events. Idempotent; call once at app
- * start. No-op in web mode (the transport is stubbed in this build).
+ * start.
  */
 export function init(): void {
-  if (initialized || !isTauri) return;
+  if (initialized) return;
   initialized = true;
 
   window.addEventListener('focus', handleFocus);
@@ -180,6 +188,8 @@ export function dispose(): void {
 
   // Drop this client's interest so the backend recomputes the union without it.
   void disconnectPrPollClient(clientId).catch(() => {});
+  selectedProjectId = null;
+  pendingBranchHints.clear();
 
   for (const projectId of [...refreshingProjects]) {
     setProjectRefreshing(projectId, false);
@@ -192,7 +202,7 @@ export function dispose(): void {
 
 /** Set the currently selected project (polls more frequently). */
 export function setSelectedProject(projectId: string | null): void {
-  if (!isTauri) return;
+  selectedProjectId = projectId;
   void setForegroundProject(clientId, projectId).catch((e) =>
     console.error('[PrPollingService] set_foreground_project failed:', e)
   );
@@ -204,15 +214,35 @@ export function updateChecksStatus(
   projectId: string,
   hasPendingChecks: boolean
 ): void {
-  if (!isTauri) return;
+  if (hasPendingChecks) {
+    pendingBranchHints.set(branchId, { branchId, projectId });
+  } else {
+    pendingBranchHints.delete(branchId);
+  }
   void setBranchPending(clientId, branchId, projectId, hasPendingChecks).catch((e) =>
     console.error('[PrPollingService] set_branch_pending failed:', e)
   );
 }
 
+/**
+ * Re-send this client's current interest after a browser WebSocket reconnect.
+ * The backend drops all interest on clean WS close, then recreates an empty
+ * client when the same id reconnects.
+ */
+export async function replayPrPollInterestHints(): Promise<void> {
+  if (!initialized) return;
+
+  await Promise.all([
+    setPrPollFocus(clientId, document.hasFocus()),
+    setForegroundProject(clientId, selectedProjectId),
+    ...Array.from(pendingBranchHints.values(), ({ branchId, projectId }) =>
+      setBranchPending(clientId, branchId, projectId, true)
+    ),
+  ]);
+}
+
 /** Trigger an immediate refresh for a specific project (e.g. after PR creation or push). */
 export function refreshNow(projectId: string): void {
-  if (!isTauri) return;
   void refreshPrStatusesNow(clientId, projectId).catch((e) =>
     console.error(`[PrPollingService] refresh_now failed for project=${projectId}:`, e)
   );