initial Commit

src/sse/client.ts

@@ -0,0 +1,243 @@
+import { SSESession, toExtendedJson } from '@xo-cash/utils';
+import { PrivateKey } from '@xo-cash/primitives'
+import { binToHex, flattenBinArray, secp256k1, sha256 } from '@bitauth/libauth'
+import type { WriteRequest } from '../types.js';
+
+export type AuthenticatedRequestHeaders = {
+  'X-Public-Key': string;
+  'X-Timestamp': string;
+  'X-Signature': string;
+};
+
+export type SSEClientOptions = {
+  onMessage: (message: string) => void;
+  onError: (error: Error) => void;
+};
+
+export class SSEClient {
+  private 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(private readonly url: string, privateKey: PrivateKey, options: Partial<SSEClientOptions> = {}) {
+    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) => {
+        // Create the request authentication headers
+        const requestBody = toExtendedJson(request.body);
+        const authHeaders = SSEClient.authenticateRequest(this.privateKey, requestBody);
+
+        // 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: body,
+      method: 'POST',
+    });
+
+    // Create a listener to re-emit the messages
+    this.sseSession.on('message', (message) => {
+      for (const listener of this.messageListeners) {
+        listener(message);
+      }
+    });
+
+    // 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 = SSEClient.makeWriteRequest(this.privateKey, resourceId, valueBytes);
+
+    const bodyStr = toExtendedJson({ resources: [resource] });
+
+    const res = await fetch(url, {
+      method: 'POST',
+
+      // Headers arent required unless we have payment service running
+      headers: {
+        'Content-Type': 'application/json',
+        // ...SSEClient.authenticateRequest(this.privateKey, bodyStr),
+      },
+
+      body: bodyStr,
+    });
+
+    if (!res.ok) {
+      throw new Error(`Failed to write resource: ${res.statusText}`);
+    }
+
+    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',
+        // ...SSEClient.authenticateRequest(this.privateKey, body),
+      },
+      body,
+    });
+
+    if (!res.ok) {
+      throw new Error(`Failed to read resource: ${res.statusText}`);
+    }
+
+    return res.json();
+  }
+
+  // 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();
+  }
+
+  /**
+   * Derives a resource-scoped private key from the root key and resource id.
+   */
+  private static deriveResourcePrivateKey(privateKey: Uint8Array, resourceId: string): Uint8Array {
+    const resourceIdBytes = new TextEncoder().encode(resourceId);
+    return sha256.hash(flattenBinArray([ privateKey, resourceIdBytes ]));
+  }
+
+  /**
+   * Signs a canonical payload string and returns the compressed public key and DER signature.
+   */
+  private static signPayload(
+    privateKey: Uint8Array,
+    payload: string,
+  ): { publicKey: string; signature: string } {
+    const payloadHash = sha256.hash(new TextEncoder().encode(payload));
+
+    const publicKey = secp256k1.derivePublicKeyCompressed(privateKey);
+    if (typeof publicKey === 'string') {
+      throw new Error('Failed to derive public key');
+    }
+
+    const signature = secp256k1.signMessageHashDER(privateKey, payloadHash);
+    if (typeof signature === 'string') {
+      throw new Error('Failed to sign message');
+    }
+
+    return {
+      publicKey: binToHex(publicKey),
+      signature: binToHex(signature),
+    };
+  }
+
+  /**
+   * Builds a signed write resource for POST /data/write.
+   *
+   * The signing key is derived per resource via sha256(privateKey + resourceId).
+   * The signature covers only the resource value, not the full request body.
+   */
+  static makeWriteRequest(
+    privateKey: PrivateKey,
+    resourceId: string,
+    value: Uint8Array,
+  ): WriteRequest {
+    const derivedKey = SSEClient.deriveResourcePrivateKey(privateKey.toBytes(), resourceId);
+    const timestamp = Date.now();
+    const payload = `${timestamp}${resourceId}${toExtendedJson(value)}`;
+    const { publicKey, signature } = SSEClient.signPayload(derivedKey, payload);
+
+    return {
+      id: resourceId,
+      value,
+      publicKey,
+      timestamp,
+      signature,
+    };
+  }
+
+  /**
+   * Signs an entire request body for transport-layer authentication headers.
+   *
+   * Unlike {@link makeWriteRequest}, this uses the root private key and signs
+   * the full serialized body rather than an individual resource value.
+   */
+  static authenticateRequest(
+    privateKey: PrivateKey,
+    body: string,
+  ): AuthenticatedRequestHeaders {
+    const timestamp = Date.now();
+    const payload = `${timestamp}${body}`;
+    const { publicKey, signature } = SSEClient.signPayload(privateKey.toBytes(), payload);
+
+    return {
+      'X-Public-Key': publicKey,
+      'X-Timestamp': timestamp.toString(),
+      'X-Signature': signature,
+    };
+  }
+}

src/sse/index.ts

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