Harvmaster/xo-stack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'wip_storage_fixes' into 'development'

Browse Source 
Store fixes

See merge request GeneralProtocols/xo/stack!42
This commit is contained in:
Jim Hamill
2025-03-15 02:49:54 +00:00
parent 3284943bea 38d0071a7b
commit bed30315f4
2 changed files with 87 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stack/src/storage/store-remote-http.ts
+16 -5
View File
@@ -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
+71 -65
View File
@@ -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();
});
}