Harvmaster/sync-server-example
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Formatting

Browse Source
This commit is contained in:
Harvmaster
2026-10-22 15:19:06 +00:00
parent 9cef3be6cb
commit 7f9bb97990
20 changed files with 12364 additions and 12277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.cspell.json
+2 -2
View File
@@ -1,4 +1,4 @@
{ {
"version": "0.1", "version": "0.1",
"import": ["@generalprotocols/cspell-dictionary/cspell.json"] "import": ["@generalprotocols/cspell-dictionary/cspell.json"]
} }
.gitignore
+3 -1
View File
@@ -1,2 +1,4 @@
/node_modules /node_modules
/dist /dist
/coverage
/docs
.prettierrc.mjs
+23
View File
@@ -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
+138 -132
View File
@@ -1,12 +1,12 @@
import { subtle, getRandomValues } from "node:crypto"; import { subtle, getRandomValues } from 'node:crypto';
import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils"; import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
import { sha256, binToBase64, binToHex } from "@bitauth/libauth"; import { sha256, binToBase64, binToHex } from '@bitauth/libauth';
import { BaseSyncClient, type Resource, type ResourcesRecord } from "../../source/shared/base-client.js"; import { BaseSyncClient, type Resource, type ResourcesRecord } from '../../source/shared/base-client.js';
type ResourceKey = { type ResourceKey = {
resourceId: string; resourceId: string;
keyMaterial: Uint8Array; keyMaterial: Uint8Array;
}; };
/** /**
@@ -17,160 +17,166 @@ type ResourceKey = {
* resource ID, so the resource ID must be unguessable (high entropy) for this to be secure. * resource ID, so the resource ID must be unguessable (high entropy) for this to be secure.
*/ */
export class EncryptedClientPerResource<T = unknown> extends BaseSyncClient<T> { export class EncryptedClientPerResource<T = unknown> extends BaseSyncClient<T> {
/** Maps the hashed (server-side) resource ID to the original resource ID and its key */ /** Maps the hashed (server-side) resource ID to the original resource ID and its key */
private readonly resourceKeys = new Map<string, ResourceKey>(); private readonly resourceKeys = new Map<string, ResourceKey>();
constructor(private readonly client: BaseSyncClient<string>) { constructor(private readonly client: BaseSyncClient<string>) {
super(); super();
this.client.on('resource-instance-updated', async (message) => { this.client.on('resource-instance-updated', async (message) => {
// Ignore updates for resources we don't hold the key for // Ignore updates for resources we don't hold the key for
const resourceKey = this.resourceKeys.get(message.resourceId); const resourceKey = this.resourceKeys.get(message.resourceId);
if (!resourceKey) return; if (!resourceKey) return;
try { try {
this.emit('resource-instance-updated', { this.emit('resource-instance-updated', {
...message, ...message,
resourceId: resourceKey.resourceId, resourceId: resourceKey.resourceId,
value: await EncryptedClientPerResource.decrypt(resourceKey.keyMaterial, message.value) as T, value: (await EncryptedClientPerResource.decrypt(resourceKey.keyMaterial, message.value)) as T,
});
} catch (error) {
this.emit('error', error instanceof Error ? error : new Error(String(error)));
}
}); });
} 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; /**
} * 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 });
* 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) => ({ return hashedResourceId;
...instance, }
resourceId,
value: await EncryptedClientPerResource.decrypt(keyMaterial, instance.value) as T,
})));
return [ resourceId, { ...resource, instances } ] as const; /**
})); * 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);
return Object.fromEntries(entries); if (!resourceKey) {
} throw new Error(`No key found for resource ${hashedResourceId}`);
}
public async connect(): Promise<void> { return resourceKey;
return this.client.connect(); }
}
public async disconnect(): Promise<void> { /**
return this.client.disconnect(); * 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);
public async readMany(resourceIds: string[]): Promise<ResourcesRecord<T>> { const instances: Resource<T>[] = await Promise.all(
const hashedResourceIds = resourceIds.map((resourceId) => this.hashResourceId(resourceId)); resource.instances.map(async (instance) => ({
...instance,
resourceId,
value: (await EncryptedClientPerResource.decrypt(keyMaterial, instance.value)) as T,
})),
);
const encryptedResources = await this.client.readMany(hashedResourceIds); return [resourceId, { ...resource, instances }] as const;
}),
);
return this.decryptResources(encryptedResources); return Object.fromEntries(entries);
} }
public async writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>> { public async connect(): Promise<void> {
const encryptedEntries = await Promise.all(Object.entries(resources).map(async ([ resourceId, value ]) => { return this.client.connect();
const hashedResourceId = this.hashResourceId(resourceId); }
const { keyMaterial } = this.getResourceKey(hashedResourceId);
return [ hashedResourceId, await EncryptedClientPerResource.encrypt(keyMaterial, value) ] as const; public async disconnect(): Promise<void> {
})); return this.client.disconnect();
}
const response = await this.client.writeMany(Object.fromEntries(encryptedEntries)); public async readMany(resourceIds: string[]): Promise<ResourcesRecord<T>> {
const hashedResourceIds = resourceIds.map((resourceId) => this.hashResourceId(resourceId));
return this.decryptResources(response); const encryptedResources = await this.client.readMany(hashedResourceIds);
}
public async subscribe(resourceId: string): Promise<void> { return this.decryptResources(encryptedResources);
return this.client.subscribe(this.hashResourceId(resourceId)); }
}
public async unsubscribe(resourceId: string): Promise<void> { public async writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>> {
return this.client.unsubscribe(this.hashResourceId(resourceId)); 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;
* 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']);
}
/** const response = await this.client.writeMany(Object.fromEntries(encryptedEntries));
* 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 return this.decryptResources(response);
const encodedValue = new TextEncoder().encode(toExtendedJson(value)); }
const encrypted = await subtle.encrypt({ name: 'AES-GCM', iv }, key, encodedValue); public async subscribe(resourceId: string): Promise<void> {
return this.client.subscribe(this.hashResourceId(resourceId));
}
// Prepend the IV so it can be recovered when decrypting public async unsubscribe(resourceId: string): Promise<void> {
return binToBase64(Buffer.concat([ iv, new Uint8Array(encrypted) ])); return this.client.unsubscribe(this.hashResourceId(resourceId));
} }
/** /**
* Decrypts a value encrypted by `encrypt` * Imports raw key material as an AES-GCM key
* @param keyMaterial - The encryption key material * @param keyMaterial - The 32 byte key material
* @param value - The IV and ciphertext as a base64 string * @returns The AES-GCM CryptoKey
* @returns The decrypted value */
*/ private static importKey(keyMaterial: Uint8Array) {
public static async decrypt(keyMaterial: Uint8Array, value: string): Promise<unknown> { return subtle.importKey('raw', Buffer.from(keyMaterial), { name: 'AES-GCM' }, false, ['encrypt', 'decrypt']);
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'); * Encrypts a value using AES-GCM with a random 12 byte IV
const iv = bytes.subarray(0, 12); * @param keyMaterial - The encryption key material
const encryptedValue = bytes.subarray(12); * @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));
const decrypted = await subtle.decrypt({ name: 'AES-GCM', iv }, key, encryptedValue); // Encode the value as extended JSON and then to UTF-8 bytes
const encodedValue = new TextEncoder().encode(toExtendedJson(value));
return fromExtendedJson(new TextDecoder().decode(decrypted)); 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
+104 -95
View File
@@ -1,127 +1,136 @@
import { subtle, getRandomValues } from "node:crypto"; import { subtle, getRandomValues } from 'node:crypto';
import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils"; import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
import { sha256, binToBase64 } from "@bitauth/libauth"; import { sha256, binToBase64 } from '@bitauth/libauth';
import { BaseSyncClient, type ResourcesRecord } from "../../source/shared/base-client.js"; import { BaseSyncClient, type ResourcesRecord } from '../../source/shared/base-client.js';
/** /**
* Encrypts every resource with a single shared key. * Encrypts every resource with a single shared key.
*/ */
export class EncryptedClient<T = unknown> extends BaseSyncClient<T> { export class EncryptedClient<T = unknown> extends BaseSyncClient<T> {
private readonly encryptionKey: Uint8Array; private readonly encryptionKey: Uint8Array;
constructor(private readonly client: BaseSyncClient<string>, encryptionKey: Uint8Array | string) { constructor(
super(); private readonly client: BaseSyncClient<string>,
encryptionKey: Uint8Array | string,
) {
super();
// Hash the encryption key into 32 bytes if it is a string // 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.encryptionKey = typeof encryptionKey === 'string' ? sha256.hash(new TextEncoder().encode(encryptionKey)) : encryptionKey;
this.client.on('resource-instance-updated', async (message) => { this.client.on('resource-instance-updated', async (message) => {
try { try {
this.emit('resource-instance-updated', { this.emit('resource-instance-updated', {
...message, ...message,
value: await EncryptedClient.decrypt(this.encryptionKey, message.value) as T, value: (await EncryptedClient.decrypt(this.encryptionKey, message.value)) as T,
});
} catch (error) {
this.emit('error', error instanceof Error ? error : new Error(String(error)));
}
}); });
} catch (error) { }
this.emit('error', error instanceof Error ? error : new Error(String(error)));
}
});
}
/** /**
* Decrypts every instance of every resource in a record. * Decrypts every instance of every resource in a record.
* @param resources - The encrypted resources record * @param resources - The encrypted resources record
* @returns The resources record with decrypted values * @returns The resources record with decrypted values
*/ */
private async decryptResources(resources: ResourcesRecord<string>): Promise<ResourcesRecord<T>> { private async decryptResources(resources: ResourcesRecord<string>): Promise<ResourcesRecord<T>> {
const entries = await Promise.all(Object.entries(resources).map(async ([ resourceId, resource ]) => { const entries = await Promise.all(
const instances = await Promise.all(resource.instances.map(async (instance) => ({ Object.entries(resources).map(async ([resourceId, resource]) => {
...instance, const instances = await Promise.all(
value: await EncryptedClient.decrypt(this.encryptionKey, instance.value) as T, resource.instances.map(async (instance) => ({
}))); ...instance,
value: (await EncryptedClient.decrypt(this.encryptionKey, instance.value)) as T,
})),
);
return [ resourceId, { ...resource, instances } ] as const; return [resourceId, { ...resource, instances }] as const;
})); }),
);
return Object.fromEntries(entries); return Object.fromEntries(entries);
} }
public async connect(): Promise<void> { public async connect(): Promise<void> {
return this.client.connect(); return this.client.connect();
} }
public async disconnect(): Promise<void> { public async disconnect(): Promise<void> {
return this.client.disconnect(); return this.client.disconnect();
} }
public async readMany(resourceIds: string[]): Promise<ResourcesRecord<T>> { public async readMany(resourceIds: string[]): Promise<ResourcesRecord<T>> {
const encryptedResources = await this.client.readMany(resourceIds); const encryptedResources = await this.client.readMany(resourceIds);
return this.decryptResources(encryptedResources); return this.decryptResources(encryptedResources);
} }
public async writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>> { public async writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>> {
const encryptedEntries = await Promise.all(Object.entries(resources).map(async ([ resourceId, value ]) => { const encryptedEntries = await Promise.all(
return [ resourceId, await EncryptedClient.encrypt(this.encryptionKey, value) ] as const; 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)); const response = await this.client.writeMany(Object.fromEntries(encryptedEntries));
return this.decryptResources(response); return this.decryptResources(response);
} }
public async subscribe(resourceId: string): Promise<void> { public async subscribe(resourceId: string): Promise<void> {
return this.client.subscribe(resourceId); return this.client.subscribe(resourceId);
} }
public async unsubscribe(resourceId: string): Promise<void> { public async unsubscribe(resourceId: string): Promise<void> {
return this.client.unsubscribe(resourceId); return this.client.unsubscribe(resourceId);
} }
/** /**
* Imports raw key material as an AES-GCM key * Imports raw key material as an AES-GCM key
* @param keyMaterial - The 32 byte key material * @param keyMaterial - The 32 byte key material
* @returns The AES-GCM CryptoKey * @returns The AES-GCM CryptoKey
*/ */
private static importKey(keyMaterial: Uint8Array) { private static importKey(keyMaterial: Uint8Array) {
return subtle.importKey('raw', Buffer.from(keyMaterial), { name: 'AES-GCM' }, false, ['encrypt', 'decrypt']); 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 * Encrypts a value using AES-GCM with a random 12 byte IV
* @param keyMaterial - The encryption key material * @param keyMaterial - The encryption key material
* @param value - The value to encrypt * @param value - The value to encrypt
* @returns The IV and ciphertext as a base64 string * @returns The IV and ciphertext as a base64 string
*/ */
public static async encrypt(keyMaterial: Uint8Array, value: unknown): Promise<string> { public static async encrypt(keyMaterial: Uint8Array, value: unknown): Promise<string> {
const key = await this.importKey(keyMaterial); const key = await this.importKey(keyMaterial);
const iv = getRandomValues(new Uint8Array(12)); const iv = getRandomValues(new Uint8Array(12));
// Encode the value as extended JSON and then to UTF-8 bytes // Encode the value as extended JSON and then to UTF-8 bytes
const encodedValue = new TextEncoder().encode(toExtendedJson(value)); const encodedValue = new TextEncoder().encode(toExtendedJson(value));
const encrypted = await subtle.encrypt({ name: 'AES-GCM', iv }, key, encodedValue); const encrypted = await subtle.encrypt({ name: 'AES-GCM', iv }, key, encodedValue);
// Prepend the IV so it can be recovered when decrypting // Prepend the IV so it can be recovered when decrypting
return binToBase64(Buffer.concat([ iv, new Uint8Array(encrypted) ])); return binToBase64(Buffer.concat([iv, new Uint8Array(encrypted)]));
} }
/** /**
* Decrypts a value encrypted by `encrypt` * Decrypts a value encrypted by `encrypt`
* @param keyMaterial - The encryption key material * @param keyMaterial - The encryption key material
* @param value - The IV and ciphertext as a base64 string * @param value - The IV and ciphertext as a base64 string
* @returns The decrypted value * @returns The decrypted value
*/ */
public static async decrypt(keyMaterial: Uint8Array, value: string): Promise<unknown> { public static async decrypt(keyMaterial: Uint8Array, value: string): Promise<unknown> {
const key = await this.importKey(keyMaterial); const key = await this.importKey(keyMaterial);
// Decode the base64 first: the IV is 12 bytes, not 12 base64 characters // Decode the base64 first: the IV is 12 bytes, not 12 base64 characters
const bytes = Buffer.from(value, 'base64'); const bytes = Buffer.from(value, 'base64');
const iv = bytes.subarray(0, 12); const iv = bytes.subarray(0, 12);
const encryptedValue = bytes.subarray(12); const encryptedValue = bytes.subarray(12);
const decrypted = await subtle.decrypt({ name: 'AES-GCM', iv }, key, encryptedValue); const decrypted = await subtle.decrypt({ name: 'AES-GCM', iv }, key, encryptedValue);
return fromExtendedJson(new TextDecoder().decode(decrypted)); return fromExtendedJson(new TextDecoder().decode(decrypted));
} }
} }
demo/shared.ts
+21 -19
View File
@@ -1,26 +1,28 @@
import type { BaseSyncClient, ResourcesRecord } from "../source/shared/base-client.js"; import type { BaseSyncClient, ResourcesRecord } from '../source/shared/base-client.js';
export const writeMany = async (client: BaseSyncClient<{ message: string }>): Promise<ResourcesRecord<{ message: string }>> => { export const writeMany = async (client: BaseSyncClient<{ message: string }>): Promise<ResourcesRecord<{ message: string }>> => {
// Create a group of resources with changes // Create a group of resources with changes
const resources = Object.fromEntries(Array.from({ length: 10 }, (_, index) => [ const resources = Object.fromEntries(
`resource-${index}`, Array.from({ length: 10 }, (_, index) => [
{ `resource-${index}`,
message: `Hello, world! this is the minimal sync client using ${client.constructor.name} ${index}`, {
}, message: `Hello, world! this is the minimal sync client using ${client.constructor.name} ${index}`,
])); },
]),
);
// Write the resources to the server // Write the resources to the server
const res = await client.writeMany(resources); const res = await client.writeMany(resources);
return res; return res;
} };
export const readMany = async (client: BaseSyncClient<{ message: string }>,): Promise<ResourcesRecord<{ message: string }>> => { export const readMany = async (client: BaseSyncClient<{ message: string }>): Promise<ResourcesRecord<{ message: string }>> => {
// Create a list of resource IDs // Create a list of resource IDs
const resources = Array.from({ length: 10 }, (_, index) => `resource-${index}`); const resources = Array.from({ length: 10 }, (_, index) => `resource-${index}`);
// Read the resources from the server // Read the resources from the server
const res = await client.readMany(resources); const res = await client.readMany(resources);
return res; return res;
} };
demo/suite.ts
+50 -46
View File
@@ -1,74 +1,78 @@
import { PrivateKey } from "@xo-cash/primitives"; import { PrivateKey } from '@xo-cash/primitives';
// Core clients // Core clients
import type { BaseSyncClient } from "../source/shared/base-client.js"; import type { BaseSyncClient } from '../source/shared/base-client.js';
import { SSEClient } from '../source/sse/client.js'; import { SSEClient } from '../source/sse/client.js';
import { WsClient } from "../source/ws/client.js"; import { WsClient } from '../source/ws/client.js';
// Experimental clients // Experimental clients
import { EncryptedClientPerResource } from "./experimental/encrypted-client-per-resource.js"; import { EncryptedClientPerResource } from './experimental/encrypted-client-per-resource.js';
import { EncryptedClient } from "./experimental/encrypted-client.js"; import { EncryptedClient } from './experimental/encrypted-client.js';
// Shared functions // Shared functions
import { writeMany, readMany } from "./shared.js"; import { writeMany, readMany } from './shared.js';
const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2'); const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
const clients = { const clients = {
sse: new SSEClient<{ message: string }>('https://dev.v2.sync.xo.harvmaster.com', privateKey), 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), 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'), 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)), encryptedPerResource: new EncryptedClientPerResource<{ message: string }>(
new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey),
),
}; };
// Test the suite for a given client // Test the suite for a given client
const testSuite = async (client: BaseSyncClient<{ message: string }>) => { const testSuite = async (client: BaseSyncClient<{ message: string }>) => {
client.on('resource-instance-updated', (message) => { client.on('resource-instance-updated', (message) => {
console.log(message); console.log(message);
}); });
// Connect to the sync server // Connect to the sync server
await client.connect(); await client.connect();
// Subscribe to the resource // Subscribe to the resource
await client.subscribe('test'); await client.subscribe('test');
// Write a new value to the resource // Write a new value to the resource
await client.write('test', {'message': 'Hello, world! this is the minimal sync client using ' + client.constructor.name}); await client.write('test', {
message: 'Hello, world! this is the minimal sync client using ' + client.constructor.name,
});
// Read the current value of the resource // Read the current value of the resource
const data = await client.read('test'); const data = await client.read('test');
console.log(data.map((resource) => resource.value.message)); console.log(data.map((resource) => resource.value.message));
// Use the shared functions to write and read many resources // Use the shared functions to write and read many resources
const writtenResources = await writeMany(client); const writtenResources = await writeMany(client);
console.dir(writtenResources, { depth: null }); console.dir(writtenResources, { depth: null });
const readResources = await readMany(client); const readResources = await readMany(client);
console.dir(readResources, { depth: null }); console.dir(readResources, { depth: null });
// If the user presses Ctrl+C, disconnect the client and exit the program // If the user presses Ctrl+C, disconnect the client and exit the program
process.on('SIGINT', async () => { process.on('SIGINT', async () => {
await client.disconnect(); await client.disconnect();
process.exit(0); process.exit(0);
}); });
} };
// Get the clients to test // Get the clients to test
const selectClients = (params: string[]): BaseSyncClient<{ message: string }>[] => { const selectClients = (params: string[]): BaseSyncClient<{ message: string }>[] => {
// If no params are provided, test all clients // If no params are provided, test all clients
const names = params.length > 0 ? params : Object.keys(clients); const names = params.length > 0 ? params : Object.keys(clients);
// Find values that are unknown to the clients object // Find values that are unknown to the clients object
const unknown = names.filter((n) => !(n in clients)); const unknown = names.filter((n) => !(n in clients));
if (unknown.length > 0) { if (unknown.length > 0) {
console.error(`Invalid client type(s): ${unknown.join(', ')}. Expected: ${Object.keys(clients).join(', ')}`); console.error(`Invalid client type(s): ${unknown.join(', ')}. Expected: ${Object.keys(clients).join(', ')}`);
process.exit(1); process.exit(1);
} }
// Return the clients that are known to the clients object // Return the clients that are known to the clients object
return names.map((n) => clients[n as keyof typeof clients]); return names.map((n) => clients[n as keyof typeof clients]);
} };
// Get the params // Get the params
const params = process.argv.slice(2); const params = process.argv.slice(2);
@@ -78,7 +82,7 @@ const clientsToTest = selectClients(params);
// Test the clients // Test the clients
for (const client of clientsToTest) { for (const client of clientsToTest) {
await testSuite(client); await testSuite(client);
} }
// Wait for the user to press Ctrl+C // Wait for the user to press Ctrl+C
@@ -86,4 +90,4 @@ await new Promise((resolve) => process.on('SIGINT', resolve));
// Exit the program // Exit the program
console.log('Exiting...'); console.log('Exiting...');
process.exit(0); process.exit(0);
demo/tui.ts
+39 -35
View File
@@ -1,56 +1,60 @@
import { PrivateKey } from "@xo-cash/primitives"; import { PrivateKey } from '@xo-cash/primitives';
import type { BaseSyncClient } from "../source/shared/base-client.js"; import type { BaseSyncClient } from '../source/shared/base-client.js';
import { SSEClient } from "../source/sse/client.js"; import { SSEClient } from '../source/sse/client.js';
import { EncryptedClientPerResource } from "./experimental/encrypted-client-per-resource.js"; import { EncryptedClientPerResource } from './experimental/encrypted-client-per-resource.js';
const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2'); const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
// Define some formatting helpers // Define some formatting helpers
const formatResourceId = (resourceId: string) => { const formatResourceId = (resourceId: string) => {
return `\x1b[32m${resourceId}\x1b[0m`; return `\x1b[32m${resourceId}\x1b[0m`;
} };
const formatMessage = (message: string) => { const formatMessage = (message: string) => {
return `\x1b[34m${message}\x1b[0m`; return `\x1b[34m${message}\x1b[0m`;
} };
const resetFormatting = () => { const resetFormatting = () => {
return `\x1b[0m`; return '\x1b[0m';
} };
const startTUI = async (client: BaseSyncClient<string>) => { const startTUI = async (client: BaseSyncClient<string>) => {
// Listen and log out the resource change // Listen and log out the resource change
client.on('resource-instance-updated', (message) => { client.on('resource-instance-updated', (message) => {
// Print the message with colours // Print the message with colours
console.log(`${formatResourceId(message.resourceId)}:${formatResourceId(message.publicKey)} updated: ${formatMessage(message.value)}${resetFormatting()}`); console.log(
}); `${formatResourceId(message.resourceId)}:${formatResourceId(message.publicKey)} updated: ${formatMessage(message.value)}${resetFormatting()}`,
);
});
// Connect to the sync server and subscribe to the resource // Connect to the sync server and subscribe to the resource
await client.connect(); await client.connect();
await client.subscribe('encrypted-test'); await client.subscribe('encrypted-test');
// Start the TUI loop
while (true) {
// Print the prompt
process.stdout.write('> ');
// Wait for the user to enter a command // Start the TUI loop
const data = await new Promise((resolve) => process.stdin.once('data', resolve)); while (true) {
// Print the prompt
process.stdout.write('> ');
// If the user entered a command, write it to the resource // Wait for the user to enter a command
if (data) { const data = await new Promise((resolve) => process.stdin.once('data', resolve));
const message = data.toString().trim();
await client.write('encrypted-test', message); // If the user entered a command, write it to the resource
if (data) {
const message = data.toString().trim();
await client.write('encrypted-test', message);
}
} }
} };
}
const clients = { const clients = {
'sse': new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey), 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)), encrypted: new EncryptedClientPerResource<string>(new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey)),
} };
const params = process.argv.slice(2); const params = process.argv.slice(2);
const client = params[0] === 'false' ? clients.sse : clients.encrypted; const client = params[0] === 'false' ? clients.sse : clients.encrypted;
startTUI(client); startTUI(client);
eslint.config.mjs
+8
View File
@@ -0,0 +1,8 @@
import baseConfig from '@xo-cash/eslint-config';
export default [
{
ignores: [ 'coverage/**', 'docs/**', 'demo/**' ],
},
...baseConfig,
];
package-lock.json
Generated
+11076 -11076
View File
File diff suppressed because it is too large Load Diff
package.json
+50 -50
View File
@@ -1,53 +1,53 @@
{ {
"name": "@xo-cash/sync-client", "name": "@xo-cash/sync-client",
"version": "0.0.1", "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": {
"analyze": "tsdown --clean --no-fixed-extension --sourcemap source/index.ts && esbuild-analyzer dist/", "analyze": "tsdown --clean --no-fixed-extension --sourcemap source/index.ts && esbuild-analyzer dist/",
"build": "tsdown --clean --sourcemap source/index.ts", "build": "tsdown --clean --sourcemap source/index.ts",
"docs": "typedoc --hideGenerator", "docs": "typedoc --hideGenerator",
"format": "prettier --write . && eslint --fix", "format": "prettier --write . && eslint --fix",
"spellcheck": "cspell 'source/**' 'test/**' 'playground/**'", "spellcheck": "cspell 'source/**' 'test/**' 'playground/**'",
"style": "eslint", "style": "eslint",
"syntax": "tsc --noEmit", "syntax": "tsc --noEmit",
"test": "vitest --dir test/ --test-timeout=15000 --passWithNoTests --run --coverage" "test": "vitest --dir test/ --test-timeout=15000 --passWithNoTests --run --coverage"
}, },
"exports": { "exports": {
".": { ".": {
"import": "./dist/src/index.js", "import": "./dist/src/index.js",
"require": "./dist/src/index.cjs" "require": "./dist/src/index.cjs"
}
},
"files": [
"dist"
],
"keywords": [],
"author": "GeneralProtocols",
"license": "MIT",
"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": "GeneralProtocols",
"license": "MIT",
"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"
}
} }
source/index.ts
+3 -3
View File
@@ -1,4 +1,4 @@
export * from './shared/base-client.js'; export * from './shared/base-client.ts';
export * from './sse/client.js'; export * from './sse/client.ts';
export * from './ws/client.js'; export * from './ws/client.ts';
source/shared/base-client.ts
+213 -214
View File
@@ -1,267 +1,266 @@
import { binToHex, flattenBinArray, secp256k1, sha256 } from "@bitauth/libauth"; import { binToHex, flattenBinArray, secp256k1, sha256 } from '@bitauth/libauth';
import { toExtendedJson, EventEmitter, fromExtendedJson } from "@xo-cash/utils"; import { toExtendedJson, EventEmitter, fromExtendedJson } from '@xo-cash/utils';
import type { WriteRequest } from "../types.js"; import type { WriteRequest } from '../types.ts';
import z from "zod"; import z from 'zod';
export type AuthenticatedRequestHeaders = { export type AuthenticatedRequestHeaders = {
'X-Public-Key': string; 'X-Public-Key': string;
'X-Timestamp': string; 'X-Timestamp': string;
'X-Signature': string; 'X-Signature': string;
}; };
export type SignedPayload = { publicKey: string; signature: string }; export type SignedPayload = { publicKey: string; signature: string };
export type XOSyncClientEventMap<T = unknown> = { export type XOSyncClientEventMap<T = unknown> = {
'connection-updated': 'connected' | 'disconnected'; 'connection-updated': 'connected' | 'disconnected';
'resource-instance-updated': Resource<T> & { 'resource-instance-updated': Resource<T> & {
resourceId: string; resourceId: string;
}; };
'subscriptions-updated': Set<string>; 'subscriptions-updated': Set<string>;
'error': Error; error: Error;
}; };
export type Resource<PayloadType = unknown> = z.infer<typeof BaseSyncClient.resourceInstanceSchema> & { export type Resource<PayloadType = unknown> = z.infer<typeof BaseSyncClient.resourceInstanceSchema> & {
value: PayloadType; value: PayloadType;
}; };
export type ResourcesRecord<PayloadType = unknown> = Record<string, { instances: Resource<PayloadType>[] }>; export type ResourcesRecord<PayloadType = unknown> = Record<string, { instances: Resource<PayloadType>[] }>;
export abstract class BaseSyncClient<T = unknown> extends EventEmitter<XOSyncClientEventMap<T>> { export abstract class BaseSyncClient<T = unknown> extends EventEmitter<XOSyncClientEventMap<T>> {
/**
* Connects to the sync server
*/
abstract connect(): Promise<void>;
/** /**
* Connects to the sync server * Disconnects from the sync server
*/ */
abstract connect(): Promise<void>; abstract disconnect(): Promise<void>;
/** /**
* Disconnects from the sync server * Sign and write one resource
*/ * @param resourceId - The resource ID to write to
abstract disconnect(): Promise<void>; * @param value - The value to write
* @returns The Resource Object with the decoded value
/** */
* Sign and write one resource async write(resourceId: string, value: T): Promise<Resource<T>[]> {
* @param resourceId - The resource ID to write to // Write the resource to the server
* @param value - The value to write const response = await this.writeMany({ [resourceId]: value });
* @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 // Get the resource from the response
if (!resource) { const resource = response[resourceId];
throw new Error(`Resource ${resourceId} not found`);
// 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;
} }
// 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
/** */
* Writes multiple values to resources abstract writeMany(resources: Record<string, T>): Promise<ResourcesRecord<T>>;
* @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 * Reads a value from a resource
* @param resourceId - The resource ID to read from * @param resourceId - The resource ID to read from
* @returns The resource with the read value * @returns The resource with the read value
*/ */
async read(resourceId: string): Promise<Resource<T>[]> { async read(resourceId: string): Promise<Resource<T>[]> {
// Read the resource from the server // Read the resource from the server
const response = await this.readMany([ resourceId ]); const response = await this.readMany([ resourceId ]);
// Get the resource from the response // Get the resource from the response
const resource = response[resourceId]; const resource = response[resourceId];
// If the resource is not found, throw an error // If the resource is not found, throw an error
if (!resource) { if (!resource) {
throw new Error(`Resource ${resourceId} not found`); throw new Error(`Resource ${resourceId} not found`);
}
// Return the resource instances
return resource.instances;
} }
// 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>>;
/** /**
* Reads multiple values from resources * Subscribes to a resource
* @param resourceIds - The resource IDs to read from * @param resourceId - The resource ID to subscribe to
* @returns The resources with the read values */
*/ abstract subscribe(resourceId: string): Promise<void>;
abstract readMany(resourceIds: string[]): Promise<ResourcesRecord<T>>;
/** /**
* Subscribes to a resource * Unsubscribes from a resource
* @param resourceId - The resource ID to subscribe to * @param resourceId - The resource ID to unsubscribe from
*/ */
abstract subscribe(resourceId: string): Promise<void>; abstract unsubscribe(resourceId: string): Promise<void>;
/** /**
* Unsubscribes from a resource * Derives the resource private key from the private key and resource ID
* @param resourceId - The resource ID to unsubscribe from * @param privateKey - The private key to derive the resource private key from
*/ * @param resourceId - The resource ID to derive the resource private key from
abstract unsubscribe(resourceId: string): Promise<void>; * @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
* Derives the resource private key from the private key and resource ID return sha256.hash(flattenBinArray([ privateKey, resourceIdBytes ]));
* @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); * 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));
// If the signature is a string, throw an error // Derive the public key
if (typeof signature === 'string') { const publicKey = secp256k1.derivePublicKeyCompressed(privateKey);
throw new Error('Failed to sign message');
// 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),
};
} }
return { /**
publicKey: binToHex(publicKey), * Signs a write request with the private key and returns the write request
signature: binToHex(signature), * @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
* Signs a write request with the private key and returns the write request const timestamp = Date.now();
* @param privateKey - The private key to sign the write request with const payload = `${timestamp}:${resourceId}:${toExtendedJson(value)}`;
* @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 // Sign the payload
const timestamp = Date.now(); const { publicKey, signature } = this.signPayload(derivedKey, payload);
const payload = `${timestamp}:${resourceId}:${toExtendedJson(value)}`;
// Sign the payload return {
const { publicKey, signature } = this.signPayload(derivedKey, payload); id: resourceId,
value,
publicKey,
timestamp,
signature,
};
}
return { /**
id: resourceId, * Signs a request with the private key and returns the public key, timestamp, and signature
value, * @param privateKey - The private key to sign the request with
publicKey, * @param path - The path of the request
timestamp, * @param body - The body of the request
signature, * @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
* Signs a request with the private key and returns the public key, timestamp, and signature const bodyStr = typeof body === 'string' ? body : toExtendedJson(body);
* @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 // Create the payload as `Timestamp:Path:Body`
const bodyStr = typeof body === 'string' ? body : toExtendedJson(body); const payload = `${timestamp}:${path}:${bodyStr}`;
// Create the payload as `Timestamp:Path:Body` // Sign the payload
const payload = `${timestamp}:${path}:${bodyStr}`; const { publicKey, signature } = this.signPayload(privateKey, payload);
// Sign the payload return {
const { publicKey, signature } = this.signPayload(privateKey, payload); 'X-Public-Key': publicKey,
'X-Timestamp': timestamp.toString(),
'X-Signature': signature,
};
}
return { /**
'X-Public-Key': publicKey, * Decodes the 'blob' field of the Resource Object from the server from a read/write
'X-Timestamp': timestamp.toString(), * @param response - The ResourceResponse Object from the server from a read/write
'X-Signature': signature, * @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
* Decodes the 'blob' field of the Resource Object from the server from a read/write const decodedInstances = instances.map((instance) => {
* @param response - The ResourceResponse Object from the server from a read/write return BaseSyncClient.decodeBlobValue<T>(instance);
* @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 // Add the decoded instances to the accumulator
const decodedInstances = instances.map((instance) => { acc[resourceId] = {
return BaseSyncClient.decodeBlobValue<T>(instance); ...resource,
}); instances: decodedInstances,
};
// Add the decoded instances to the accumulator // Return the accumulator
acc[resourceId] = { return acc;
...resource, }, {});
instances: decodedInstances, }
};
// Return the accumulator static decodeBlobValue<T = unknown>(instance: z.infer<typeof BaseSyncClient.resourceInstanceSchema>): Resource<T> {
return acc; const decodedBlob = new TextDecoder().decode(instance.blob);
}, {}); const value = fromExtendedJson(decodedBlob) as T;
}
static decodeBlobValue<T = unknown>(instance: z.infer<typeof BaseSyncClient.resourceInstanceSchema>): Resource<T> { return {
const decodedBlob = new TextDecoder().decode(instance.blob); ...instance,
const value = fromExtendedJson(decodedBlob) as T; value,
};
}
return { static resourceInstanceSchema = z.object({
...instance, resourceId: z.string(),
value, blob: z.instanceof(Uint8Array),
}; timestamp: z.number(),
} publicKey: z.string(),
signature: z.string(),
});
static resourceInstanceSchema = z.object({ static rawResourcesSchema = z.object({
resourceId: z.string(), instances: z.array(BaseSyncClient.resourceInstanceSchema),
blob: z.instanceof(Uint8Array), });
timestamp: z.number(),
publicKey: z.string(),
signature: z.string(),
});
static rawResourcesSchema = z.object({ static resourcesResponseSchema = z.object({
instances: z.array(BaseSyncClient.resourceInstanceSchema), resources: z.record(
}); // Resource ID
z.string(),
static resourcesResponseSchema = z.object({ // Resource instances
resources: z.record( BaseSyncClient.rawResourcesSchema,
// Resource ID ),
z.string(), });
// Resource instances
BaseSyncClient.rawResourcesSchema,
),
});
} }
source/sse/client.ts
+230 -221
View File
@@ -1,254 +1,263 @@
import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils'; import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
import { PrivateKey } from '@xo-cash/primitives' import type { PrivateKey } from '@xo-cash/primitives';
import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.js'; import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.ts';
export class SSEClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> { export class SSEClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
private readonly url: string; private readonly url: string;
public sseSession: SSESession | undefined; public sseSession: SSESession | undefined;
private readonly subscriptions: Set<string> = new Set(); private readonly subscriptions: Set<string> = new Set();
private privateKey: PrivateKey; private privateKey: PrivateKey;
private textEncoder = new TextEncoder(); private textEncoder = new TextEncoder();
private get subscriptionUrl() { private get subscriptionUrl(): string {
return `${this.url}/data/subscribe`; 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() {
// 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(); /**
* 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();
await this.sseSession.connect(); 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
* Disconnects from the server // We add a small debounce to this to prevent parallel subscriptions from causing multiple reconnects
*/ this.on(
async disconnect() { 'subscriptions-updated',
await this.sseSession?.disconnect(); () => {
} // 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,
);
}
/** /**
* Reconnects to the server * Connects to the server
*/ */
async reconnect() { async connect(): Promise<void> {
// Close the current session. Its subscription list is fixed in the request body, so it can't be reused // The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet.
this.sseSession?.close(); // subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session.
this.sseSession = undefined; if (this.subscriptions.size === 0) {
return;
// Connect a new session with the current subscriptions
await this.connect();
}
/**
* Creates a new SSE session
* @returns The SSE session
*/
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 = 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 this.sseSession ??= this.createSession();
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 await this.sseSession.connect();
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 = 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); * Disconnects from the server
*/
// Return the decoded resources async disconnect(): Promise<void> {
return decodedResources; this.sseSession?.disconnect();
}
/**
* 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); * 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;
// Return the decoded resources // Connect a new session with the current subscriptions
return decodedResources; await this.connect();
} }
/** /**
* Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list * Creates a new SSE session
* @param resourceId - The resource ID to add to the subscription list * @returns The SSE session
*/ */
async subscribe(resourceId: string) { private createSession(): SSESession {
this.subscriptions.add(resourceId); // Build the request body
const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] });
this.emit('subscriptions-updated', 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
* Destroy the current SSE Session, then re-create it without the resource ID in the subscription list const authHeaders = BaseSyncClient.signRequest(this.privateKey.toBytes(), path, request.body);
* @param resourceId - The resource ID to remove from the subscription list
*/
async unsubscribe(resourceId: string) {
this.subscriptions.delete(resourceId);
this.emit('subscriptions-updated', this.subscriptions); // Initialize the request headers if they don't exist
} request.headers ??= {};
async handleOnConnectionUpdated(value: 'connected' | 'disconnected') { // Add the authentication headers to the request
this.emit('connection-updated', value); request.headers = {
} ...request.headers,
...authHeaders,
};
/** // Return the request with the authentication headers
* Sends a request to the server return request;
* @param path - The path to send the request to },
* @param body - The body of the request headers: {
* @returns The response from the server 'Content-Type': 'application/json',
*/ },
private async sendRequest(path: string, body: unknown): Promise<Response> { body,
// Build the URL method: 'POST',
const url = `${this.url}${path}`; // Reconnect when the server ends the stream (e.g. a restart or a proxy idle timeout). By default only errors are retried
persistent: true,
});
// Format the body as Extended JSON // Create a listener to re-emit the connected and disconnected events
const formattedBody = toExtendedJson(body); session.on('connected', this.handleOnConnectionUpdated.bind(this, 'connected'));
session.on('disconnected', this.handleOnConnectionUpdated.bind(this, 'disconnected'));
// Send the request to the server // Create a listener to re-emit the messages
const res = await fetch(url, { session.on('message', (message): void => {
method: 'POST', const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(fromExtendedJson(message.data));
headers: { if (!parsedMessage.success) {
'Content-Type': 'application/json', return;
...BaseSyncClient.signRequest(this.privateKey.toBytes(), path, body), }
},
body: formattedBody,
});
// Return the response this.emit('resource-instance-updated', {
return res; ...parsedMessage.data,
} value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType,
});
});
/** // Create a listener to re-emit the errors
* Decodes the blobs on the response body to the type T session.on('error', (error) => {
* @param response - The response to decode this.emit('error', error);
* @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 // Return the session
const resources = BaseSyncClient.resourcesResponseSchema.parse(text); return session;
}
// Decode the blobs and convert them from extended json to the type T /**
const decodedResources = BaseSyncClient.decodeResourcesResponse<PayloadType>(resources); * 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 decodedResources; 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
+1 -1
View File
@@ -1 +1 @@
export * from './client.js'; export * from './client.ts';
source/types.ts
+8 -8
View File
@@ -1,11 +1,11 @@
export type WriteRequest = { export type WriteRequest = {
id: string; id: string;
publicKey: string; publicKey: string;
timestamp: number; timestamp: number;
/** /**
* sha256(timestamp + id + toExtendedJson(value)) * sha256(timestamp + id + toExtendedJson(value))
*/ */
signature: string; signature: string;
value: unknown; value: unknown;
}; };
source/ws/client.ts
+349 -334
View File
@@ -1,28 +1,29 @@
import { ExponentialBackoff, fromExtendedJson, toExtendedJson } from '@xo-cash/utils'; import { ExponentialBackoff, fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
import type { PrivateKey } from '@xo-cash/primitives'; import type { PrivateKey } from '@xo-cash/primitives';
import { WsMessageSchema, WsSuccessResponseSchema, WsErrorResponseSchema, type WsMessage } from './types.js'; import { WsMessageSchema, WsSuccessResponseSchema, WsErrorResponseSchema, type WsMessage } from './types.ts';
import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.js'; import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.ts';
/** A request sent over the sync server's WebSocket connection. */ /** A request sent over the sync server's WebSocket connection. */
type WsRequest = { type WsRequest = {
id?: string; id?: string;
path: string; path: string;
body?: unknown; body?: unknown;
headers?: Record<string, string>; headers?: Record<string, string>;
}; };
type PendingRequest = { type PendingRequest = {
resolve: (body: unknown) => void; resolve: (body: unknown) => void;
reject: (error: Error) => void; reject: (error: Error) => void;
}; };
export type WsClientOptions = { export type WsClientOptions = {
/** Called for server-pushed events and uncorrelated server errors. */
onMessage: (message: WsMessage) => void;
/** Called for transport errors and malformed server messages. */ /** Called for server-pushed events and uncorrelated server errors. */
onError: (error: Error) => void; onMessage: (message: WsMessage) => void;
/** Called for transport errors and malformed server messages. */
onError: (error: Error) => void;
}; };
/** /**
@@ -39,357 +40,371 @@ export type WsClientOptions = {
* Heartbeats and request timeouts are not implemented. * Heartbeats and request timeouts are not implemented.
*/ */
export class WsClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> { export class WsClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
private readonly url: string; private readonly url: string;
private readonly privateKey: PrivateKey; private readonly privateKey: PrivateKey;
private readonly pendingRequests = new Map<string, PendingRequest>(); private readonly pendingRequests = new Map<string, PendingRequest>();
private readonly messageListeners = new Set<(message: WsMessage) => void>(); private readonly messageListeners = new Set<(message: WsMessage) => void>();
private readonly errorListeners = new Set<(error: Error) => void>(); private readonly errorListeners = new Set<(error: Error) => void>();
private readonly textEncoder = new TextEncoder(); private readonly textEncoder = new TextEncoder();
/** Resource IDs to (re)subscribe to whenever a socket opens. */
private readonly subscriptions = new Set<string>();
/** /** Resource IDs to (re)subscribe to whenever a socket opens. */
* The current connection: resolves to an open socket once a connection attempt succeeds. private readonly subscriptions = new Set<string>();
* 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 }); * 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;
constructor( /** Retries a failed connection attempt forever (maxAttempts 0), waiting up to 30s between attempts. */
url: string, private readonly connectBackoff = new ExponentialBackoff({
privateKey: PrivateKey, maxAttempts: 0,
options: Partial<WsClientOptions> = {}, maxDelay: 30_000,
) {
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. constructor(url: string, privateKey: PrivateKey, options: Partial<WsClientOptions> = {}) {
// Mark the rejection as handled so it isn't reported as unhandled. Callers awaiting `attempt` still receive it. super();
attempt.catch(() => {});
return attempt; this.url = url;
} this.privateKey = privateKey;
this.messageListeners.add(options.onMessage ?? ((): void => {}));
async readMany(resourceIds: string[]): Promise<ResourcesRecord<PayloadType>> { this.errorListeners.add(options.onError ?? ((): void => {}));
// 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] }); /** Connect, and keep reconnecting on drops until `disconnect()` is called. */
} async connect(): Promise<void> {
await this.getWebSocket();
/** 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)); * Close the socket and stop reconnecting.
const request: WsRequest = { id, path, body, headers }; *
* 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;
socket.send(toExtendedJson(request)); this.rejectPendingRequests(new Error('WebSocket client disconnected'));
}
/** // If the attempt is still retrying, it sees it is no longer current after its next handshake,
* Resolve with the socket once it opens, or reject if the handshake fails. // closes that socket and stops. Its rejection is handled here.
* @param socket - The WebSocket to wait for attempt?.then(
* @returns The WebSocket (socket) => socket.close(),
*/ () => {},
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.
* @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. * The current socket, starting a connection attempt if there isn't one.
const pending = message.id ? this.pendingRequests.get(message.id) : undefined; *
* A new socket has no topics on the server, so starting an attempt also queues a subscribe for
// If the message has no outstanding ID, it is a subscription event so we will just notify the listeners * every remembered subscription. They are sent through send(), so they go out once the socket opens.
if (!pending) { */
for (const listener of this.messageListeners) { private getWebSocket(): Promise<WebSocket> {
listener(message); if (this.webSocket) {
} return this.webSocket;
// 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', { this.webSocket = this.openWebSocket();
...parsedMessage.data,
value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType, 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,
}); });
}
return; // 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;
} }
// Delete the pending request async writeMany(resourcesToWrite: Record<string, PayloadType>): Promise<ResourcesRecord<PayloadType>> {
this.pendingRequests.delete(message.id!); // Sign the resources
const signedResources = Object.entries(resourcesToWrite).map(([ resourceId, value ]) => {
const valueStr = toExtendedJson(value);
const valueBytes = this.textEncoder.encode(valueStr);
// If the message is an error, reject the pending request return BaseSyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
if (message.type === 'error') { });
const error = WsErrorResponseSchema.parse(message);
pending.reject(new Error(`${error.error} (${error.statusCode})`)); // Write the resources to the server
return; 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;
} }
// If the message is a response, resolve the pending request /**
if (message.type === 'response') { * Subscribe to one resource topic, now and on every future socket.
const response = WsSuccessResponseSchema.parse(message); *
* 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 ] });
// If the response is not successful, reject the pending request // Remembered only after sending: if send() started a new socket, getWebSocket() replays the
if (response.statusCode < 200 || response.statusCode >= 300) { // existing subscriptions, and this one isn't among them yet. Sending it twice in the same
pending.reject(new Error(`Request failed (${response.statusCode})`)); // millisecond gives identical signatures, which the server rejects as a replay.
return; this.subscriptions.add(resourceId);
}
// 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(); /**
} * 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.
* Helper to convert an HTTP URL to a WebSocket URL if (!this.webSocket) {
* @param httpUrl - The HTTP URL to convert return;
* @returns The WebSocket URL }
*/
static httpToWsUrl(httpUrl: string): string {
const url = new URL(httpUrl);
if (url.protocol === 'http:') { await this.request('/data/unsubscribe', { resourceId: [ resourceId ] });
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. /** Send a request and wait for the response carrying the same ID. */
const pathname = url.pathname.replace(/\/+$/, ''); private async request(path: string, body: unknown): Promise<unknown> {
url.pathname = pathname.endsWith('/ws') ? pathname : `${pathname}/ws`; const id = crypto.randomUUID();
return url.toString(); 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
+1 -1
View File
@@ -1 +1 @@
export * from './client.js'; export * from './client.ts';
source/ws/types.ts
+14 -14
View File
@@ -2,24 +2,24 @@ import { z } from 'zod';
/** A normal, correlated reply to a WebSocket request. */ /** A normal, correlated reply to a WebSocket request. */
export const WsSuccessResponseSchema = z.object({ 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>; export type WsSuccessResponse = z.infer<typeof WsSuccessResponseSchema>;
/** A public error returned by the server for one request. */ /** A public error returned by the server for one request. */
export const WsErrorResponseSchema = z.object({ 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 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>; export type WsResponse = z.infer<typeof WsResponseSchema>;
/** /**
@@ -30,11 +30,11 @@ export type WsResponse = z.infer<typeof WsResponseSchema>;
* whenever the server adds another event. * whenever the server adds another event.
*/ */
export const WsEventSchema = z.object({ 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 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>; export type WsMessage = z.infer<typeof WsMessageSchema>;
tsconfig.json
+31 -25
View File
@@ -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
},
"exclude": ["demo/**"]
} }