Improve tui demo

Publish as public package

See merge request GeneralProtocols/xo/sync-client!3

.cspell.json

@@ -0,0 +1,4 @@
+{
+    "version": "0.1",
+    "import": ["@generalprotocols/cspell-dictionary/cspell.json"]
+}

.gitignore

@@ -1,2 +1,4 @@
 /node_modules
 /dist
+/coverage
+/docs

.gitlab-ci.yml

@@ -0,0 +1,3 @@
+# Use the default NPM publishing template for the `@xo-cash/*` packages.
+include:
+    - component: $CI_SERVER_FQDN/GeneralProtocols/xo/pipelines/npm@v1.2.0

.prettierrc.mjs

@@ -0,0 +1,23 @@
+// Prettier configuration aligned with @xo-cash/eslint-config
+export default {
+    // Matches @stylistic/max-len': [ 'warn', 150, ...]
+    printWidth: 150,
+
+    // Matches @stylistic/no-tabs': [ 'error' ]
+    useTabs: false,
+
+    // Matches @stylistic/indent': [ 'error', 4, ...]
+    tabWidth: 4,
+
+    // Matches @stylistic/array-bracket-spacing': [ 'error', 'always', ...]
+    bracketSpacing: true,
+
+    // Matches @stylistic/quotes: ['error', 'single', { avoidEscape: true }], from @chalp/eslint-airbnb/base/stylistic.js
+    singleQuote: true,
+
+    // Matches @stylistic/semi: ['error', 'always'], from @chalp/eslint-airbnb/base/stylistic.js
+    semi: true,
+
+    // Matches @stylistic/arrow-parens: ['error', 'always'], from @chalp/eslint-airbnb/base/stylistic.js
+    arrowParens: 'always',
+};

demo/experimental/encrypted-client-per-resource.ts

@@ -0,0 +1,182 @@
+import { subtle, getRandomValues } from 'node:crypto';
+import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
+import { sha256, binToBase64, binToHex } from '@bitauth/libauth';
+
+import { BaseSyncClient, type Resource, type ResourcesRecord } from '../../source/shared/base-client.js';
+
+type ResourceKey = {
+    resourceId: string;
+    keyMaterial: Uint8Array;
+};
+
+/**
+ * Encrypts each resource with its own key, where the resource ID itself is the secret.
+ *
+ * The caller-facing resource ID is hashed into a 32 byte AES key, and that key is hashed again
+ * to get the resource ID sent to the server. The server never sees the AES key or the original
+ * resource ID, so the resource ID must be unguessable (high entropy) for this to be secure.
+ */
+export class EncryptedClientPerResource<T = unknown> extends BaseSyncClient<T> {
+    /** Maps the hashed (server-side) resource ID to the original resource ID and its key */
+    private readonly resourceKeys = new Map<string, ResourceKey>();
+
+    constructor(private readonly client: BaseSyncClient<string>) {
+        super();
+
+        this.client.on('resource-instance-updated', async (message) => {
+            // Ignore updates for resources we don't hold the key for
+            const resourceKey = this.resourceKeys.get(message.resourceId);
+            if (!resourceKey) return;
+
+            try {
+                this.emit('resource-instance-updated', {
+                    ...message,
+                    resourceId: resourceKey.resourceId,
+                    value: (await EncryptedClientPerResource.decrypt(resourceKey.keyMaterial, message.value)) as T,
+                });
+            } catch (error) {
+                this.emit('error', error instanceof Error ? error : new Error(String(error)));
+            }
+        });
+    }
+
+    /**
+     * Derives the AES key and server-side resource ID from a resource ID, and remembers the mapping.
+     * @param resourceId - The resource ID (the secret)
+     * @returns The hashed resource ID to send to the server
+     */
+    private hashResourceId(resourceId: string): string {
+        const keyMaterial = sha256.hash(new TextEncoder().encode(resourceId));
+        const hashedResourceId = binToHex(sha256.hash(keyMaterial));
+
+        this.resourceKeys.set(hashedResourceId, { resourceId, keyMaterial });
+
+        return hashedResourceId;
+    }
+
+    /**
+     * Gets the original resource ID and key for a hashed resource ID.
+     * @param hashedResourceId - The hashed (server-side) resource ID
+     * @returns The original resource ID and its key material
+     */
+    private getResourceKey(hashedResourceId: string): ResourceKey {
+        const resourceKey = this.resourceKeys.get(hashedResourceId);
+
+        if (!resourceKey) {
+            throw new Error(`No key found for resource ${hashedResourceId}`);
+        }
+
+        return resourceKey;
+    }
+
+    /**
+     * Decrypts every instance of every resource in a record, and maps the hashed resource IDs back to the original resource IDs.
+     * @param resources - The encrypted resources record, keyed by hashed resource ID
+     * @returns The resources record with decrypted values, keyed by original resource ID
+     */
+    private async decryptResources(resources: ResourcesRecord<string>): Promise<ResourcesRecord<T>> {
+        const entries = await Promise.all(
+            Object.entries(resources).map(async ([hashedResourceId, resource]) => {
+                const { resourceId, keyMaterial } = this.getResourceKey(hashedResourceId);
+
+                const instances: Resource<T>[] = await Promise.all(
+                    resource.instances.map(async (instance) => ({
+                        ...instance,
+                        resourceId,
+                        value: (await EncryptedClientPerResource.decrypt(keyMaterial, instance.value)) as T,
+                    })),
+                );
+
+                return [resourceId, { ...resource, instances }] as const;
+            }),
+        );
+
+        return Object.fromEntries(entries);
+    }
+
+    public async connect(): Promise<void> {
+        return this.client.connect();
+    }
+
+    public async disconnect(): Promise<void> {
+        return this.client.disconnect();
+    }
+
+    public async readMany(resourceIds: string[]): Promise<ResourcesRecord<T>> {
+        const hashedResourceIds = resourceIds.map((resourceId) => this.hashResourceId(resourceId));
+
+        const encryptedResources = await this.client.readMany(hashedResourceIds);
+
+        return this.decryptResources(encryptedResources);
+    }
+
+    public async writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>> {
+        const encryptedEntries = await Promise.all(
+            Object.entries(resources).map(async ([resourceId, value]) => {
+                const hashedResourceId = this.hashResourceId(resourceId);
+                const { keyMaterial } = this.getResourceKey(hashedResourceId);
+
+                return [hashedResourceId, await EncryptedClientPerResource.encrypt(keyMaterial, value)] as const;
+            }),
+        );
+
+        const response = await this.client.writeMany(Object.fromEntries(encryptedEntries));
+
+        return this.decryptResources(response);
+    }
+
+    public async subscribe(resourceId: string): Promise<void> {
+        return this.client.subscribe(this.hashResourceId(resourceId));
+    }
+
+    public async unsubscribe(resourceId: string): Promise<void> {
+        return this.client.unsubscribe(this.hashResourceId(resourceId));
+    }
+
+    /**
+     * Imports raw key material as an AES-GCM key
+     * @param keyMaterial - The 32 byte key material
+     * @returns The AES-GCM CryptoKey
+     */
+    private static importKey(keyMaterial: Uint8Array) {
+        return subtle.importKey('raw', Buffer.from(keyMaterial), { name: 'AES-GCM' }, false, ['encrypt', 'decrypt']);
+    }
+
+    /**
+     * Encrypts a value using AES-GCM with a random 12 byte IV
+     * @param keyMaterial - The encryption key material
+     * @param value - The value to encrypt
+     * @returns The IV and ciphertext as a base64 string
+     */
+    public static async encrypt(keyMaterial: Uint8Array, value: unknown): Promise<string> {
+        const key = await this.importKey(keyMaterial);
+        const iv = getRandomValues(new Uint8Array(12));
+
+        // Encode the value as extended JSON and then to UTF-8 bytes
+        const encodedValue = new TextEncoder().encode(toExtendedJson(value));
+
+        const encrypted = await subtle.encrypt({ name: 'AES-GCM', iv }, key, encodedValue);
+
+        // Prepend the IV so it can be recovered when decrypting
+        return binToBase64(Buffer.concat([iv, new Uint8Array(encrypted)]));
+    }
+
+    /**
+     * Decrypts a value encrypted by `encrypt`
+     * @param keyMaterial - The encryption key material
+     * @param value - The IV and ciphertext as a base64 string
+     * @returns The decrypted value
+     */
+    public static async decrypt(keyMaterial: Uint8Array, value: string): Promise<unknown> {
+        const key = await this.importKey(keyMaterial);
+
+        // Decode the base64 first: the IV is 12 bytes, not 12 base64 characters
+        const bytes = Buffer.from(value, 'base64');
+        const iv = bytes.subarray(0, 12);
+        const encryptedValue = bytes.subarray(12);
+
+        const decrypted = await subtle.decrypt({ name: 'AES-GCM', iv }, key, encryptedValue);
+
+        return fromExtendedJson(new TextDecoder().decode(decrypted));
+    }
+}

demo/experimental/encrypted-client.ts

@@ -0,0 +1,136 @@
+import { subtle, getRandomValues } from 'node:crypto';
+import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
+import { sha256, binToBase64 } from '@bitauth/libauth';
+
+import { BaseSyncClient, type ResourcesRecord } from '../../source/shared/base-client.js';
+
+/**
+ * Encrypts every resource with a single shared key.
+ */
+export class EncryptedClient<T = unknown> extends BaseSyncClient<T> {
+    private readonly encryptionKey: Uint8Array;
+
+    constructor(
+        private readonly client: BaseSyncClient<string>,
+        encryptionKey: Uint8Array | string,
+    ) {
+        super();
+
+        // Hash the encryption key into 32 bytes if it is a string
+        this.encryptionKey = typeof encryptionKey === 'string' ? sha256.hash(new TextEncoder().encode(encryptionKey)) : encryptionKey;
+
+        this.client.on('resource-instance-updated', async (message) => {
+            try {
+                this.emit('resource-instance-updated', {
+                    ...message,
+                    value: (await EncryptedClient.decrypt(this.encryptionKey, message.value)) as T,
+                });
+            } catch (error) {
+                this.emit('error', error instanceof Error ? error : new Error(String(error)));
+            }
+        });
+    }
+
+    /**
+     * Decrypts every instance of every resource in a record.
+     * @param resources - The encrypted resources record
+     * @returns The resources record with decrypted values
+     */
+    private async decryptResources(resources: ResourcesRecord<string>): Promise<ResourcesRecord<T>> {
+        const entries = await Promise.all(
+            Object.entries(resources).map(async ([resourceId, resource]) => {
+                const instances = await Promise.all(
+                    resource.instances.map(async (instance) => ({
+                        ...instance,
+                        value: (await EncryptedClient.decrypt(this.encryptionKey, instance.value)) as T,
+                    })),
+                );
+
+                return [resourceId, { ...resource, instances }] as const;
+            }),
+        );
+
+        return Object.fromEntries(entries);
+    }
+
+    public async connect(): Promise<void> {
+        return this.client.connect();
+    }
+
+    public async disconnect(): Promise<void> {
+        return this.client.disconnect();
+    }
+
+    public async readMany(resourceIds: string[]): Promise<ResourcesRecord<T>> {
+        const encryptedResources = await this.client.readMany(resourceIds);
+
+        return this.decryptResources(encryptedResources);
+    }
+
+    public async writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>> {
+        const encryptedEntries = await Promise.all(
+            Object.entries(resources).map(async ([resourceId, value]) => {
+                return [resourceId, await EncryptedClient.encrypt(this.encryptionKey, value)] as const;
+            }),
+        );
+
+        const response = await this.client.writeMany(Object.fromEntries(encryptedEntries));
+
+        return this.decryptResources(response);
+    }
+
+    public async subscribe(resourceId: string): Promise<void> {
+        return this.client.subscribe(resourceId);
+    }
+
+    public async unsubscribe(resourceId: string): Promise<void> {
+        return this.client.unsubscribe(resourceId);
+    }
+
+    /**
+     * Imports raw key material as an AES-GCM key
+     * @param keyMaterial - The 32 byte key material
+     * @returns The AES-GCM CryptoKey
+     */
+    private static importKey(keyMaterial: Uint8Array) {
+        return subtle.importKey('raw', Buffer.from(keyMaterial), { name: 'AES-GCM' }, false, ['encrypt', 'decrypt']);
+    }
+
+    /**
+     * Encrypts a value using AES-GCM with a random 12 byte IV
+     * @param keyMaterial - The encryption key material
+     * @param value - The value to encrypt
+     * @returns The IV and ciphertext as a base64 string
+     */
+    public static async encrypt(keyMaterial: Uint8Array, value: unknown): Promise<string> {
+        const key = await this.importKey(keyMaterial);
+        const iv = getRandomValues(new Uint8Array(12));
+
+        // Encode the value as extended JSON and then to UTF-8 bytes
+        const encodedValue = new TextEncoder().encode(toExtendedJson(value));
+
+        const encrypted = await subtle.encrypt({ name: 'AES-GCM', iv }, key, encodedValue);
+
+        // Prepend the IV so it can be recovered when decrypting
+        return binToBase64(Buffer.concat([iv, new Uint8Array(encrypted)]));
+    }
+
+    /**
+     * Decrypts a value encrypted by `encrypt`
+     * @param keyMaterial - The encryption key material
+     * @param value - The IV and ciphertext as a base64 string
+     * @returns The decrypted value
+     */
+    public static async decrypt(keyMaterial: Uint8Array, value: string): Promise<unknown> {
+        const key = await this.importKey(keyMaterial);
+
+        // Decode the base64 first: the IV is 12 bytes, not 12 base64 characters
+        const bytes = Buffer.from(value, 'base64');
+        const iv = bytes.subarray(0, 12);
+        const encryptedValue = bytes.subarray(12);
+
+        const decrypted = await subtle.decrypt({ name: 'AES-GCM', iv }, key, encryptedValue);
+
+        return fromExtendedJson(new TextDecoder().decode(decrypted));
+    }
+}

demo/shared.ts

@@ -0,0 +1,28 @@
+import type { BaseSyncClient, ResourcesRecord } from '../source/shared/base-client.js';
+
+export const writeMany = async (client: BaseSyncClient<{ message: string }>): Promise<ResourcesRecord<{ message: string }>> => {
+    // Create a group of resources with changes
+    const resources = Object.fromEntries(
+        Array.from({ length: 10 }, (_, index) => [
+            `resource-${index}`,
+            {
+                message: `Hello, world! this is the minimal sync client using ${client.constructor.name} ${index}`,
+            },
+        ]),
+    );
+
+    // Write the resources to the server
+    const res = await client.writeMany(resources);
+
+    return res;
+};
+
+export const readMany = async (client: BaseSyncClient<{ message: string }>): Promise<ResourcesRecord<{ message: string }>> => {
+    // Create a list of resource IDs
+    const resources = Array.from({ length: 10 }, (_, index) => `resource-${index}`);
+
+    // Read the resources from the server
+    const res = await client.readMany(resources);
+
+    return res;
+};

demo/sse.ts

@@ -1,33 +0,0 @@
-import { PrivateKey } from "@xo-cash/primitives";
-
-import { SSEClient } from '../src/sse/client.js';
-
-const testSSE = async () => {
-  const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
-  const client = new SSEClient('https://v2.sync.xo.harvmaster.com', privateKey);
-
-  client.on('message', (message) => {
-    console.log(message);
-  });
-
-  // Connect to the sync server
-  await client.connect();
-
-  // Subscribe to the resource
-  await client.subscribe('test');
-
-  // Write a new value to the resource
-  await client.write('test', {'message': 'Hello, world! this is the minimal sync client using SSE'});
-
-  // Read the current value of the resource
-  const data = await client.read('test');
-  console.log(data);
-
-  // If the user presses Ctrl+C, disconnect the client and exit the program
-  process.on('SIGINT', async () => {
-    await client.disconnect();
-    process.exit(0);
-  });
-}
-
-testSSE();

demo/suite.ts

@@ -0,0 +1,93 @@
+import { PrivateKey } from '@xo-cash/primitives';
+
+// Core clients
+import type { BaseSyncClient } from '../source/shared/base-client.js';
+import { SSEClient } from '../source/sse/client.js';
+import { WsClient } from '../source/ws/client.js';
+
+// Experimental clients
+import { EncryptedClientPerResource } from './experimental/encrypted-client-per-resource.js';
+import { EncryptedClient } from './experimental/encrypted-client.js';
+
+// Shared functions
+import { writeMany, readMany } from './shared.js';
+
+const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
+
+const clients = {
+    sse: new SSEClient<{ message: string }>('https://dev.v2.sync.xo.harvmaster.com', privateKey),
+    ws: new WsClient<{ message: string }>('wss://dev.v2.sync.xo.harvmaster.com', privateKey),
+    encrypted: new EncryptedClient<{ message: string }>(new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey), 'secret'),
+    encryptedPerResource: new EncryptedClientPerResource<{ message: string }>(
+        new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey),
+    ),
+};
+
+// Test the suite for a given client
+const testSuite = async (client: BaseSyncClient<{ message: string }>) => {
+    client.on('resource-instance-updated', (message) => {
+        console.log(message);
+    });
+
+    // Connect to the sync server
+    await client.connect();
+
+    // Subscribe to the resource
+    await client.subscribe('test');
+
+    // Write a new value to the resource
+    await client.write('test', {
+        message: 'Hello, world! this is the minimal sync client using ' + client.constructor.name,
+    });
+
+    // Read the current value of the resource
+    const data = await client.read('test');
+    console.log(data.map((resource) => resource.value.message));
+
+    // Use the shared functions to write and read many resources
+    const writtenResources = await writeMany(client);
+    console.dir(writtenResources, { depth: null });
+
+    const readResources = await readMany(client);
+    console.dir(readResources, { depth: null });
+
+    // If the user presses Ctrl+C, disconnect the client and exit the program
+    process.on('SIGINT', async () => {
+        await client.disconnect();
+        process.exit(0);
+    });
+};
+
+// Get the clients to test
+const selectClients = (params: string[]): BaseSyncClient<{ message: string }>[] => {
+    // If no params are provided, test all clients
+    const names = params.length > 0 ? params : Object.keys(clients);
+
+    // Find values that are unknown to the clients object
+    const unknown = names.filter((n) => !(n in clients));
+    if (unknown.length > 0) {
+        console.error(`Invalid client type(s): ${unknown.join(', ')}. Expected: ${Object.keys(clients).join(', ')}`);
+        process.exit(1);
+    }
+
+    // Return the clients that are known to the clients object
+    return names.map((n) => clients[n as keyof typeof clients]);
+};
+
+// Get the params
+const params = process.argv.slice(2);
+
+// Get the clients to test
+const clientsToTest = selectClients(params);
+
+// Test the clients
+for (const client of clientsToTest) {
+    await testSuite(client);
+}
+
+// Wait for the user to press Ctrl+C
+await new Promise((resolve) => process.on('SIGINT', resolve));
+
+// Exit the program
+console.log('Exiting...');
+process.exit(0);

demo/tui.ts

@@ -0,0 +1,100 @@
+/// <reference types="node" />
+import readline from 'node:readline';
+
+import { PrivateKey } from '@xo-cash/primitives';
+
+import type { BaseSyncClient } from '../source/shared/base-client.js';
+
+import { SSEClient } from '../source/sse/client.js';
+import { EncryptedClientPerResource } from './experimental/encrypted-client-per-resource.js';
+
+const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
+
+// Define some formatting helpers
+const formatResourceId = (resourceId: string) => {
+    return `\x1b[32m${resourceId}\x1b[0m`;
+};
+
+const formatMessage = (message: string) => {
+    return `\x1b[34m${message}\x1b[0m`;
+};
+
+const resetFormatting = () => {
+    return '\x1b[0m';
+};
+
+// readline lets us get both whole lines and individual keypresses (for Escape)
+// escapeCodeTimeout: how long to wait after Escape to see if it's part of a key sequence (default 500ms)
+const rl = readline.createInterface({ input: process.stdin, output: process.stdout, escapeCodeTimeout: 0 });
+
+const startTUI = async (client: BaseSyncClient<string>) => {
+    // Listen and log out the resource change
+    client.on('resource-instance-updated', (message) => {
+        // Print the message with colours
+        console.log(
+            `${formatResourceId(message.resourceId)}:${formatResourceId(message.publicKey)} updated: ${formatMessage(message.value)}${resetFormatting()}`,
+        );
+    });
+
+    // Connect to the sync server and subscribe to the resource
+    await client.connect();
+
+    const writeResourceLoop = async (resourceId: string) => {
+        while (true) {
+            // Print the prompt
+            process.stdout.write('> ');
+
+            // Wait for the user to enter a command
+            // Resolves with the typed line, or null if Escape was pressed
+            const onKeypress = (_char: string, key: readline.Key) => key?.name === 'escape' && rl.emit('line', null);
+            process.stdin.on('keypress', onKeypress);
+            const line = await new Promise<string | null>((resolve) => rl.once('line', resolve));
+            process.stdin.off('keypress', onKeypress);
+
+            // Escape exits the loop
+            if (line === null) return;
+
+            if (line.trim()) {
+                await client.write(resourceId, line.trim());
+            }
+        }
+    };
+
+    const selectResource = async () => {
+        while (true) {
+            console.log('Enter a resource:');
+
+            // Print the prompt
+            process.stdout.write('> ');
+
+            // Wait for the user to enter a command
+            const line = await new Promise<string>((resolve) => rl.once('line', resolve));
+            const resourceId = line.trim();
+
+            if (resourceId) {
+                // Subscribe to the resource
+                await client.subscribe(resourceId);
+
+                // Read the current instances
+                const instance = await client.read(resourceId);
+                console.dir(instance, { depth: null });
+
+                // Write to the resource
+                await writeResourceLoop(resourceId);
+            }
+        }
+    };
+
+    // Start the TUI loop
+    await selectResource();
+};
+
+const clients = {
+    sse: new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey),
+    encrypted: new EncryptedClientPerResource<string>(new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey)),
+};
+
+const params = process.argv.slice(2);
+const client = params[0] === 'false' ? clients.sse : clients.encrypted;
+
+startTUI(client);

demo/ws.ts

@@ -1,40 +0,0 @@
-import { PrivateKey } from "@xo-cash/primitives";
-
-import { WsClient } from '../src/ws/client.js';
-
-const testWS = async () => {
-  const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
-  const client = new WsClient('https://v2.sync.xo.harvmaster.com', privateKey, {
-    // onMessage: (message) => {
-    //   console.log(toExtendedJson(message));
-    // },
-    onError: (error) => {
-      console.error(error);
-    },
-  });
-
-  client.on('message', (message) => {
-    console.log(message);
-  });
-
-  // Connect to the sync server
-  await client.connect();
-
-  // Subscribe to the resource
-  client.subscribe('test');
-
-  // Write a new value to the resource
-  await client.write('test', {'message': 'Hello, world! this is the minimal sync client using WS'});
-  
-  // Read the current value of the resource
-  const data = await client.read('test');
-  console.log(data);
-
-  // If the user presses Ctrl+C, disconnect the client and exit the program
-  process.on('SIGINT', async () => {
-    await client.disconnect();
-    process.exit(0);
-  });
-}
-
-testWS();

eslint.config.mjs

@@ -0,0 +1,8 @@
+import baseConfig from '@xo-cash/eslint-config';
+
+export default [
+    {
+        ignores: [ 'coverage/**', 'docs/**', 'demo/**' ],
+    },
+    ...baseConfig,
+];

package.json

@@ -1,35 +1,56 @@
 {
-  "name": "@xo-cash/sync-client",
+    "name": "@xo-cash/sync-client",
-  "version": "1.0.0",
+    "version": "0.0.1",
-  "description": "A minimal sync client for the XO Cash Sync Server",
+    "description": "A minimal sync client for the XO Cash Sync Server",
-  "main": "index.js",
+    "main": "index.js",
-  "type": "module",
+    "type": "module",
-  "scripts": {
+    "scripts": {
-    "build": "tsc",
+        "analyze": "tsdown --clean --no-fixed-extension --sourcemap source/index.ts && esbuild-analyzer dist/",
-    "start": "tsx src/index.ts",
+        "build": "tsdown --clean --sourcemap source/index.ts",
-    "test": "echo \"Error: no test specified\" && exit 1"
+        "docs": "typedoc --hideGenerator",
-  },
+        "format": "prettier --write . && eslint --fix",
-  "exports": {
+        "spellcheck": "cspell 'source/**' 'test/**' 'playground/**'",
-    ".": {
+        "style": "eslint",
-      "import": "./dist/src/index.js",
+        "syntax": "tsc --noEmit",
-      "require": "./dist/src/index.cjs"
+        "test": "vitest --dir test/ --test-timeout=15000 --passWithNoTests --run --coverage"
+    },
+    "exports": {
+        ".": {
+            "import": "./dist/index.mjs",
+            "require": "./dist/index.cjs"
+        }
+    },
+    "files": [
+        "dist"
+    ],
+    "keywords": [],
+    "author": "GeneralProtocols",
+    "license": "MIT",
+    "publishConfig": {
+        "access": "public"
+    },
+    "dependencies": {
+        "@bitauth/libauth": "^3.0.0",
+        "@xo-cash/primitives": "^0.0.2",
+        "@xo-cash/utils": "^0.0.8",
+        "zod": "^4.4.3"
+    },
+    "devDependencies": {
+        "@generalprotocols/cspell-dictionary": "^1.0.1",
+        "@types/node": "^26.6.1",
+        "@vitest/coverage-v8": "^5.0.5",
+        "@viz-kit/esbuild-analyzer": "^1.0.0",
+        "@xo-cash/eslint-config": "^1.0.2",
+        "cspell": "^10.4.0",
+        "prettier": "^3.9.10",
+        "tsdown": "^0.23.1",
+        "tsx": "^4.23.1",
+        "typedoc": "^0.28.20",
+        "typedoc-plugin-coverage": "^4.0.3",
+        "typescript": "^6.0.3",
+        "vitest": "^5.0.5"
+    },
+    "overrides": {
+        "echarts": "6.1.0"
     }
-  },
-  "files": [
-    "dist"
-  ],
-  "keywords": [],
-  "author": "",
-  "license": "ISC",
-  "devDependencies": {
-    "@types/node": "^26.6.1",
-    "tsx": "^4.23.1",
-    "typescript": "^7.0.2"
-  },
-  "dependencies": {
-    "@bitauth/libauth": "^3.0.0",
-    "@xo-cash/primitives": "^0.0.2",
-    "@xo-cash/utils": "^0.0.8",
-    "zod": "^4.4.3"
-  }
 }

readme.md

@@ -1,20 +1,19 @@
 # Installation
 
 ```
-git clone https://git.harvmaster.com/Harvmaster/sync-server-example.git
+git clone hgit@gitlab.com:GeneralProtocols/xo/sync-client.git
-cd sync-server-example
+cd sync-client
 npm ci
 ```
 
 # Running
 
-There are 2 examples in the `src/index.ts` file.
+There are 2 examples in the `demo` folder.
 
-- `testSSE()` - Example using the SSE client
+- The `suite` will run general read/write operations on various clients.
-- `testWS()` - Example using the WebSocket client
+- The `tui` will allow you to write values into the console as well as monitor for updates.
+
+- `npx tsx ./demo/suite.ts` - Optionally, add `sse` | `ws` | `encryped` | `encryptedPerResource` to select a specific implementation to test. Defaults to all clients.
+- `npx tsx ./demo/tui.ts` - Optionally add `false` to use a SSE without encryption. Default is `EncryptedClientPerResource`
 
-To run the examples, you can use the following command:
 
-```
-npm run start
-```

source/index.ts

@@ -0,0 +1,4 @@
+export * from './shared/base-client.ts';
+
+export * from './sse/client.ts';
+export * from './ws/client.ts';

source/shared/base-client.ts

@@ -0,0 +1,266 @@
+import { binToHex, flattenBinArray, secp256k1, sha256 } from '@bitauth/libauth';
+
+import { toExtendedJson, EventEmitter, fromExtendedJson } from '@xo-cash/utils';
+import type { WriteRequest } from '../types.ts';
+import z from 'zod';
+
+export type AuthenticatedRequestHeaders = {
+    'X-Public-Key': string;
+    'X-Timestamp': string;
+    'X-Signature': string;
+};
+
+export type SignedPayload = { publicKey: string; signature: string };
+
+export type XOSyncClientEventMap<T = unknown> = {
+    'connection-updated': 'connected' | 'disconnected';
+    'resource-instance-updated': Resource<T> & {
+        resourceId: string;
+    };
+    'subscriptions-updated': Set<string>;
+    error: Error;
+};
+
+export type Resource<PayloadType = unknown> = z.infer<typeof BaseSyncClient.resourceInstanceSchema> & {
+    value: PayloadType;
+};
+
+export type ResourcesRecord<PayloadType = unknown> = Record<string, { instances: Resource<PayloadType>[] }>;
+
+export abstract class BaseSyncClient<T = unknown> extends EventEmitter<XOSyncClientEventMap<T>> {
+    /**
+     * Connects to the sync server
+     */
+    abstract connect(): Promise<void>;
+
+    /**
+     * Disconnects from the sync server
+     */
+    abstract disconnect(): Promise<void>;
+
+    /**
+     * Sign and write one resource
+     * @param resourceId - The resource ID to write to
+     * @param value - The value to write
+     * @returns The Resource Object with the decoded value
+     */
+    async write(resourceId: string, value: T): Promise<Resource<T>[]> {
+        // Write the resource to the server
+        const response = await this.writeMany({ [resourceId]: value });
+
+        // Get the resource from the response
+        const resource = response[resourceId];
+
+        // If the resource is not found, throw an error
+        if (!resource) {
+            throw new Error(`Resource ${resourceId} not found`);
+        }
+
+        // Return the resource instances
+        return resource.instances;
+    }
+
+    /**
+     * Writes multiple values to resources
+     * @param resources - The resources to write to
+     * @returns The resources with the written values
+     */
+    abstract writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>>;
+
+    /**
+     * Reads a value from a resource
+     * @param resourceId - The resource ID to read from
+     * @returns The resource with the read value
+     */
+    async read(resourceId: string): Promise<Resource<T>[]> {
+        // Read the resource from the server
+        const response = await this.readMany([ resourceId ]);
+
+        // Get the resource from the response
+        const resource = response[resourceId];
+
+        // If the resource is not found, throw an error
+        if (!resource) {
+            throw new Error(`Resource ${resourceId} not found`);
+        }
+
+        // Return the resource instances
+        return resource.instances;
+    }
+
+    /**
+     * Reads multiple values from resources
+     * @param resourceIds - The resource IDs to read from
+     * @returns The resources with the read values
+     */
+    abstract readMany(resourceIds: string[]): Promise<ResourcesRecord<T>>;
+
+    /**
+     * Subscribes to a resource
+     * @param resourceId - The resource ID to subscribe to
+     */
+    abstract subscribe(resourceId: string): Promise<void>;
+
+    /**
+     * Unsubscribes from a resource
+     * @param resourceId - The resource ID to unsubscribe from
+     */
+    abstract unsubscribe(resourceId: string): Promise<void>;
+
+    /**
+     * Derives the resource private key from the private key and resource ID
+     * @param privateKey - The private key to derive the resource private key from
+     * @param resourceId - The resource ID to derive the resource private key from
+     * @returns The resource private key
+     */
+    static deriveResourcePrivateKey(privateKey: Uint8Array, resourceId: string): Uint8Array {
+        // Convert the resource ID to bytes
+        const resourceIdBytes = new TextEncoder().encode(resourceId);
+
+        // Hash the private key and resource ID
+        return sha256.hash(flattenBinArray([ privateKey, resourceIdBytes ]));
+    }
+
+    /**
+     * Signs a payload with the private key and returns the public key and signature
+     * @param privateKey - The private key to sign the payload with
+     * @param payload - The payload to sign
+     * @returns The public key and signature
+     */
+    static signPayload(privateKey: Uint8Array, payload: string): SignedPayload {
+        // Convert the payload to Binary and hash it
+        const payloadHash = sha256.hash(new TextEncoder().encode(payload));
+
+        // Derive the public key
+        const publicKey = secp256k1.derivePublicKeyCompressed(privateKey);
+
+        // If the public key is a string, throw an error
+        if (typeof publicKey === 'string') {
+            throw new Error('Failed to derive public key');
+        }
+
+        // Sign the payload
+        const signature = secp256k1.signMessageHashDER(privateKey, payloadHash);
+
+        // If the signature is a string, throw an error
+        if (typeof signature === 'string') {
+            throw new Error('Failed to sign message');
+        }
+
+        return {
+            publicKey: binToHex(publicKey),
+            signature: binToHex(signature),
+        };
+    }
+
+    /**
+     * Signs a write request with the private key and returns the write request
+     * @param privateKey - The private key to sign the write request with
+     * @param resourceId - The resource ID to write to
+     * @param value - The value to write
+     * @returns The write request
+     */
+    static signWriteRequest(privateKey: Uint8Array, resourceId: string, value: Uint8Array): WriteRequest {
+        // Derive the resource private key
+        const derivedKey = this.deriveResourcePrivateKey(privateKey, resourceId);
+
+        // Create the payload
+        const timestamp = Date.now();
+        const payload = `${timestamp}:${resourceId}:${toExtendedJson(value)}`;
+
+        // Sign the payload
+        const { publicKey, signature } = this.signPayload(derivedKey, payload);
+
+        return {
+            id: resourceId,
+            value,
+            publicKey,
+            timestamp,
+            signature,
+        };
+    }
+
+    /**
+     * Signs a request with the private key and returns the public key, timestamp, and signature
+     * @param privateKey - The private key to sign the request with
+     * @param path - The path of the request
+     * @param body - The body of the request
+     * @returns The public key, timestamp, and signature
+     */
+    static signRequest(privateKey: Uint8Array, path: string, body: unknown): AuthenticatedRequestHeaders {
+        // Create the payload
+        const timestamp = Date.now();
+
+        // 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);
+
+        return {
+            'X-Public-Key': publicKey,
+            'X-Timestamp': timestamp.toString(),
+            'X-Signature': signature,
+        };
+    }
+
+    /**
+     * Decodes the 'blob' field of the Resource Object from the server from a read/write
+     * @param response - The ResourceResponse Object from the server from a read/write
+     * @returns The Resource Object with the decoded value
+     */
+    static decodeResourcesResponse<T = unknown>(response: z.infer<typeof BaseSyncClient.resourcesResponseSchema>): ResourcesRecord<T> {
+        return Object.entries(response.resources).reduce<ResourcesRecord<T>>((acc, [ resourceId, resource ]) => {
+            // Get the instances
+            const instances = resource.instances;
+
+            // Decode the instances
+            const decodedInstances = instances.map((instance) => {
+                return BaseSyncClient.decodeBlobValue<T>(instance);
+            });
+
+            // Add the decoded instances to the accumulator
+            acc[resourceId] = {
+                ...resource,
+                instances: decodedInstances,
+            };
+
+            // Return the accumulator
+            return acc;
+        }, {});
+    }
+
+    static decodeBlobValue<T = unknown>(instance: z.infer<typeof BaseSyncClient.resourceInstanceSchema>): Resource<T> {
+        const decodedBlob = new TextDecoder().decode(instance.blob);
+        const value = fromExtendedJson(decodedBlob) as T;
+
+        return {
+            ...instance,
+            value,
+        };
+    }
+
+    static resourceInstanceSchema = z.object({
+        resourceId: z.string(),
+        blob: z.instanceof(Uint8Array),
+        timestamp: z.number(),
+        publicKey: z.string(),
+        signature: z.string(),
+    });
+
+    static rawResourcesSchema = z.object({
+        instances: z.array(BaseSyncClient.resourceInstanceSchema),
+    });
+
+    static resourcesResponseSchema = z.object({
+        resources: z.record(
+            // Resource ID
+            z.string(),
+            // Resource instances
+            BaseSyncClient.rawResourcesSchema,
+        ),
+    });
+}

source/sse/client.ts

@@ -0,0 +1,263 @@
+import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
+import type { PrivateKey } from '@xo-cash/primitives';
+
+import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.ts';
+
+export class SSEClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
+    private readonly url: string;
+    public sseSession: SSESession | undefined;
+    private readonly subscriptions: Set<string> = new Set();
+
+    private privateKey: PrivateKey;
+    private textEncoder = new TextEncoder();
+
+    private get subscriptionUrl(): string {
+        return `${this.url}/data/subscribe`;
+    }
+
+    /**
+     * Creates a new SSE client
+     * @param url - The URL of the server
+     * @param privateKey - The private key to use for authentication
+     */
+    constructor(url: string, privateKey: PrivateKey) {
+        super();
+
+        this.url = url;
+        this.privateKey = privateKey;
+
+        // Whenever we get a new subscription, we need to reconnect the session with the new subscriptions in the body
+        // We add a small debounce to this to prevent parallel subscriptions from causing multiple reconnects
+        this.on(
+            'subscriptions-updated',
+            () => {
+                // Nothing awaits this listener, so report failures as an error event instead of an unhandled rejection
+                this.reconnect().catch((error) => this.emit('error', error));
+            },
+            25,
+        );
+    }
+
+    /**
+     * Connects to the server
+     */
+    async connect(): Promise<void> {
+        // The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet.
+        // subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session.
+        if (this.subscriptions.size === 0) {
+            return;
+        }
+
+        this.sseSession ??= this.createSession();
+
+        await this.sseSession.connect();
+    }
+
+    /**
+     * Disconnects from the server
+     */
+    async disconnect(): Promise<void> {
+        this.sseSession?.disconnect();
+    }
+
+    /**
+     * Reconnects to the server
+     */
+    async reconnect(): Promise<void> {
+        // Close the current session. Its subscription list is fixed in the request body, so it can't be reused
+        this.sseSession?.close();
+        this.sseSession = undefined;
+
+        // Connect a new session with the current subscriptions
+        await this.connect();
+    }
+
+    /**
+     * Creates a new SSE session
+     * @returns The SSE session
+     */
+    private createSession(): SSESession {
+        // Build the request body
+        const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] });
+
+        // Create a new SSE session
+        const session = new SSESession(this.subscriptionUrl, {
+            onRequest: async (request): Promise<RequestInit> => {
+                // Get the path from the URL
+                const url = new URL(this.subscriptionUrl);
+                const path = url.pathname;
+
+                // Sign the request body
+                const authHeaders = BaseSyncClient.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',
+            // Reconnect when the server ends the stream (e.g. a restart or a proxy idle timeout). By default only errors are retried
+            persistent: true,
+        });
+
+        // Create a listener to re-emit the connected and disconnected events
+        session.on('connected', this.handleOnConnectionUpdated.bind(this, 'connected'));
+        session.on('disconnected', this.handleOnConnectionUpdated.bind(this, 'disconnected'));
+
+        // Create a listener to re-emit the messages
+        session.on('message', (message): void => {
+            const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(fromExtendedJson(message.data));
+            if (!parsedMessage.success) {
+                return;
+            }
+
+            this.emit('resource-instance-updated', {
+                ...parsedMessage.data,
+                value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType,
+            });
+        });
+
+        // Create a listener to re-emit the errors
+        session.on('error', (error) => {
+            this.emit('error', error);
+        });
+
+        // Return the session
+        return session;
+    }
+
+    /**
+     * Send a POST /data/write request to the server
+     * @param resources - The resources to write
+     * @returns The Resource Object with the decoded value
+     */
+    async writeMany(resources: Record<string, PayloadType>): Promise<ResourcesRecord<PayloadType>> {
+        // Sign the resource write requests
+        const signedResources = Object.entries(resources).map(([ resourceId, value ]) => {
+            const valueStr = toExtendedJson(value);
+            const valueBytes = this.textEncoder.encode(valueStr);
+
+            return BaseSyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
+        });
+
+        // Send the resource write requests to the server
+        const res = await this.sendRequest('/data/write', {
+            resources: signedResources,
+        });
+
+        // If the response is not ok, throw an error
+        if (!res.ok) {
+            const error = await res.json();
+            throw new Error(`Failed to write resources [${Object.keys(resources).join(', ')}] (${res.statusText}): ${JSON.stringify(error)}`);
+        }
+
+        // Decode the response and return the resources
+        const decodedResources = await this.decodeResponse(res);
+
+        // Return the decoded resources
+        return decodedResources;
+    }
+
+    /**
+     * Send a POST /data/read request to the server
+     * @param resourceIds - The resource IDs to read
+     * @returns The resources with the read values
+     */
+    async readMany(resourceIds: string[]): Promise<ResourcesRecord<PayloadType>> {
+        // Send the resource read requests to the server
+        const res = await this.sendRequest('/data/get', {
+            resourceId: resourceIds,
+        });
+
+        // If the response is not ok, throw an error
+        if (!res.ok) {
+            throw new Error(`Failed to read resources [${resourceIds.join(', ')}] (${res.statusText})`);
+        }
+
+        // Decode the response and return the resources
+        const decodedResources = await this.decodeResponse(res);
+
+        // Return the decoded resources
+        return decodedResources;
+    }
+
+    /**
+     * Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list
+     * @param resourceId - The resource ID to add to the subscription list
+     */
+    async subscribe(resourceId: string): Promise<void> {
+        this.subscriptions.add(resourceId);
+
+        this.emit('subscriptions-updated', this.subscriptions);
+    }
+
+    /**
+     * Destroy the current SSE Session, then re-create it without the resource ID in the subscription list
+     * @param resourceId - The resource ID to remove from the subscription list
+     */
+    async unsubscribe(resourceId: string): Promise<void> {
+        this.subscriptions.delete(resourceId);
+
+        this.emit('subscriptions-updated', this.subscriptions);
+    }
+
+    async handleOnConnectionUpdated(value: 'connected' | 'disconnected'): Promise<void> {
+        this.emit('connection-updated', value);
+    }
+
+    /**
+     * Sends a request to the server
+     * @param path - The path to send the request to
+     * @param body - The body of the request
+     * @returns The response from the server
+     */
+    private async sendRequest(path: string, body: unknown): Promise<Response> {
+        // Build the URL
+        const url = `${this.url}${path}`;
+
+        // Format the body as Extended JSON
+        const formattedBody = toExtendedJson(body);
+
+        // Send the request to the server
+        const res = await fetch(url, {
+            method: 'POST',
+            headers: {
+                'Content-Type': 'application/json',
+                ...BaseSyncClient.signRequest(this.privateKey.toBytes(), path, body),
+            },
+            body: formattedBody,
+        });
+
+        // Return the response
+        return res;
+    }
+
+    /**
+     * Decodes the blobs on the response body to the type T
+     * @param response - The response to decode
+     * @returns The decoded resources
+     */
+    private async decodeResponse(response: Response): Promise<ResourcesRecord<PayloadType>> {
+        // Decode the response from Extended JSON
+        const text = fromExtendedJson(await response.text());
+
+        // Parse the response as a record of resources
+        const resources = BaseSyncClient.resourcesResponseSchema.parse(text);
+
+        // Decode the blobs and convert them from extended json to the type T
+        const decodedResources = BaseSyncClient.decodeResourcesResponse<PayloadType>(resources);
+
+        return decodedResources;
+    }
+}

source/sse/index.ts

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

source/types.ts

@@ -0,0 +1,11 @@
+export type WriteRequest = {
+    id: string;
+    publicKey: string;
+    timestamp: number;
+
+    /**
+     * sha256(timestamp + id + toExtendedJson(value))
+     */
+    signature: string;
+    value: unknown;
+};

source/ws/client.ts

@@ -0,0 +1,410 @@
+import { ExponentialBackoff, fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
+import type { PrivateKey } from '@xo-cash/primitives';
+
+import { WsMessageSchema, WsSuccessResponseSchema, WsErrorResponseSchema, type WsMessage } from './types.ts';
+import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.ts';
+
+/** A request sent over the sync server's WebSocket connection. */
+type WsRequest = {
+    id?: string;
+    path: string;
+    body?: unknown;
+    headers?: Record<string, string>;
+};
+
+type PendingRequest = {
+    resolve: (body: unknown) => void;
+    reject: (error: Error) => void;
+};
+
+export type WsClientOptions = {
+
+    /** Called for server-pushed events and uncorrelated server errors. */
+    onMessage: (message: WsMessage) => void;
+
+    /** Called for transport errors and malformed server messages. */
+    onError: (error: Error) => void;
+};
+
+/**
+ * Minimal WebSocket client for the XO sync server.
+ *
+ * - one connection is opened at `/ws`, on demand: every request awaits the
+ *   connection first, so callers never need to wait for one themselves;
+ * - request IDs correlate concurrent read/write/unsubscribe calls;
+ * - subscriptions are remembered and re-sent on every new socket, because the
+ *   server drops a socket's topics when it closes; and
+ * - a dropped connection fails pending requests and is reopened with backoff
+ *   until `disconnect()` is called. Requests made in the meantime wait for it.
+ *
+ * Heartbeats and request timeouts are not implemented.
+ */
+export class WsClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
+    private readonly url: string;
+    private readonly privateKey: PrivateKey;
+    private readonly pendingRequests = new Map<string, PendingRequest>();
+    private readonly messageListeners = new Set<(message: WsMessage) => void>();
+    private readonly errorListeners = new Set<(error: Error) => void>();
+    private readonly textEncoder = new TextEncoder();
+
+    /** Resource IDs to (re)subscribe to whenever a socket opens. */
+    private readonly subscriptions = new Set<string>();
+
+    /**
+     * The current connection: resolves to an open socket once a connection attempt succeeds.
+     * Undefined until `connect()` and after `disconnect()`, which also means "don't reconnect".
+     */
+    private webSocket: Promise<WebSocket> | undefined;
+
+    /** Retries a failed connection attempt forever (maxAttempts 0), waiting up to 30s between attempts. */
+    private readonly connectBackoff = new ExponentialBackoff({
+        maxAttempts: 0,
+        maxDelay: 30_000,
+    });
+
+    constructor(url: string, privateKey: PrivateKey, options: Partial<WsClientOptions> = {}) {
+        super();
+
+        this.url = url;
+        this.privateKey = privateKey;
+        this.messageListeners.add(options.onMessage ?? ((): void => {}));
+        this.errorListeners.add(options.onError ?? ((): void => {}));
+    }
+
+    /** Connect, and keep reconnecting on drops until `disconnect()` is called. */
+    async connect(): Promise<void> {
+        await this.getWebSocket();
+    }
+
+    /**
+     * Close the socket and stop reconnecting.
+     *
+     * Closing a WebSocket automatically removes all of its server-side topics,
+     * so disconnect does not need to send an unsubscribe request first.
+     * Subscriptions are kept, and are replayed if the client connects again.
+     *
+     * Resolves without waiting for the socket to finish closing; `connection-updated` fires when it does.
+     */
+    async disconnect(): Promise<void> {
+        const attempt = this.webSocket;
+        this.webSocket = undefined;
+
+        this.rejectPendingRequests(new Error('WebSocket client disconnected'));
+
+        // If the attempt is still retrying, it sees it is no longer current after its next handshake,
+        // closes that socket and stops. Its rejection is handled here.
+        attempt?.then(
+            (socket) => socket.close(),
+            () => {},
+        );
+    }
+
+    /**
+     * The current socket, starting a connection attempt if there isn't one.
+     *
+     * A new socket has no topics on the server, so starting an attempt also queues a subscribe for
+     * every remembered subscription. They are sent through send(), so they go out once the socket opens.
+     */
+    private getWebSocket(): Promise<WebSocket> {
+        if (this.webSocket) {
+            return this.webSocket;
+        }
+
+        this.webSocket = this.openWebSocket();
+
+        for (const resourceId of this.subscriptions) {
+            // A failure here means the attempt was stopped by disconnect() or the socket closed before the
+            // frame was sent. A close starts a new attempt, which replays this again, so there is nothing to do.
+            this.send('/data/subscribe', { resourceId: [ resourceId ] }).catch(() => {});
+        }
+
+        return this.webSocket;
+    }
+
+    /**
+     * Retry opening a socket until one opens. Use getWebSocket() instead of calling this directly.
+     *
+     * `this.webSocket === attempt` tells whether this attempt is still the current connection.
+     * It stops being current when `disconnect()` is called.
+     */
+    private openWebSocket(): Promise<WebSocket> {
+        const attempt: Promise<WebSocket> = this.connectBackoff.run(
+            async ({ stopRetries }) => {
+                // `attempt` is only read after this await: run() calls this function before `attempt` is assigned.
+                const socket = await WsClient.waitForOpen(new WebSocket(WsClient.httpToWsUrl(this.url)));
+
+                // disconnect() was called while connecting. The thrown error rejects `attempt`.
+                if (this.webSocket !== attempt) {
+                    socket.close();
+                    stopRetries('WebSocket client disconnected');
+                    throw new Error('WebSocket client disconnected');
+                }
+
+                socket.onmessage = (event): void => this.handleMessage(String(event.data));
+                socket.onerror = (): void => this.emitError(new Error('WebSocket error'));
+                socket.onclose = async (): Promise<void> => {
+                    this.rejectPendingRequests(new Error('WebSocket connection closed'));
+                    this.emit('connection-updated', 'disconnected');
+
+                    // Reconnect, unless disconnect() was called
+                    if (this.webSocket === attempt) {
+                        this.webSocket = undefined;
+                        await this.getWebSocket();
+                    }
+                };
+
+                this.emit('connection-updated', 'connected');
+
+                return socket;
+            },
+            {
+                // Report failed handshakes, but not the error thrown above after disconnect()
+                onError: (error) => {
+                    if (this.webSocket === attempt) {
+                        this.emitError(error);
+                    }
+                },
+            },
+        );
+
+        // Nothing may be awaiting this attempt (e.g. a reconnect started by onclose), and it rejects if disconnect() stops it.
+        // Mark the rejection as handled so it isn't reported as unhandled. Callers awaiting `attempt` still receive it.
+        attempt.catch(() => {});
+
+        return attempt;
+    }
+
+    async readMany(resourceIds: string[]): Promise<ResourcesRecord<PayloadType>> {
+        // Read the resources from the server
+        const response = await this.request('/data/get', {
+            resourceId: resourceIds,
+        });
+
+        // Parse the response as a record of resources
+        const resources = BaseSyncClient.resourcesResponseSchema.parse(response);
+
+        // Decode the resources and convert the blobs from Uint8Arrays to the type T
+        const decodedResources = BaseSyncClient.decodeResourcesResponse<PayloadType>(resources);
+
+        return decodedResources;
+    }
+
+    async writeMany(resourcesToWrite: Record<string, PayloadType>): Promise<ResourcesRecord<PayloadType>> {
+        // Sign the resources
+        const signedResources = Object.entries(resourcesToWrite).map(([ resourceId, value ]) => {
+            const valueStr = toExtendedJson(value);
+            const valueBytes = this.textEncoder.encode(valueStr);
+
+            return BaseSyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
+        });
+
+        // Write the resources to the server
+        const response = await this.request('/data/write', {
+            resources: signedResources,
+        });
+
+        // Parse the response as a record of resources
+        const resources = BaseSyncClient.resourcesResponseSchema.parse(response);
+
+        // Decode the resources and convert the blobs from Uint8Arrays to the type T
+        const decodedResources = BaseSyncClient.decodeResourcesResponse<PayloadType>(resources);
+
+        return decodedResources;
+    }
+
+    /**
+     * Subscribe to one resource topic, now and on every future socket.
+     *
+     * The server never replies to a successful subscribe (the route stays pending until the topic is
+     * removed), so this resolves once the frame is sent. A failed subscribe comes back as an `error`
+     * message that matches no pending request, which handleMessage() reports through emitError().
+     *
+     * @param resourceId - The resource ID to subscribe to
+     */
+    async subscribe(resourceId: string): Promise<void> {
+        await this.send('/data/subscribe', { resourceId: [ resourceId ] });
+
+        // Remembered only after sending: if send() started a new socket, getWebSocket() replays the
+        // existing subscriptions, and this one isn't among them yet. Sending it twice in the same
+        // millisecond gives identical signatures, which the server rejects as a replay.
+        this.subscriptions.add(resourceId);
+    }
+
+    /**
+     * Unsubscribe from one resource topic.
+     * @param resourceId - The resource ID to unsubscribe from
+     */
+    async unsubscribe(resourceId: string): Promise<void> {
+        this.subscriptions.delete(resourceId);
+
+        // Not connected: the server holds no topics for us, and future sockets won't replay this one.
+        if (!this.webSocket) {
+            return;
+        }
+
+        await this.request('/data/unsubscribe', { resourceId: [ resourceId ] });
+    }
+
+    /** Send a request and wait for the response carrying the same ID. */
+    private async request(path: string, body: unknown): Promise<unknown> {
+        const id = crypto.randomUUID();
+
+        await this.send(path, body, id);
+
+        // Registered after sending, so only requests that were actually sent can be rejected by a close.
+        // The reply can't arrive first: socket messages are delivered as later events.
+        return new Promise<unknown>((resolve, reject) => {
+            this.pendingRequests.set(id, { resolve, reject });
+        });
+    }
+
+    /**
+     * Connect if needed, then sign and send one request envelope using the shared Extended JSON codec.
+     * @param path - The path of the request
+     * @param body - The body of the request
+     * @param id - The ID of the request
+     */
+    private async send(path: string, body: unknown, id: string = crypto.randomUUID()): Promise<void> {
+        const socket = await this.getWebSocket();
+
+        // The socket can close between being awaited and being used. A closed socket drops frames silently.
+        if (socket.readyState !== WebSocket.OPEN) {
+            throw new Error('WebSocket is not connected');
+        }
+
+        // Signed after connecting: the server rejects signatures older than its timestamp window (5 minutes by default).
+        const headers = BaseSyncClient.signRequest(this.privateKey.toBytes(), path, toExtendedJson(body));
+        const request: WsRequest = { id, path, body, headers };
+
+        socket.send(toExtendedJson(request));
+    }
+
+    /**
+     * Resolve with the socket once it opens, or reject if the handshake fails.
+     * @param socket - The WebSocket to wait for
+     * @returns The WebSocket
+     */
+    private static waitForOpen(socket: WebSocket): Promise<WebSocket> {
+        return new Promise((resolve, reject) => {
+            socket.onopen = (): void => resolve(socket);
+            socket.onerror = (): void => reject(new Error('WebSocket connection failed'));
+        });
+    }
+
+    /**
+     * Decode and route one frame received from the sync server.
+     * @param raw - The raw message received from the sync server
+     */
+    private handleMessage(raw: string): void {
+        const { success, data: message } = WsMessageSchema.safeParse(fromExtendedJson(raw));
+        if (!success) {
+            this.emitError(new Error('Invalid WebSocket message'));
+
+            return;
+        }
+
+        // Only response/error messages with an outstanding ID are RPC replies.
+        // Application events are left for message listeners, even if they have IDs.
+        const pending = message.id ? this.pendingRequests.get(message.id) : undefined;
+
+        // If the message has no outstanding ID, it is a subscription event so we will just notify the listeners
+        if (!pending) {
+            for (const listener of this.messageListeners) {
+                listener(message);
+            }
+
+            // Errors with no pending request are failed subscribes (see sendSubscribe) or uncorrelated server errors.
+            const error = WsErrorResponseSchema.safeParse(message);
+            if (error.success) {
+                this.emitError(new Error(`${error.data.error} (${error.data.statusCode})`));
+
+                return;
+            }
+
+            if (message.type === 'instance-changed') {
+                const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(message.data);
+                if (!parsedMessage.success) {
+                    return;
+                }
+
+                this.emit('resource-instance-updated', {
+                    ...parsedMessage.data,
+                    value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType,
+                });
+            }
+
+            return;
+        }
+
+        // Delete the pending request
+        this.pendingRequests.delete(message.id!);
+
+        // If the message is an error, reject the pending request
+        if (message.type === 'error') {
+            const error = WsErrorResponseSchema.parse(message);
+            pending.reject(new Error(`${error.error} (${error.statusCode})`));
+
+            return;
+        }
+
+        // If the message is a response, resolve the pending request
+        if (message.type === 'response') {
+            const response = WsSuccessResponseSchema.parse(message);
+
+            // If the response is not successful, reject the pending request
+            if (response.statusCode < 200 || response.statusCode >= 300) {
+                pending.reject(new Error(`Request failed (${response.statusCode})`));
+
+                return;
+            }
+
+            // If the response is successful, resolve the pending request
+            pending.resolve(response.body);
+        }
+    }
+
+    /**
+     * Helper to emit an error to all error listeners
+     * @param error - The error to emit
+     */
+    private emitError(error: Error): void {
+        for (const listener of this.errorListeners) {
+            listener(error);
+        }
+    }
+
+    /**
+     * Helper for when the client disconnects
+     * @param error - The error to reject the pending requests with
+     */
+    private rejectPendingRequests(error: Error): void {
+        for (const pending of this.pendingRequests.values()) {
+            pending.reject(error);
+        }
+
+        this.pendingRequests.clear();
+    }
+
+    /**
+     * Helper to convert an HTTP URL to a WebSocket URL
+     * @param httpUrl - The HTTP URL to convert
+     * @returns The WebSocket URL
+     */
+    static httpToWsUrl(httpUrl: string): string {
+        const url = new URL(httpUrl);
+
+        if (url.protocol === 'http:') {
+            url.protocol = 'ws:';
+        } else if (url.protocol === 'https:') {
+            url.protocol = 'wss:';
+        } else if (url.protocol !== 'ws:' && url.protocol !== 'wss:') {
+            throw new Error(`Unsupported sync server protocol: ${url.protocol}`);
+        }
+
+        // The constructor accepts either the server root or the complete `/ws` URL.
+        const pathname = url.pathname.replace(/\/+$/, '');
+        url.pathname = pathname.endsWith('/ws') ? pathname : `${pathname}/ws`;
+
+        return url.toString();
+    }
+}

source/ws/index.ts

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

source/ws/types.ts

@@ -2,24 +2,24 @@ import { z } from 'zod';
 
 /** A normal, correlated reply to a WebSocket request. */
 export const WsSuccessResponseSchema = z.object({
-  id: z.string().optional(),
+    id: z.string().optional(),
-  type: z.literal('response'),
+    type: z.literal('response'),
-  statusCode: z.number(),
+    statusCode: z.number(),
-  body: z.unknown(),
+    body: z.unknown(),
 });
 export type WsSuccessResponse = z.infer<typeof WsSuccessResponseSchema>;
 
 /** A public error returned by the server for one request. */
 export const WsErrorResponseSchema = z.object({
-  id: z.string().optional(),
+    id: z.string().optional(),
-  type: z.literal('error'),
+    type: z.literal('error'),
-  statusCode: z.number(),
+    statusCode: z.number(),
-  error: z.string(),
+    error: z.string(),
-  details: z.unknown().optional(),
+    details: z.unknown().optional(),
 });
 export type WsErrorResponse = z.infer<typeof WsErrorResponseSchema>;
 
-export const WsResponseSchema = z.discriminatedUnion('type', [WsSuccessResponseSchema, WsErrorResponseSchema]);
+export const WsResponseSchema = z.discriminatedUnion('type', [ WsSuccessResponseSchema, WsErrorResponseSchema ]);
 export type WsResponse = z.infer<typeof WsResponseSchema>;
 
 /**
@@ -30,11 +30,11 @@ export type WsResponse = z.infer<typeof WsResponseSchema>;
  * whenever the server adds another event.
  */
 export const WsEventSchema = z.object({
-  id: z.string().optional(),
+    id: z.string().optional(),
-  type: z.string(),
+    type: z.string(),
-  data: z.unknown(),
+    data: z.unknown(),
 });
 export type WsEvent = z.infer<typeof WsEventSchema>;
 
-export const WsMessageSchema = z.union([WsResponseSchema, WsEventSchema]);
+export const WsMessageSchema = z.union([ WsResponseSchema, WsEventSchema ]);
 export type WsMessage = z.infer<typeof WsMessageSchema>;

src/index.ts

@@ -1,4 +0,0 @@
-export * from './shared/client.js';
-
-export * from './sse/client.js';
-export * from './ws/client.js';

src/shared/client.ts

@@ -1,142 +0,0 @@
-import { binToHex, flattenBinArray, secp256k1, sha256 } from "@bitauth/libauth";
-
-import { toExtendedJson, EventEmitter } from "@xo-cash/utils";
-import type { WriteRequest } from "../types.js";
-import z from "zod";
-
-export type AuthenticatedRequestHeaders = {
-  'X-Public-Key': string;
-  'X-Timestamp': string;
-  'X-Signature': string;
-};
-
-export type SignedPayload = { publicKey: string; signature: string };
-
-export type XOSyncClientEventMap = {
-  'connection-updated': 'connected' | 'disconnected';
-  'message': {
-    blob: unknown;
-    timestamp: number;
-    publicKey: string;
-    signature: string;
-  };
-  'subscriptions-updated': Set<string>;
-  'error': Error;
-};
-
-export abstract class SyncClient extends EventEmitter<XOSyncClientEventMap> {
-
-  abstract connect(): Promise<void>;
-  abstract disconnect(): Promise<void>;
-  abstract write(resourceId: string, value: Record<string, unknown>): Promise<unknown>;
-  abstract read(resourceId: string): Promise<unknown>;
-  abstract subscribe(resourceId: string): Promise<void>;
-  abstract unsubscribe(resourceId: string): Promise<unknown>;
-
-  /**
-   * Derives the resource private key from the private key and resource ID
-   * @param privateKey - The private key to derive the resource private key from
-   * @param resourceId - The resource ID to derive the resource private key from
-   * @returns The resource private key
-   */
-  static deriveResourcePrivateKey(privateKey: Uint8Array, resourceId: string): Uint8Array {
-    // Convert the resource ID to bytes
-    const resourceIdBytes = new TextEncoder().encode(resourceId);
-
-    // Hash the private key and resource ID
-    return sha256.hash(flattenBinArray([ privateKey, resourceIdBytes ]));
-  }
-
-  /**
-   * Signs a payload with the private key and returns the public key and signature
-   * @param privateKey - The private key to sign the payload with
-   * @param payload - The payload to sign
-   * @returns The public key and signature
-   */
-  static signPayload(privateKey: Uint8Array, payload: string): SignedPayload {
-    // Convert the payload to Binary and hash it
-    const payloadHash = sha256.hash(new TextEncoder().encode(payload));
-
-    // Derive the public key
-    const publicKey = secp256k1.derivePublicKeyCompressed(privateKey);
-    
-    // If the public key is a string, throw an error
-    if (typeof publicKey === 'string') {
-      throw new Error('Failed to derive public key');
-    }
-
-    // Sign the payload
-    const signature = secp256k1.signMessageHashDER(privateKey, payloadHash);
-
-    // If the signature is a string, throw an error
-    if (typeof signature === 'string') {
-      throw new Error('Failed to sign message');
-    }
-
-    return {
-      publicKey: binToHex(publicKey),
-      signature: binToHex(signature),
-    };
-  }
-
-  /**
-   * Signs a write request with the private key and returns the write request
-   * @param privateKey - The private key to sign the write request with
-   * @param resourceId - The resource ID to write to
-   * @param value - The value to write
-   * @returns The write request
-   */
-  static signWriteRequest(privateKey: Uint8Array, resourceId: string, value: Uint8Array): WriteRequest {
-    // Derive the resource private key
-    const derivedKey = this.deriveResourcePrivateKey(privateKey, resourceId);
-
-    // Create the payload
-    const timestamp = Date.now();
-    const payload = `${timestamp}:${resourceId}:${toExtendedJson(value)}`;
-
-    // Sign the payload
-    const { publicKey, signature } = this.signPayload(derivedKey, payload);
-
-    return { 
-      id: resourceId,
-      value,
-      publicKey,
-      timestamp,
-      signature,
-    };
-  }
-
-  /**
-   * Signs a request with the private key and returns the public key, timestamp, and signature
-   * @param privateKey - The private key to sign the request with
-   * @param path - The path of the request
-   * @param body - The body of the request
-   * @returns The public key, timestamp, and signature
-   */
-  static signRequest(privateKey: Uint8Array, path: string, body: unknown): AuthenticatedRequestHeaders {
-    // Create the payload
-    const timestamp = Date.now();
-
-    // 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);
-
-    return {
-      'X-Public-Key': publicKey,
-      'X-Timestamp': timestamp.toString(),
-      'X-Signature': signature,
-    };
-  }
-
-  static messageSchema = z.object({
-    blob: z.unknown(),
-    timestamp: z.number(),
-    publicKey: z.string(),
-    signature: z.string(),
-  });
-}

src/sse/client.ts

@@ -1,187 +0,0 @@
-import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
-import { PrivateKey } from '@xo-cash/primitives'
-
-import { SyncClient } from '../shared/client.js';
-
-export class SSEClient extends SyncClient {
-  private readonly url: string;
-  public sseSession: SSESession | undefined;
-  private readonly subscriptions: Set<string> = new Set();
-
-  private privateKey: PrivateKey;
-  private textEncoder = new TextEncoder();
-
-  private get subscriptionUrl() {
-    return `${this.url}/data/subscribe`;
-  }
-
-  constructor(url: string, privateKey: PrivateKey) {
-    super();
-
-    this.url = url;
-    this.privateKey = privateKey;
-
-    // Whenever we get a new subscription, we need to reconnect the session with the new subscriptions in the body
-    // We add a small debounce to this to prevent parallel subscriptions from causing multiple reconnects
-    this.on('subscriptions-updated', () => {
-      // Nothing awaits this listener, so report failures as an error event instead of an unhandled rejection
-      this.reconnect().catch((error) => this.emit('error', error));
-    }, 25);
-  }
-
-  async connect() {
-    // The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet.
-    // subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session.
-    if (this.subscriptions.size === 0) {
-      return;
-    }
-
-    this.sseSession ??= this.createSession();
-
-    await this.sseSession.connect();
-  }
-
-  async disconnect() {
-    await this.sseSession?.disconnect();
-  }
-
-  async reconnect() {
-    // Close the current session. Its subscription list is fixed in the request body, so it can't be reused
-    this.sseSession?.close();
-    this.sseSession = undefined;
-
-    // Connect a new session with the current subscriptions
-    await this.connect();
-  }
-
-  private createSession() {
-    // Build the request body
-    const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] });
-
-    // Create a new SSE session
-    const session = 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',
-      // Reconnect when the server ends the stream (e.g. a restart or a proxy idle timeout). By default only errors are retried
-      persistent: true,
-    });
-
-    // Create a listener to re-emit the connected and disconnected events
-    session.on('connected', this.handleOnConnectionUpdated.bind(this, 'connected'));
-    session.on('disconnected', this.handleOnConnectionUpdated.bind(this, 'disconnected'));
-
-    // Create a listener to re-emit the messages
-    session.on('message', (message) => {
-      const parsedMessage = SyncClient.messageSchema.safeParse(fromExtendedJson(message.data));
-      if (!parsedMessage.success) {
-        return;
-      }
-
-      this.emit('message', parsedMessage.data);
-    });
-
-    // Create a listener to re-emit the errors
-    session.on('error', (error) => {
-      this.emit('error', error);
-    });
-
-    // Return the session
-    return session;
-  }
-  
-  // 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)}`);
-    }
-
-    const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[];
-
-    return text;
-  }
-
-  // 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}`);
-    }
-
-    const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[];
-
-    return 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);
-
-    this.emit('subscriptions-updated', this.subscriptions);
-  }
-
-  // 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);
-
-    this.emit('subscriptions-updated', this.subscriptions);
-  }
-
-  async handleOnConnectionUpdated(value: 'connected' | 'disconnected') {
-    this.emit('connection-updated', value);
-  }
-}

src/sse/index.ts

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

src/types.ts

@@ -1,11 +0,0 @@
-export type WriteRequest = {
-  id: string;
-  publicKey: string;
-  timestamp: number;
-
-  /**
-   * sha256(timestamp + id + toExtendedJson(value))
-   */
-  signature: string;
-  value: unknown;
-};

src/ws/client.ts

@@ -1,348 +0,0 @@
-import { ExponentialBackoff, fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
-import type { PrivateKey } from '@xo-cash/primitives';
-
-import { WsMessageSchema, WsSuccessResponseSchema, WsErrorResponseSchema, type WsMessage } from './types.js';
-import { SyncClient } from '../shared/client.js';
-
-/** A request sent over the sync server's WebSocket connection. */
-type WsRequest = {
-  id?: string;
-  path: string;
-  body?: unknown;
-  headers?: Record<string, string>;
-};
-
-type PendingRequest = {
-  resolve: (body: unknown) => void;
-  reject: (error: Error) => void;
-};
-
-export type WsClientOptions = {
-  /** Called for server-pushed events and uncorrelated server errors. */
-  onMessage: (message: WsMessage) => void;
-
-  /** Called for transport errors and malformed server messages. */
-  onError: (error: Error) => void;
-};
-
-/**
- * Minimal WebSocket client for the XO sync server.
- *
- * - one connection is opened at `/ws`, on demand: every request awaits the
- *   connection first, so callers never need to wait for one themselves;
- * - request IDs correlate concurrent read/write/unsubscribe calls;
- * - subscriptions are remembered and re-sent on every new socket, because the
- *   server drops a socket's topics when it closes; and
- * - a dropped connection fails pending requests and is reopened with backoff
- *   until `disconnect()` is called. Requests made in the meantime wait for it.
- *
- * Heartbeats and request timeouts are not implemented.
- */
-export class WsClient extends SyncClient {
-  private readonly url: string;
-  private readonly privateKey: PrivateKey;
-  private readonly pendingRequests = new Map<string, PendingRequest>();
-  private readonly messageListeners = new Set<(message: WsMessage) => void>();
-  private readonly errorListeners = new Set<(error: Error) => void>();
-
-  /** Resource IDs to (re)subscribe to whenever a socket opens. */
-  private readonly subscriptions = new Set<string>();
-
-  /**
-   * The current connection: resolves to an open socket once a connection attempt succeeds.
-   * Undefined until `connect()` and after `disconnect()`, which also means "don't reconnect".
-   */
-  private webSocket: Promise<WebSocket> | undefined;
-
-  /** Retries a failed connection attempt forever (maxAttempts 0), waiting up to 30s between attempts. */
-  private readonly connectBackoff = new ExponentialBackoff({ maxAttempts: 0, maxDelay: 30_000 });
-
-  constructor(
-    url: string,
-    privateKey: PrivateKey,
-    options: Partial<WsClientOptions> = {},
-  ) {
-    super();
-
-    this.url = url;
-    this.privateKey = privateKey;
-    this.messageListeners.add(options.onMessage ?? (() => {}));
-    this.errorListeners.add(options.onError ?? (() => {}));
-  }
-
-  /** Connect, and keep reconnecting on drops until `disconnect()` is called. */
-  async connect(): Promise<void> {
-    await this.getWebSocket();
-  }
-
-  /**
-   * Close the socket and stop reconnecting.
-   *
-   * Closing a WebSocket automatically removes all of its server-side topics,
-   * so disconnect does not need to send an unsubscribe request first.
-   * Subscriptions are kept, and are replayed if the client connects again.
-   *
-   * Resolves without waiting for the socket to finish closing; `connection-updated` fires when it does.
-   */
-  async disconnect(): Promise<void> {
-    const attempt = this.webSocket;
-    this.webSocket = undefined;
-
-    this.rejectPendingRequests(new Error('WebSocket client disconnected'));
-
-    // If the attempt is still retrying, it sees it is no longer current after its next handshake,
-    // closes that socket and stops. Its rejection is handled here.
-    attempt?.then((socket) => socket.close(), () => {});
-  }
-
-  /**
-   * The current socket, starting a connection attempt if there isn't one.
-   *
-   * A new socket has no topics on the server, so starting an attempt also queues a subscribe for
-   * every remembered subscription. They are sent through send(), so they go out once the socket opens.
-   */
-  private getWebSocket(): Promise<WebSocket> {
-    if (this.webSocket) {
-      return this.webSocket;
-    }
-
-    this.webSocket = this.openWebSocket();
-
-    for (const resourceId of this.subscriptions) {
-      // A failure here means the attempt was stopped by disconnect() or the socket closed before the
-      // frame was sent. A close starts a new attempt, which replays this again, so there is nothing to do.
-      this.send('/data/subscribe', { resourceId: [resourceId] }).catch(() => {});
-    }
-
-    return this.webSocket;
-  }
-
-  /**
-   * Retry opening a socket until one opens. Use getWebSocket() instead of calling this directly.
-   *
-   * `this.webSocket === attempt` tells whether this attempt is still the current connection.
-   * It stops being current when `disconnect()` is called.
-   */
-  private openWebSocket(): Promise<WebSocket> {
-    const attempt: Promise<WebSocket> = this.connectBackoff.run(async ({ stopRetries }) => {
-      // `attempt` is only read after this await: run() calls this function before `attempt` is assigned.
-      const socket = await WsClient.waitForOpen(new WebSocket(WsClient.httpToWsUrl(this.url)));
-
-      // disconnect() was called while connecting. The thrown error rejects `attempt`.
-      if (this.webSocket !== attempt) {
-        socket.close();
-        stopRetries('WebSocket client disconnected');
-        throw new Error('WebSocket client disconnected');
-      }
-
-      socket.onmessage = (event) => this.handleMessage(String(event.data));
-      socket.onerror = () => this.emitError(new Error('WebSocket error'));
-      socket.onclose = () => {
-        this.rejectPendingRequests(new Error('WebSocket connection closed'));
-        this.emit('connection-updated', 'disconnected');
-
-        // Reconnect, unless disconnect() was called
-        if (this.webSocket === attempt) {
-          this.webSocket = undefined;
-          this.getWebSocket();
-        }
-      };
-
-      this.emit('connection-updated', 'connected');
-
-      return socket;
-    }, {
-      // Report failed handshakes, but not the error thrown above after disconnect()
-      onError: (error) => {
-        if (this.webSocket === attempt) {
-          this.emitError(error);
-        }
-      },
-    });
-
-    // Nothing may be awaiting this attempt (e.g. a reconnect started by onclose), and it rejects if disconnect() stops it.
-    // Mark the rejection as handled so it isn't reported as unhandled. Callers awaiting `attempt` still receive it.
-    attempt.catch(() => {});
-
-    return attempt;
-  }
-
-  /** Read every stored instance of one resource. */
-  async read(resourceId: string): Promise<unknown> {
-    return this.request('/data/get', { resourceId: [resourceId] });
-  }
-
-  /**
-   * Sign and write one resource, using the same wire format as the SSE client.
-   */
-  async write(resourceId: string, value: Record<string, unknown>): Promise<unknown> {
-    const valueBytes = new TextEncoder().encode(toExtendedJson(value));
-    const resource = SyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
-
-    return this.request('/data/write', { resources: [resource] });
-  }
-
-  /**
-   * Subscribe to one resource topic, now and on every future socket.
-   *
-   * The server never replies to a successful subscribe (the route stays pending until the topic is
-   * removed), so this resolves once the frame is sent. A failed subscribe comes back as an `error`
-   * message that matches no pending request, which handleMessage() reports through emitError().
-   */
-  async subscribe(resourceId: string): Promise<void> {
-    await this.send('/data/subscribe', { resourceId: [resourceId] });
-
-    // Remembered only after sending: if send() started a new socket, getWebSocket() replays the
-    // existing subscriptions, and this one isn't among them yet. Sending it twice in the same
-    // millisecond gives identical signatures, which the server rejects as a replay.
-    this.subscriptions.add(resourceId);
-  }
-
-  /**
-   * Unsubscribe from one resource topic.
-   */
-  async unsubscribe(resourceId: string): Promise<void> {
-    this.subscriptions.delete(resourceId);
-
-    // Not connected: the server holds no topics for us, and future sockets won't replay this one.
-    if (!this.webSocket) {
-      return;
-    }
-
-    await this.request('/data/unsubscribe', { resourceId: [resourceId] });
-  }
-
-  /** Send a request and wait for the response carrying the same ID. */
-  private async request(path: string, body: unknown): Promise<unknown> {
-    const id = crypto.randomUUID();
-
-    await this.send(path, body, id);
-
-    // Registered after sending, so only requests that were actually sent can be rejected by a close.
-    // The reply can't arrive first: socket messages are delivered as later events.
-    return new Promise<unknown>((resolve, reject) => {
-      this.pendingRequests.set(id, { resolve, reject });
-    });
-  }
-
-  /** Connect if needed, then sign and send one request envelope using the shared Extended JSON codec. */
-  private async send(path: string, body: unknown, id: string = crypto.randomUUID()): Promise<void> {
-    const socket = await this.getWebSocket();
-
-    // The socket can close between being awaited and being used. A closed socket drops frames silently.
-    if (socket.readyState !== WebSocket.OPEN) {
-      throw new Error('WebSocket is not connected');
-    }
-
-    // Signed after connecting: the server rejects signatures older than its timestamp window (5 minutes by default).
-    const headers = SyncClient.signRequest(this.privateKey.toBytes(), path, toExtendedJson(body));
-    const request: WsRequest = { id, path, body, headers };
-
-    socket.send(toExtendedJson(request));
-  }
-
-  /** Resolve with the socket once it opens, or reject if the handshake fails. */
-  private static waitForOpen(socket: WebSocket): Promise<WebSocket> {
-    return new Promise((resolve, reject) => {
-      socket.onopen = () => resolve(socket);
-      socket.onerror = () => reject(new Error('WebSocket connection failed'));
-    });
-  }
-
-  /** Decode and route one frame received from the sync server. */
-  private handleMessage(raw: string): void {
-    const { success, data: message } = WsMessageSchema.safeParse(fromExtendedJson(raw));
-    if (!success) {
-      this.emitError(new Error('Invalid WebSocket message'));
-      return;
-    }
-
-    // Only response/error messages with an outstanding ID are RPC replies.
-    // Application events are left for message listeners, even if they have IDs.
-    const pending = message.id ? this.pendingRequests.get(message.id) : undefined;
-
-    // If the message has no outstanding ID, it is a subscription event so we will just notify the listeners
-    if (!pending) {
-      for (const listener of this.messageListeners) {
-        listener(message);
-      }
-
-      // Errors with no pending request are failed subscribes (see sendSubscribe) or uncorrelated server errors.
-      const error = WsErrorResponseSchema.safeParse(message);
-      if (error.success) {
-        this.emitError(new Error(`${error.data.error} (${error.data.statusCode})`));
-        return;
-      }
-
-      if (message.type === 'instance-changed') {
-        const parsedMessage = SyncClient.messageSchema.safeParse(message.data);
-        if (!parsedMessage.success) {
-          return;
-        }
-
-        this.emit('message', parsedMessage.data);
-      }
-
-      return;
-    }
-
-    // Delete the pending request
-    this.pendingRequests.delete(message.id!);
-
-    // If the message is an error, reject the pending request
-    if (message.type === 'error') {
-      const error = WsErrorResponseSchema.parse(message);
-      pending.reject(new Error(`${error.error} (${error.statusCode})`));
-      return;
-    }
-
-    // If the message is a response, resolve the pending request
-    if (message.type === 'response') {
-      const response = WsSuccessResponseSchema.parse(message);
-
-      // If the response is not successful, reject the pending request
-      if (response.statusCode < 200 || response.statusCode >= 300) {
-        pending.reject(new Error(`Request failed (${response.statusCode})`));
-        return;
-      }
-
-      // If the response is successful, resolve the pending request
-      pending.resolve(response.body);
-    }
-  }
-
-  // Helper to emit an error to all error listeners
-  private emitError(error: Error): void {
-    for (const listener of this.errorListeners) {
-      listener(error);
-    }
-  }
-
-  // Helper for when the client disconnects
-  private rejectPendingRequests(error: Error): void {
-    for (const pending of this.pendingRequests.values()) {
-      pending.reject(error);
-    }
-
-    this.pendingRequests.clear();
-  }
-
-  // Helper to convert an HTTP URL to a WebSocket URL
-  static httpToWsUrl(httpUrl: string): string {
-    const url = new URL(httpUrl);
-
-    if (url.protocol === 'http:') {
-      url.protocol = 'ws:';
-    } else if (url.protocol === 'https:') {
-      url.protocol = 'wss:';
-    } else if (url.protocol !== 'ws:' && url.protocol !== 'wss:') {
-      throw new Error(`Unsupported sync server protocol: ${url.protocol}`);
-    }
-
-    // The constructor accepts either the server root or the complete `/ws` URL.
-    const pathname = url.pathname.replace(/\/+$/, '');
-    url.pathname = pathname.endsWith('/ws') ? pathname : `${pathname}/ws`;
-
-    return url.toString();
-  }
-}

src/ws/index.ts

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

tsconfig.json

@@ -1,31 +1,37 @@
 {
-  // Visit https://aka.ms/tsconfig to read more about this file
+    // Visit https://aka.ms/tsconfig to read more about this file
-  "compilerOptions": {
+    "compilerOptions": {
-    // File Layout
+        // File Layout
-    "outDir": "./dist",
+        "rootDir": "./source",
+        "outDir": "./dist",
 
-    // Environment Settings
+        // Environment Settings
-    // See also https://aka.ms/tsconfig/module
+        // See also https://aka.ms/tsconfig/module
-    "module": "nodenext",
+        "module": "nodenext",
-    "target": "esnext",
+        "target": "esnext",
-    "types": ["node"],
+        "types": ["node"],
 
-    // Other Outputs
+        // Other Outputs
-    "sourceMap": true,
+        "sourceMap": true,
-    "declaration": true,
+        "declaration": true,
-    "declarationMap": true,
+        "declarationMap": true,
 
-    // Stricter Typechecking Options
+        // Stricter Typechecking Options
-    "noUncheckedIndexedAccess": true,
+        "noUncheckedIndexedAccess": true,
-    // "exactOptionalPropertyTypes": true,
+        "exactOptionalPropertyTypes": true,
 
-    // Recommended Options
+        // Recommended Options
-    "strict": true,
+        "strict": true,
-    "jsx": "react-jsx",
+        "jsx": "react-jsx",
-    "verbatimModuleSyntax": true,
+        "verbatimModuleSyntax": true,
-    "isolatedModules": true,
+        "isolatedModules": true,
-    "noUncheckedSideEffectImports": true,
+        "noUncheckedSideEffectImports": true,
-    "moduleDetection": "force",
+        "moduleDetection": "force",
-    "skipLibCheck": true,
+        "skipLibCheck": true,
-  }
+
+        // GP
+        "allowImportingTsExtensions": true,
+        "rewriteRelativeImportExtensions": true
+    },
+    "include": ["source/**/*"]
 }