sync-server-example/source/sse/client.ts

import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
import type { PrivateKey } from '@xo-cash/primitives';

import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.ts';

export class SSEClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
    private readonly url: string;
    public sseSession: SSESession | undefined;
    private readonly subscriptions: Set<string> = new Set();

    private privateKey: PrivateKey;
    private textEncoder = new TextEncoder();

    private get subscriptionUrl(): string {
        return `${this.url}/data/subscribe`;
    }

    /**
     * Creates a new SSE client
     * @param url - The URL of the server
     * @param privateKey - The private key to use for authentication
     */
    constructor(url: string, privateKey: PrivateKey) {
        super();

        this.url = url;
        this.privateKey = privateKey;

        // Whenever we get a new subscription, we need to reconnect the session with the new subscriptions in the body
        // We add a small debounce to this to prevent parallel subscriptions from causing multiple reconnects
        this.on(
            'subscriptions-updated',
            () => {
                // Nothing awaits this listener, so report failures as an error event instead of an unhandled rejection
                this.reconnect().catch((error) => this.emit('error', error));
            },
            25,
        );
    }

    /**
     * Connects to the server
     */
    async connect(): Promise<void> {
        // The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet.
        // subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session.
        if (this.subscriptions.size === 0) {
            return;
        }

        this.sseSession ??= this.createSession();

        await this.sseSession.connect();
    }

    /**
     * Disconnects from the server
     */
    async disconnect(): Promise<void> {
        this.sseSession?.disconnect();
    }

    /**
     * Reconnects to the server
     */
    async reconnect(): Promise<void> {
        // Close the current session. Its subscription list is fixed in the request body, so it can't be reused
        this.sseSession?.close();
        this.sseSession = undefined;

        // Connect a new session with the current subscriptions
        await this.connect();
    }

    /**
     * Creates a new SSE session
     * @returns The SSE session
     */
    private createSession(): SSESession {
        // Build the request body
        const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] });

        // Create a new SSE session
        const session = new SSESession(this.subscriptionUrl, {
            onRequest: async (request): Promise<RequestInit> => {
                // Get the path from the URL
                const url = new URL(this.subscriptionUrl);
                const path = url.pathname;

                // Sign the request body
                const authHeaders = BaseSyncClient.signRequest(this.privateKey.toBytes(), path, request.body);

                // Initialize the request headers if they don't exist
                request.headers ??= {};

                // Add the authentication headers to the request
                request.headers = {
                    ...request.headers,
                    ...authHeaders,
                };

                // Return the request with the authentication headers
                return request;
            },
            headers: {
                'Content-Type': 'application/json',
            },
            body,
            method: 'POST',
            // Reconnect when the server ends the stream (e.g. a restart or a proxy idle timeout). By default only errors are retried
            persistent: true,
        });

        // Create a listener to re-emit the connected and disconnected events
        session.on('connected', this.handleOnConnectionUpdated.bind(this, 'connected'));
        session.on('disconnected', this.handleOnConnectionUpdated.bind(this, 'disconnected'));

        // Create a listener to re-emit the messages
        session.on('message', (message): void => {
            const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(fromExtendedJson(message.data));
            if (!parsedMessage.success) {
                return;
            }

            this.emit('resource-instance-updated', {
                ...parsedMessage.data,
                value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType,
            });
        });

        // Create a listener to re-emit the errors
        session.on('error', (error) => {
            this.emit('error', error);
        });

        // Return the session
        return session;
    }

    /**
     * Send a POST /data/write request to the server
     * @param resources - The resources to write
     * @returns The Resource Object with the decoded value
     */
    async writeMany(resources: Record<string, PayloadType>): Promise<ResourcesRecord<PayloadType>> {
        // Sign the resource write requests
        const signedResources = Object.entries(resources).map(([ resourceId, value ]) => {
            const valueStr = toExtendedJson(value);
            const valueBytes = this.textEncoder.encode(valueStr);

            return BaseSyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
        });

        // Send the resource write requests to the server
        const res = await this.sendRequest('/data/write', {
            resources: signedResources,
        });

        // If the response is not ok, throw an error
        if (!res.ok) {
            const error = await res.json();
            throw new Error(`Failed to write resources [${Object.keys(resources).join(', ')}] (${res.statusText}): ${JSON.stringify(error)}`);
        }

        // Decode the response and return the resources
        const decodedResources = await this.decodeResponse(res);

        // Return the decoded resources
        return decodedResources;
    }

    /**
     * Send a POST /data/read request to the server
     * @param resourceIds - The resource IDs to read
     * @returns The resources with the read values
     */
    async readMany(resourceIds: string[]): Promise<ResourcesRecord<PayloadType>> {
        // Send the resource read requests to the server
        const res = await this.sendRequest('/data/get', {
            resourceId: resourceIds,
        });

        // If the response is not ok, throw an error
        if (!res.ok) {
            throw new Error(`Failed to read resources [${resourceIds.join(', ')}] (${res.statusText})`);
        }

        // Decode the response and return the resources
        const decodedResources = await this.decodeResponse(res);

        // Return the decoded resources
        return decodedResources;
    }

    /**
     * Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list
     * @param resourceId - The resource ID to add to the subscription list
     */
    async subscribe(resourceId: string): Promise<void> {
        this.subscriptions.add(resourceId);

        this.emit('subscriptions-updated', this.subscriptions);
    }

    /**
     * Destroy the current SSE Session, then re-create it without the resource ID in the subscription list
     * @param resourceId - The resource ID to remove from the subscription list
     */
    async unsubscribe(resourceId: string): Promise<void> {
        this.subscriptions.delete(resourceId);

        this.emit('subscriptions-updated', this.subscriptions);
    }

    async handleOnConnectionUpdated(value: 'connected' | 'disconnected'): Promise<void> {
        this.emit('connection-updated', value);
    }

    /**
     * Sends a request to the server
     * @param path - The path to send the request to
     * @param body - The body of the request
     * @returns The response from the server
     */
    private async sendRequest(path: string, body: unknown): Promise<Response> {
        // Build the URL
        const url = `${this.url}${path}`;

        // Format the body as Extended JSON
        const formattedBody = toExtendedJson(body);

        // Send the request to the server
        const res = await fetch(url, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
                ...BaseSyncClient.signRequest(this.privateKey.toBytes(), path, body),
            },
            body: formattedBody,
        });

        // Return the response
        return res;
    }

    /**
     * Decodes the blobs on the response body to the type T
     * @param response - The response to decode
     * @returns The decoded resources
     */
    private async decodeResponse(response: Response): Promise<ResourcesRecord<PayloadType>> {
        // Decode the response from Extended JSON
        const text = fromExtendedJson(await response.text());

        // Parse the response as a record of resources
        const resources = BaseSyncClient.resourcesResponseSchema.parse(text);

        // Decode the blobs and convert them from extended json to the type T
        const decodedResources = BaseSyncClient.decodeResourcesResponse<PayloadType>(resources);

        return decodedResources;
    }
}