diff --git a/demo/sse.ts b/demo/sse.ts index 03f007b..3f9b565 100644 --- a/demo/sse.ts +++ b/demo/sse.ts @@ -1,17 +1,13 @@ import { PrivateKey } from "@xo-cash/primitives"; -import { toExtendedJson } from "@xo-cash/utils"; 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, { - onMessage: (message) => { - console.log(toExtendedJson(message)); - }, - onError: (error) => { - console.error(error); - }, + const client = new SSEClient('https://v2.sync.xo.harvmaster.com', privateKey); + + client.on('message', (message) => { + console.log(message); }); // Connect to the sync server diff --git a/demo/ws.ts b/demo/ws.ts index a0edd2d..ed5b7c1 100644 --- a/demo/ws.ts +++ b/demo/ws.ts @@ -1,19 +1,22 @@ import { PrivateKey } from "@xo-cash/primitives"; -import { toExtendedJson } from "@xo-cash/utils"; 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)); - }, + // 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(); diff --git a/package-lock.json b/package-lock.json index 208a53d..c1ed3a6 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,17 +1,17 @@ { - "name": "minimal-sync", + "name": "@xo-cash/sync-client", "version": "1.0.0", "lockfileVersion": 3, "requires": true, "packages": { "": { - "name": "minimal-sync", + "name": "@xo-cash/sync-client", "version": "1.0.0", "license": "ISC", "dependencies": { "@bitauth/libauth": "^3.0.0", "@xo-cash/primitives": "^0.0.2", - "@xo-cash/utils": "git+https://git.harvmaster.com/Harvmaster/xo-cash-utils.git", + "@xo-cash/utils": "^0.0.8", "zod": "^4.4.3" }, "devDependencies": { @@ -840,9 +840,9 @@ } }, "node_modules/@xo-cash/types": { - "version": "0.0.4", - "resolved": "https://registry.npmjs.org/@xo-cash/types/-/types-0.0.4.tgz", - "integrity": "sha512-XBB/jNC1orDcMUi5Q0g+fnPXdKdb+n7OyqmJ8pxFrfM0y9OAdLjfddMih/JWfOWQtBkvY0CtdaZapgBSrvLeSw==", + "version": "0.0.6", + "resolved": "https://verdaccio.harvmaster.com/@xo-cash/types/-/types-0.0.6.tgz", + "integrity": "sha512-SZBY8RA6V6lhdr5qwJtcQZVWqeT++pHXwOTHJeeBT3LDABIP7wcmoniKYuQehtmqV8Ttb8LGL4d0XemBd+6U7Q==", "license": "MIT", "dependencies": { "@bitauth/libauth": "^3.1.0-next.8" @@ -850,7 +850,7 @@ }, "node_modules/@xo-cash/types/node_modules/@bitauth/libauth": { "version": "3.1.0-next.8", - "resolved": "https://registry.npmjs.org/@bitauth/libauth/-/libauth-3.1.0-next.8.tgz", + "resolved": "https://verdaccio.harvmaster.com/@bitauth/libauth/-/libauth-3.1.0-next.8.tgz", "integrity": "sha512-Pm+Ju+YP3JeBLLTiVrBnia2wwE4G17r4XqpvPRMcklElJTe8J6x3JgKRg1by0Xm3ZY6UFxACkEAoSA+x419/zA==", "license": "MIT", "engines": { @@ -858,13 +858,14 @@ } }, "node_modules/@xo-cash/utils": { - "version": "0.0.3", - "resolved": "git+https://git.harvmaster.com/Harvmaster/xo-cash-utils.git#fb3362783c749265630e0881c07614777c147050", + "version": "0.0.8", + "resolved": "https://verdaccio.harvmaster.com/@xo-cash/utils/-/utils-0.0.8.tgz", + "integrity": "sha512-HTD0TzrpqGb3vGNUXHh19uPAwj/kphcg04JIinzhc4mdNI1Jv7t5VPVMFQ3OIAqNmULQCYzf2b0yGQ6stVrhHA==", "license": "MIT", "dependencies": { "@bitauth/libauth": "^3.1.0-next.8", "@xo-cash/primitives": "0.0.2", - "@xo-cash/types": "0.0.4", + "@xo-cash/types": "0.0.6", "zod": "^4.3.6" } }, diff --git a/package.json b/package.json index 282cf8b..58887b9 100644 --- a/package.json +++ b/package.json @@ -11,10 +11,13 @@ }, "exports": { ".": { - "import": "./dist/index.js", - "require": "./dist/index.cjs" + "import": "./dist/src/index.js", + "require": "./dist/src/index.cjs" } }, + "files": [ + "dist" + ], "keywords": [], "author": "", "license": "ISC", @@ -26,7 +29,7 @@ "dependencies": { "@bitauth/libauth": "^3.0.0", "@xo-cash/primitives": "^0.0.2", - "@xo-cash/utils": "git+https://git.harvmaster.com/Harvmaster/xo-cash-utils.git", + "@xo-cash/utils": "^0.0.8", "zod": "^4.4.3" } } diff --git a/src/shared/client.ts b/src/shared/client.ts index 02b0acb..a194ef5 100644 --- a/src/shared/client.ts +++ b/src/shared/client.ts @@ -1,7 +1,8 @@ import { binToHex, flattenBinArray, secp256k1, sha256 } from "@bitauth/libauth"; -import { toExtendedJson } from "@xo-cash/utils"; +import { toExtendedJson, EventEmitter } from "@xo-cash/utils"; import type { WriteRequest } from "../types.js"; +import z from "zod"; export type AuthenticatedRequestHeaders = { 'X-Public-Key': string; @@ -11,7 +12,19 @@ export type AuthenticatedRequestHeaders = { export type SignedPayload = { publicKey: string; signature: string }; -export abstract class SyncClient { +export type XOSyncClientEventMap = { + 'connection-updated': 'connected' | 'disconnected'; + 'message': { + blob: unknown; + timestamp: number; + publicKey: string; + signature: string; + }; + 'subscriptions-updated': Set; + 'error': Error; +}; + +export abstract class SyncClient extends EventEmitter { abstract connect(): Promise; abstract disconnect(): Promise; @@ -119,4 +132,11 @@ export abstract class SyncClient { 'X-Signature': signature, }; } + + static messageSchema = z.object({ + blob: z.unknown(), + timestamp: z.number(), + publicKey: z.string(), + signature: z.string(), + }); } diff --git a/src/sse/client.ts b/src/sse/client.ts index 73b26dd..906b641 100644 --- a/src/sse/client.ts +++ b/src/sse/client.ts @@ -3,17 +3,10 @@ import { PrivateKey } from '@xo-cash/primitives' import { SyncClient } from '../shared/client.js'; -export type SSEClientOptions = { - onMessage: (message: unknown) => void; - onError: (error: Error) => void; -}; - export class SSEClient extends SyncClient { private readonly url: string; - public sseSession: SSESession; + public sseSession: SSESession | undefined; private readonly subscriptions: Set = new Set(); - private readonly messageListeners = new Set<(message: unknown) => void>(); - private readonly errorListeners = new Set<(error: Error) => void>(); private privateKey: PrivateKey; private textEncoder = new TextEncoder(); @@ -22,38 +15,51 @@ export class SSEClient extends SyncClient { return `${this.url}/data/subscribe`; } - constructor(url: string, privateKey: PrivateKey, options: Partial = {}) { + constructor(url: string, privateKey: PrivateKey) { super(); this.url = url; this.privateKey = privateKey; - this.sseSession = new SSESession(`${this.url}/data/subscribe`, { - method: 'POST', - body: JSON.stringify({ resourceId: [ ...this.subscriptions ] }), - }); - this.messageListeners.add(options.onMessage ?? (() => {})); - this.errorListeners.add(options.onError ?? (() => {})); + + // Whenever we get a new subscription, we need to reconnect the session with the new subscriptions in the body + // We add a small debounce to this to prevent parallel subscriptions from causing multiple reconnects + this.on('subscriptions-updated', () => { + // Nothing awaits this listener, so report failures as an error event instead of an unhandled rejection + this.reconnect().catch((error) => this.emit('error', error)); + }, 25); } async connect() { + // The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet. + // subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session. + if (this.subscriptions.size === 0) { + return; + } + + this.sseSession ??= this.createSession(); + await this.sseSession.connect(); } async disconnect() { - await this.sseSession.disconnect(); + await this.sseSession?.disconnect(); } async reconnect() { - // Disconnect the current session - await this.sseSession.disconnect(); - this.sseSession.off('message'); - this.sseSession.off('error'); + // Close the current session. Its subscription list is fixed in the request body, so it can't be reused + this.sseSession?.close(); + this.sseSession = undefined; + // Connect a new session with the current subscriptions + await this.connect(); + } + + private createSession() { // Build the request body const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] }); - // Re-create the session with the additional resource ID in the subscription list - this.sseSession = new SSESession(this.subscriptionUrl, { + // 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); @@ -79,24 +85,31 @@ export class SSEClient extends SyncClient { }, body, method: 'POST', + // Reconnect when the server ends the stream (e.g. a restart or a proxy idle timeout). By default only errors are retried + persistent: true, }); + // Create a listener to re-emit the connected and disconnected events + session.on('connected', this.handleOnConnectionUpdated.bind(this, 'connected')); + session.on('disconnected', this.handleOnConnectionUpdated.bind(this, 'disconnected')); + // Create a listener to re-emit the messages - this.sseSession.on('message', (message) => { - for (const listener of this.messageListeners) { - listener(fromExtendedJson(message.data)); + session.on('message', (message) => { + const parsedMessage = SyncClient.messageSchema.safeParse(fromExtendedJson(message.data)); + if (!parsedMessage.success) { + return; } + + this.emit('message', parsedMessage.data); }); // Create a listener to re-emit the errors - this.sseSession.on('error', (error) => { - for (const listener of this.errorListeners) { - listener(error); - } + session.on('error', (error) => { + this.emit('error', error); }); - // Connect to the server - await this.connect(); + // Return the session + return session; } // Send a POST /data/write request to the server @@ -126,7 +139,9 @@ export class SSEClient extends SyncClient { throw new Error(`Failed to write resource [${resourceId}] (${res.statusText}): ${JSON.stringify(error)}`); } - return fromExtendedJson(await res.text()); + const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[]; + + return text; } // Send a POST /data/read request to the server @@ -147,20 +162,26 @@ export class SSEClient extends SyncClient { throw new Error(`Failed to read resource: ${res.statusText}`); } - return fromExtendedJson(await res.text()); + const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[]; + + return text; } // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list async subscribe(resourceId: string) { this.subscriptions.add(resourceId); - await this.reconnect(); + this.emit('subscriptions-updated', this.subscriptions); } // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list async unsubscribe(resourceId: string) { this.subscriptions.delete(resourceId); - await this.reconnect(); + this.emit('subscriptions-updated', this.subscriptions); + } + + async handleOnConnectionUpdated(value: 'connected' | 'disconnected') { + this.emit('connection-updated', value); } } diff --git a/src/ws/client.ts b/src/ws/client.ts index c936f75..a54c7bd 100644 --- a/src/ws/client.ts +++ b/src/ws/client.ts @@ -1,4 +1,4 @@ -import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils'; +import { ExponentialBackoff, fromExtendedJson, toExtendedJson } from '@xo-cash/utils'; import type { PrivateKey } from '@xo-cash/primitives'; import { WsMessageSchema, WsSuccessResponseSchema, WsErrorResponseSchema, type WsMessage } from './types.js'; @@ -28,25 +28,35 @@ export type WsClientOptions = { /** * Minimal WebSocket client for the XO sync server. * - * This class intentionally owns only the basic WebSocket protocol: - * - * - one connection is opened at `/ws`; + * - one connection is opened at `/ws`, on demand: every request awaits the + * connection first, so callers never need to wait for one themselves; * - request IDs correlate concurrent read/write/unsubscribe calls; - * - subscriptions stay on that connection and receive pushed events; and - * - a dropped connection fails pending requests but is not reconnected. + * - subscriptions are remembered and re-sent on every new socket, because the + * server drops a socket's topics when it closes; and + * - a dropped connection fails pending requests and is reopened with backoff + * until `disconnect()` is called. Requests made in the meantime wait for it. * - * A production client would normally wrap this class with retry, heartbeat, - * resubscription, and request-timeout behavior. Those concerns are omitted - * here to keep the transport demo easy to follow. + * Heartbeats and request timeouts are not implemented. */ export class WsClient extends SyncClient { private readonly url: string; private readonly privateKey: PrivateKey; - private socket: WebSocket | undefined; private readonly pendingRequests = new Map(); private readonly messageListeners = new Set<(message: WsMessage) => void>(); private readonly errorListeners = new Set<(error: Error) => void>(); + /** Resource IDs to (re)subscribe to whenever a socket opens. */ + private readonly subscriptions = new Set(); + + /** + * 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 | undefined; + + /** Retries a failed connection attempt forever (maxAttempts 0), waiting up to 30s between attempts. */ + private readonly connectBackoff = new ExponentialBackoff({ maxAttempts: 0, maxDelay: 30_000 }); + constructor( url: string, privateKey: PrivateKey, @@ -60,59 +70,101 @@ export class WsClient extends SyncClient { this.errorListeners.add(options.onError ?? (() => {})); } - /** Open the socket and resolve once the WebSocket handshake completes. */ + /** Connect, and keep reconnecting on drops until `disconnect()` is called. */ async connect(): Promise { - if (this.socket?.readyState === WebSocket.OPEN) { - return; - } - - await new Promise((resolve, reject) => { - const socket = new WebSocket(WsClient.httpToWsUrl(this.url)); - this.socket = socket; - - socket.onopen = () => resolve(); - - socket.onmessage = (event) => { - this.handleMessage(String(event.data)); - }; - - socket.onerror = (ev) => { - const error = new Error('WebSocket connection failed'); - this.emitError(error); - reject(error); - }; - - socket.onclose = () => { - if (this.socket === socket) { - this.socket = undefined; - } - - this.rejectPendingRequests(new Error('WebSocket connection closed')); - }; - }); + await this.getWebSocket(); } /** - * Close the socket. + * 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 { - const socket = this.socket; - this.socket = undefined; + const attempt = this.webSocket; + this.webSocket = undefined; this.rejectPendingRequests(new Error('WebSocket client disconnected')); - if (!socket || socket.readyState === WebSocket.CLOSED) { - return; + // 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 { + if (this.webSocket) { + return this.webSocket; } - // Wait for `close` so callers know the underlying connection is gone. - await new Promise((resolve) => { - socket.onclose = () => resolve(); - socket.close(); + 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 { + const attempt: Promise = this.connectBackoff.run(async ({ stopRetries }) => { + // `attempt` is only read after this await: run() calls this function before `attempt` is assigned. + const socket = await WsClient.waitForOpen(new WebSocket(WsClient.httpToWsUrl(this.url))); + + // disconnect() was called while connecting. The thrown error rejects `attempt`. + if (this.webSocket !== attempt) { + socket.close(); + stopRetries('WebSocket client disconnected'); + throw new Error('WebSocket client disconnected'); + } + + socket.onmessage = (event) => this.handleMessage(String(event.data)); + socket.onerror = () => this.emitError(new Error('WebSocket error')); + socket.onclose = () => { + this.rejectPendingRequests(new Error('WebSocket connection closed')); + this.emit('connection-updated', 'disconnected'); + + // Reconnect, unless disconnect() was called + if (this.webSocket === attempt) { + this.webSocket = undefined; + this.getWebSocket(); + } + }; + + this.emit('connection-updated', 'connected'); + + return socket; + }, { + // Report failed handshakes, but not the error thrown above after disconnect() + onError: (error) => { + if (this.webSocket === attempt) { + this.emitError(error); + } + }, }); + + // Nothing may be awaiting this attempt (e.g. a reconnect started by onclose), and it rejects if disconnect() stops it. + // Mark the rejection as handled so it isn't reported as unhandled. Callers awaiting `attempt` still receive it. + attempt.catch(() => {}); + + return attempt; } /** Read every stored instance of one resource. */ @@ -131,52 +183,70 @@ export class WsClient extends SyncClient { } /** - * Register one resource topic on the current connection. + * Subscribe to one resource topic, now and on every future socket. * - * The subscribe route is intentionally long-running and therefore does not - * send an acknowledgement. This method resolves after the request frame has - * been handed to the socket; future updates arrive through `onMessage`. + * The server never replies to a successful subscribe (the route stays pending until the topic is + * removed), so this resolves once the frame is sent. A failed subscribe comes back as an `error` + * message that matches no pending request, which handleMessage() reports through emitError(). */ async subscribe(resourceId: string): Promise { - await this.request('/data/subscribe', { resourceId: [resourceId] }); + 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 on the current connection. + * Unsubscribe from one resource topic. */ async unsubscribe(resourceId: string): Promise { + this.subscriptions.delete(resourceId); + + // Not connected: the server holds no topics for us, and future sockets won't replay this one. + if (!this.webSocket) { + return; + } + await this.request('/data/unsubscribe', { resourceId: [resourceId] }); } /** Send a request and wait for the response carrying the same ID. */ - private async request(path: string, body?: unknown): Promise { + private async request(path: string, body: unknown): Promise { const id = crypto.randomUUID(); - const response = new Promise((resolve, reject) => { + 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((resolve, reject) => { this.pendingRequests.set(id, { resolve, reject }); }); - - // Sign the request body - const headers = SyncClient.signRequest(this.privateKey.toBytes(), path, toExtendedJson(body)); - - try { - this.send({ id, path, body, headers }); - } catch (error) { - // Avoid leaving a promise in the map when the socket was not open. - this.pendingRequests.delete(id); - throw error; - } - - return response; } - /** Encode one complete request envelope using the shared Extended JSON codec. */ - private send(request: WsRequest): void { - if (!this.socket || this.socket.readyState !== WebSocket.OPEN) { + /** Connect if needed, then sign and send one request envelope using the shared Extended JSON codec. */ + private async send(path: string, body: unknown, id: string = crypto.randomUUID()): Promise { + 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'); } - this.socket.send(toExtendedJson(request)); + // Signed after connecting: the server rejects signatures older than its timestamp window (5 minutes by default). + const headers = SyncClient.signRequest(this.privateKey.toBytes(), path, toExtendedJson(body)); + const request: WsRequest = { id, path, body, headers }; + + socket.send(toExtendedJson(request)); + } + + /** Resolve with the socket once it opens, or reject if the handshake fails. */ + private static waitForOpen(socket: WebSocket): Promise { + 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. */ @@ -197,6 +267,22 @@ export class WsClient extends SyncClient { listener(message); } + // Errors with no pending request are failed subscribes (see sendSubscribe) or uncorrelated server errors. + const error = WsErrorResponseSchema.safeParse(message); + if (error.success) { + this.emitError(new Error(`${error.data.error} (${error.data.statusCode})`)); + return; + } + + if (message.type === 'instance-changed') { + const parsedMessage = SyncClient.messageSchema.safeParse(message.data); + if (!parsedMessage.success) { + return; + } + + this.emit('message', parsedMessage.data); + } + return; }