diff --git a/package.json b/package.json index ffcd4ca5..19d5dcb5 100644 --- a/package.json +++ b/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", diff --git a/src/client/streaming/AbstractStreamingClient.ts b/src/client/streaming/AbstractStreamingClient.ts new file mode 100644 index 00000000..2060f701 --- /dev/null +++ b/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; +}; + +export abstract class AbstractStreamingClient extends TypedEventEmitter { + readonly #url: string; + readonly #transport: StreamingTransport; + #reconcilePromise: Promise | 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 { + 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(), + }; + this.#desired = desired; + this.#reconcile(); + return desired.waiter.promise; + } + + async close(): Promise { + 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; + + protected onSupersede(): void {} + + protected ctx( + phase: string, + extra?: Partial, + ): 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; + } + })(); + } +} diff --git a/src/client/streaming/SseParser.ts b/src/client/streaming/SseParser.ts new file mode 100644 index 00000000..06925a23 --- /dev/null +++ b/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, + }); + } +} diff --git a/src/client/streaming/Streaming.transactions.live.spec.ts b/src/client/streaming/Streaming.transactions.live.spec.ts new file mode 100644 index 00000000..ec934581 --- /dev/null +++ b/src/client/streaming/Streaming.transactions.live.spec.ts @@ -0,0 +1,186 @@ +/** + * Live smoke test that compares transaction streaming across all four + * provider + transport combinations (toncenter-ws, toncenter-sse, + * tonapi-ws, tonapi-sse) on mainnet. + * + * All four clients subscribe to the same address and wait until every source + * has received at least one transaction event (up to a ~95 s timeout). + * After that the suite verifies that: + * - every source received at least one transaction event + * - no streaming errors were emitted + * + * Required env vars: + * TONCENTER_API_KEY – mainnet Toncenter API key + * TONAPI_API_KEY – mainnet TonAPI key + * STREAMING_TEST_ADDRESS – (optional) address to watch; defaults to a + * well-known high-activity address + */ + +import { TonWsClient } from "./TonWsClient"; +import { TonSseClient } from "./TonSseClient"; +import type { StreamingClient } from "./types"; + +const TONCENTER_API_KEY = process.env.TONCENTER_API_KEY; +const TONAPI_API_KEY = process.env.TONAPI_API_KEY; +const TEST_ADDRESS = + process.env.STREAMING_TEST_ADDRESS ?? + "EQCS4UEa5UaJLzOyyKieqQOQ2P9M-7kXpkO5HnP3Bv250cN3"; +const WATCH_TIMEOUT_MS = 95_000; +const POLL_INTERVAL_MS = 250; +const CLOSE_SETTLE_MS = 1_500; + +const describeLive = + TONCENTER_API_KEY && TONAPI_API_KEY ? describe : describe.skip; + +const SOURCES = [ + "toncenter-ws", + "toncenter-sse", + "tonapi-ws", + "tonapi-sse", +] as const; + +type Source = (typeof SOURCES)[number]; + +type SeenEvent = { + source: Source; +}; + +describeLive("streaming live transaction watch", () => { + jest.setTimeout(130_000); + + const clients = {} as Record; + const errors = {} as Record; + const events: SeenEvent[] = []; + const detachHandlers: (() => void)[] = []; + + beforeAll(async () => { + clients["toncenter-ws"] = new TonWsClient({ + endpoint: "wss://toncenter.com/api/streaming/v2/ws", + apiKey: TONCENTER_API_KEY, + }); + clients["tonapi-ws"] = new TonWsClient({ + endpoint: "wss://tonapi.io/streaming/v2/ws", + apiKey: TONAPI_API_KEY, + apiKeyParam: "token", + }); + clients["toncenter-sse"] = new TonSseClient({ + endpoint: "https://toncenter.com/api/streaming/v2/sse", + apiKey: TONCENTER_API_KEY, + }); + clients["tonapi-sse"] = new TonSseClient({ + endpoint: "https://tonapi.io/streaming/v2/sse", + apiKey: TONAPI_API_KEY, + apiKeyParam: "token", + }); + + for (const source of SOURCES) { + errors[source] = []; + detachHandlers.push( + attachCollectors( + clients[source], + source, + events, + errors[source], + ), + ); + } + + const subscription = { + addresses: [TEST_ADDRESS], + types: ["transactions"] as const, + }; + + await Promise.all( + SOURCES.map((s) => clients[s].subscribe(subscription)), + ); + + for (const source of SOURCES) { + expect(clients[source].ready).toBe(true); + } + + process.stdout.write( + `Waiting for transactions on address ${TEST_ADDRESS} from ${SOURCES.length} sources (timeout ${WATCH_TIMEOUT_MS / 1000}s)...\n`, + ); + + await waitUntilAllSourcesSeen(events, WATCH_TIMEOUT_MS); + }); + + afterAll(async () => { + for (const detach of detachHandlers) { + detach(); + } + + await Promise.all(SOURCES.map((s) => clients[s].close())); + await delay(CLOSE_SETTLE_MS); + }); + + it("receives events from all four sources", () => { + expect(events.length).toBeGreaterThan(0); + + const seenSources = new Set(events.map((e) => e.source)); + expect(seenSources).toEqual(new Set(SOURCES)); + }); + + it("reports no streaming errors", () => { + for (const source of SOURCES) { + expect(errors[source]).toEqual([]); + } + }); +}); + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function attachCollectors( + target: StreamingClient, + source: Source, + events: SeenEvent[], + errors: Error[], +): () => void { + let first = true; + const detachers = [ + target.on("transactions", () => { + if (first) { + first = false; + process.stdout.write(` ✓ ${source}: first transaction received\n`); + } + events.push({ source }); + }), + target.on("error", (error: Error) => { + errors.push(error); + }), + ]; + + return () => detachers.forEach((d) => d()); +} + +async function waitUntilAllSourcesSeen( + events: readonly SeenEvent[], + timeoutMs: number, +): Promise { + const deadline = Date.now() + timeoutMs; + + const startedAt = Date.now(); + + while (Date.now() < deadline) { + const seen = new Set(events.map((e) => e.source)); + if (SOURCES.every((s) => seen.has(s))) { + process.stdout.write( + `All sources received transactions in ${((Date.now() - startedAt) / 1000).toFixed(1)}s\n`, + ); + return; + } + await delay(POLL_INTERVAL_MS); + } + + const seen = new Set(events.map((e) => e.source)); + const missing = SOURCES.filter((s) => !seen.has(s)); + throw new Error( + `Timed out waiting for transaction events from: ${missing.join(", ")}`, + ); +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/src/client/streaming/Streaming.wallet.e2e.live.spec.ts b/src/client/streaming/Streaming.wallet.e2e.live.spec.ts new file mode 100644 index 00000000..21c8a76b --- /dev/null +++ b/src/client/streaming/Streaming.wallet.e2e.live.spec.ts @@ -0,0 +1,963 @@ +/** + * Live end-to-end test for the streaming clients (SSE / WebSocket) against the TON testnet. + * + * The suite derives a wallet from a mnemonic, opens two streaming connections + * (one for address-level events, one for trace-level events), and then: + * + * 1. Sends a small TON self-transfer and verifies that the address stream + * delivers matching `transactions`, `actions`, `account_state_change` + * events and that the trace stream delivers a corresponding `trace` event + * with a consistent `trace_external_hash_norm` across all of them. + * + * 2. Discovers a jetton with a non-zero balance on the wallet (via TonAPI), + * sends a minimal jetton transfer, and — in addition to the checks above — + * verifies that a `jettons_change` event arrives with the correct owner. + * If no transferable jetton is found the test logs a warning and returns + * early rather than failing. + * + * Configuration is read from environment variables (use `node --env-file=.env` + * to load from a file). The suite is skipped entirely when + * `RUN_STREAMING_WALLET_E2E` is not set or the required env vars are missing. + * + * Required env vars: + * WALLET_MNEMONIC – space-separated mnemonic words + * WALLET_VERSION – one of "v3r1", "v3r2", "v4", "v5" + * TONCENTER_API_KEY_TESTNET – Toncenter testnet API key + * TONAPI_API_KEY – TonAPI key (used for jetton discovery and optionally as streaming key) + * STREAMING_PROVIDER – "toncenter" or "tonapi" + * CONNECTION_TYPE – "sse" or "websocket" (default "sse") + * RUN_STREAMING_WALLET_E2E – "1" or "true" to enable the suite + */ + +import fs from "node:fs"; +import path from "node:path"; +import { + Address, + beginCell, + Cell, + comment, + external, + internal, + Message, + MessageRelaxed, + SendMode, + toNano, +} from "@ton/core"; +import { mnemonicToPrivateKey } from "@ton/crypto"; +import axios from "axios"; +import { z } from "zod"; +import { JettonMaster } from "../../jetton/JettonMaster"; +import { JettonWallet } from "../../jetton/JettonWallet"; +import { WalletContractV3R1 } from "../../wallets/WalletContractV3R1"; +import { WalletContractV3R2 } from "../../wallets/WalletContractV3R2"; +import { WalletContractV4 } from "../../wallets/WalletContractV4"; +import { WalletContractV5R1 } from "../../wallets/WalletContractV5R1"; +import { TonClient } from "../TonClient"; +import { TonSseClient } from "./TonSseClient"; +import { TonWsClient } from "./TonWsClient"; +import type { + StreamingAccountStateEvent, + StreamingActionsEvent, + StreamingClient, + StreamingJettonsEvent, + StreamingTraceEvent, + StreamingTransactionsEvent, +} from "./types"; + +const liveEnvSchema = z.object({ + WALLET_MNEMONIC: z.string().min(1), + WALLET_VERSION: z.enum(["v3r1", "v3r2", "v4", "v5"]), + TONCENTER_API_KEY: z.string().min(1), + TONAPI_API_KEY: z.string().min(1), + STREAMING_PROVIDER: z.enum(["toncenter", "tonapi"]), + CONNECTION_TYPE: z.enum(["sse", "websocket"]), +}); + +const liveEnvResult = liveEnvSchema.safeParse({ + WALLET_MNEMONIC: process.env.WALLET_MNEMONIC, + WALLET_VERSION: process.env.WALLET_VERSION, + TONCENTER_API_KEY: process.env.TONCENTER_API_KEY_TESTNET, + TONAPI_API_KEY: process.env.TONAPI_API_KEY, + STREAMING_PROVIDER: process.env.STREAMING_PROVIDER, + CONNECTION_TYPE: normalizeConnectionType( + process.env.CONNECTION_TYPE ?? "sse", + ), +}); + +const RUN_STREAMING_WALLET_E2E = + process.env.RUN_STREAMING_WALLET_E2E === "1" || + process.env.RUN_STREAMING_WALLET_E2E === "true"; +const describeLive = + liveEnvResult.success && RUN_STREAMING_WALLET_E2E + ? describe + : describe.skip; + +const TESTNET_TONCENTER_RPC = "https://testnet.toncenter.com/api/v2/jsonRPC"; +const TONAPI_TESTNET_ACCOUNTS = "https://testnet.tonapi.io/v2/accounts"; + +const TON_TRANSFER_VALUE = toNano("0.001"); +const JETTON_TRANSFER_MESSAGE_VALUE = toNano("0.03"); +const TRACE_TIMEOUT_MS = 5_000; +const POLL_INTERVAL_MS = 500; +const ADDRESS_STREAM_TYPES = [ + "transactions", + "actions", + "account_state_change", + "jettons_change", +] as const; + +type SupportedWalletVersion = z.infer["WALLET_VERSION"]; +type SupportedWallet = + | WalletContractV3R1 + | WalletContractV3R2 + | WalletContractV4 + | WalletContractV5R1; + +type WalletContext = { + version: SupportedWalletVersion; + contract: SupportedWallet; + secretKey: Buffer; +}; + +type ObservedAddressEvents = { + transactions: StreamingTransactionsEvent[]; + actions: StreamingActionsEvent[]; + traces: StreamingTraceEvent[]; + accountStates: StreamingAccountStateEvent[]; + jettons: StreamingJettonsEvent[]; + errors: Error[]; +}; + +type AddressEventCursor = { + transactions: number; + actions: number; + traces: number; + accountStates: number; + jettons: number; +}; + +type SendObservation = { + cursor: AddressEventCursor; + transactions: StreamingTransactionsEvent; + actions: StreamingActionsEvent; + trace: StreamingTraceEvent; + accountState: StreamingAccountStateEvent; +}; + +type StreamLogSource = "address-stream" | "trace-stream"; + +type StreamLogEntry = { + timestamp: string; + source: StreamLogSource; + type: + | "transactions" + | "actions" + | "trace" + | "account_state_change" + | "jettons_change" + | "error"; + summary: string; + payload: unknown; +}; + +type StreamingLogWriter = { + filePath: string; + recordEvent( + source: StreamLogSource, + type: StreamLogEntry["type"], + event: unknown, + ): void; + recordError(source: StreamLogSource, error: Error): void; + flush(): void; +}; + +const tonApiJettonsSchema = z.object({ + balances: z.array( + z.object({ + balance: z.string(), + wallet_address: z.object({ + address: z.string(), + }), + jetton: z.object({ + address: z.string(), + symbol: z.string().optional(), + name: z.string().optional(), + }), + }), + ), +}); + +describeLive("streaming wallet testnet e2e", () => { + jest.setTimeout(30_000); + + let env: z.infer; + let wallet: WalletContext; + let auxRecipient: WalletContractV4; + let tonClient: TonClient; + let addressStream: StreamingClient; + let traceStream: StreamingClient; + let addressEvents: ObservedAddressEvents; + let logWriter: StreamingLogWriter; + let detachAddressHandlers: () => void; + let detachTraceHandlers: () => void; + + beforeAll(async () => { + if (!liveEnvResult.success) { + throw new Error( + `Invalid live streaming environment: ${liveEnvResult.error.message}`, + ); + } + + env = liveEnvResult.data; + const mnemonicWords = env.WALLET_MNEMONIC.trim().split(/\s+/); + const keyPair = await mnemonicToPrivateKey(mnemonicWords); + + wallet = createWalletContext(env.WALLET_VERSION, keyPair); + auxRecipient = WalletContractV4.create({ + workchain: 0, + publicKey: keyPair.publicKey, + walletId: 0, + }); + + tonClient = new TonClient({ + endpoint: TESTNET_TONCENTER_RPC, + apiKey: env.TONCENTER_API_KEY, + }); + + const streamingApiKey = + env.STREAMING_PROVIDER === "tonapi" + ? env.TONAPI_API_KEY + : env.TONCENTER_API_KEY; + + addressStream = createStreamingClient( + env.CONNECTION_TYPE, + env.STREAMING_PROVIDER, + streamingApiKey, + ); + traceStream = createStreamingClient( + env.CONNECTION_TYPE, + env.STREAMING_PROVIDER, + streamingApiKey, + ); + + addressEvents = { + transactions: [], + actions: [], + traces: [], + accountStates: [], + jettons: [], + errors: [], + }; + + logWriter = createStreamingLogWriter({ + connectionType: env.CONNECTION_TYPE, + provider: env.STREAMING_PROVIDER, + walletAddress: wallet.contract.address, + }); + + detachAddressHandlers = attachEventCollectors( + addressStream, + "address-stream", + addressEvents, + logWriter, + ); + detachTraceHandlers = attachEventCollectors( + traceStream, + "trace-stream", + addressEvents, + logWriter, + ); + + const walletProvider = tonClient.provider( + wallet.contract.address, + wallet.contract.init, + ); + const balance = await wallet.contract.getBalance(walletProvider); + expect(balance).toBeGreaterThan(TON_TRANSFER_VALUE); + + await addressStream.subscribe( + addressSubscription(wallet.contract.address), + ); + expect(addressStream.ready).toBe(true); + }); + + afterAll(async () => { + detachAddressHandlers(); + detachTraceHandlers(); + await Promise.all([ + addressStream.close(), + traceStream.close(), + ]); + logWriter.flush(); + }); + + afterAll(() => { + expect(addressEvents.errors).toEqual([]); + }); + + it("streams address updates for a TON self-transfer", async () => { + const observation = await sendAndObserve({ + addressEvents, + traceStreaming: traceStream, + tonClient, + wallet, + buildMessages: () => [ + internal({ + to: wallet.contract.address, + value: TON_TRANSFER_VALUE, + bounce: false, + body: "streaming wallet e2e ton transfer", + }), + ], + }); + + expectObservedSend(observation, wallet.contract.address); + }); + + it("streams address and jetton updates for a jetton transfer", async () => { + const transferableJetton = await discoverTransferableJetton( + env.TONAPI_API_KEY, + wallet.contract.address, + ); + + if (!transferableJetton) { + console.warn( + "No transferable jetton found on testnet wallet — skipping jetton assertions", + ); + return; + } + + const walletProvider = tonClient.provider( + wallet.contract.address, + wallet.contract.init, + ); + const balance = await wallet.contract.getBalance(walletProvider); + expect(balance).toBeGreaterThan( + TON_TRANSFER_VALUE + JETTON_TRANSFER_MESSAGE_VALUE, + ); + + const jettonMaster = JettonMaster.create( + Address.parse(transferableJetton.jettonAddress), + ); + const jettonWalletAddress = Address.parse( + transferableJetton.walletAddress, + ); + const resolvedWalletAddress = await tonClient + .open(jettonMaster) + .getWalletAddress(wallet.contract.address); + expect(resolvedWalletAddress.equals(jettonWalletAddress)).toBeTruthy(); + + const jettonWallet = JettonWallet.create(jettonWalletAddress); + const jettonBalance = await tonClient.open(jettonWallet).getBalance(); + expect(jettonBalance).toBeGreaterThan(0n); + const jettonTransferAmount = jettonBalance > 1n ? 1n : jettonBalance; + + const observation = await sendAndObserve({ + addressEvents, + traceStreaming: traceStream, + tonClient, + wallet, + buildMessages: () => [ + internal({ + to: jettonWalletAddress, + value: JETTON_TRANSFER_MESSAGE_VALUE, + bounce: true, + body: createJettonTransferBody({ + amount: jettonTransferAmount, + destination: auxRecipient.address, + responseDestination: wallet.contract.address, + commentText: "streaming wallet e2e jetton transfer", + }), + }), + ], + }); + + expectObservedSend(observation, wallet.contract.address); + + const jettonsEvent = await waitForNextEvent( + "jettons_change event", + addressEvents.jettons, + observation.cursor.jettons, + (event) => sameAddress(event.jetton.address, jettonWalletAddress), + ); + + expect( + sameAddress(jettonsEvent.jetton.owner, wallet.contract.address), + ).toBe(true); + }); +}); + +function normalizeConnectionType(value: string): "sse" | "websocket" { + if (value === "ws" || value === "websocket") { + return "websocket"; + } + + if (value === "sse") { + return "sse"; + } + + return value as "sse" | "websocket"; +} + +// --------------------------------------------------------------------------- +// Wallet helpers +// --------------------------------------------------------------------------- + +function createWalletContext( + version: SupportedWalletVersion, + keyPair: { publicKey: Buffer; secretKey: Buffer }, +): WalletContext { + switch (version) { + case "v3r1": + return { + version, + contract: WalletContractV3R1.create({ + workchain: 0, + publicKey: keyPair.publicKey, + }), + secretKey: keyPair.secretKey, + }; + case "v3r2": + return { + version, + contract: WalletContractV3R2.create({ + workchain: 0, + publicKey: keyPair.publicKey, + }), + secretKey: keyPair.secretKey, + }; + case "v4": + return { + version, + contract: WalletContractV4.create({ + workchain: 0, + publicKey: keyPair.publicKey, + }), + secretKey: keyPair.secretKey, + }; + case "v5": + return { + version, + contract: WalletContractV5R1.create({ + publicKey: keyPair.publicKey, + walletId: { + networkGlobalId: -3, + context: { + workchain: 0, + walletVersion: "v5r1", + subwalletNumber: 0, + }, + }, + }), + secretKey: keyPair.secretKey, + }; + } +} + +async function createTransferBody( + wallet: WalletContext, + tonClient: TonClient, + messages: MessageRelaxed[], +): Promise { + const provider = tonClient.provider( + wallet.contract.address, + wallet.contract.init, + ); + const seqno = await wallet.contract.getSeqno(provider); + const baseArgs = { + seqno, + secretKey: wallet.secretKey, + messages, + sendMode: SendMode.PAY_GAS_SEPARATELY, + }; + + switch (wallet.version) { + case "v3r1": + return (wallet.contract as WalletContractV3R1).createTransfer( + baseArgs, + ) as Cell; + case "v3r2": + return (wallet.contract as WalletContractV3R2).createTransfer( + baseArgs, + ) as Cell; + case "v4": + return (wallet.contract as WalletContractV4).createTransfer( + baseArgs, + ) as Cell; + case "v5": + return (await ( + wallet.contract as WalletContractV5R1 + ).createTransfer(baseArgs)) as Cell; + } +} + +// --------------------------------------------------------------------------- +// Jetton helpers +// --------------------------------------------------------------------------- + +async function discoverTransferableJetton( + tonApiApiKey: string, + ownerAddress: Address, +): Promise<{ + walletAddress: string; + jettonAddress: string; +} | null> { + const response = await axios.get( + `${TONAPI_TESTNET_ACCOUNTS}/${encodeURIComponent(toTestnetFriendly(ownerAddress))}/jettons`, + { + headers: { + Authorization: `Bearer ${tonApiApiKey}`, + }, + timeout: 5_000, + }, + ); + + const parsed = tonApiJettonsSchema.parse(response.data); + const firstNonZero = parsed.balances.find( + (balance) => BigInt(balance.balance) > 0n, + ); + if (!firstNonZero) { + return null; + } + + return { + walletAddress: firstNonZero.wallet_address.address, + jettonAddress: firstNonZero.jetton.address, + }; +} + +function createJettonTransferBody(args: { + amount: bigint; + destination: Address; + responseDestination: Address; + commentText: string; +}): Cell { + return beginCell() + .storeUint(0x0f8a7ea5, 32) + .storeUint(BigInt(Date.now()), 64) + .storeCoins(args.amount) + .storeAddress(args.destination) + .storeAddress(args.responseDestination) + .storeBit(0) + .storeCoins(1n) + .storeBit(1) + .storeRef(comment(args.commentText)) + .endCell(); +} + +// --------------------------------------------------------------------------- +// Streaming: send + observe +// --------------------------------------------------------------------------- + +async function sendAndObserve(args: { + addressEvents: ObservedAddressEvents; + traceStreaming: StreamingClient; + tonClient: TonClient; + wallet: WalletContext; + buildMessages: () => MessageRelaxed[]; +}): Promise { + const cursor = markAddressEvents(args.addressEvents); + + const messages = args.buildMessages(); + const transferBody = await createTransferBody( + args.wallet, + args.tonClient, + messages, + ); + const deployed = await args.tonClient.isContractDeployed( + args.wallet.contract.address, + ); + const externalMessage = external({ + to: args.wallet.contract.address, + init: + !deployed && args.wallet.contract.init + ? args.wallet.contract.init + : undefined, + body: transferBody, + }); + + await args.traceStreaming.subscribe( + traceSubscription(getExternalMessageHashNorm(externalMessage)), + ); + + await args.tonClient.sendMessage(externalMessage); + + const [transactions, actions, accountState] = await Promise.all([ + waitForNextEvent( + "transactions event", + args.addressEvents.transactions, + cursor.transactions, + ), + waitForNextEvent( + "actions event", + args.addressEvents.actions, + cursor.actions, + ), + waitForNextEvent( + "account_state_change event", + args.addressEvents.accountStates, + cursor.accountStates, + ), + ]); + const trace = await waitForNextEvent( + "trace event", + args.addressEvents.traces, + cursor.traces, + ); + + return { cursor, transactions, actions, trace, accountState }; +} + +// --------------------------------------------------------------------------- +// Streaming: subscriptions + event collection +// --------------------------------------------------------------------------- + +function attachEventCollectors( + target: StreamingClient, + source: StreamLogSource, + sink: ObservedAddressEvents, + logWriter: StreamingLogWriter, +): () => void { + const detachers: (() => void)[] = []; + + if (source === "address-stream") { + detachers.push( + target.on("transactions", (event) => { + logWriter.recordEvent(source, event.type, event); + sink.transactions.push(event); + }), + target.on("actions", (event) => { + logWriter.recordEvent(source, event.type, event); + sink.actions.push(event); + }), + target.on("account_state_change", (event) => { + logWriter.recordEvent(source, event.type, event); + sink.accountStates.push(event); + }), + target.on("jettons_change", (event) => { + logWriter.recordEvent(source, event.type, event); + sink.jettons.push(event); + }), + ); + } + + if (source === "trace-stream") { + detachers.push( + target.on("trace", (event) => { + logWriter.recordEvent(source, event.type, event); + sink.traces.push(event); + }), + ); + } + + detachers.push( + target.on("error", (error) => { + logWriter.recordError(source, error); + if (!isExpectedStreamingAbort(error)) { + sink.errors.push(error); + } + }), + ); + + return () => detachers.forEach((d) => d()); +} + +function markAddressEvents(events: ObservedAddressEvents): AddressEventCursor { + return { + transactions: events.transactions.length, + actions: events.actions.length, + traces: events.traces.length, + accountStates: events.accountStates.length, + jettons: events.jettons.length, + }; +} + +function addressSubscription(address: Address) { + return { + addresses: [toTestnetFriendly(address)], + types: ADDRESS_STREAM_TYPES, + minFinality: "confirmed" as const, + includeAddressBook: true, + includeMetadata: true, + }; +} + +function traceSubscription(traceExternalHashNorm: string) { + return { + traceExternalHashNorms: [traceExternalHashNorm], + types: ["trace"] as const, + minFinality: "confirmed" as const, + includeAddressBook: true, + includeMetadata: true, + }; +} + +function getExternalMessageHashNorm(message: Message): string { + if (message.info.type !== "external-in") { + return message.body.hash().toString("hex"); + } + + return beginCell() + .storeUint(2, 2) + .storeUint(0, 2) + .storeAddress(message.info.dest) + .storeUint(0, 4) + .storeBit(false) + .storeBit(true) + .storeRef(message.body) + .endCell() + .hash() + .toString("base64"); +} + +function isExpectedStreamingAbort(error: Error): boolean { + return /aborted/i.test(error.message); +} + +// --------------------------------------------------------------------------- +// Streaming: client lifecycle +// --------------------------------------------------------------------------- + +function createStreamingClient( + connectionType: "sse" | "websocket", + service: "toncenter" | "tonapi", + apiKey: string, +): StreamingClient { + if (connectionType === "websocket") { + return new TonWsClient({ + service, + network: "testnet", + apiKey, + }); + } + + return new TonSseClient({ + service, + network: "testnet", + apiKey, + }); +} + +// --------------------------------------------------------------------------- +// Assertions +// --------------------------------------------------------------------------- + +function expectObservedSend( + observation: SendObservation, + walletAddress: Address, +) { + expect(observation.transactions.transactions.length).toBeGreaterThan(0); + expect(observation.actions.actions.length).toBeGreaterThan(0); + expect(Object.keys(observation.trace.transactions).length).toBeGreaterThan( + 0, + ); + + const traceHash = observation.transactions.trace_external_hash_norm; + expect(traceHash).toBeTruthy(); + expect(traceHash).toBe(observation.actions.trace_external_hash_norm); + expect(traceHash).toBe(observation.trace.trace_external_hash_norm); + expect( + sameAddress(observation.accountState.account, walletAddress), + ).toBeTruthy(); +} + +// --------------------------------------------------------------------------- +// Logging +// --------------------------------------------------------------------------- + +function createStreamingLogWriter(args: { + connectionType: "sse" | "websocket"; + provider: "toncenter" | "tonapi"; + walletAddress: Address; +}): StreamingLogWriter { + const logsDir = path.join( + process.cwd(), + ".tmp", + "streaming-wallet-e2e-logs", + ); + fs.mkdirSync(logsDir, { recursive: true }); + + const filePath = path.join( + logsDir, + `streaming-wallet-e2e-${Date.now()}-${args.connectionType}.json`, + ); + const entries: StreamLogEntry[] = []; + const meta = { + createdAt: new Date().toISOString(), + connectionType: args.connectionType, + provider: args.provider, + walletAddress: toTestnetFriendly(args.walletAddress), + }; + + writeStreamingLine( + `Streaming raw log: ${path.relative(process.cwd(), filePath)}`, + ); + + return { + filePath, + recordEvent(source, type, event) { + const summary = summarizeStreamingEvent(type, event); + entries.push({ + timestamp: new Date().toISOString(), + source, + type, + summary, + payload: event, + }); + writeStreamingLine( + `${formatStreamSourceLabel(source)}: ${summary}`, + ); + }, + recordError(source, error) { + const payload = { + name: error.name, + message: error.message, + stack: error.stack, + }; + const summary = `error name=${error.name} message=${error.message}`; + entries.push({ + timestamp: new Date().toISOString(), + source, + type: "error", + summary, + payload, + }); + writeStreamingLine( + `${formatStreamSourceLabel(source)}: ${summary}`, + ); + }, + flush() { + fs.writeFileSync( + filePath, + JSON.stringify({ meta, entries }, null, 2), + "utf8", + ); + }, + }; +} + +function writeStreamingLine(line: string) { + process.stdout.write(`${line}\n`); +} + +function formatStreamSourceLabel(source: StreamLogSource): string { + return source === "address-stream" ? "Address stream" : "Trace stream"; +} + +function summarizeStreamingEvent( + type: StreamLogEntry["type"], + event: unknown, +): string { + switch (type) { + case "transactions": { + const payload = event as StreamingTransactionsEvent; + return [ + "transactions", + `finality=${payload.finality}`, + `trace=${shortHash(payload.trace_external_hash_norm)}`, + `count=${payload.transactions.length}`, + ].join(" "); + } + case "actions": { + const payload = event as StreamingActionsEvent; + return [ + "actions", + `finality=${payload.finality}`, + `trace=${shortHash(payload.trace_external_hash_norm)}`, + `count=${payload.actions.length}`, + ].join(" "); + } + case "trace": { + const payload = event as StreamingTraceEvent; + return [ + "trace", + `finality=${payload.finality}`, + `trace=${shortHash(payload.trace_external_hash_norm)}`, + `txs=${Object.keys(payload.transactions).length}`, + `actions=${payload.actions?.length ?? 0}`, + ].join(" "); + } + case "account_state_change": { + const payload = event as StreamingAccountStateEvent; + return [ + "account_state_change", + `finality=${payload.finality}`, + `account=${payload.account}`, + `status=${payload.state.account_status}`, + `balance=${payload.state.balance}`, + ].join(" "); + } + case "jettons_change": { + const payload = event as StreamingJettonsEvent; + return [ + "jettons_change", + `finality=${payload.finality}`, + `wallet=${payload.jetton.address}`, + `owner=${payload.jetton.owner}`, + `balance=${payload.jetton.balance}`, + ].join(" "); + } + case "error": + return "error"; + } +} + +// --------------------------------------------------------------------------- +// Generic utilities +// --------------------------------------------------------------------------- + +async function waitForNextEvent( + label: string, + events: readonly T[], + startIndex: number, + match: (event: T) => boolean = () => true, +): Promise { + return waitFor(label, TRACE_TIMEOUT_MS, () => { + for (let i = startIndex; i < events.length; i++) { + if (match(events[i])) return events[i]; + } + return undefined; + }); +} + +async function waitFor( + label: string, + timeoutMs: number, + probe: () => T | undefined | Promise, +): Promise { + const deadline = Date.now() + timeoutMs; + + while (Date.now() < deadline) { + const result = await probe(); + if (result !== undefined) { + return result; + } + await delay(POLL_INTERVAL_MS); + } + + throw new Error(`Timed out waiting for ${label}`); +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function shortHash(value: string): string { + if (value.length <= 16) { + return value; + } + + return `${value.slice(0, 8)}...${value.slice(-6)}`; +} + +function toTestnetFriendly(address: Address): string { + return address.toString({ + bounceable: true, + urlSafe: true, + testOnly: true, + }); +} + +function sameAddress(left: string, right: Address): boolean { + return Address.parse(left).equals(right); +} diff --git a/src/client/streaming/TonSseClient.ts b/src/client/streaming/TonSseClient.ts new file mode 100644 index 00000000..cafdada6 --- /dev/null +++ b/src/client/streaming/TonSseClient.ts @@ -0,0 +1,370 @@ +/** + * 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 { + AbstractStreamingClient, + type DesiredSubscription, +} from "./AbstractStreamingClient"; +import { + StreamingClosedError, + StreamingError, + StreamingHandshakeError, + wrapStreamingError, +} from "./errors"; +import { parseStreamingEvent } from "./protocol"; +import { SseParser } from "./SseParser"; +import { serializeSubscription } from "./subscriptionState"; +import type { StreamingEventMap, StreamingSseParameters } from "./types"; +import { + type Deferred, + buildStreamingUrl, + deferred, + describeHttpError, + describeUnexpectedMessage, + isAbortError, + isRecord, +} from "./utils"; + +type SseSession = { + abort: AbortController; + subscribed: Deferred; + closed: Deferred; + isReady: boolean; +}; + +// SSE subscriptions are immutable for the lifetime of the HTTP stream, +// so each subscribe() replaces the desired snapshot and reconnects. +export class TonSseClient extends AbstractStreamingClient { + readonly #fetchFn: typeof fetch; + readonly #headers: Record; + + #session: SseSession | null = null; + #closingPromise: Promise | null = null; + + constructor(parameters: StreamingSseParameters) { + super("sse", buildStreamingUrl("sse", parameters)); + + this.#fetchFn = + parameters.fetch ?? + (globalThis as { fetch?: typeof fetch }).fetch ?? + (() => { + throw new Error( + "fetch is not available. Pass a fetch function via parameters or use Node 18+.", + ); + }); + this.#headers = { + "Content-Type": "application/json", + Accept: "text/event-stream", + "Cache-Control": "no-cache", + ...(parameters.headers ?? {}), + }; + } + + protected get isSessionReady(): boolean { + return this.#session?.isReady === true; + } + + // SSE can't change subscription mid-stream, so abort any in-flight connection. + protected onSupersede(): void { + if (this.#session && !this.#session.isReady) { + this.#session.abort.abort(); + } + } + + protected applySubscription( + desired: DesiredSubscription, + ): Promise<"ready" | "replaced"> { + return this.#startSession(desired); + } + + protected async closeTransport(error: StreamingError): Promise { + await this.#closeActiveSession(error); + } + + async #startSession( + desired: DesiredSubscription, + ): Promise<"ready" | "replaced"> { + if (this.#session) { + await this.#closeActiveSession( + new StreamingClosedError( + "Streaming subscription is being replaced", + this.ctx("close"), + ), + ); + } else if (this.#closingPromise) { + await this.#closingPromise; + } + + const session: SseSession = { + abort: new AbortController(), + subscribed: deferred(), + closed: deferred(), + isReady: false, + }; + this.#session = session; + + let response; + try { + response = await this.#fetchFn(this.url, { + method: "POST", + headers: this.#headers, + body: JSON.stringify(serializeSubscription(desired.snapshot)), + signal: session.abort.signal, + }); + } catch (error) { + this.#teardown(session); + if (isAbortError(error)) { + if (this.isSuperseded(desired)) { + return "replaced"; + } + throw new StreamingClosedError( + "Streaming transport is closing", + this.ctx("close"), + { cause: error }, + ); + } + throw wrapStreamingError( + error, + this.ctx("connect"), + "Streaming SSE connection failed", + ); + } + + if (!response.ok) { + this.#teardown(session); + throw new StreamingError( + `Streaming SSE connection failed: ${await describeHttpError(response)}`, + this.ctx("connect", { rawPayload: response }), + ); + } + + if (!response.body || typeof response.body.getReader !== "function") { + this.#teardown(session); + throw new StreamingError( + "SSE response does not expose a readable body", + this.ctx("connect", { rawPayload: response }), + ); + } + const body = response.body; + + if (this.#session !== session || this.isSuperseded(desired)) { + void body.cancel(); + this.#teardown(session); + return "replaced"; + } + + void this.#readStream(body, session); + + try { + await session.subscribed.promise; + } catch (error) { + if (this.isSuperseded(desired)) { + return "replaced"; + } + throw error; + } + + return "ready"; + } + + #closeActiveSession(error: StreamingError): Promise { + const session = this.#session; + if (!session) { + return this.#closingPromise ?? Promise.resolve(); + } + + const wasReady = session.isReady; + this.#session = null; + session.isReady = false; + session.abort.abort(); + session.subscribed.reject(error); + + if (wasReady) { + this.emit("close", undefined); + } + + const closingPromise = session.closed.promise.finally(() => { + if (this.#closingPromise === closingPromise) { + this.#closingPromise = null; + } + }); + this.#closingPromise = closingPromise; + return closingPromise; + } + + #teardown(session: SseSession): void { + if (this.#session === session) { + this.#session = null; + } + session.closed.resolve(); + } + + #rejectOrEmitError(session: SseSession, error: StreamingError): void { + if (!session.subscribed.settled) { + session.subscribed.reject(error); + session.abort.abort(); + } else { + this.emit("error", error); + } + } + + async #readStream( + body: ReadableStream, + session: SseSession, + ): Promise { + const parser = new SseParser((sseEvent) => { + let payload: unknown; + try { + payload = JSON.parse(sseEvent.data) as unknown; + } catch (error) { + this.#rejectOrEmitError( + session, + wrapStreamingError( + error, + this.ctx("message", { rawPayload: sseEvent.data }), + "Failed to parse streaming SSE event", + ), + ); + return; + } + + if (!isRecord(payload)) { + this.#rejectOrEmitError( + session, + new StreamingError( + `Unexpected streaming SSE payload: ${describeUnexpectedMessage(payload)}`, + this.ctx("message", { rawPayload: payload }), + ), + ); + return; + } + + if (this.#session !== session) { + return; + } + + if (typeof payload.status === "string") { + if (payload.status === "subscribed") { + if (!session.isReady) { + session.isReady = true; + this.emit("open", undefined); + } + session.subscribed.resolve(); + return; + } + + this.#rejectOrEmitError( + session, + payload.error !== undefined + ? wrapStreamingError( + payload.error, + this.ctx("subscription_confirmation", { + rawPayload: payload, + }), + `Streaming SSE request failed with status ${payload.status}`, + ) + : new StreamingError( + `Unexpected streaming SSE status message: ${describeUnexpectedMessage(payload)}`, + this.ctx("subscription_confirmation", { + rawPayload: payload, + }), + ), + ); + return; + } + + try { + const event = parseStreamingEvent(payload); + this.emit( + event.type as keyof StreamingEventMap, + event as never, + ); + } catch (error) { + this.#rejectOrEmitError( + session, + wrapStreamingError( + error, + this.ctx("notification", { rawPayload: payload }), + "Invalid streaming SSE notification", + ), + ); + } + }); + + const reader = body.getReader(); + let endedNormally = false; + + const decoder = new TextDecoder(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + parser.feed(decoder.decode(value, { stream: true })); + } + + const tail = decoder.decode(); + if (tail) { + parser.feed(tail); + } + parser.finish(); + endedNormally = true; + } catch (error) { + if (!session.abort.signal.aborted && !isAbortError(error)) { + this.#rejectOrEmitError( + session, + wrapStreamingError( + error, + this.ctx("stream"), + "Streaming SSE stream terminated unexpectedly", + ), + ); + } + } finally { + reader.releaseLock(); + + if (this.#session !== session) { + this.#teardown(session); + return; + } + + const wasReady = session.isReady; + this.#session = null; + session.isReady = false; + + const preReadyError = session.abort.signal.aborted + ? new StreamingClosedError( + "Streaming transport is closing", + this.ctx("close"), + ) + : new StreamingHandshakeError( + "Streaming SSE connection closed before subscription confirmation", + this.ctx("subscription_confirmation"), + ); + + session.subscribed.reject(preReadyError); + + if (endedNormally && wasReady) { + this.emit( + "error", + new StreamingClosedError( + "Streaming SSE stream closed by server", + this.ctx("stream"), + ), + ); + } + + if (wasReady) { + this.emit("close", undefined); + } + + this.#teardown(session); + } + } +} diff --git a/src/client/streaming/TonWsClient.ts b/src/client/streaming/TonWsClient.ts new file mode 100644 index 00000000..9daa1bfc --- /dev/null +++ b/src/client/streaming/TonWsClient.ts @@ -0,0 +1,547 @@ +/** + * 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 { + AbstractStreamingClient, + type DesiredSubscription, +} from "./AbstractStreamingClient"; +import { + StreamingClosedError, + StreamingError, + StreamingHandshakeError, + StreamingRequestTimeoutError, + wrapStreamingError, +} from "./errors"; +import { parseStreamingEvent } from "./protocol"; +import type { ResolvedStreamingSubscription } from "./subscriptionState"; +import { serializeSubscription } from "./subscriptionState"; +import type { + IWebSocket, + IWebSocketConstructor, + StreamingEventMap, + StreamingWebSocketParameters, +} from "./types"; +import { + type Deferred, + DEFAULT_PING_INTERVAL_MS, + DEFAULT_REQUEST_TIMEOUT_MS, + buildStreamingUrl, + deferred, + describeUnexpectedMessage, + isRecord, + normalizeTimeoutMs, +} from "./utils"; + +type PendingRequest = { + resolve: (value: StreamingResponse) => void; + reject: (reason: Error) => void; + timeout: ReturnType; +}; + +type StreamingResponse = { + id?: string | number; + status?: string; + error?: unknown; + [key: string]: unknown; +}; + +type WsSession = { + socket: IWebSocket; + state: "connecting" | "open" | "closing"; + connected: Deferred; + closed: Deferred; + isReady: boolean; + pingInterval: ReturnType | null; + pendingRequests: Map; + requestId: number; +}; + +const textDecoder = new TextDecoder(); + +function parseWebSocketMessage(rawMessage: unknown): unknown { + if (typeof rawMessage === "string") { + return JSON.parse(rawMessage) as unknown; + } + + if (rawMessage instanceof ArrayBuffer || ArrayBuffer.isView(rawMessage)) { + return JSON.parse(textDecoder.decode(rawMessage)) as unknown; + } + + throw new Error(`Unexpected WebSocket message type: ${typeof rawMessage}`); +} + +export class TonWsClient extends AbstractStreamingClient { + readonly #wsCtor: IWebSocketConstructor; + readonly #wsHeaders: Record; + readonly #hasCustomHeaders: boolean; + readonly #requestTimeoutMs: number; + readonly #pingIntervalMs: number; + + #session: WsSession | null = null; + + constructor(parameters: StreamingWebSocketParameters) { + super("ws", buildStreamingUrl("ws", parameters)); + + this.#wsHeaders = { ...(parameters.headers ?? {}) }; + this.#hasCustomHeaders = Object.keys(this.#wsHeaders).length > 0; + + const wsCtor = + parameters.WebSocket ?? + (globalThis as { WebSocket?: IWebSocketConstructor }).WebSocket; + if (!wsCtor) { + throw new Error( + "WebSocket is not available. Pass a WebSocket constructor via parameters.", + ); + } + this.#wsCtor = wsCtor; + + if (this.#hasCustomHeaders && parameters.WebSocket === undefined) { + throw new Error( + "Custom headers require a custom WebSocket constructor. " + + "Browser WebSocket does not support arbitrary headers.", + ); + } + + this.#requestTimeoutMs = normalizeTimeoutMs( + parameters.requestTimeoutMs, + DEFAULT_REQUEST_TIMEOUT_MS, + "parameters.requestTimeoutMs", + ); + this.#pingIntervalMs = normalizeTimeoutMs( + parameters.pingIntervalMs, + DEFAULT_PING_INTERVAL_MS, + "parameters.pingIntervalMs", + ); + } + + protected get isSessionReady(): boolean { + return this.#session?.isReady === true; + } + + protected async applySubscription( + desired: DesiredSubscription, + ): Promise<"ready" | "replaced"> { + await this.#ensureSocketOpen(); + await this.#sendSubscribe(desired.snapshot); + return "ready"; + } + + protected async closeTransport(error: StreamingError): Promise { + await this.#closeActiveSession(error); + } + + #closeActiveSession(error: StreamingError): Promise { + const session = this.#session; + if (!session || session.state === "closing") { + return session?.closed.promise ?? Promise.resolve(); + } + + const wasConnecting = session.state === "connecting"; + const wasSubscribed = session.isReady; + + session.state = "closing"; + session.isReady = false; + this.#stopPing(session); + this.#rejectAllPending(session, error); + + if (wasConnecting) { + session.connected.reject( + new StreamingClosedError( + "WebSocket connection was closed", + this.ctx("connect"), + ), + ); + } + + if (wasSubscribed) { + this.emit("close", undefined); + } + + try { + session.socket.close(); + } catch { + // socket.close() threw — cleanup below handles it + } + + this.#cleanupSession(session); + return session.closed.promise; + } + + async #ensureSocketOpen(): Promise { + if (this.#session?.state === "open") { + return; + } + if (this.#session?.state === "closing") { + await this.#session.closed.promise; + } + if (this.#session?.state === "connecting") { + return this.#session.connected.promise; + } + + const ws = this.#hasCustomHeaders + ? new this.#wsCtor(this.url, { headers: this.#wsHeaders }) + : new this.#wsCtor(this.url); + + const session: WsSession = { + socket: ws, + state: "connecting", + connected: deferred(), + closed: deferred(), + isReady: false, + pingInterval: null, + pendingRequests: new Map(), + requestId: 0, + }; + this.#session = session; + + ws.onopen = () => { + if (this.#session !== session || session.state !== "connecting") { + return; + } + session.state = "open"; + session.connected.resolve(); + }; + + ws.onerror = () => { + if (this.#session !== session) { + return; + } + + const error = new StreamingError( + "WebSocket connection error", + this.ctx("connect"), + ); + + if (session.state === "connecting") { + this.#cleanupSession(session); + session.connected.reject(error); + try { + ws.close(); + } catch {} + return; + } + + if (session.isReady) { + this.emit("error", error); + } + }; + + ws.onmessage = (event) => { + if (this.#session !== session) { + return; + } + + try { + this.#handleMessage(session, event.data); + } catch (error) { + this.emit( + "error", + wrapStreamingError( + error, + this.ctx("message", { rawPayload: event.data }), + "Failed to handle streaming message", + ), + ); + } + }; + + ws.onclose = () => { + if (this.#session !== session) { + return; + } + + const wasConnecting = session.state === "connecting"; + const wasSubscribed = session.isReady; + + this.#cleanupSession(session); + + if (wasConnecting) { + session.connected.reject( + new StreamingHandshakeError( + "WebSocket connection closed before opening", + this.ctx("connect"), + ), + ); + return; + } + + if (wasSubscribed) { + this.emit( + "error", + new StreamingClosedError( + "Streaming WebSocket stream closed by server", + this.ctx("stream"), + ), + ); + this.emit("close", undefined); + } + }; + + return session.connected.promise; + } + + async #sendSubscribe( + resolved: ResolvedStreamingSubscription, + ): Promise { + const session = this.#session; + if (!session || session.state !== "open") { + throw new StreamingError( + "WebSocket is not connected", + this.ctx("send"), + ); + } + + const id = this.#nextRequestId(session); + const response = await this.#sendRequest(session, id, { + operation: "subscribe", + id, + ...serializeSubscription(resolved), + }); + + if (response.status !== "subscribed") { + throw new StreamingError( + `Subscription failed: ${describeUnexpectedMessage(response)}`, + this.ctx("subscription_confirmation", { + requestId: id, + rawPayload: response, + }), + ); + } + + if (!session.isReady) { + session.isReady = true; + this.#startPing(session); + this.emit("open", undefined); + } + } + + #nextRequestId(session: WsSession): string { + session.requestId += 1; + return String(session.requestId); + } + + async #sendRequest( + session: WsSession, + id: string, + message: Record, + ): Promise { + if ( + session.state !== "open" || + session.socket.readyState !== this.#wsCtor.OPEN + ) { + throw new StreamingError( + "WebSocket is not connected", + this.ctx("send", { requestId: id, rawPayload: message }), + ); + } + + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + if (!session.pendingRequests.has(id)) { + return; + } + session.pendingRequests.delete(id); + reject( + new StreamingRequestTimeoutError( + `Streaming request ${id} timed out`, + this.ctx("request", { requestId: id }), + ), + ); + }, this.#requestTimeoutMs); + + session.pendingRequests.set(id, { + resolve: (value) => { + clearTimeout(timeout); + resolve(value); + }, + reject: (reason) => { + clearTimeout(timeout); + reject(reason); + }, + timeout, + }); + + try { + session.socket.send(JSON.stringify(message)); + } catch (error) { + session.pendingRequests.delete(id); + clearTimeout(timeout); + reject( + wrapStreamingError( + error, + this.ctx("send", { + requestId: id, + rawPayload: message, + }), + "Failed to send WebSocket message", + ), + ); + } + }); + } + + #rejectAllPending(session: WsSession, error: Error): void { + for (const pending of session.pendingRequests.values()) { + clearTimeout(pending.timeout); + pending.reject(error); + } + session.pendingRequests.clear(); + } + + #handleMessage(session: WsSession, rawMessage: unknown): void { + const data = parseWebSocketMessage(rawMessage); + + if (this.#handleResponse(session, data)) { + return; + } + + if (this.#handleNotification(data)) { + return; + } + + this.emit( + "error", + new StreamingError( + `Unexpected streaming message: ${describeUnexpectedMessage(data)}`, + this.ctx("message", { rawPayload: data }), + ), + ); + } + + #handleResponse( + session: WsSession, + payload: unknown, + ): payload is StreamingResponse { + if (!isRecord(payload) || payload.id === undefined) { + return false; + } + + const requestId = String(payload.id); + const pending = session.pendingRequests.get(requestId); + if (!pending) { + return false; + } + + session.pendingRequests.delete(requestId); + if (payload.error !== undefined) { + pending.reject( + wrapStreamingError( + payload.error, + this.ctx("request", { requestId, rawPayload: payload }), + `Streaming request ${requestId} failed`, + ), + ); + } else { + pending.resolve(payload); + } + return true; + } + + #handleNotification(payload: unknown): boolean { + if (!isRecord(payload) || typeof payload.type !== "string") { + return false; + } + + try { + const event = parseStreamingEvent(payload); + this.emit(event.type as keyof StreamingEventMap, event as never); + } catch (error) { + this.emit( + "error", + wrapStreamingError( + error, + this.ctx("notification", { rawPayload: payload }), + "Invalid streaming notification", + ), + ); + } + + return true; + } + + #startPing(session: WsSession): void { + if (this.#pingIntervalMs === 0) { + return; + } + + this.#stopPing(session); + session.pingInterval = setInterval(() => { + if (this.#session !== session) { + return; + } + + const id = this.#nextRequestId(session); + void this.#sendRequest(session, id, { + operation: "ping", + id, + }).catch((error) => { + this.#handleHeartbeatFailure(session, id, error); + }); + }, this.#pingIntervalMs); + } + + #handleHeartbeatFailure( + session: WsSession, + requestId: string, + reason: unknown, + ): void { + if (this.#session !== session || session.state !== "open") { + return; + } + + const cause = reason instanceof Error ? reason : undefined; + const message = cause?.message ?? "WebSocket heartbeat failed"; + const error = new StreamingError( + message, + this.ctx("heartbeat", { requestId }), + { cause }, + ); + const wasSubscribed = session.isReady; + + this.emit("error", error); + this.#cleanupSession(session); + if (wasSubscribed) { + this.emit("close", undefined); + } + + try { + session.socket.close(); + } catch {} + } + + #stopPing(session: WsSession): void { + if (!session.pingInterval) { + return; + } + clearInterval(session.pingInterval); + session.pingInterval = null; + } + + #cleanupSession(session: WsSession): void { + if (this.#session === session) { + this.#session = null; + } + this.#stopPing(session); + this.#rejectAllPending( + session, + new StreamingClosedError( + "Connection closed", + this.ctx("transport"), + ), + ); + session.socket.onopen = null; + session.socket.onclose = null; + session.socket.onmessage = null; + session.socket.onerror = null; + try { + session.socket.terminate?.(); + } catch {} + session.isReady = false; + session.closed.resolve(); + } +} diff --git a/src/client/streaming/TypedEventEmitter.ts b/src/client/streaming/TypedEventEmitter.ts new file mode 100644 index 00000000..ef13cba3 --- /dev/null +++ b/src/client/streaming/TypedEventEmitter.ts @@ -0,0 +1,70 @@ +/** + * 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 class TypedEventEmitter> { + #listeners = new Map< + keyof TEvents, + Set<(data: TEvents[keyof TEvents]) => void> + >(); + + on( + event: K, + handler: (data: TEvents[K]) => void, + ): () => void { + const handlers = this.#getOrCreateHandlers(event); + handlers.add(handler as (data: TEvents[keyof TEvents]) => void); + return () => this.off(event, handler); + } + + off( + event: K, + handler: (data: TEvents[K]) => void, + ): void { + const handlers = this.#listeners.get(event); + if (!handlers) { + return; + } + + handlers.delete(handler as (data: TEvents[keyof TEvents]) => void); + if (handlers.size === 0) { + this.#listeners.delete(event); + } + } + + protected emit(event: K, data: TEvents[K]): void { + const handlers = this.#listeners.get(event); + if (!handlers || handlers.size === 0) { + return; + } + + for (const handler of [...handlers]) { + try { + handler(data as TEvents[keyof TEvents]); + } catch (error) { + queueMicrotask(() => { + throw error; + }); + } + } + } + + protected removeAllListeners(): void { + this.#listeners.clear(); + } + + #getOrCreateHandlers( + event: K, + ): Set<(data: TEvents[keyof TEvents]) => void> { + let handlers = this.#listeners.get(event); + if (!handlers) { + handlers = new Set(); + this.#listeners.set(event, handlers); + } + return handlers; + } +} diff --git a/src/client/streaming/errors.ts b/src/client/streaming/errors.ts new file mode 100644 index 00000000..40c8ddab --- /dev/null +++ b/src/client/streaming/errors.ts @@ -0,0 +1,75 @@ +/** + * 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 { isRecord } from "./utils"; + +export type StreamingTransport = "sse" | "ws"; + +export type StreamingErrorContext = { + transport: StreamingTransport; + endpoint?: string; + phase?: string; + requestId?: string; + rawPayload?: unknown; +}; + +export class StreamingError extends Error { + readonly context: Readonly; + declare readonly cause: unknown; + + constructor( + message: string, + context: StreamingErrorContext, + options?: { cause?: unknown }, + ) { + super(message); + this.name = new.target.name; + this.context = Object.freeze({ ...context }); + + if (options?.cause !== undefined) { + Object.defineProperty(this, "cause", { + configurable: true, + enumerable: false, + value: options.cause, + writable: false, + }); + } + } +} + +export class StreamingRequestTimeoutError extends StreamingError {} + +export class StreamingClosedError extends StreamingError {} + +export class StreamingHandshakeError extends StreamingError {} + +export class StreamingSupersededError extends StreamingError {} + +export function wrapStreamingError( + reason: unknown, + context: StreamingErrorContext, + fallback?: string, +): StreamingError { + if (reason instanceof StreamingError) { + return reason; + } + + let message = fallback ?? String(reason); + if (reason instanceof Error) { + message = reason.message; + } else if (typeof reason === "string" && reason) { + message = reason; + } else if (isRecord(reason)) { + const msg = reason.message ?? reason.error; + if (typeof msg === "string" && msg) { + message = msg; + } + } + + return new StreamingError(message, context, { cause: reason }); +} diff --git a/src/client/streaming/index.ts b/src/client/streaming/index.ts new file mode 100644 index 00000000..160078ab --- /dev/null +++ b/src/client/streaming/index.ts @@ -0,0 +1,54 @@ +/** + * 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 { + StreamingClosedError, + StreamingError, + StreamingHandshakeError, + StreamingRequestTimeoutError, + StreamingSupersededError, +} from "./errors"; +export { TonSseClient } from "./TonSseClient"; +export { TonWsClient } from "./TonWsClient"; + +export type { + StreamingErrorContext, + StreamingTransport, +} from "./errors"; +export type { + Finality, + NonPendingFinality, + SubscribableEventType, + StreamingService, + StreamingNetwork, + StreamingAction, + StreamingAccountStateEvent, + StreamingAddressBookEntry, + StreamingActionsEvent, + StreamingBaseParameters, + StreamingBlockRef, + StreamingClient, + StreamingDecodedMessage, + StreamingEvent, + StreamingEventMap, + StreamingJettonsEvent, + StreamingLifecycleEvents, + StreamingMessage, + StreamingMessageContent, + StreamingMetadataEntry, + StreamingSseParameters, + StreamingSubscription, + StreamingTrace, + StreamingTraceEvent, + StreamingTraceInvalidatedEvent, + StreamingTraceNode, + StreamingTransaction, + StreamingTransactionAccountState, + StreamingTransactionsEvent, + StreamingWebSocketParameters, +} from "./types"; diff --git a/src/client/streaming/protocol.ts b/src/client/streaming/protocol.ts new file mode 100644 index 00000000..6c7470b8 --- /dev/null +++ b/src/client/streaming/protocol.ts @@ -0,0 +1,265 @@ +/** + * 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 type { + JsonObject, + StreamingAccountStateEvent, + StreamingAction, + StreamingActionsEvent, + StreamingAddressBookEntry, + StreamingEvent, + StreamingJettonsEvent, + StreamingMetadataEntry, + StreamingTrace, + StreamingTraceEvent, + StreamingTraceInvalidatedEvent, + StreamingTransaction, + StreamingTransactionsEvent, +} from "./types"; +import type { Finality, NonPendingFinality } from "./utils"; +import { FINALITY_SET, NON_PENDING_FINALITY_SET, isRecord } from "./utils"; + +function expectRecord( + value: unknown, + fieldName: string, +): Record { + if (!isRecord(value)) { + throw new Error(`${fieldName} must be an object`); + } + return value; +} + +function expectString(value: unknown, fieldName: string): string { + if (typeof value !== "string") { + throw new Error(`${fieldName} must be a string`); + } + return value; +} + +function expectFinality(value: unknown, fieldName: string): Finality { + const finality = expectString(value, fieldName); + if (!FINALITY_SET.has(finality)) { + throw new Error(`${fieldName} has unsupported value: ${finality}`); + } + return finality as Finality; +} + +function expectNonPendingFinality( + value: unknown, + fieldName: string, +): NonPendingFinality { + const finality = expectString(value, fieldName); + if (!NON_PENDING_FINALITY_SET.has(finality)) { + throw new Error(`${fieldName} has unsupported value: ${finality}`); + } + return finality as NonPendingFinality; +} + +function expectArray(value: unknown, fieldName: string): unknown[] { + if (!Array.isArray(value)) { + throw new Error(`${fieldName} must be an array`); + } + return value; +} + +function expectJsonObject(value: unknown, fieldName: string): JsonObject { + return expectRecord(value, fieldName) as JsonObject; +} + +function castJsonObjectArray( + value: unknown, + fieldName: string, +): T[] { + const arr = expectArray(value, fieldName); + for (let i = 0; i < arr.length; i++) { + if (!isRecord(arr[i])) { + throw new Error(`${fieldName}[${i}] must be an object`); + } + } + return arr as T[]; +} + +function castJsonObjectRecord( + value: unknown, + fieldName: string, +): Record { + const record = expectRecord(value, fieldName); + for (const [key, entry] of Object.entries(record)) { + if (!isRecord(entry)) { + throw new Error(`${fieldName}.${key} must be an object`); + } + } + return record as Record; +} + +function castOptionalJsonObjectRecord( + value: unknown, + fieldName: string, +): Record | undefined { + return value === undefined + ? undefined + : castJsonObjectRecord(value, fieldName); +} + +function parseAddressBook( + value: unknown, + fieldName: string, +): Record | undefined { + return castOptionalJsonObjectRecord( + value, + fieldName, + ); +} + +function parseMetadata( + value: unknown, + fieldName: string, +): Record | undefined { + return castOptionalJsonObjectRecord( + value, + fieldName, + ); +} + +function parseTraceCommonFields( + payload: Record, + prefix: string, +) { + return { + finality: expectFinality(payload.finality, `${prefix}.finality`), + trace_external_hash_norm: expectString( + payload.trace_external_hash_norm, + `${prefix}.trace_external_hash_norm`, + ), + address_book: parseAddressBook( + payload.address_book, + `${prefix}.address_book`, + ), + metadata: parseMetadata(payload.metadata, `${prefix}.metadata`), + }; +} + +function parseTransactionsEvent( + payload: Record, +): StreamingTransactionsEvent { + return { + type: "transactions", + ...parseTraceCommonFields(payload, "transactions"), + transactions: castJsonObjectArray( + payload.transactions, + "transactions.transactions", + ), + }; +} + +function parseActionsEvent( + payload: Record, +): StreamingActionsEvent { + return { + type: "actions", + ...parseTraceCommonFields(payload, "actions"), + actions: castJsonObjectArray( + payload.actions, + "actions.actions", + ), + }; +} + +function parseTraceEvent( + payload: Record, +): StreamingTraceEvent { + return { + type: "trace", + ...parseTraceCommonFields(payload, "trace"), + trace: expectJsonObject(payload.trace, "trace.trace") as StreamingTrace, + transactions: castJsonObjectRecord( + payload.transactions, + "trace.transactions", + ), + actions: + payload.actions === undefined + ? undefined + : castJsonObjectArray( + payload.actions, + "trace.actions", + ), + }; +} + +function parseAccountStateEvent( + payload: Record, +): StreamingAccountStateEvent { + return { + type: "account_state_change", + finality: expectNonPendingFinality( + payload.finality, + "account_state_change.finality", + ), + account: expectString(payload.account, "account_state_change.account"), + state: expectJsonObject( + payload.state, + "account_state_change.state", + ) as StreamingAccountStateEvent["state"], + }; +} + +function parseJettonsEvent( + payload: Record, +): StreamingJettonsEvent { + return { + type: "jettons_change", + finality: expectNonPendingFinality( + payload.finality, + "jettons_change.finality", + ), + jetton: expectJsonObject( + payload.jetton, + "jettons_change.jetton", + ) as StreamingJettonsEvent["jetton"], + address_book: parseAddressBook( + payload.address_book, + "jettons_change.address_book", + ), + metadata: parseMetadata(payload.metadata, "jettons_change.metadata"), + }; +} + +function parseTraceInvalidatedEvent( + payload: Record, +): StreamingTraceInvalidatedEvent { + return { + type: "trace_invalidated", + trace_external_hash_norm: expectString( + payload.trace_external_hash_norm, + "trace_invalidated.trace_external_hash_norm", + ), + }; +} + +export function parseStreamingEvent( + payload: Record, +): StreamingEvent { + const type = expectString(payload.type, "streaming message.type"); + + switch (type) { + case "transactions": + return parseTransactionsEvent(payload); + case "actions": + return parseActionsEvent(payload); + case "trace": + return parseTraceEvent(payload); + case "account_state_change": + return parseAccountStateEvent(payload); + case "jettons_change": + return parseJettonsEvent(payload); + case "trace_invalidated": + return parseTraceInvalidatedEvent(payload); + default: + throw new Error(`Unexpected streaming event type: ${type}`); + } +} diff --git a/src/client/streaming/subscriptionState.ts b/src/client/streaming/subscriptionState.ts new file mode 100644 index 00000000..2dd69a6c --- /dev/null +++ b/src/client/streaming/subscriptionState.ts @@ -0,0 +1,164 @@ +/** + * 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 type { StreamingSubscription } from "./types"; +import type { Finality, SubscribableEventType } from "./utils"; +import { + FINALITY_SET, + SUBSCRIBABLE_EVENT_TYPE_SET, + requireStringList, + sanitizeStringList, +} from "./utils"; + +export type ResolvedStreamingSubscription = { + addresses: string[]; + traceExternalHashNorms: string[]; + types: SubscribableEventType[]; + minFinality: Finality; + includeAddressBook: boolean; + includeMetadata: boolean; + actionTypes: string[]; + supportedActionTypes: string[]; +}; + +export function resolveStreamingSubscription( + params: StreamingSubscription, +): ResolvedStreamingSubscription { + const types = requireStringList(params.types, "params.types").map( + (type) => { + if (!SUBSCRIBABLE_EVENT_TYPE_SET.has(type)) { + throw new Error(`Unsupported streaming event type: ${type}`); + } + return type as SubscribableEventType; + }, + ); + types.sort(); + + const addresses = sanitizeStringList(params.addresses) ?? []; + addresses.sort(); + + const traceExternalHashNorms = + sanitizeStringList(params.traceExternalHashNorms) ?? []; + traceExternalHashNorms.sort(); + + const actionTypes = sanitizeStringList(params.actionTypes) ?? []; + actionTypes.sort(); + + const supportedActionTypes = + sanitizeStringList(params.supportedActionTypes) ?? []; + supportedActionTypes.sort(); + + const minFinality: Finality = params.minFinality ?? "finalized"; + if (!FINALITY_SET.has(minFinality)) { + throw new Error(`Unsupported finality level: ${minFinality}`); + } + + const includeAddressBook = params.includeAddressBook ?? false; + const includeMetadata = params.includeMetadata ?? false; + + const hasTraceSubscription = types.includes("trace"); + const hasAddressBoundSubscription = types.some((type) => type !== "trace"); + + if (hasTraceSubscription && traceExternalHashNorms.length === 0) { + throw new Error( + 'traceExternalHashNorms are required when subscribing to "trace" events', + ); + } + + if (hasAddressBoundSubscription && addresses.length === 0) { + throw new Error( + "addresses are required when subscribing to non-trace streaming events", + ); + } + + if (actionTypes.length > 0 && !types.includes("actions")) { + throw new Error( + 'actionTypes can only be used with the "actions" event type', + ); + } + + if ( + supportedActionTypes.length > 0 && + !types.some((type) => type === "actions" || type === "trace") + ) { + throw new Error( + 'supportedActionTypes can only be used with the "actions" or "trace" event types', + ); + } + + return { + addresses, + traceExternalHashNorms, + types, + minFinality, + includeAddressBook, + includeMetadata, + actionTypes, + supportedActionTypes, + }; +} + +export function sameSubscription( + a: ResolvedStreamingSubscription, + b: ResolvedStreamingSubscription, +): boolean { + return ( + a.minFinality === b.minFinality && + a.includeAddressBook === b.includeAddressBook && + a.includeMetadata === b.includeMetadata && + arraysEqual(a.types, b.types) && + arraysEqual(a.addresses, b.addresses) && + arraysEqual(a.traceExternalHashNorms, b.traceExternalHashNorms) && + arraysEqual(a.actionTypes, b.actionTypes) && + arraysEqual(a.supportedActionTypes, b.supportedActionTypes) + ); +} + +function arraysEqual(a: readonly string[], b: readonly string[]): boolean { + if (a.length !== b.length) { + return false; + } + for (let i = 0; i < a.length; i += 1) { + if (a[i] !== b[i]) { + return false; + } + } + return true; +} + +export function serializeSubscription( + resolved: ResolvedStreamingSubscription, +): Record { + const result: Record = { + types: resolved.types, + }; + + if (resolved.addresses.length > 0) { + result.addresses = resolved.addresses; + } + if (resolved.traceExternalHashNorms.length > 0) { + result.trace_external_hash_norms = resolved.traceExternalHashNorms; + } + if (resolved.minFinality !== "finalized") { + result.min_finality = resolved.minFinality; + } + if (resolved.includeAddressBook) { + result.include_address_book = true; + } + if (resolved.includeMetadata) { + result.include_metadata = true; + } + if (resolved.actionTypes.length > 0) { + result.action_types = resolved.actionTypes; + } + if (resolved.supportedActionTypes.length > 0) { + result.supported_action_types = resolved.supportedActionTypes; + } + + return result; +} diff --git a/src/client/streaming/types.ts b/src/client/streaming/types.ts new file mode 100644 index 00000000..3b0ae1fc --- /dev/null +++ b/src/client/streaming/types.ts @@ -0,0 +1,308 @@ +/** + * 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 type { StreamingError } from "./errors"; +import type { + Finality, + NonPendingFinality, + SubscribableEventType, + StreamingService, + StreamingNetwork, +} from "./utils"; + +export type { + Finality, + NonPendingFinality, + SubscribableEventType, + StreamingService, + StreamingNetwork, +} from "./utils"; + +export type JsonPrimitive = string | number | boolean | null; +export type JsonValue = JsonPrimitive | JsonObject | JsonValue[]; +export type JsonObject = { + [key: string]: JsonValue; +}; + +export interface IWebSocket { + readonly readyState: number; + send(data: string): void; + close(): void; + /** Immediately destroy the underlying socket (available in Node.js `ws` and native WebSocket). */ + terminate?(): void; + onopen: ((event: unknown) => void) | null; + onclose: ((event: unknown) => void) | null; + onmessage: ((event: { data: unknown }) => void) | null; + onerror: ((event: unknown) => void) | null; +} + +export interface IWebSocketConstructor { + new (url: string, options?: unknown): IWebSocket; + readonly OPEN: number; +} + +export type StreamingLifecycleEvents = { + error: StreamingError; + close: undefined; + open: undefined; +}; + +export type StreamingBaseParameters = { + /** When set, endpoint and API key query parameter are inferred. Cannot be combined with `endpoint`. */ + service?: StreamingService; + /** @default "mainnet" */ + network?: StreamingNetwork; + /** Required when `service` is omitted. Cannot be combined with `service`. */ + endpoint?: string; + apiKey?: string; + /** @default "api_key" */ + apiKeyParam?: string; +}; + +export type StreamingWebSocketParameters = StreamingBaseParameters & { + /** Custom WebSocket constructor (for Node.js < 22 use the `ws` package). */ + WebSocket?: IWebSocketConstructor; + /** Requires a custom `WebSocket` constructor; browser WebSocket does not support arbitrary headers. */ + headers?: Record; + /** @default 5000 */ + requestTimeoutMs?: number; + /** Set to 0 to disable. @default 15000 */ + pingIntervalMs?: number; +}; + +export type StreamingSubscription = { + /** Wallet or contract addresses to monitor. */ + addresses?: readonly string[]; + /** Trace external hashes to monitor (required for the "trace" event type). */ + traceExternalHashNorms?: readonly string[]; + /** Event types to receive. */ + types: readonly SubscribableEventType[]; + /** Minimum finality level. Default: "finalized". */ + minFinality?: Finality; + /** Include DNS-resolved and friendly names for addresses. */ + includeAddressBook?: boolean; + /** Include metadata for known token contracts. */ + includeMetadata?: boolean; + /** Filter actions by type (for example ["jetton_transfer", "ton_transfer"]). */ + actionTypes?: readonly string[]; + /** Advertise which action classification types the client understands. */ + supportedActionTypes?: readonly string[]; +}; + +export type StreamingAddressBookEntry = JsonObject & { + user_friendly?: string; + domain?: string | null; + interfaces?: string[]; +}; + +export type StreamingMetadataEntry = JsonObject; + +export type StreamingDecodedMessage = JsonObject & { + "@type": string; +}; + +export type StreamingMessageContent = JsonObject & { + hash: string; + body: string; + decoded: StreamingDecodedMessage | null; +}; + +// Many fields are null for external messages (source, value, fwd_fee, etc.) +export type StreamingMessage = JsonObject & { + hash: string; + source: string | null; + destination: string | null; + value: string | null; + value_extra_currencies: JsonObject | null; + fwd_fee: string | null; + ihr_fee: string | null; + extra_flags: string | null; + created_lt: string | null; + created_at: string | null; + opcode: string | null; + decoded_opcode: string | null; + ihr_disabled: boolean | null; + bounce: boolean | null; + bounced: boolean | null; + import_fee: string | null; + message_content: StreamingMessageContent | null; + init_state: JsonObject | null; +}; + +export type StreamingTransactionAccountState = JsonObject & { + hash: string; + balance: string | null; + extra_currencies: JsonValue | null; + account_status: string | null; + frozen_hash: string | null; + data_hash: string | null; + code_hash: string | null; +}; + +export type StreamingBlockRef = JsonObject & { + workchain: number; + shard: string; + seqno: number; +}; + +export type StreamingTransaction = JsonObject & { + account: string; + hash: string; + lt: string; + now: number; + mc_block_seqno: number; + trace_id: string; + prev_trans_hash: string; + prev_trans_lt: string; + orig_status: string; + end_status: string; + total_fees: string; + total_fees_extra_currencies: JsonObject; + description: JsonObject & { + type: string; + }; + out_msgs: StreamingMessage[]; + block_ref?: StreamingBlockRef; + account_state_before?: StreamingTransactionAccountState; + account_state_after?: StreamingTransactionAccountState; + in_msg?: StreamingMessage | null; + finality?: Finality; + emulated?: boolean; +}; + +export type StreamingTraceNode = JsonObject & { + tx_hash: string; + children: StreamingTraceNode[]; + in_msg_hash?: string; + transaction?: StreamingTransaction | null; +}; + +export type StreamingTrace = StreamingTraceNode; + +export type StreamingAction = JsonObject & { + trace_id: string; + action_id: string; + start_lt: string; + end_lt: string; + start_utime: number; + end_utime: number; + trace_end_lt: string; + trace_end_utime: number; + trace_mc_seqno_end: number; + transactions: string[]; + success: boolean; + type: string; + details: JsonObject; + trace_external_hash?: string; + trace_external_hash_norm?: string; + accounts: string[]; + finality?: Finality; +}; + +export type StreamingTransactionsEvent = { + type: "transactions"; + finality: Finality; + trace_external_hash_norm: string; + transactions: StreamingTransaction[]; + address_book?: Record; + metadata?: Record; +}; + +export type StreamingActionsEvent = { + type: "actions"; + finality: Finality; + trace_external_hash_norm: string; + actions: StreamingAction[]; + address_book?: Record; + metadata?: Record; +}; + +export type StreamingTraceEvent = { + type: "trace"; + finality: Finality; + trace_external_hash_norm: string; + trace: StreamingTrace; + transactions: Record; + actions?: StreamingAction[]; + address_book?: Record; + metadata?: Record; +}; + +export type StreamingAccountStateEvent = { + type: "account_state_change"; + finality: NonPendingFinality; + account: string; + state: JsonObject & { + hash: string; + balance: string; + account_status: string; + data_hash?: string; + code_hash?: string; + }; +}; + +export type StreamingJettonsEvent = { + type: "jettons_change"; + finality: NonPendingFinality; + jetton: JsonObject & { + address: string; + balance: string; + owner: string; + jetton: string; + last_transaction_lt: string; + }; + address_book?: Record; + metadata?: Record; +}; + +export type StreamingTraceInvalidatedEvent = { + type: "trace_invalidated"; + trace_external_hash_norm: string; +}; + +export type StreamingEvent = + | StreamingTransactionsEvent + | StreamingActionsEvent + | StreamingTraceEvent + | StreamingAccountStateEvent + | StreamingJettonsEvent + | StreamingTraceInvalidatedEvent; + +export type StreamingEventMap = StreamingLifecycleEvents & { + transactions: StreamingTransactionsEvent; + actions: StreamingActionsEvent; + trace: StreamingTraceEvent; + account_state_change: StreamingAccountStateEvent; + jettons_change: StreamingJettonsEvent; + trace_invalidated: StreamingTraceInvalidatedEvent; +}; + +/** + * A streaming client does NOT auto-reconnect. When the connection drops the client + * emits an `"error"` event followed by `"close"`. To resume, create a new client + * (or call `subscribe()` again on a WebSocket client) and re-attach listeners. + */ +export type StreamingClient = { + subscribe(params: StreamingSubscription): Promise; + on( + event: K, + handler: (data: StreamingEventMap[K]) => void, + ): () => void; + off( + event: K, + handler: (data: StreamingEventMap[K]) => void, + ): void; + close(): Promise; + readonly ready: boolean; +}; + +export type StreamingSseParameters = StreamingBaseParameters & { + /** Defaults to `globalThis.fetch`. Must support streaming responses. */ + fetch?: typeof fetch; + headers?: Record; +}; diff --git a/src/client/streaming/utils.ts b/src/client/streaming/utils.ts new file mode 100644 index 00000000..8f2a21ac --- /dev/null +++ b/src/client/streaming/utils.ts @@ -0,0 +1,228 @@ +/** + * 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 Deferred = { + promise: Promise; + resolve(value: T): void; + reject(error: unknown): void; + settled: boolean; +}; + +export function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + const d: Deferred = { + promise, + settled: false, + resolve(value: T) { + if (!d.settled) { + d.settled = true; + resolve(value); + } + }, + reject(error: unknown) { + if (!d.settled) { + d.settled = true; + reject(error); + } + }, + }; + return d; +} + +export const FINALITIES = ["pending", "confirmed", "finalized"] as const; +export type Finality = (typeof FINALITIES)[number]; +export const FINALITY_SET: ReadonlySet = new Set(FINALITIES); + +export const NON_PENDING_FINALITIES = ["confirmed", "finalized"] as const; +export type NonPendingFinality = (typeof NON_PENDING_FINALITIES)[number]; +export const NON_PENDING_FINALITY_SET: ReadonlySet = new Set( + NON_PENDING_FINALITIES, +); + +export const SUBSCRIBABLE_EVENT_TYPES = [ + "transactions", + "actions", + "trace", + "account_state_change", + "jettons_change", +] as const; +export type SubscribableEventType = (typeof SUBSCRIBABLE_EVENT_TYPES)[number]; +export const SUBSCRIBABLE_EVENT_TYPE_SET: ReadonlySet = new Set( + SUBSCRIBABLE_EVENT_TYPES, +); + +export type StreamingService = "tonapi" | "toncenter"; +export type StreamingNetwork = "mainnet" | "testnet"; + +export const DEFAULT_REQUEST_TIMEOUT_MS = 5_000; +export const DEFAULT_PING_INTERVAL_MS = 15_000; + +export function isAbortError(reason: unknown): boolean { + return reason instanceof Error && reason.name === "AbortError"; +} + +export function sanitizeStringList( + values?: readonly string[], +): string[] | undefined { + if (!values) { + return undefined; + } + + const seen = new Set(); + const normalized: string[] = []; + + for (const rawValue of values) { + if (typeof rawValue !== "string") { + throw new TypeError("Expected a string value"); + } + + const value = rawValue.trim(); + if (!value || seen.has(value)) { + continue; + } + + seen.add(value); + normalized.push(value); + } + + return normalized.length > 0 ? normalized : undefined; +} + +export function requireStringList( + values: readonly string[] | undefined, + fieldName: string, +): string[] { + const normalized = sanitizeStringList(values); + if (!normalized) { + throw new Error( + `${fieldName} must contain at least one non-empty value`, + ); + } + return normalized; +} + +export function normalizeTimeoutMs( + value: number | undefined, + defaultValue: number, + fieldName: string, +): number { + if (value === undefined) { + return defaultValue; + } + + if (!Number.isFinite(value) || value < 0) { + throw new Error(`${fieldName} must be a non-negative finite number`); + } + + return Math.floor(value); +} + +export function appendQueryParameter( + endpoint: string, + key: string, + value: string, +): string { + const url = new URL(endpoint); + url.searchParams.set(key, value); + return url.toString(); +} + +export async function describeHttpError(response: { + status: number; + statusText: string; + text?(): Promise; +}): Promise { + const statusPart = `${response.status} ${response.statusText}`.trim(); + let bodyText = ""; + + try { + if (typeof response.text === "function") { + bodyText = (await response.text()).trim(); + } + } catch {} + + return bodyText ? `${statusPart} — ${bodyText}` : statusPart; +} + +export function describeUnexpectedMessage(payload: unknown): string { + try { + return JSON.stringify(payload); + } catch { + return String(payload); + } +} + +export function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +export function buildStreamingUrl( + transport: "sse" | "ws", + params: { + service?: StreamingService; + network?: StreamingNetwork; + endpoint?: string; + apiKey?: string; + apiKeyParam?: string; + }, +): string { + const resolved = resolveProviderEndpoint( + transport, + params.service, + params.network, + params.endpoint, + params.apiKeyParam, + ); + return params.apiKey + ? appendQueryParameter( + resolved.endpoint, + resolved.apiKeyParam, + params.apiKey, + ) + : resolved.endpoint; +} + +export function resolveProviderEndpoint( + transport: "sse" | "ws", + service: StreamingService | undefined, + network: StreamingNetwork | undefined, + endpoint: string | undefined, + apiKeyParam: string | undefined, +): { endpoint: string; apiKeyParam: string } { + if (service && endpoint) { + throw new Error( + "Cannot specify both 'service' and 'endpoint'. Use one or the other.", + ); + } + + if (service) { + const domain = service === "tonapi" ? "tonapi.io" : "toncenter.com"; + const host = + (network ?? "mainnet") === "testnet" ? `testnet.${domain}` : domain; + const prefix = service === "toncenter" ? "/api" : ""; + const proto = transport === "ws" ? "wss" : "https"; + + return { + endpoint: `${proto}://${host}${prefix}/streaming/v2/${transport}`, + apiKeyParam: service === "tonapi" ? "token" : "api_key", + }; + } + + if (!endpoint) { + throw new Error( + "Streaming endpoint is required when service is not specified", + ); + } + + return { endpoint, apiKeyParam: apiKeyParam ?? "api_key" }; +} diff --git a/src/index.ts b/src/index.ts index 69009f2d..6502e2d5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -24,6 +24,52 @@ export { TonClient4Parameters, } from "./client/TonClient4"; +// +// Streaming Clients +// + +export { + TonWsClient, + TonSseClient, + StreamingClosedError, + StreamingError, + StreamingHandshakeError, + StreamingRequestTimeoutError, +} from "./client/streaming"; +export type { + Finality, + StreamingAction, + StreamingAccountStateEvent, + StreamingAddressBookEntry, + StreamingActionsEvent, + StreamingBlockRef, + StreamingDecodedMessage, + StreamingEvent, + StreamingEventMap, + SubscribableEventType, + StreamingJettonsEvent, + StreamingLifecycleEvents, + StreamingMessage, + StreamingMessageContent, + StreamingMetadataEntry, + StreamingErrorContext, + StreamingSubscription, + StreamingTransport, + StreamingClient, + StreamingService, + StreamingNetwork, + StreamingSseParameters, + StreamingWebSocketParameters, + StreamingBaseParameters, + StreamingTrace, + StreamingTraceEvent, + StreamingTraceInvalidatedEvent, + StreamingTraceNode, + StreamingTransaction, + StreamingTransactionAccountState, + StreamingTransactionsEvent, +} from "./client/streaming"; + // // Wallets //