Fix WS Syncing. Remove requirement for connection. Clean ups. Fix blobs.

src/sse/client.ts

@@ -3,17 +3,10 @@ import { PrivateKey } from '@xo-cash/primitives'
 
 import { SyncClient } from '../shared/client.js';
 
-export type SSEClientOptions = {
-  onMessage: (message: unknown) => void;
-  onError: (error: Error) => void;
-};
-
 export class SSEClient extends SyncClient {
   private readonly url: string;
-  public sseSession: SSESession;
+  public sseSession: SSESession | undefined;
   private readonly subscriptions: Set<string> = new Set();
-  private readonly messageListeners = new Set<(message: unknown) => void>();
-  private readonly errorListeners = new Set<(error: Error) => void>();
 
   private privateKey: PrivateKey;
   private textEncoder = new TextEncoder();
@@ -22,38 +15,51 @@ export class SSEClient extends SyncClient {
     return `${this.url}/data/subscribe`;
   }
 
-  constructor(url: string, privateKey: PrivateKey, options: Partial<SSEClientOptions> = {}) {
+  constructor(url: string, privateKey: PrivateKey) {
     super();
 
     this.url = url;
     this.privateKey = privateKey;
-    this.sseSession = new SSESession(`${this.url}/data/subscribe`, {
-      method: 'POST',
-      body: JSON.stringify({ resourceId: [ ...this.subscriptions ] }),
-    });
-    this.messageListeners.add(options.onMessage ?? (() => {}));
-    this.errorListeners.add(options.onError ?? (() => {}));
+
+    // Whenever we get a new subscription, we need to reconnect the session with the new subscriptions in the body
+    // We add a small debounce to this to prevent parallel subscriptions from causing multiple reconnects
+    this.on('subscriptions-updated', () => {
+      // Nothing awaits this listener, so report failures as an error event instead of an unhandled rejection
+      this.reconnect().catch((error) => this.emit('error', error));
+    }, 25);
   }
 
   async connect() {
+    // The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet.
+    // subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session.
+    if (this.subscriptions.size === 0) {
+      return;
+    }
+
+    this.sseSession ??= this.createSession();
+
     await this.sseSession.connect();
   }
 
   async disconnect() {
-    await this.sseSession.disconnect();
+    await this.sseSession?.disconnect();
   }
 
   async reconnect() {
-    // Disconnect the current session
-    await this.sseSession.disconnect();
-    this.sseSession.off('message');
-    this.sseSession.off('error');
+    // Close the current session. Its subscription list is fixed in the request body, so it can't be reused
+    this.sseSession?.close();
+    this.sseSession = undefined;
 
+    // Connect a new session with the current subscriptions
+    await this.connect();
+  }
+
+  private createSession() {
     // Build the request body
     const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] });
 
-    // Re-create the session with the additional resource ID in the subscription list
-    this.sseSession = new SSESession(this.subscriptionUrl, {
+    // Create a new SSE session
+    const session = new SSESession(this.subscriptionUrl, {
       onRequest: async (request) => {
         // Get the path from the URL
         const url = new URL(this.subscriptionUrl);
@@ -79,24 +85,31 @@ export class SSEClient extends SyncClient {
       },
       body,
       method: 'POST',
+      // Reconnect when the server ends the stream (e.g. a restart or a proxy idle timeout). By default only errors are retried
+      persistent: true,
     });
 
+    // Create a listener to re-emit the connected and disconnected events
+    session.on('connected', this.handleOnConnectionUpdated.bind(this, 'connected'));
+    session.on('disconnected', this.handleOnConnectionUpdated.bind(this, 'disconnected'));
+
     // Create a listener to re-emit the messages
-    this.sseSession.on('message', (message) => {
-      for (const listener of this.messageListeners) {
-        listener(fromExtendedJson(message.data));
+    session.on('message', (message) => {
+      const parsedMessage = SyncClient.messageSchema.safeParse(fromExtendedJson(message.data));
+      if (!parsedMessage.success) {
+        return;
       }
+
+      this.emit('message', parsedMessage.data);
     });
 
     // Create a listener to re-emit the errors
-    this.sseSession.on('error', (error) => {
-      for (const listener of this.errorListeners) {
-        listener(error);
-      }
+    session.on('error', (error) => {
+      this.emit('error', error);
     });
 
-    // Connect to the server
-    await this.connect();
+    // Return the session
+    return session;
   }
   
   // Send a POST /data/write request to the server
@@ -126,7 +139,9 @@ export class SSEClient extends SyncClient {
       throw new Error(`Failed to write resource [${resourceId}] (${res.statusText}): ${JSON.stringify(error)}`);
     }
 
-    return fromExtendedJson(await res.text());
+    const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[];
+
+    return text;
   }
 
   // Send a POST /data/read request to the server
@@ -147,20 +162,26 @@ export class SSEClient extends SyncClient {
       throw new Error(`Failed to read resource: ${res.statusText}`);
     }
 
-    return fromExtendedJson(await res.text());
+    const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[];
+
+    return text;
   }
 
   // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list
   async subscribe(resourceId: string) {
     this.subscriptions.add(resourceId);
 
-    await this.reconnect();
+    this.emit('subscriptions-updated', this.subscriptions);
   }
 
   // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list
   async unsubscribe(resourceId: string) {
     this.subscriptions.delete(resourceId);
 
-    await this.reconnect();
+    this.emit('subscriptions-updated', this.subscriptions);
+  }
+
+  async handleOnConnectionUpdated(value: 'connected' | 'disconnected') {
+    this.emit('connection-updated', value);
   }
 }