xo-cli/src/utils/syncing/sse/client.ts

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();
  }
}