Fix Web Sockets. Form into library. Move demo to demo folder. Split demo between sse and websocket.

src/index.ts

@@ -1,53 +1,4 @@
-import { PrivateKey } from "@xo-cash/primitives";
+export * from './shared/client.js';
 
-import { SSEClient } from "./sse/client.js";
-import { WsClient } from './ws/client.js';
-
-const testSSE = async () => {
-  const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
-  const client = new SSEClient('https://v2.sync.xo.harvmaster.com', privateKey, {
-    onMessage: (message) => {
-      console.log(message);
-    },
-    onError: (error) => {
-      console.error(error);
-    },
-  });
-
-  await client.subscribe('test');
-
-  // await client.connect();
-
-  await client.write('test', {'message': 'Hello, world! this is the minimal sync client using SSE'});
-
-  const data = await client.read('test');
-  console.log(data);
-
-  // await client.disconnect();
-}
-
-const testWS = async () => {
-  const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
-  const client = new WsClient('wss://v2.sync.xo.harvmaster.com', privateKey, {
-    onMessage: (message) => {
-      console.log(message);
-    },
-    onError: (error) => {
-      console.error(error);
-    },
-  });
-
-  await client.connect();
-  await client.subscribe('test');
-
-  await client.write('test', {'message': 'Hello, world! this is the minimal sync client using WS'});
-  
-  const data = await client.read('test');
-  console.log(data);
-
-  // await client.unsubscribe('test');
-  // await client.disconnect();
-}
-
-testSSE();
-// testWS();
+export * from './sse/client.js';
+export * from './ws/client.js';

src/shared/client.ts

@@ -79,7 +79,7 @@ export abstract class SyncClient {
 
     // Create the payload
     const timestamp = Date.now();
-    const payload = `${timestamp}${resourceId}${toExtendedJson(value)}`;
+    const payload = `${timestamp}:${resourceId}:${toExtendedJson(value)}`;
 
     // Sign the payload
     const { publicKey, signature } = this.signPayload(derivedKey, payload);
@@ -103,7 +103,12 @@ export abstract class SyncClient {
   static signRequest(privateKey: Uint8Array, path: string, body: unknown): AuthenticatedRequestHeaders {
     // Create the payload
     const timestamp = Date.now();
-    const payload = `${timestamp}:${path}:${toExtendedJson(body)}`;
+
+    // Convert the body to a string if it isn't already a string
+    const bodyStr = typeof body === 'string' ? body : toExtendedJson(body);
+
+    // Create the payload as `Timestamp:Path:Body`
+    const payload = `${timestamp}:${path}:${bodyStr}`;
 
     // Sign the payload
     const { publicKey, signature } = this.signPayload(privateKey, payload);
@@ -114,4 +119,4 @@ export abstract class SyncClient {
       'X-Signature': signature,
     };
   }
-}
+}

src/sse/client.ts

@@ -1,17 +1,18 @@
-import { SSESession, toExtendedJson } from '@xo-cash/utils';
+import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
 import { PrivateKey } from '@xo-cash/primitives'
 
 import { SyncClient } from '../shared/client.js';
 
 export type SSEClientOptions = {
-  onMessage: (message: string) => void;
+  onMessage: (message: unknown) => void;
   onError: (error: Error) => void;
 };
 
 export class SSEClient extends SyncClient {
-  private sseSession: SSESession;
+  private readonly url: string;
+  public sseSession: SSESession;
   private readonly subscriptions: Set<string> = new Set();
-  private readonly messageListeners = new Set<(message: string) => void>();
+  private readonly messageListeners = new Set<(message: unknown) => void>();
   private readonly errorListeners = new Set<(error: Error) => void>();
 
   private privateKey: PrivateKey;
@@ -21,9 +22,10 @@ export class SSEClient extends SyncClient {
     return `${this.url}/data/subscribe`;
   }
 
-  constructor(private readonly url: string, privateKey: PrivateKey, options: Partial<SSEClientOptions> = {}) {
+  constructor(url: string, privateKey: PrivateKey, options: Partial<SSEClientOptions> = {}) {
     super();
 
+    this.url = url;
     this.privateKey = privateKey;
     this.sseSession = new SSESession(`${this.url}/data/subscribe`, {
       method: 'POST',
@@ -53,15 +55,12 @@ export class SSEClient extends SyncClient {
     // Re-create the session with the additional resource ID in the subscription list
     this.sseSession = new SSESession(this.subscriptionUrl, {
       onRequest: async (request) => {
-        // Create the request authentication headers
-        const requestBody = toExtendedJson(request.body);
-
         // Get the path from the URL
-        const url = new URL(this.url);
+        const url = new URL(this.subscriptionUrl);
         const path = url.pathname;
 
         // Sign the request body
-        const authHeaders = SyncClient.signRequest(this.privateKey.toBytes(), path, requestBody);
+        const authHeaders = SyncClient.signRequest(this.privateKey.toBytes(), path, request.body);
 
         // Initialize the request headers if they don't exist
         request.headers ??= {};
@@ -78,14 +77,14 @@ export class SSEClient extends SyncClient {
       headers: {
         'Content-Type': 'application/json',
       },
-      body: body,
+      body,
       method: 'POST',
     });
 
     // Create a listener to re-emit the messages
     this.sseSession.on('message', (message) => {
       for (const listener of this.messageListeners) {
-        listener(message.data);
+        listener(fromExtendedJson(message.data));
       }
     });
 
@@ -108,7 +107,7 @@ export class SSEClient extends SyncClient {
     const valueBytes = this.textEncoder.encode(valueStr);
     const resource = SyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
 
-    const bodyStr = toExtendedJson({ resources: [resource] });
+    const body = toExtendedJson({ resources: [resource] });
 
     const res = await fetch(url, {
       method: 'POST',
@@ -116,10 +115,10 @@ export class SSEClient extends SyncClient {
       // Headers arent required unless we have payment service running
       headers: {
         'Content-Type': 'application/json',
-        // ...SSEClient.authenticateRequest(this.privateKey, bodyStr),
+        ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/write', body),
       },
 
-      body: bodyStr,
+      body,
     });
 
     if (!res.ok) {
@@ -127,7 +126,7 @@ export class SSEClient extends SyncClient {
       throw new Error(`Failed to write resource [${resourceId}] (${res.statusText}): ${JSON.stringify(error)}`);
     }
 
-    return res.json();
+    return fromExtendedJson(await res.text());
   }
 
   // Send a POST /data/read request to the server
@@ -139,7 +138,7 @@ export class SSEClient extends SyncClient {
       method: 'POST',
       headers: {
         'Content-Type': 'application/json',
-        // ...SSEClient.authenticateRequest(this.privateKey, body),
+        ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/get', body),
       },
       body,
     });
@@ -148,7 +147,7 @@ export class SSEClient extends SyncClient {
       throw new Error(`Failed to read resource: ${res.statusText}`);
     }
 
-    return res.json();
+    return fromExtendedJson(await res.text());
   }
 
   // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list

src/sse/index.ts

@@ -1 +1 @@
-export * from './client.js';
+export * from './client.js';

src/ws/client.ts

@@ -40,18 +40,22 @@ export type WsClientOptions = {
  * here to keep the transport demo easy to follow.
  */
 export class WsClient extends SyncClient {
+  private readonly url: string;
+  private readonly privateKey: PrivateKey;
   private socket: WebSocket | undefined;
   private readonly pendingRequests = new Map<string, PendingRequest>();
   private readonly messageListeners = new Set<(message: WsMessage) => void>();
   private readonly errorListeners = new Set<(error: Error) => void>();
 
   constructor(
-    private readonly url: string,
-    private readonly privateKey: PrivateKey,
+    url: string,
+    privateKey: PrivateKey,
     options: Partial<WsClientOptions> = {},
   ) {
     super();
 
+    this.url = url;
+    this.privateKey = privateKey;
     this.messageListeners.add(options.onMessage ?? (() => {}));
     this.errorListeners.add(options.onError ?? (() => {}));
   }
@@ -72,7 +76,7 @@ export class WsClient extends SyncClient {
         this.handleMessage(String(event.data));
       };
 
-      socket.onerror = () => {
+      socket.onerror = (ev) => {
         const error = new Error('WebSocket connection failed');
         this.emitError(error);
         reject(error);
@@ -134,15 +138,14 @@ export class WsClient extends SyncClient {
    * been handed to the socket; future updates arrive through `onMessage`.
    */
   async subscribe(resourceId: string): Promise<void> {
-    this.send({
-      path: '/data/subscribe',
-      body: { resourceId: [resourceId] },
-    });
+    await this.request('/data/subscribe', { resourceId: [resourceId] });
   }
 
-  /** Remove one resource topic without closing the shared connection. */
-  async unsubscribe(resourceId: string): Promise<unknown> {
-    return this.request('/data/unsubscribe', { resourceId: [resourceId] });
+  /**
+   * Unsubscribe from one resource topic on the current connection.
+   */
+  async unsubscribe(resourceId: string): Promise<void> {
+    await this.request('/data/unsubscribe', { resourceId: [resourceId] });
   }
 
   /** Send a request and wait for the response carrying the same ID. */