From edc26ce1cad31904d07688d1c4c7715e945939e5 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sat, 29 Aug 2026 17:57:33 +0200 Subject: [PATCH] fix: reconnect live updates after network blips, token refresh, account switch and app resume --- App.tsx | 127 +++++++++++--- src/api/push-stream.ts | 374 +++++++++++++++++++++++++++++++++++++++++ src/api/push.ts | 125 +++----------- 3 files changed, 499 insertions(+), 127 deletions(-) create mode 100644 src/api/push-stream.ts diff --git a/App.tsx b/App.tsx index cbbae0d2..a3c2e13e 100644 --- a/App.tsx +++ b/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 diff --git a/src/api/push-stream.ts b/src/api/push-stream.ts new file mode 100644 index 00000000..131eee5b --- /dev/null +++ b/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; 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 = {}; + 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 = {}; + 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 { + 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 | null = null; + let watchdog: ReturnType | null = null; + let recycleTimer: ReturnType | 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(); + }, + }; +} diff --git a/src/api/push.ts b/src/api/push.ts index 41d80e71..4724dea7 100644 --- a/src/api/push.ts +++ b/src/api/push.ts @@ -117,123 +117,40 @@ export async function destroyPushSubscription(id: string): Promise { ); } -/** - * 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 = {}; - - 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 = {}; - 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; } }