fix(web-admin): consolidate OAuth SSE onto shared admin hub

Standalone EventSource connections starved oauth-consent-request delivery
through the CF tunnel HTTP/1.1 connection cap. Route OAuth, meta-tool, and
backend event channels through admin-sse-hub and enable SSE at sync start.

Signed-off-by: crimsonsunset <jsangio1@gmail.com>

Author: crimsonsunset

apps/desktop/src/hooks/useDataSync.ts

@@ -16,6 +16,9 @@ export function useDataSync() {
   useEffect(() => {
     async function syncData() {
       console.log('[useDataSync] Starting data sync...');
+      if (!isTauri()) {
+        enableAdminSse();
+      }
       setLoading('spaces', true);
       try {
         // Refresh OAuth tokens first (before connecting servers)
@@ -38,9 +41,6 @@ export function useDataSync() {
         console.error('[useDataSync] Failed to sync:', error);
       } finally {
         setLoading('spaces', false);
-        if (!isTauri()) {
-          enableAdminSse();
-        }
         console.log('[useDataSync] Data sync complete');
       }
     }

apps/desktop/src/lib/backend/events/admin-sse-hub.ts

@@ -37,6 +37,8 @@ const allHandlers = new Set<AllEventsCallback>();
 const lastEventListeners = new Set<(event: { channel: DomainEventChannel; payload: DomainEventPayload }) => void>();
 /** Handlers for non-domain channels (workspace, meta-tool, etc.) sharing the same SSE connection. */
 const rawChannelHandlers = new Map<string, Set<RawChannelHandler>>();
+/** Raw channels that already have a single dispatcher on `sharedSource`. */
+const rawChannelsAttached = new Set<string>();
 
 /**
  * Dispatch an SSE frame to all registered handlers.
@@ -70,18 +72,29 @@ function ensureSharedSource(): void {
     });
   }
 
-  for (const [channel, handlers] of rawChannelHandlers) {
-    source.addEventListener(channel, (event: MessageEvent<string>) => {
-      try {
-        const payload = JSON.parse(event.data) as unknown;
-        handlers.forEach((handler) => handler(payload));
-      } catch {
-        // ignore malformed frames
-      }
-    });
+  for (const channel of rawChannelHandlers.keys()) {
+    attachRawChannelListener(channel);
   }
 }
 
+/**
+ * Attach one SSE listener per raw channel; dispatches to all handlers in the set.
+ */
+function attachRawChannelListener(channel: string): void {
+  if (!sharedSource || rawChannelsAttached.has(channel)) {
+    return;
+  }
+  rawChannelsAttached.add(channel);
+  sharedSource.addEventListener(channel, (event: MessageEvent<string>) => {
+    try {
+      const payload = JSON.parse(event.data) as unknown;
+      rawChannelHandlers.get(channel)?.forEach((handler) => handler(payload));
+    } catch {
+      // ignore malformed frames
+    }
+  });
+}
+
 /**
  * Close the shared SSE connection when the last consumer detaches.
  */
@@ -91,10 +104,11 @@ function releaseSharedSource(): void {
   }
   sharedSource.close();
   sharedSource = null;
+  rawChannelsAttached.clear();
 }
 
 /**
- * Open the shared SSE connection after startup sync (listSpaces) has finished.
+ * Open the shared SSE connection once web admin startup sync begins.
  */
 export function enableAdminSse(): void {
   if (isTauri()) {
@@ -178,16 +192,7 @@ export function subscribeAdminSseRaw(channel: string, handler: RawChannelHandler
     rawChannelHandlers.set(channel, new Set());
   }
   rawChannelHandlers.get(channel)!.add(handler);
-
-  if (sharedSource) {
-    sharedSource.addEventListener(channel, (event: MessageEvent<string>) => {
-      try {
-        handler(JSON.parse(event.data) as unknown);
-      } catch {
-        // ignore malformed frames
-      }
-    });
-  }
+  attachRawChannelListener(channel);
 
   return () => {
     rawChannelHandlers.get(channel)?.delete(handler);

apps/desktop/src/lib/backend/events/use-backend-event-subscription.ts

@@ -2,6 +2,11 @@ import { useEffect } from 'react';
 
 import { isTauri } from '../data/transport';
 
+import {
+  acquireAdminSseConsumer,
+  releaseAdminSseConsumer,
+  subscribeAdminSseRaw,
+} from './admin-sse-hub';
 import { listenWhenTauri } from './tauri-adapter';
 
 /** Options for {@link useBackendEventSubscription}. */
@@ -45,19 +50,19 @@ export function useBackendEventSubscription<T>(
       return;
     }
 
-    const source = new EventSource('/api/v1/events');
-    const onMessage = (event: MessageEvent<string>) => {
+    acquireAdminSseConsumer();
+
+    const unsubscribe = subscribeAdminSseRaw(channel, (payload) => {
       try {
-        callback(JSON.parse(event.data) as T);
+        callback(payload as T);
       } catch {
         // ignore malformed frames
       }
-    };
-    source.addEventListener(channel, onMessage);
+    });
 
     return () => {
-      source.removeEventListener(channel, onMessage);
-      source.close();
+      unsubscribe();
+      releaseAdminSseConsumer();
     };
   }, [channel, callback, sse]);
 }

apps/desktop/src/lib/backend/events/useMetaToolEventsWeb.ts

@@ -8,8 +8,14 @@ import type { MetaToolAuditEvent } from '@/lib/api/metaTools';
 
 import { isTauri } from '../data/transport';
 
+import {
+  acquireAdminSseConsumer,
+  releaseAdminSseConsumer,
+  subscribeAdminSseRaw,
+} from './admin-sse-hub';
+
 /**
- * Subscribe to `meta-tool-invoked` over SSE in web admin mode.
+ * Subscribe to `meta-tool-invoked` over the shared admin SSE hub in web admin mode.
  */
 export function useMetaToolEventsWeb() {
   const handlersRef = useRef<Set<(event: MetaToolAuditEvent) => void>>(new Set());
@@ -18,18 +24,22 @@ export function useMetaToolEventsWeb() {
     if (isTauri()) {
       return;
     }
-    const source = new EventSource('/api/v1/events');
 
-    source.addEventListener('meta-tool-invoked', (event: MessageEvent<string>) => {
+    acquireAdminSseConsumer();
+
+    const unsubscribe = subscribeAdminSseRaw('meta-tool-invoked', (payload) => {
       try {
-        const payload = JSON.parse(event.data) as MetaToolAuditEvent;
-        handlersRef.current.forEach((handler) => handler(payload));
+        const data = payload as MetaToolAuditEvent;
+        handlersRef.current.forEach((handler) => handler(data));
       } catch {
         // ignore malformed frames
       }
     });
 
-    return () => source.close();
+    return () => {
+      unsubscribe();
+      releaseAdminSseConsumer();
+    };
   }, []);
 
   /**

apps/desktop/src/lib/backend/events/useOAuthClientEventsWeb.ts

@@ -6,10 +6,16 @@ import { useCallback, useEffect, useRef } from 'react';
 
 import { isTauri } from '../data/transport';
 
+import {
+  acquireAdminSseConsumer,
+  releaseAdminSseConsumer,
+  subscribeAdminSseRaw,
+} from './admin-sse-hub';
+
 import type { OAuthClientChangedPayload } from './useOAuthClientEvents';
 
 /**
- * Subscribe to `oauth-client-changed` over SSE in web admin mode.
+ * Subscribe to `oauth-client-changed` over the shared admin SSE hub in web admin mode.
  */
 export function useOAuthClientEventsWeb() {
   const handlersRef = useRef<Set<(payload: OAuthClientChangedPayload) => void>>(new Set());
@@ -18,18 +24,22 @@ export function useOAuthClientEventsWeb() {
     if (isTauri()) {
       return;
     }
-    const source = new EventSource('/api/v1/events');
 
-    source.addEventListener('oauth-client-changed', (event: MessageEvent<string>) => {
+    acquireAdminSseConsumer();
+
+    const unsubscribe = subscribeAdminSseRaw('oauth-client-changed', (payload) => {
       try {
-        const payload = JSON.parse(event.data) as OAuthClientChangedPayload;
-        handlersRef.current.forEach((handler) => handler(payload));
+        const data = payload as OAuthClientChangedPayload;
+        handlersRef.current.forEach((handler) => handler(data));
       } catch {
         // ignore malformed frames
       }
     });
 
-    return () => source.close();
+    return () => {
+      unsubscribe();
+      releaseAdminSseConsumer();
+    };
   }, []);
 
   /**

apps/desktop/src/lib/backend/events/useWorkspaceEventsWeb.ts

@@ -2,8 +2,7 @@
  * SSE workspace event channels for web admin mode.
  *
  * Uses the shared admin-sse-hub connection instead of opening a second
- * EventSource, avoiding HTTP/1.1 connection starvation and ensuring workspace
- * events are gated behind enableAdminSse() like all other domain channels.
+ * EventSource, avoiding HTTP/1.1 connection starvation.
  */
 
 import { useCallback, useEffect, useRef } from 'react';

apps/desktop/src/lib/backend/shell/index.ts

@@ -8,6 +8,11 @@ import type { ExportConfigRequest } from '@/lib/api/configExport';
 import type { AdminWebSettings } from '@/lib/api/settings';
 
 import { apiCall, isTauri } from '../data/transport';
+import {
+  acquireAdminSseConsumer,
+  releaseAdminSseConsumer,
+  subscribeAdminSseRaw,
+} from '../events/admin-sse-hub';
 
 export { isTauri };
 export type { Event, UnlistenFn, Update };
@@ -201,27 +206,23 @@ export function subscribeOAuthConsentEvents(
     };
   }
 
-  console.log('[OAuth] subscribeOAuthConsentEvents: using SSE /api/v1/events');
-  const source = new EventSource('/api/v1/events');
-  source.onopen = () => console.log('[OAuth] SSE connected');
-  source.onerror = (err) => console.warn('[OAuth] SSE error:', err);
-  const onConsentRequest = (event: MessageEvent<string>) => {
-    console.log('[OAuth] SSE consent event raw:', event.data);
-    try {
-      const payload = JSON.parse(event.data) as OAuthConsentDeepLinkPayload;
-      if (payload.requestId) {
-        console.log('[OAuth] SSE consent event parsed:', payload);
-        handler(payload);
-      }
-    } catch {
-      console.warn('[OAuth] SSE consent event: malformed payload');
+  console.log('[OAuth] subscribeOAuthConsentEvents: using shared admin SSE hub');
+  acquireAdminSseConsumer();
+
+  const onConsentRequest = (payload: unknown) => {
+    const data = payload as OAuthConsentDeepLinkPayload;
+    if (data.requestId) {
+      console.log('[OAuth] SSE consent event parsed:', data);
+      handler(data);
     }
   };
-  source.addEventListener('oauth-consent-request', onConsentRequest);
+
+  const unsubscribe = subscribeAdminSseRaw('oauth-consent-request', onConsentRequest);
+
   return () => {
-    console.log('[OAuth] Closing SSE consent listener');
-    source.removeEventListener('oauth-consent-request', onConsentRequest);
-    source.close();
+    console.log('[OAuth] Unsubscribing shared SSE consent listener');
+    unsubscribe();
+    releaseAdminSseConsumer();
   };
 }
 

apps/desktop/src/stores/metaToolActivityStore.ts

@@ -12,6 +12,11 @@
 import { create } from 'zustand';
 import { listen, type UnlistenFn } from '@tauri-apps/api/event';
 import type { MetaToolAuditEvent } from '@/lib/api/metaTools';
+import {
+  acquireAdminSseConsumer,
+  releaseAdminSseConsumer,
+  subscribeAdminSseRaw,
+} from '@/lib/backend/events/admin-sse-hub';
 import { isTauri } from '@/lib/backend/data/transport';
 
 /** Ring-buffer size — most recent N invocations kept in memory. */
@@ -38,6 +43,7 @@ export const useMetaToolActivityStore = create<MetaToolActivityState>((set) => (
 // changes) and is wired exactly once regardless of how many callers init it.
 let listening = false;
 let unlistenPromise: Promise<UnlistenFn> | null = null;
+let sseUnsubscribe: (() => void) | null = null;
 
 /**
  * Start the app-wide `meta-tool-invoked` listener (idempotent). Call once near
@@ -55,21 +61,28 @@ export function startMetaToolActivityListener(): void {
     return;
   }
 
-  const source = new EventSource('/api/v1/events');
-  source.addEventListener('meta-tool-invoked', (event: MessageEvent<string>) => {
+  acquireAdminSseConsumer();
+
+  sseUnsubscribe = subscribeAdminSseRaw('meta-tool-invoked', (payload) => {
     try {
-      const payload = JSON.parse(event.data) as MetaToolAuditEvent;
-      useMetaToolActivityStore.getState().push(payload);
+      const data = payload as MetaToolAuditEvent;
+      useMetaToolActivityStore.getState().push(data);
     } catch {
       // ignore malformed frames
     }
   });
-  unlistenPromise = Promise.resolve(() => source.close());
+
+  unlistenPromise = Promise.resolve(() => {
+    sseUnsubscribe?.();
+    sseUnsubscribe = null;
+    releaseAdminSseConsumer();
+  });
 }
 
 /** Tear down the listener (mainly for tests / hot-reload hygiene). */
 export function stopMetaToolActivityListener(): void {
   listening = false;
   void unlistenPromise?.then((fn) => fn()).catch(() => {});
   unlistenPromise = null;
+  sseUnsubscribe = null;
 }