Add experiment encyption clients. Rewrite demos. Reduce implementation code. Add value decoding. Rename to base-client. Type inference and generics

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

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

demo/experimental/encrypted-client.ts

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

demo/shared.ts

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

demo/sse.ts

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

demo/suite.ts

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

demo/tui.ts

@@ -0,0 +1,56 @@
+import { PrivateKey } from "@xo-cash/primitives";
+
+import type { BaseSyncClient } from "../src/shared/base-client.js";
+
+import { SSEClient } from "../src/sse/client.js";
+import { EncryptedClientPerResource } from "./experimental/encrypted-client-per-resource.js";
+
+const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2');
+
+// Define some formatting helpers
+const formatResourceId = (resourceId: string) => {
+  return `\x1b[32m${resourceId}\x1b[0m`;
+}
+const formatMessage = (message: string) => {
+  return `\x1b[34m${message}\x1b[0m`;
+}
+const resetFormatting = () => {
+  return `\x1b[0m`;
+}
+
+const startTUI = async (client: BaseSyncClient<string>) => {
+  // Listen and log out the resource change
+  client.on('resource-instance-updated', (message) => {
+    // Print the message with colours
+    console.log(`${formatResourceId(message.resourceId)}:${formatResourceId(message.publicKey)} updated: ${formatMessage(message.value)}${resetFormatting()}`);
+  });
+
+  // Connect to the sync server and subscribe to the resource
+  await client.connect();
+  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
+    const data = await new Promise((resolve) => process.stdin.once('data', resolve));
+
+    // 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 = {
+  'sse': new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey),
+  'encrypted': new EncryptedClientPerResource<string>(new SSEClient<string>('https://dev.v2.sync.xo.harvmaster.com', privateKey)),
+}
+
+const params = process.argv.slice(2);
+const client = params[0] === 'false' ? clients.sse : clients.encrypted;
+
+startTUI(client);

demo/ws.ts

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

src/shared/base-client.ts

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

src/shared/client.ts

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

src/sse/client.ts

@@ -1,9 +1,9 @@
 import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
 import { PrivateKey } from '@xo-cash/primitives'
 
-import { SyncClient } from '../shared/client.js';
+import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.js';
 
-export class SSEClient extends SyncClient {
+export class SSEClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
   private readonly url: string;
   public sseSession: SSESession | undefined;
   private readonly subscriptions: Set<string> = new Set();
@@ -15,6 +15,11 @@ export class SSEClient extends SyncClient {
     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();
 
@@ -29,6 +34,9 @@ export class SSEClient extends SyncClient {
     }, 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.
@@ -41,10 +49,16 @@ export class SSEClient extends SyncClient {
     await this.sseSession.connect();
   }
 
+  /**
+   * Disconnects from the server
+   */
   async disconnect() {
     await this.sseSession?.disconnect();
   }
 
+  /**
+   * Reconnects to the server
+   */
   async reconnect() {
     // Close the current session. Its subscription list is fixed in the request body, so it can't be reused
     this.sseSession?.close();
@@ -54,6 +68,10 @@ export class SSEClient extends SyncClient {
     await this.connect();
   }
 
+  /**
+   * Creates a new SSE session
+   * @returns The SSE session
+   */
   private createSession() {
     // Build the request body
     const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] });
@@ -66,7 +84,7 @@ export class SSEClient extends SyncClient {
         const path = url.pathname;
 
         // Sign the request body
-        const authHeaders = SyncClient.signRequest(this.privateKey.toBytes(), path, request.body);
+        const authHeaders = BaseSyncClient.signRequest(this.privateKey.toBytes(), path, request.body);
 
         // Initialize the request headers if they don't exist
         request.headers ??= {};
@@ -95,12 +113,15 @@ export class SSEClient extends SyncClient {
 
     // Create a listener to re-emit the messages
     session.on('message', (message) => {
-      const parsedMessage = SyncClient.messageSchema.safeParse(fromExtendedJson(message.data));
+      const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(fromExtendedJson(message.data));
       if (!parsedMessage.success) {
         return;
       }
 
-      this.emit('message', parsedMessage.data);
+      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
@@ -112,69 +133,70 @@ export class SSEClient extends SyncClient {
     return session;
   }
   
-  // Send a POST /data/write request to the server
-  async write(resourceId: string, value: Record<string, unknown>) {
-    const url = `${this.url}/data/write`;
-
-    const valueStr = toExtendedJson(value);
-    const valueBytes = this.textEncoder.encode(valueStr);
-    const resource = SyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
-
-    const body = toExtendedJson({ resources: [resource] });
-
-    const res = await fetch(url, {
-      method: 'POST',
-
-      // Headers arent required unless we have payment service running
-      headers: {
-        'Content-Type': 'application/json',
-        ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/write', body),
-      },
-
-      body,
+  /**
+   * 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 resource [${resourceId}] (${res.statusText}): ${JSON.stringify(error)}`);
+      throw new Error(`Failed to write resources [${Object.keys(resources).join(', ')}] (${res.statusText}): ${JSON.stringify(error)}`);
     }
 
-    const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[];
+    // Decode the response and return the resources
+    const decodedResources = await this.decodeResponse(res);
 
-    return text;
+    // Return the decoded resources
+    return decodedResources;
   }
 
-  // Send a POST /data/read request to the server
-  async read(resourceId: string) {
-    const url = `${this.url}/data/get`;
-    const body = toExtendedJson({ resourceId: [ resourceId ] });
-
-    const res = await fetch(url, {
-      method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-        ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/get', body),
-      },
-      body,
-    });
+  /**
+   * 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 resource: ${res.statusText}`);
+      throw new Error(`Failed to read resources [${resourceIds.join(', ')}] (${res.statusText})`);
     }
 
-    const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[];
+    // Decode the response and return the resources
+    const decodedResources = await this.decodeResponse(res);
 
-    return text;
+    // Return the decoded resources
+    return decodedResources;
   }
 
-  // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list
+  /**
+   * 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) {
     this.subscriptions.add(resourceId);
 
     this.emit('subscriptions-updated', this.subscriptions);
   }
 
-  // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list
+  /**
+   * 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) {
     this.subscriptions.delete(resourceId);
 
@@ -184,4 +206,49 @@ export class SSEClient extends SyncClient {
   async handleOnConnectionUpdated(value: 'connected' | 'disconnected') {
     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;
+  }
 }

src/ws/client.ts

@@ -2,7 +2,7 @@ import { ExponentialBackoff, fromExtendedJson, toExtendedJson } from '@xo-cash/u
 import type { PrivateKey } from '@xo-cash/primitives';
 
 import { WsMessageSchema, WsSuccessResponseSchema, WsErrorResponseSchema, type WsMessage } from './types.js';
-import { SyncClient } from '../shared/client.js';
+import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.js';
 
 /** A request sent over the sync server's WebSocket connection. */
 type WsRequest = {
@@ -38,13 +38,14 @@ export type WsClientOptions = {
  *
  * Heartbeats and request timeouts are not implemented.
  */
-export class WsClient extends SyncClient {
+export class WsClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
   private readonly url: string;
   private readonly privateKey: PrivateKey;
   private readonly pendingRequests = new Map<string, PendingRequest>();
   private readonly messageListeners = new Set<(message: WsMessage) => void>();
   private readonly errorListeners = new Set<(error: Error) => void>();
-
+  private readonly textEncoder = new TextEncoder();
+  
   /** Resource IDs to (re)subscribe to whenever a socket opens. */
   private readonly subscriptions = new Set<string>();
 
@@ -167,19 +168,37 @@ export class WsClient extends SyncClient {
     return attempt;
   }
 
-  /** Read every stored instance of one resource. */
-  async read(resourceId: string): Promise<unknown> {
-    return this.request('/data/get', { resourceId: [resourceId] });
+  async readMany(resourceIds: string[]): Promise<ResourcesRecord<PayloadType>> {
+    // Read the resources from the server
+    const response = await this.request('/data/get', { resourceId: resourceIds });
+
+    // Parse the response as a record of resources
+    const resources = BaseSyncClient.resourcesResponseSchema.parse(response);
+
+    // Decode the resources and convert the blobs from Uint8Arrays to the type T
+    const decodedResources = BaseSyncClient.decodeResourcesResponse<PayloadType>(resources);
+
+    return decodedResources;
   }
 
-  /**
-   * Sign and write one resource, using the same wire format as the SSE client.
-   */
-  async write(resourceId: string, value: Record<string, unknown>): Promise<unknown> {
-    const valueBytes = new TextEncoder().encode(toExtendedJson(value));
-    const resource = SyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
+  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);
+    });
 
-    return this.request('/data/write', { resources: [resource] });
+    // 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;
   }
 
   /**
@@ -188,6 +207,8 @@ export class WsClient extends SyncClient {
    * 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] });
@@ -200,6 +221,7 @@ export class WsClient extends SyncClient {
 
   /**
    * Unsubscribe from one resource topic.
+   * @param resourceId - The resource ID to unsubscribe from
    */
   async unsubscribe(resourceId: string): Promise<void> {
     this.subscriptions.delete(resourceId);
@@ -225,7 +247,12 @@ export class WsClient extends SyncClient {
     });
   }
 
-  /** Connect if needed, then sign and send one request envelope using the shared Extended JSON codec. */
+  /**
+   * 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();
 
@@ -235,13 +262,17 @@ export class WsClient extends SyncClient {
     }
 
     // Signed after connecting: the server rejects signatures older than its timestamp window (5 minutes by default).
-    const headers = SyncClient.signRequest(this.privateKey.toBytes(), path, toExtendedJson(body));
+    const 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. */
+  /**
+   * 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 = () => resolve(socket);
@@ -249,7 +280,10 @@ export class WsClient extends SyncClient {
     });
   }
 
-  /** Decode and route one frame received from the sync server. */
+  /**
+   * 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) {
@@ -275,12 +309,15 @@ export class WsClient extends SyncClient {
       }
 
       if (message.type === 'instance-changed') {
-        const parsedMessage = SyncClient.messageSchema.safeParse(message.data);
+        const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(message.data);
         if (!parsedMessage.success) {
           return;
         }
 
-        this.emit('message', parsedMessage.data);
+        this.emit('resource-instance-updated', {
+          ...parsedMessage.data,
+          value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType,
+        });
       }
 
       return;
@@ -311,14 +348,20 @@ export class WsClient extends SyncClient {
     }
   }
 
-  // Helper to emit an error to all error listeners
+  /**
+   * 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
+  /**
+   * 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);
@@ -327,7 +370,11 @@ export class WsClient extends SyncClient {
     this.pendingRequests.clear();
   }
 
-  // Helper to convert an HTTP URL to a WebSocket URL
+  /**
+   * 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);