Toncenter Streaming API v2 clients

package.json

@@ -18,6 +18,7 @@
         "build": "rm -fr dist && tsc -p ./tsconfig.build.json",
         "typecheck": "tsc",
         "test": "jest --verbose",
+        "test:streaming:live": "node --env-file=.env node_modules/.bin/jest --runInBand --forceExit --verbose src/client/streaming/Streaming.transactions.live.spec.ts src/client/streaming/Streaming.wallet.e2e.live.spec.ts",
         "format": "biome format --write .",
         "format:check": "biome format .",
         "coverage": "jest -c ./jest-coverage.config.js",

src/client/streaming/AbstractStreamingClient.ts

@@ -0,0 +1,178 @@
+/**
+ * Copyright (c) Whales Corp.
+ * All Rights Reserved.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+import { StreamingClosedError, StreamingSupersededError } from "./errors";
+import type { StreamingErrorContext, StreamingTransport } from "./errors";
+import type { StreamingError } from "./errors";
+import {
+    type ResolvedStreamingSubscription,
+    resolveStreamingSubscription,
+    sameSubscription,
+} from "./subscriptionState";
+import { TypedEventEmitter } from "./TypedEventEmitter";
+import type { StreamingEventMap, StreamingSubscription } from "./types";
+import { type Deferred, deferred } from "./utils";
+
+export type DesiredSubscription = {
+    snapshot: ResolvedStreamingSubscription;
+    waiter: Deferred<void>;
+};
+
+export abstract class AbstractStreamingClient extends TypedEventEmitter<StreamingEventMap> {
+    readonly #url: string;
+    readonly #transport: StreamingTransport;
+    #reconcilePromise: Promise<void> | null = null;
+    #applied: ResolvedStreamingSubscription | null = null;
+    #desired: DesiredSubscription | null = null;
+
+    constructor(transport: StreamingTransport, url: string) {
+        super();
+        this.#transport = transport;
+        this.#url = url;
+    }
+
+    subscribe(params: StreamingSubscription): Promise<void> {
+        const snapshot = resolveStreamingSubscription(params);
+        if (
+            this.ready &&
+            this.#applied &&
+            sameSubscription(this.#applied, snapshot)
+        ) {
+            return Promise.resolve();
+        }
+
+        if (
+            this.#desired &&
+            !this.#desired.waiter.settled &&
+            sameSubscription(this.#desired.snapshot, snapshot)
+        ) {
+            return this.#desired.waiter.promise;
+        }
+
+        // onSupersede() must run before the waiter rejection so that SSE
+        // can abort the in-flight fetch before the rejection propagates.
+        this.onSupersede();
+
+        this.#desired?.waiter.reject(
+            new StreamingSupersededError(
+                "Streaming subscribe was superseded by a newer snapshot",
+                this.ctx("subscribe"),
+            ),
+        );
+
+        const desired: DesiredSubscription = {
+            snapshot,
+            waiter: deferred<void>(),
+        };
+        this.#desired = desired;
+        this.#reconcile();
+        return desired.waiter.promise;
+    }
+
+    async close(): Promise<void> {
+        const error = new StreamingClosedError(
+            "Streaming transport is closing",
+            this.ctx("close"),
+        );
+        this.#applied = null;
+        this.#rejectDesiredWaiter(error);
+        await this.closeTransport(error);
+        await this.#reconcilePromise;
+        this.removeAllListeners();
+    }
+
+    get ready(): boolean {
+        return this.isSessionReady;
+    }
+
+    protected get url(): string {
+        return this.#url;
+    }
+
+    protected abstract get isSessionReady(): boolean;
+
+    protected abstract applySubscription(
+        desired: DesiredSubscription,
+    ): Promise<"ready" | "replaced">;
+
+    protected abstract closeTransport(error: StreamingError): Promise<void>;
+
+    protected onSupersede(): void {}
+
+    protected ctx(
+        phase: string,
+        extra?: Partial<StreamingErrorContext>,
+    ): StreamingErrorContext {
+        return {
+            transport: this.#transport,
+            endpoint: this.#url,
+            phase,
+            ...extra,
+        };
+    }
+
+    protected isSuperseded(desired: DesiredSubscription): boolean {
+        return this.#desired !== desired;
+    }
+
+    #rejectDesiredWaiter(reason: unknown): void {
+        const desired = this.#desired;
+        this.#desired = null;
+        desired?.waiter.reject(reason);
+    }
+
+    #reconcile(): void {
+        if (this.#reconcilePromise) {
+            return;
+        }
+
+        this.#reconcilePromise = (async () => {
+            try {
+                while (this.#desired) {
+                    const target = this.#desired;
+                    if (
+                        this.ready &&
+                        this.#applied &&
+                        sameSubscription(this.#applied, target.snapshot)
+                    ) {
+                        if (this.#desired !== target) {
+                            continue;
+                        }
+                        target.waiter.resolve();
+                        break;
+                    }
+
+                    let outcome: "ready" | "replaced";
+                    try {
+                        outcome = await this.applySubscription(target);
+                    } catch (error) {
+                        if (this.isSuperseded(target)) {
+                            continue;
+                        }
+                        throw error;
+                    }
+
+                    if (outcome === "replaced") {
+                        continue;
+                    }
+
+                    this.#applied = target.snapshot;
+                    if (this.#desired === target) {
+                        this.#desired = null;
+                        target.waiter.resolve();
+                        break;
+                    }
+                }
+            } catch (error) {
+                this.#rejectDesiredWaiter(error);
+            } finally {
+                this.#reconcilePromise = null;
+            }
+        })();
+    }
+}

src/client/streaming/SseParser.ts

@@ -0,0 +1,148 @@
+/**
+ * Copyright (c) Whales Corp.
+ * All Rights Reserved.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+export type SseEvent = {
+    event?: string;
+    data: string;
+    id?: string;
+};
+
+const MAX_BUFFER_SIZE = 4 * 1024 * 1024; // 4 MB
+
+export class SseParser {
+    #buffer = "";
+    #onEvent: (event: SseEvent) => void;
+    #isStartOfStream = true;
+
+    constructor(onEvent: (event: SseEvent) => void) {
+        this.#onEvent = onEvent;
+    }
+
+    feed(chunk: string): void {
+        if (chunk.length === 0) {
+            return;
+        }
+
+        if (this.#isStartOfStream) {
+            this.#isStartOfStream = false;
+            if (chunk.charCodeAt(0) === 0xfeff) {
+                chunk = chunk.slice(1);
+            }
+        }
+
+        this.#buffer += chunk;
+
+        if (this.#buffer.length > MAX_BUFFER_SIZE) {
+            throw new Error(
+                `SSE buffer exceeded ${MAX_BUFFER_SIZE} bytes without an event boundary`,
+            );
+        }
+
+        while (true) {
+            const boundary = this.#findBoundary();
+            if (boundary === null) {
+                break;
+            }
+
+            const part = this.#normalizeChunk(
+                this.#buffer.slice(0, boundary.index),
+            );
+            this.#buffer = this.#buffer.slice(boundary.index + boundary.length);
+            this.#dispatch(part);
+        }
+    }
+
+    // SSE spec requires dispatching trailing events without a blank-line delimiter.
+    finish(): void {
+        if (this.#buffer.length === 0) {
+            return;
+        }
+
+        const part = this.#normalizeChunk(this.#buffer);
+        this.#buffer = "";
+        this.#dispatch(part);
+    }
+
+    #findBoundary(): { index: number; length: number } | null {
+        const nnIndex = this.#buffer.indexOf("\n\n");
+        // \n\n is the overwhelmingly common SSE delimiter.
+        // Only scan for rare \r\n\r\n and \r\r if \r exists in the buffer.
+        if (this.#buffer.indexOf("\r") === -1) {
+            return nnIndex === -1 ? null : { index: nnIndex, length: 2 };
+        }
+
+        const crlfIndex = this.#buffer.indexOf("\r\n\r\n");
+        const crIndex = this.#buffer.indexOf("\r\r");
+
+        let best: { index: number; length: number } | null = null;
+        if (nnIndex !== -1) {
+            best = { index: nnIndex, length: 2 };
+        }
+        if (crlfIndex !== -1 && (best === null || crlfIndex < best.index)) {
+            best = { index: crlfIndex, length: 4 };
+        }
+        if (crIndex !== -1 && (best === null || crIndex < best.index)) {
+            best = { index: crIndex, length: 2 };
+        }
+
+        return best;
+    }
+
+    #normalizeChunk(part: string): string {
+        return part.includes("\r") ? part.replace(/\r\n?/g, "\n") : part;
+    }
+
+    #dispatch(part: string): void {
+        if (!part) {
+            return;
+        }
+
+        let event: string | undefined;
+        let id: string | undefined;
+        const dataLines: string[] = [];
+
+        for (const line of part.split("\n")) {
+            if (!line || line.startsWith(":")) {
+                continue;
+            }
+
+            const colonIndex = line.indexOf(":");
+            const field = colonIndex === -1 ? line : line.slice(0, colonIndex);
+            let value = colonIndex === -1 ? "" : line.slice(colonIndex + 1);
+            if (value.startsWith(" ")) {
+                value = value.slice(1);
+            }
+
+            switch (field) {
+                case "event":
+                    event = value;
+                    break;
+                case "data":
+                    dataLines.push(value);
+                    break;
+                case "id":
+                    if (!value.includes("\u0000")) {
+                        id = value;
+                    }
+                    break;
+                default:
+                    break;
+            }
+        }
+
+        if (dataLines.length === 0) {
+            return;
+        }
+
+        this.#onEvent({
+            event,
+            data: dataLines.join("\n"),
+            id,
+        });
+    }
+}