fix: reconnect live updates after network blips, token refresh, account switch and app resume

App.tsx

@@ -1,5 +1,5 @@
 import React from 'react';
-import { ActivityIndicator, StyleSheet, Text, View } from 'react-native';
+import { ActivityIndicator, AppState, StyleSheet, Text, View } from 'react-native';
 import { StatusBar } from 'expo-status-bar';
 import { useColorScheme } from 'react-native';
 import { NavigationContainer, createNavigationContainerRef } from '@react-navigation/native';
@@ -7,7 +7,9 @@ import { createBottomTabNavigator } from '@react-navigation/bottom-tabs';
 import { createNativeStackNavigator, type NativeStackScreenProps } from '@react-navigation/native-stack';
 import { Mail, Calendar, BookUser, HardDrive, Settings } from 'lucide-react-native';
 
-import { startPushUpdates } from './src/api/push';
+import { startLiveUpdates, type LiveUpdatesHandle } from './src/api/push-stream';
+import { jmapClient } from './src/api/jmap-client';
+import type { StateChange } from './src/api/types';
 import {
   addMessageListener,
   addNotificationTapListener,
@@ -388,45 +390,124 @@ export default function App() {
     };
   }, [client, isAuthenticated, emailNotificationsEnabled, activeAccountId]);
 
+  // Live updates (SSE with polling fallback), re-armed on every account
+  // switch and every re-established session — the singleton `client` object
+  // never changes identity, so it cannot be the dependency on its own.
+  // Backgrounding closes the stream (no socket + server pings while the app
+  // is asleep); foregrounding reconnects it and refreshes what the user is
+  // looking at, since events that happened in between are gone for good.
   React.useEffect(() => {
-    if (!isAuthenticated || !client) {
+    if (!isAuthenticated || !client || !haveLiveSession) {
       return;
     }
 
     let mounted = true;
-    let cleanup: (() => void) | null = null;
+    let handle: LiveUpdatesHandle | null = null;
+    let appActive = AppState.currentState !== 'background' && AppState.currentState !== 'inactive';
 
-    void (async () => {
+    const onStateChange = async (change: StateChange) => {
+      await Promise.all([
+        useEmailStore.getState().handleStateChange(change),
+        useContactsStore.getState().handleStateChange(change),
+        useCalendarStore.getState().handleStateChange(change),
+      ]);
+    };
+
+    const start = async () => {
       try {
-        const stopPushUpdates = await startPushUpdates(client, {
-          onStateChange: async (change) => {
-            await Promise.all([
-              useEmailStore.getState().handleStateChange(change),
-              useContactsStore.getState().handleStateChange(change),
-              useCalendarStore.getState().handleStateChange(change),
-            ]);
-          },
-          onError: (error) => {
-            console.warn(error.message);
-          },
+        const next = await startLiveUpdates({
+          onStateChange: (change) => { void onStateChange(change); },
+          onError: (error) => { console.warn('[push]', error.message); },
+          onFallback: (reason) => { console.warn('[push] falling back to polling:', reason); },
+          isActive: () => appActive,
         });
-
         if (!mounted) {
-          stopPushUpdates();
+          next.close();
           return;
         }
-
-        cleanup = stopPushUpdates;
+        handle = next;
       } catch (error) {
         console.warn(error instanceof Error ? error.message : 'Failed to start JMAP push updates');
       }
-    })();
+    };
+
+    const refreshAfterResume = () => {
+      const email = useEmailStore.getState();
+      void email.fetchMailboxes();
+      if (email.currentMailboxId) void email.refreshEmails();
+      void useOutboxStore.getState().flush();
+    };
+
+    void start();
+
+    const subscription = AppState.addEventListener('change', (state) => {
+      const nowActive = state === 'active';
+      if (nowActive === appActive) return;
+      appActive = nowActive;
+      if (!nowActive) {
+        handle?.close();
+        handle = null;
+        return;
+      }
+      // Coming back: verify the session is still alive before trusting the
+      // stream, then reconnect and catch up on what was missed.
+      void (async () => {
+        const alive = await jmapClient.ping().catch(() => false);
+        if (!mounted) return;
+        if (!alive && useNetworkStore.getState().online) {
+          const ok = await useAuthStore.getState().retrySession().catch(() => false);
+          if (!ok || !mounted) return;
+        }
+        if (!handle) await start();
+        else handle.reconnect();
+        refreshAfterResume();
+      })();
+    });
+
+    // Reconnect when the network comes back while foregrounded.
+    const unsubscribeNetwork = useNetworkStore.subscribe((state, prev) => {
+      if (state.online && !prev.online && appActive) {
+        if (handle) handle.reconnect();
+        else void start();
+      }
+    });
 
     return () => {
       mounted = false;
-      cleanup?.();
+      subscription.remove();
+      unsubscribeNetwork();
+      handle?.close();
+      handle = null;
     };
-  }, [client, isAuthenticated]);
+  }, [client, isAuthenticated, haveLiveSession, activeAccountId]);
+
+  // Foreground keep-alive: a `Core/echo` every 30 s tells us when the server
+  // is unreachable even though the device is online (NetInfo cannot), and
+  // drives the per-account connection dot.
+  React.useEffect(() => {
+    if (!isAuthenticated || !haveLiveSession || !activeAccountId) return;
+    let cancelled = false;
+    let failures = 0;
+    const timer = setInterval(() => {
+      if (AppState.currentState !== 'active') return;
+      if (!useNetworkStore.getState().online) return;
+      void jmapClient.ping().then((ok) => {
+        if (cancelled) return;
+        failures = ok ? 0 : failures + 1;
+        const account = useAccountStore.getState().getAccountById(activeAccountId);
+        if (!account) return;
+        // One missed echo can be a blip; two in a row is a lost connection.
+        const connected = failures < 2;
+        if (account.isConnected !== connected) {
+          useAccountStore.getState().updateAccount(activeAccountId, { isConnected: connected });
+        }
+      }).catch(() => undefined);
+    }, 30_000);
+    return () => {
+      cancelled = true;
+      clearInterval(timer);
+    };
+  }, [isAuthenticated, haveLiveSession, activeAccountId]);
 
   // Skip the "Restoring session" flash for returning users: if we already
   // have a persisted active account, render the main UI immediately with

src/api/push-stream.ts

@@ -0,0 +1,374 @@
+// Live updates from the JMAP EventSource (SSE), with the lifecycle the plain
+// `react-native-sse` EventSource lacks:
+//
+// - reconnect with back-off after a network error (the library only re-polls
+//   when the XHR reaches DONE with a status; a dropped socket - iOS
+//   suspension, Wi-Fi → cellular - fires `error` with status 0 and then
+//   nothing ever happens again);
+// - a fresh Authorization header on every (re)connect, and an immediate
+//   reconnect when the OAuth token rotates (the library reuses the headers it
+//   was constructed with, so after a refresh every retry sent the stale
+//   bearer: 401 → error → retry every 5 s forever);
+// - a 90 s ping watchdog and a periodic recycle so the accumulated
+//   `responseText` cannot grow without bound;
+// - fallback to polling when the stream is refused (proxy without SSE, TLS
+//   client certificate that XHR cannot present, repeated failures).
+
+import { jmapClient, AuthenticationError } from './jmap-client';
+import { CAPABILITIES } from './types';
+import type { StateChange } from './types';
+import { getClientCertAlias } from '../lib/client-cert';
+
+export type StateChangeHandler = (change: StateChange) => void;
+
+// Server pings every 30 s; consider the stream dead after three missed ones.
+const PING_SECONDS = 30;
+const PING_TIMEOUT_MS = 90_000;
+// Recycle a healthy stream periodically to bound the XHR buffer.
+const RECYCLE_MS = 30 * 60_000;
+const RECONNECT_DELAYS_MS = [2_000, 5_000, 10_000, 20_000, 40_000, 60_000];
+// After this many consecutive failed connects the stream is given up on and
+// polling takes over for the rest of the session.
+const MAX_CONSECUTIVE_FAILURES = 6;
+
+export const POLL_INTERVAL_MS = 5_000;
+export const SLOW_POLL_INTERVAL_MS = 20_000;
+
+interface SseEvent {
+  type?: string;
+  data?: string;
+  message?: string;
+  xhrStatus?: number;
+}
+
+interface EventSourceLike {
+  addEventListener: (type: string, handler: (event: SseEvent) => void) => void;
+  removeEventListener: (type: string, handler: (event: SseEvent) => void) => void;
+  close: () => void;
+}
+
+type EventSourceCtor = new (
+  url: string,
+  options: { headers: Record<string, string>; pollingInterval?: number; timeoutBeforeConnection?: number },
+) => EventSourceLike;
+
+function eventSourceImpl(): EventSourceCtor {
+  // eslint-disable-next-line @typescript-eslint/no-require-imports
+  return require('react-native-sse').default as EventSourceCtor;
+}
+
+export interface LiveUpdatesOptions {
+  onStateChange: StateChangeHandler;
+  onError?: (error: Error) => void;
+  /** Called when the stream is abandoned for polling. */
+  onFallback?: (reason: string) => void;
+  /** Returns false while the app is backgrounded; polling pauses. */
+  isActive?: () => boolean;
+}
+
+export interface LiveUpdatesHandle {
+  /** Tear everything down (stream, timers, polling). */
+  close: () => void;
+  /** Drop the current stream and connect again (foreground resume). */
+  reconnect: () => void;
+  /** 'sse' | 'polling' | 'closed' */
+  readonly mode: 'sse' | 'polling' | 'closed';
+}
+
+function buildEventSourceUrl(template: string): string {
+  return template
+    .replace('{types}', '*')
+    .replace('{closeafter}', 'no')
+    .replace('{ping}', String(PING_SECONDS));
+}
+
+/**
+ * Polling fallback: `Mailbox/get` + `Email/get` with empty id lists return
+ * the current state tokens; a change in either is reported as a StateChange
+ * for the primary account. One request in flight at a time, paused while
+ * backgrounded.
+ */
+export function startPolling(
+  onStateChange: StateChangeHandler,
+  interval = POLL_INTERVAL_MS,
+  isActive: () => boolean = () => true,
+): () => void {
+  let accountId: string;
+  try {
+    accountId = jmapClient.accountId;
+  } catch {
+    return () => undefined;
+  }
+  const stateCache: Record<string, string> = {};
+  let inFlight = false;
+  let stopped = false;
+
+  const tick = async () => {
+    if (stopped || inFlight || !isActive()) return;
+    if ('isConnected' in jmapClient && !jmapClient.isConnected) return;
+    inFlight = true;
+    try {
+      const res = await jmapClient.request(
+        [
+          ['Mailbox/get', { accountId, ids: [] }, 'm'],
+          ['Email/get', { accountId, ids: [] }, 'e'],
+        ],
+        [CAPABILITIES.CORE, CAPABILITIES.MAIL],
+      );
+
+      const changed: Record<string, string> = {};
+      for (const [method, result] of res.methodResponses) {
+        if (method === 'error') continue;
+        const type = method.replace('/get', '');
+        const newState = (result as { state?: string }).state;
+        if (newState && stateCache[type] && stateCache[type] !== newState) {
+          changed[type] = newState;
+        }
+        if (newState) stateCache[type] = newState;
+      }
+
+      if (Object.keys(changed).length > 0) {
+        onStateChange({ '@type': 'StateChange', changed: { [accountId]: changed } });
+      }
+    } catch {
+      // Silently retry on next interval
+    } finally {
+      inFlight = false;
+    }
+  };
+
+  const timer = setInterval(() => { void tick(); }, interval);
+  return () => {
+    stopped = true;
+    clearInterval(timer);
+  };
+}
+
+/**
+ * Start live updates: SSE when the session advertises an eventSourceUrl and
+ * no client certificate is in play, polling otherwise. Resolves to a handle
+ * that can be closed or told to reconnect.
+ */
+export async function startLiveUpdates(opts: LiveUpdatesOptions): Promise<LiveUpdatesHandle> {
+  const isActive = opts.isActive ?? (() => true);
+  const safeOnStateChange: StateChangeHandler = (change) => {
+    try {
+      opts.onStateChange(change);
+    } catch (err) {
+      opts.onError?.(err instanceof Error ? err : new Error(String(err)));
+    }
+  };
+
+  let mode: 'sse' | 'polling' | 'closed' = 'closed';
+  let stopPolling: (() => void) | null = null;
+  let es: EventSourceLike | null = null;
+  let esHandlers: Array<[string, (event: SseEvent) => void]> = [];
+  let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  let watchdog: ReturnType<typeof setTimeout> | null = null;
+  let recycleTimer: ReturnType<typeof setTimeout> | null = null;
+  let consecutiveFailures = 0;
+  let closed = false;
+  let unsubscribeTokenRefresh: (() => void) | null = null;
+
+  const session = jmapClient.currentSession;
+  const template = session?.eventSourceUrl;
+  const certAlias = await getClientCertAlias().catch(() => null);
+
+  const clearTimers = () => {
+    if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; }
+    if (watchdog) { clearTimeout(watchdog); watchdog = null; }
+    if (recycleTimer) { clearTimeout(recycleTimer); recycleTimer = null; }
+  };
+
+  const dropStream = () => {
+    if (es) {
+      for (const [type, handler] of esHandlers) {
+        try { es.removeEventListener(type, handler); } catch { /* ignore */ }
+      }
+      esHandlers = [];
+      try { es.close(); } catch { /* ignore */ }
+      es = null;
+    }
+    if (watchdog) { clearTimeout(watchdog); watchdog = null; }
+    if (recycleTimer) { clearTimeout(recycleTimer); recycleTimer = null; }
+  };
+
+  const fallbackToPolling = (reason: string) => {
+    if (closed) return;
+    dropStream();
+    if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; }
+    if (mode !== 'polling') {
+      mode = 'polling';
+      opts.onFallback?.(reason);
+      stopPolling = startPolling(safeOnStateChange, SLOW_POLL_INTERVAL_MS, isActive);
+    }
+  };
+
+  const scheduleReconnect = () => {
+    if (closed || reconnectTimer) return;
+    if (consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
+      fallbackToPolling('stream failed repeatedly');
+      return;
+    }
+    const delay = RECONNECT_DELAYS_MS[Math.min(consecutiveFailures, RECONNECT_DELAYS_MS.length - 1)];
+    reconnectTimer = setTimeout(() => {
+      reconnectTimer = null;
+      void connect();
+    }, delay);
+  };
+
+  const armWatchdog = () => {
+    if (watchdog) clearTimeout(watchdog);
+    watchdog = setTimeout(() => {
+      // No ping for 90 s: the socket is dead even though the XHR never said so.
+      dropStream();
+      consecutiveFailures += 1;
+      scheduleReconnect();
+    }, PING_TIMEOUT_MS);
+  };
+
+  const connect = async () => {
+    if (closed || !template) return;
+    if (!isActive()) return; // resumed later by reconnect()
+    dropStream();
+    try {
+      await jmapClient.ensureFreshToken();
+    } catch {
+      // the 401 path below handles a stale token
+    }
+    let authHeader: string;
+    try {
+      authHeader = jmapClient.authHeader;
+    } catch {
+      return;
+    }
+    const url = buildEventSourceUrl(template);
+    let source: EventSourceLike;
+    try {
+      const Impl = eventSourceImpl();
+      // pollingInterval 0 disables the library's own re-poll on DONE; every
+      // reconnect goes through scheduleReconnect so headers stay fresh.
+      source = new Impl(url, {
+        headers: { Authorization: authHeader },
+        pollingInterval: 0,
+        timeoutBeforeConnection: 0,
+      });
+    } catch (err) {
+      opts.onError?.(err instanceof Error ? err : new Error(String(err)));
+      consecutiveFailures += 1;
+      scheduleReconnect();
+      return;
+    }
+    es = source;
+    mode = 'sse';
+
+    const on = (type: string, handler: (event: SseEvent) => void) => {
+      source.addEventListener(type, handler);
+      esHandlers.push([type, handler]);
+    };
+
+    on('open', () => {
+      consecutiveFailures = 0;
+      armWatchdog();
+      if (recycleTimer) clearTimeout(recycleTimer);
+      recycleTimer = setTimeout(() => {
+        if (closed || es !== source) return;
+        dropStream();
+        void connect();
+      }, RECYCLE_MS);
+    });
+    on('ping', () => armWatchdog());
+    on('state', (event) => {
+      armWatchdog();
+      try {
+        const data: StateChange = JSON.parse(typeof event === 'string' ? event : event.data ?? '');
+        safeOnStateChange(data);
+      } catch {
+        // Ignore malformed events
+      }
+    });
+    on('close', () => {
+      if (closed || es !== source) return;
+      dropStream();
+      scheduleReconnect();
+    });
+    on('error', (event) => {
+      if (closed || es !== source) return;
+      dropStream();
+      const status = event?.xhrStatus ?? 0;
+      if (status === 401) {
+        // Stale bearer: refresh and come straight back with the new header.
+        void jmapClient.forceRefreshToken().then(
+          (refreshed) => {
+            if (closed) return;
+            if (refreshed) {
+              consecutiveFailures = 0;
+              void connect();
+            } else {
+              consecutiveFailures += 1;
+              opts.onError?.(new AuthenticationError('Live updates rejected: session expired'));
+              scheduleReconnect();
+            }
+          },
+          () => {
+            consecutiveFailures += 1;
+            scheduleReconnect();
+          },
+        );
+        return;
+      }
+      if (status >= 400) {
+        // The endpoint answered but refused the stream (proxy without SSE,
+        // rate limit, server error): don't hammer it.
+        consecutiveFailures += 2;
+        opts.onError?.(new Error(`Live updates stream refused: ${status}`));
+        scheduleReconnect();
+        return;
+      }
+      // status 0: socket dropped / no network. Back off and retry.
+      consecutiveFailures += 1;
+      scheduleReconnect();
+    });
+  };
+
+  if (!template || certAlias) {
+    mode = 'polling';
+    opts.onFallback?.(certAlias ? 'client certificate in use' : 'server has no eventSourceUrl');
+    stopPolling = startPolling(safeOnStateChange, POLL_INTERVAL_MS, isActive);
+  } else {
+    unsubscribeTokenRefresh = jmapClient.onTokenRefresh(() => {
+      if (closed || mode !== 'sse') return;
+      // New bearer: the open stream still carries the old one for its
+      // lifetime, which is fine, but every reconnect from now on must use
+      // the new header - recycle now so a pending stale retry can't loop.
+      dropStream();
+      consecutiveFailures = 0;
+      void connect();
+    });
+    await connect();
+  }
+
+  return {
+    get mode() {
+      return mode;
+    },
+    close: () => {
+      closed = true;
+      mode = 'closed';
+      clearTimers();
+      dropStream();
+      unsubscribeTokenRefresh?.();
+      unsubscribeTokenRefresh = null;
+      stopPolling?.();
+      stopPolling = null;
+    },
+    reconnect: () => {
+      if (closed) return;
+      if (mode === 'polling') return; // polling keeps running on its own
+      if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; }
+      consecutiveFailures = 0;
+      dropStream();
+      void connect();
+    },
+  };
+}

src/api/push.ts

@@ -117,123 +117,40 @@ export async function destroyPushSubscription(id: string): Promise<void> {
   );
 }
 
-/**
- * Connect to the JMAP EventSource (SSE) endpoint for real-time push updates.
- * Returns a cleanup function to close the connection.
- */
-export function connectEventSource(
-  onStateChange: StateChangeHandler,
-): (() => void) | null {
-  const session = jmapClient.currentSession;
-  if (!session?.eventSourceUrl) return null;
+// ─── Live updates ────────────────────────────────────────
+// The EventSource lifecycle (reconnect with back-off, fresh bearer on every
+// connect, ping watchdog, polling fallback) lives in ./push-stream. These
+// re-exports keep the historical entry points.
 
-  const url = session.eventSourceUrl
-    .replace('{types}', '*')
-    .replace('{closeafter}', 'no')
-    .replace('{ping}', '30');
-
-  // Use native EventSource or react-native-sse
-  const EventSourceImpl = typeof EventSource !== 'undefined'
-    ? EventSource
-    : require('react-native-sse').default;
-
-  const es = new EventSourceImpl(url, {
-    headers: { Authorization: jmapClient.authHeader },
-  });
-
-  const handler = (event: any) => {
-    try {
-      const data: StateChange = JSON.parse(
-        typeof event === 'string' ? event : event.data,
-      );
-      onStateChange(data);
-    } catch {
-      // Ignore malformed events
-    }
-  };
-
-  es.addEventListener('state', handler);
-
-  return () => {
-    es.removeEventListener('state', handler);
-    es.close();
-  };
-}
-
-/**
- * Polling fallback when SSE is not available.
- * Returns a cleanup function to stop polling.
- */
-export function startPolling(
-  onStateChange: StateChangeHandler,
-  interval = 5000,
-): () => void {
-  const accountId = jmapClient.accountId;
-  const stateCache: Record<string, string> = {};
-
-  const timer = setInterval(async () => {
-    try {
-      const res = await jmapClient.request(
-        [
-          ['Mailbox/get', { accountId, ids: [] }, 'm'],
-          ['Email/get', { accountId, ids: [] }, 'e'],
-        ],
-        [CAPABILITIES.CORE, CAPABILITIES.MAIL],
-      );
-
-      const changed: Record<string, string> = {};
-      for (const [method, result] of res.methodResponses) {
-        const type = method.replace('/get', '');
-        const newState = result.state;
-        if (newState && stateCache[type] && stateCache[type] !== newState) {
-          changed[type] = newState;
-        }
-        if (newState) stateCache[type] = newState;
-      }
-
-      if (Object.keys(changed).length > 0) {
-        onStateChange({
-          '@type': 'StateChange',
-          changed: { [accountId]: changed },
-        });
-      }
-    } catch {
-      // Silently retry on next interval
-    }
-  }, interval);
-
-  return () => clearInterval(timer);
-}
+export {
+  startLiveUpdates,
+  startPolling,
+  type LiveUpdatesHandle,
+  type LiveUpdatesOptions,
+} from './push-stream';
 
 export interface StartPushOptions {
   onStateChange: StateChangeHandler;
   onError?: (error: Error) => void;
+  onFallback?: (reason: string) => void;
+  isActive?: () => boolean;
 }
 
 /**
- * Start real-time updates, preferring SSE with polling fallback. The `_client`
- * argument is unused today (the underlying event source uses the singleton
- * `jmapClient`) but is part of the call signature so the host App can pass
- * the active client reference and trigger React re-runs on account switch.
+ * Start real-time updates, preferring SSE with polling fallback. Resolves to
+ * a cleanup function; prefer `startLiveUpdates` when the handle's
+ * `reconnect()` is needed (foreground resume).
  */
-export function startPushUpdates(
+export async function startPushUpdates(
   _client: unknown,
   opts: StartPushOptions,
-): () => void {
-  const onError = opts.onError;
-  const safeOnStateChange: StateChangeHandler = (change) => {
-    try {
-      opts.onStateChange(change);
-    } catch (err) {
-      if (onError) onError(err instanceof Error ? err : new Error(String(err)));
-    }
-  };
+): Promise<() => void> {
+  const { startLiveUpdates: start } = await import('./push-stream');
   try {
-    const cleanup = connectEventSource(safeOnStateChange);
-    if (cleanup) return cleanup;
-    return startPolling(safeOnStateChange);
+    const handle = await start(opts);
+    return () => handle.close();
   } catch (err) {
-    if (onError) onError(err instanceof Error ? err : new Error(String(err)));
+    opts.onError?.(err instanceof Error ? err : new Error(String(err)));
     return () => undefined;
   }
 }