Update to latest engine. Add sync-v2. Various fixes.

src/utils/syncing/sse/client.ts

@@ -0,0 +1,166 @@
+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;
+  onError: (error: Error) => void;
+};
+
+export class SSEClient extends SyncClient {
+  private readonly url: string;
+  public sseSession: SSESession;
+  private readonly subscriptions: Set<string> = new Set();
+  private readonly messageListeners = new Set<(message: string) => void>();
+  private readonly errorListeners = new Set<(error: Error) => void>();
+
+  private privateKey: PrivateKey;
+  private textEncoder = new TextEncoder();
+
+  private get subscriptionUrl() {
+    return `${this.url}/data/subscribe`;
+  }
+
+  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',
+      body: JSON.stringify({ resourceId: [ ...this.subscriptions ] }),
+    });
+    this.messageListeners.add(options.onMessage ?? (() => {}));
+    this.errorListeners.add(options.onError ?? (() => {}));
+  }
+
+  async connect() {
+    await this.sseSession.connect();
+  }
+
+  async disconnect() {
+    await this.sseSession.disconnect();
+  }
+
+  async reconnect() {
+    // Disconnect the current session
+    await this.sseSession.disconnect();
+    this.sseSession.off('message');
+    this.sseSession.off('error');
+
+    // 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, {
+      onRequest: async (request) => {
+        // Get the path from the URL
+        const url = new URL(this.subscriptionUrl);
+        const path = url.pathname;
+
+        // Sign the request body
+        const authHeaders = SyncClient.signRequest(this.privateKey.toBytes(), path, request.body);
+
+        // Initialize the request headers if they don't exist
+        request.headers ??= {};
+
+        // Add the authentication headers to the request
+        request.headers = {
+          ...request.headers,
+          ...authHeaders,
+        }
+
+        // Return the request with the authentication headers
+        return request;
+      },
+      headers: {
+        'Content-Type': 'application/json',
+      },
+      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);
+      }
+    });
+
+    // Create a listener to re-emit the errors
+    this.sseSession.on('error', (error) => {
+      for (const listener of this.errorListeners) {
+        listener(error);
+      }
+    });
+
+    // Connect to the server
+    await this.connect();
+  }
+  
+  // Send a POST /data/write request to the server
+  async write(resourceId: string, value: Record<string, unknown>) {
+    const url = `${this.url}/data/write`;
+
+    const valueStr = toExtendedJson(value);
+    const valueBytes = this.textEncoder.encode(valueStr);
+    const resource = SyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
+
+    const body = toExtendedJson({ resources: [resource] });
+
+    const res = await fetch(url, {
+      method: 'POST',
+
+      // Headers arent required unless we have payment service running
+      headers: {
+        'Content-Type': 'application/json',
+        ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/write', body),
+      },
+
+      body,
+    });
+
+    if (!res.ok) {
+      const error = await res.json();
+      throw new Error(`Failed to write resource [${resourceId}] (${res.statusText}): ${JSON.stringify(error)}`);
+    }
+
+    return res.json();
+  }
+
+  // Send a POST /data/read request to the server
+  async read(resourceId: string) {
+    const url = `${this.url}/data/get`;
+    const body = toExtendedJson({ resourceId: [ resourceId ] });
+
+    const res = await fetch(url, {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/get', body),
+      },
+      body,
+    });
+
+    if (!res.ok) {
+      throw new Error(`Failed to read resource: ${res.statusText}`);
+    }
+
+    return fromExtendedJson(await res.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();
+  }
+
+  // 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();
+  }
+}