Store fixes

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

@@ -1,6 +1,11 @@
 import { BaseStore } from './base-store.js';
 import { Encrypter, SSEClient } from '../utils';
-import { Bytes, PrivateKey, PublicKey } from '@xocash/primitives';
+import {
+  Bytes,
+  PrivateKey,
+  PublicKey,
+  decodeExtendedJson,
+} from '@xocash/primitives';
 import { pack, unpack } from 'msgpackr';
 import { z } from 'zod';
 
@@ -68,6 +73,7 @@ export class StoreRemoteHTTP extends BaseStore {
   }
 
   async subscribeEvents() {
+    // Compute our signature (to authenticate with the SSE endpoint).
     const timestamp = Math.floor(Date.now() / 1000);
     const signature = await StoreRemoteHTTP.signRequest(
       this.privateKey.toHex(),
@@ -76,6 +82,7 @@ export class StoreRemoteHTTP extends BaseStore {
       {},
     );
 
+    // Define our headers.
     const headers = {
       'Content-Type': 'application/json',
       Accept: 'application/json',
@@ -84,13 +91,19 @@ export class StoreRemoteHTTP extends BaseStore {
       'X-Timestamp': timestamp.toString(),
     };
 
-    const sseClient = new SSEClient(`${this.baseURL}events`, {
+    // Create our SSE Client.
+    // 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`, {
       method: 'POST',
       headers,
       onMessage: (e) => {
         // Validate the payload using Zod.
         // TODO: Use msgpackr
-        const payload = StoreRemoteHTTPEventSchema.parse(JSON.parse(e.data));
+        const payload = StoreRemoteHTTPEventSchema.parse(
+          decodeExtendedJson(e.data),
+        );
 
         // Emit the event (update) to any listeners.
         this.emit(payload.action, {
@@ -100,8 +113,6 @@ export class StoreRemoteHTTP extends BaseStore {
       },
       body: JSON.stringify({}),
     });
-
-    await sseClient.connect();
   }
 
   private async makeRequest(endpoint: string, payload: object = {}) {

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

@@ -77,79 +77,85 @@ export class SSEClient {
       cache: 'no-store',
     };
 
-    // Using an async IIFE to handle the connection without blocking
-    return new Promise(async (resolve) => {
-      try {
-        const response = await fetch(this.url, fetchOptions);
+    // Using a regular Promise executor (not async)
+    return new Promise((resolve, reject) => {
+      // Move the async logic inside a separate function
+      const establishConnection = async () => {
+        try {
+          const response = await fetch(this.url, fetchOptions);
 
-        if (!response.ok) {
-          throw new Error(`HTTP error! Status: ${response.status}`);
-        }
-
-        if (!response.body) {
-          throw new Error('Response body is null');
-        }
-
-        // Notify that the connection has been established
-        if (this.options.onOpen) {
-          this.options.onOpen();
-        }
-
-        const reader = response.body.getReader();
-        const decoder = new TextDecoder();
-        let buffer = '';
-
-        // Start the read here so we can resolve and then start handlng the stream
-        // We will await this at the start of the loop, and redeclare it at the end of each loop iteration
-        let readPromise = reader.read();
-
-        // Resolve before starting the loop so that we can start peforming other actions
-        // Keep this out of the loop to avoid flags or `already resolved` errors.
-        resolve();
-
-        while (true) {
-          const { done, value } = await readPromise;
-
-          if (done) {
-            this.connected = false;
-            if (this.options.onClose) {
-              this.options.onClose();
-            }
-            break;
+          if (!response.ok) {
+            throw new Error(`HTTP error! Status: ${response.status}`);
           }
 
-          // Decode the received chunk and add it to the buffer
-          const chunk = decoder.decode(value, { stream: true });
-          buffer += chunk;
-
-          // Process any complete events in the buffer
-          const events = this.parseEvents(buffer);
-          buffer = events.remainder;
-
-          // Handle each complete event
-          for (const event of events.parsed) {
-            if (this.options.onMessage) {
-              this.options.onMessage(event);
-            }
+          if (!response.body) {
+            throw new Error('Response body is null');
           }
 
-          // Start reading the next chunk
-          readPromise = reader.read();
-        }
-      } catch (error) {
-        this.connected = false;
+          // Notify that the connection has been established
+          if (this.options.onOpen) {
+            this.options.onOpen();
+          }
 
-        if (this.options.onError) {
-          this.options.onError(error);
-        }
+          const reader = response.body.getReader();
+          const decoder = new TextDecoder();
+          let buffer = '';
 
-        if (this.options.onClose) {
-          this.options.onClose();
-        }
+          // Start the read here so we can resolve and then start handling the stream
+          // We will await this at the start of the loop, and redeclare it at the end of each loop iteration
+          let readPromise = reader.read();
 
-        // Doesnt seem like you want to propagate the error here.
-        resolve();
-      }
+          // Resolve before starting the loop so that we can start performing other actions
+          // Keep this out of the loop to avoid flags or `already resolved` errors.
+          resolve();
+
+          while (true) {
+            const { done, value } = await readPromise;
+
+            if (done) {
+              this.connected = false;
+              if (this.options.onClose) {
+                this.options.onClose();
+              }
+              break;
+            }
+
+            // Decode the received chunk and add it to the buffer
+            const chunk = decoder.decode(value, { stream: true });
+            buffer += chunk;
+
+            // Process any complete events in the buffer
+            const events = this.parseEvents(buffer);
+            buffer = events.remainder;
+
+            // Handle each complete event
+            for (const event of events.parsed) {
+              if (this.options.onMessage) {
+                this.options.onMessage(event);
+              }
+            }
+
+            // Start reading the next chunk
+            readPromise = reader.read();
+          }
+        } catch (error) {
+          this.connected = false;
+
+          if (this.options.onError) {
+            this.options.onError(error);
+          }
+
+          if (this.options.onClose) {
+            this.options.onClose();
+          }
+
+          // Now we'll reject the promise instead of resolving
+          reject(error);
+        }
+      };
+
+      // Execute the async function
+      establishConnection();
     });
   }