From 38d0071a7b864fd413590d37c97f51a7ba4e8c2b Mon Sep 17 00:00:00 2001 From: Jim Hamill Date: Sat, 15 Mar 2025 02:49:54 +0000 Subject: [PATCH] Store fixes --- .../stack/src/storage/store-remote-http.ts | 21 ++- packages/stack/src/utils/sse-client.ts | 136 +++++++++--------- 2 files changed, 87 insertions(+), 70 deletions(-) diff --git a/packages/stack/src/storage/store-remote-http.ts b/packages/stack/src/storage/store-remote-http.ts index 106824a..0f0e461 100644 --- a/packages/stack/src/storage/store-remote-http.ts +++ b/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 = {}) { diff --git a/packages/stack/src/utils/sse-client.ts b/packages/stack/src/utils/sse-client.ts index 065ac1e..3ce2dab 100644 --- a/packages/stack/src/utils/sse-client.ts +++ b/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(); }); }