Merge branch 'sse-session-reconnect' into 'development'

Feat(SSE): SSE Session auto-reconnect

See merge request GeneralProtocols/xo/stack!90

packages/stack/src/storage/store-remote-http.ts

@@ -1,5 +1,5 @@
 import { BaseStore } from './base-store.js';
-import { SSEClient } from '../utils/index.js';
+import { SSESession } from '../utils/index.js';
 import {
   Bytes,
   Encrypter,
@@ -96,7 +96,7 @@ export class StoreRemoteHTTP extends BaseStore {
     // NOTE: SSE only supports text-encoding, so we use Extended JSON instead of msgpackr (which is binary).
     // TODO: It would be really nice to somehow be able to support msgpackr for consistency though.
     //       But, from a bandwidth point of view, this is not crucial: These payloads are "live" payloads (thus, there's not much data).
-    await SSEClient.from(`${this.baseURL}events`, {
+    await SSESession.from(`${this.baseURL}events`, {
       method: 'POST',
       headers,
       onMessage: (e) => {

packages/stack/src/utils/index.ts

@@ -3,6 +3,6 @@ export * from './ext-error.js';
 export * from './inverse-bloom-filter.js';
 export * from './memoizer.js';
 export * from './misc.js';
-export * from './sse-client.js';
+export * from './sse-session.js';
-export * from './sse-client-msgpackr.js';
+export * from './sse-session-msgpackr.js';
 export * from './uuid.js';

packages/stack/src/utils/sse-session-msgpackr.ts

@@ -1,18 +1,18 @@
-import { SSEClient, SSEClientOptions, SSEvent } from './sse-client.js';
+import { SSESession, SSESessionOptions, SSEvent } from './sse-session.js';
 
 import { unpackMultiple } from 'msgpackr';
 
 /**
  * SSEClientMsgpackr is a subclass of SSEClient that uses msgpackr to parse the events.
  */
-export class SSEClientMsgpackr extends SSEClient {
+export class SSEClientMsgpackr extends SSESession {
   /**
    * Creates a new SSEClientMsgpackr instance.
    * @param {string} url The URL to connect to
    * @param {SSEClientOptions} options The options for the SSEClient
    * @returns {SSEClientMsgpackr} A new SSEClientMsgpackr instance
    */
-  public static async from(url: string, options: SSEClientOptions) {
+  public static async from(url: string, options: SSESessionOptions) {
     const client = new SSEClientMsgpackr(url, options);
     await client.connect();
     return client;

packages/stack/src/utils/sse-session.ts

@@ -1,25 +1,19 @@
-// TODO: 1. Handle reconnection logic.
-//       2. Allow specifying custom
-//       3. Try to make it so that connect DOES NOT resolve until connection is actually establish.
-//          There's a very slim chance of a race-condition otherwise where we might miss an event.
-//          Note that this IS NOT trivial. Claude will probably NOT be able to do this for you!
-
 /**
  * A Server-Sent Events client implementation using fetch API.
  * Supports custom headers, POST requests, and is non-blocking.
  */
-export class SSEClient {
+export class SSESession {
   /**
-   * Creates and connects a new SSEClient instance.
+   * Creates and connects a new SSESession instance.
    * @param url The URL to connect to
    * @param options Configuration options
-   * @returns A new connected SSEClient instance
+   * @returns A new connected SSESession instance
    */
   public static async from(
     url: string,
-    options: Partial<SSEClientOptions> = {},
+    options: Partial<SSESessionOptions> = {},
-  ): Promise<SSEClient> {
+  ): Promise<SSESession> {
-    const client = new SSEClient(url, options);
+    const client = new SSESession(url, options);
     await client.connect();
     return client;
   }
@@ -28,35 +22,68 @@ export class SSEClient {
   private url: string;
   private controller: AbortController;
   private connected: boolean = false;
-  protected options: SSEClientOptions;
+  protected options: SSESessionOptions;
   protected messageBuffer: Uint8Array = new Uint8Array();
 
+  // Listener for when the tab is hidden or shown.
+  private visibilityChangeHandler: ((event: Event) => void) | null = null;
+
+  // Keep track of the current reconnection attempt to use for exponential backoff.
+  protected currentAttempt: number = 0;
+
   // Text decoders and encoders for parsing the message buffer.
   private textDecoder: TextDecoder = new TextDecoder();
   private textEncoder: TextEncoder = new TextEncoder();
 
   /**
-   * Creates a new SSEClient instance.
+   * Creates a new SSESession instance.
    * @param url The URL to connect to
    * @param options Configuration options
    */
-  constructor(url: string, options: Partial<SSEClientOptions> = {}) {
+  constructor(url: string, options: Partial<SSESessionOptions> = {}) {
     this.url = url;
     this.options = {
       // Use default fetch function.
-      fetch: fetch,
+      fetch: (...args) => fetch(...args),
       method: 'GET',
       headers: {
         Accept: 'text/event-stream',
         'Cache-Control': 'no-cache',
       },
-      onOpen: () => {},
+      onConnected: () => {},
       onMessage: () => {},
-      onError: (error) => console.error('SSEClient error:', error),
+      onError: (error) => console.error('SSESession error:', error),
-      onClose: () => {},
+      onDisconnected: () => {},
+
+      // Reconnection options
+      attemptReconnect: true,
+      retryDelay: 1000,
+      persistent: false,
       ...options,
     };
     this.controller = new AbortController();
+    
+    // Set up visibility change handling if in mobile browser environment
+    if (typeof document !== 'undefined') {
+      this.visibilityChangeHandler = this.handleVisibilityChange.bind(this);
+      document.addEventListener('visibilitychange', this.visibilityChangeHandler);
+    }
+  }
+
+  /**
+   * Handles visibility change events in the browser.
+   */
+  private async handleVisibilityChange(): Promise<void> {
+    // When going to background, close the current connection cleanly
+    // This allows us to reconnect mobile devices when they come back after leaving the tab or browser app.
+    if (document.visibilityState === 'hidden') {
+      this.controller.abort();
+    }
+
+    // When coming back to foreground, attempt to reconnect if not connected
+    if (document.visibilityState === 'visible' && !this.connected) {
+      await this.attemptReconnect();
+    }
   }
 
   /**
@@ -66,6 +93,9 @@ export class SSEClient {
     if (this.connected) return;
 
     this.connected = true;
+    const abortController = new AbortController();
+    this.controller = abortController;
+
     const { method, headers, body } = this.options;
 
     const fetchOptions: RequestInit = {
@@ -92,10 +122,11 @@ export class SSEClient {
             throw new Error('Response body is null');
           }
 
-          // Notify that the connection has been established
+          // Reset the current attempt.
-          if (this.options.onOpen) {
+          this.currentAttempt = 0;
-            this.options.onOpen();
+
-          }
+          // Call the onConnected callback.
+          this.handleCallback(this.options.onConnected, undefined);
 
           const reader = response.body.getReader();
 
@@ -112,9 +143,15 @@ export class SSEClient {
 
             if (done) {
               this.connected = false;
-              if (this.options.onClose) {
+
-                this.options.onClose();
+              // Call the onDisconnected callback.
+              this.handleCallback(this.options.onDisconnected, undefined);
+
+              // If the connection was closed by the server, we want to attempt a reconnect if the connection should be persistent.
+              if (this.options.persistent) {
+                await this.attemptReconnect();
               }
+
               break;
             }
 
@@ -124,7 +161,7 @@ export class SSEClient {
             // Handle each complete event
             for (const event of events) {
               if (this.options.onMessage) {
-                this.options.onMessage(event);
+                this.handleCallback(this.options.onMessage, event);
               }
             }
 
@@ -134,15 +171,28 @@ export class SSEClient {
         } catch (error) {
           this.connected = false;
 
-          if (this.options.onError) {
+          // Call the onDisconnected callback.
+          this.handleCallback(this.options.onDisconnected, error);
+
+          // If the connection was aborted using the controller, we don't need to call onError.
+          if (abortController.signal.aborted) {
+            return;
+          }
+
+          // Call the onError callback.
+          // NOTE: we dont use the handleCallback here because it would result in 2 error callbacks.
+          try {
             this.options.onError(error);
+          } catch (error) {
+            console.log(`SSE Session: onError callback error:`, error);
           }
 
-          if (this.options.onClose) {
+          // Attempt to reconnect if enabled
-            this.options.onClose();
+          if (this.options.attemptReconnect) {
+            return await this.attemptReconnect();
           }
 
-          // Now we'll reject the promise instead of resolving
+          // Reject the promise
           reject(error);
         }
       };
@@ -214,13 +264,45 @@ export class SSEClient {
   }
 
   /**
-   * Closes the SSE connection.
+   * Attempts to reconnect to the SSE endpoint.
+   * @returns void
+   */
+  protected attemptReconnect(): void {
+    // Don't attempt to reconnect if it's not enabled or if we are already connected
+    if (!this.options.attemptReconnect || this.connected) return;
+
+    // Exponential backoff with a maximum of 10 seconds
+    // TODO: We should update this to use an Exponential Backoff utility when we have one.
+    this.currentAttempt++;
+    const delay = Math.min(this.options.retryDelay * Math.pow(1.3, this.currentAttempt - 1), 10000);
+    console.log(`Attempting to reconnect in ${delay}ms`);
+
+    setTimeout(async () => {
+      await this.connect();
+    }, delay);
+  }
+
+  /**
+   * Override the onMessage callback.
+   *
+   * @param onMessage The callback to set.
+   */
+  public setOnMessage(onMessage: (event: SSEvent) => void): void {
+    this.options.onMessage = onMessage;
+  }
+
+  /**
+   * Closes the SSE connection and cleans up event listeners.
    */
   public close(): void {
-    if (!this.connected) return;
+    // Clean up everything including the visibility handler
-
     this.controller.abort();
-    this.connected = false;
+
+    // Remove the visibility handler (This is only required on browsers)
+    if (this.visibilityChangeHandler && typeof document !== 'undefined') {
+      document.removeEventListener('visibilitychange', this.visibilityChangeHandler);
+      this.visibilityChangeHandler = null;
+    }
   }
 
   /**
@@ -230,12 +312,31 @@ export class SSEClient {
   public isConnected(): boolean {
     return this.connected;
   }
+
+  /**
+   * Will handle thrown errors from the callback and call the onError callback.
+   * This is to avoid the sse-session from disconnecting from errors that are not a result of the sse-session itself.
+   * 
+   * @param callback The callback to handle.
+   * @param args The arguments to pass to the callback.
+   */
+  private handleCallback<T>(callback: (args: T) => void, args: T): void {
+    try {
+      callback(args);
+    } catch (error) {
+      try {
+        this.options.onError(error);
+      } catch (error) {
+        console.log(`SSE Session: onError callback error:`, error);
+      }
+    }
+  }
 }
 
 /**
- * Configuration options for the SSEClient.
+ * Configuration options for the SSESession.
  */
-export interface SSEClientOptions {
+export interface SSESessionOptions {
   /**
    * The fetch function to use.
    *
@@ -264,22 +365,37 @@ export interface SSEClientOptions {
   /**
    * Called when the connection is established.
    */
-  onOpen?: () => void;
+  onConnected: () => void;
 
   /**
    * Called when a message is received.
    */
-  onMessage?: (event: SSEvent) => void;
+  onMessage: (event: SSEvent) => void;
 
   /**
    * Called when an error occurs.
    */
-  onError?: (error: unknown) => void;
+  onError: (error: unknown) => void;
 
   /**
    * Called when the connection is closed.
    */
-  onClose?: () => void;
+  onDisconnected: () => void;
+
+  /**
+   * Whether to attempt to reconnect.
+   */
+  attemptReconnect: boolean;
+
+  /**
+   * The delay in milliseconds between reconnection attempts.
+   */
+  retryDelay: number;
+
+  /**
+   * Whether to reconnect when the session is terminated by the server.
+   */
+  persistent: boolean;
 }
 
 /**