diff --git a/.changeset/long-cougars-play.md b/.changeset/long-cougars-play.md new file mode 100644 index 000000000..9a1b3781d --- /dev/null +++ b/.changeset/long-cougars-play.md @@ -0,0 +1,5 @@ +--- +"@knocklabs/client": minor +--- + +Connect sockets via WebSocket gateway diff --git a/.changeset/ripe-pandas-eat.md b/.changeset/ripe-pandas-eat.md new file mode 100644 index 000000000..c22d566de --- /dev/null +++ b/.changeset/ripe-pandas-eat.md @@ -0,0 +1,15 @@ +--- +"@knocklabs/client": minor +--- + +Add `maxRetries` option to `ApiRequestConfig` + +When making a request to the Knock API using `ApiClient`'s `makeRequest` function, the `maxRetries` option of `ApiRequestConfig` can now be used to set the maximum number of times the request should be retried when a retryable error occurs. To disable retries altogether, set `maxRetries` to 0. + +```js +const result = await knock.client().makeRequest({ + method: "GET", + url: "/v1/users", + maxRetries: 0, +}); +``` diff --git a/packages/client/src/api.ts b/packages/client/src/api.ts index 878bcbf5f..6fc4f44df 100644 --- a/packages/client/src/api.ts +++ b/packages/client/src/api.ts @@ -1,6 +1,10 @@ import { Socket } from "phoenix"; -import { exponentialBackoffFullJitter } from "./helpers"; +import { + createGatewayTransport, + fetchSocketConnectUrl, +} from "./gatewayTransport"; +import { cacheUntilExpired, exponentialBackoffFullJitter } from "./helpers"; import { PageVisibilityManager } from "./pageVisibility"; // Ceiling for socket reconnect backoff. This is intentionally much larger than @@ -19,6 +23,10 @@ const STABLE_CONNECTION_THRESHOLD_MS = 30_000; // counting. const MAX_UNSTABLE_CONNECTION_COUNT = 15; +// The max number of times a request is retried after a retryable failure, +// unless overridden by `ApiRequestConfig.maxRetries`. +const DEFAULT_MAX_REQUEST_RETRIES = 3; + type ApiClientOptions = { host: string; apiKey: string; @@ -44,6 +52,8 @@ export type ApiRequestConfig = { data?: unknown; headers?: HeadersInit; signal?: AbortSignal; + /** Max number of retries for retryable failures. Set this to 0 to disable retries. */ + maxRetries?: number; }; type FetchClient = ( @@ -117,7 +127,21 @@ class ApiClient { }); if (typeof window !== "undefined" && typeof WebSocket !== "undefined") { + // Connect sockets via the WebSocket gateway API endpoint, which returns a signed ws:// or + // wss:// connect URL. This signed URL is cached and used to reconnect until it expires. + const connectUrlCache = cacheUntilExpired((signal: AbortSignal) => + // If this fetch rejects, the error is handled by the gateway transport, and Phoenix + // considers it a failed connection attempt + fetchSocketConnectUrl({ apiClient: this, signal }), + ); + const transport = createGatewayTransport({ + baseTransport: WebSocket, + fetchConnectUrl: connectUrlCache.get, + onConnectUrlInvalid: connectUrlCache.invalidate, + }); + this.socket = new Socket(`${this.host.replace("http", "ws")}/ws/v1`, { + transport: transport, params: { user_token: this.userToken, api_key: this.apiKey, @@ -249,9 +273,11 @@ class ApiClient { private async requestWithRetries(req: ApiRequestConfig) { let lastError: unknown; + const { maxRetries = DEFAULT_MAX_REQUEST_RETRIES } = req; + // Sequential retry loop: each attempt awaits in order and returns early on // success or a non-retryable error, so it doesn't reduce to an array method. - for (let attempt = 0; attempt <= 3; attempt++) { + for (let attempt = 0; attempt <= maxRetries; attempt++) { let retryAfterMs = 0; try { @@ -279,7 +305,7 @@ class ApiClient { } } - if (attempt < 3) { + if (attempt < maxRetries) { await this.delay(this.getRetryDelay(attempt + 1, retryAfterMs)); } } diff --git a/packages/client/src/gatewayTransport.ts b/packages/client/src/gatewayTransport.ts new file mode 100644 index 000000000..a1a229766 --- /dev/null +++ b/packages/client/src/gatewayTransport.ts @@ -0,0 +1,272 @@ +import type ApiClient from "./api"; +import { appendQueryString, getQueryString } from "./helpers"; + +/** + * The signature that Phoenix expects socket transports to conform to, based on the native browser + * `WebSocket`. Only fields and methods that Phoenix relies on are included here. + */ +export type WebSocketLike = Pick< + WebSocket, + | "readyState" + | "bufferedAmount" + | "binaryType" + | "onopen" + | "onclose" + | "onmessage" + | "onerror" + | "send" + | "close" +>; + +export type TransportConstructor = new ( + url: string, + protocols?: string | string[], +) => WebSocketLike; + +export type GatewayConnectResult = { + signedUrl: string; + expiresAt: number; +}; + +export type GatewayTransportOptions = { + /** + * The underlying transport constructor. An instance of this transport will be created once a + * connect URL has been fetched. + */ + baseTransport: TransportConstructor; + /** + * Resolves the signed WebSocket URL to connect to. Rejects on failure. + */ + fetchConnectUrl: (signal: AbortSignal) => Promise; + /** + * Called when a connection made with the fetched URL closed before it ever + * opened, which most likely means the URL was rejected (though a network blip + * could look the same). + */ + onConnectUrlInvalid?: () => void; +}; + +/** + * All possible socket states. Copied from Phoenix, based on the `readyState` property of + * `WebSocket`. + * @see {@link https://github.com/phoenixframework/phoenix/blob/1bae3a639f84cd97ccb6b984a8ef4fcdc7aa48ae/assets/js/phoenix/constants.js#L5} + */ +export const SOCKET_STATES = { + connecting: 0, + open: 1, + closing: 2, + closed: 3, +} as const; + +// WebSocket close event status codes. +// See https://www.rfc-editor.org/info/rfc6455/#section-7.4.1 +const CLOSE_CODE_NORMAL = 1000; +const CLOSE_CODE_ABNORMAL = 1006; + +/** + * Requests a signed WebSocket URL from the WebSocket gateway API endpoint. Rejects on any HTTP + * error or malformed body so the transport treats it like a failed connection attempt. + */ +export async function fetchSocketConnectUrl({ + apiClient, + signal, +}: { + apiClient: ApiClient; + signal: AbortSignal; +}): Promise { + const result = await apiClient.makeRequest({ + method: "POST", + url: "/v1/sockets/connect", + signal, + // Set maxRetries to 0, so that the API client's own retry logic is never used. Phoenix + // handles retries automatically via the socket's reconnect backoff. + maxRetries: 0, + }); + + if (result.statusCode !== "ok") { + throw ( + result.error ?? + new Error(`Socket connect request failed with status ${result.status}`) + ); + } + + const body = result.body as + | Partial<{ url: string; expires_at: string }> + | undefined; + + const expiresAt = + typeof body?.expires_at === "string" ? Date.parse(body.expires_at) : NaN; + + if (typeof body?.url !== "string" || !body.url || Number.isNaN(expiresAt)) { + throw new Error("Invalid socket connect response"); + } + + return { signedUrl: body.url, expiresAt }; +} + +export function createGatewayTransport({ + baseTransport, + fetchConnectUrl, + onConnectUrlInvalid, +}: GatewayTransportOptions): TransportConstructor { + // Use a named class, since Phoenix logs using the class name + return class GatewayTransport implements WebSocketLike { + binaryType: BinaryType = "blob"; + + onopen: ((event: Event) => void) | null = null; + onclose: ((event: CloseEvent) => void) | null = null; + onmessage: ((event: MessageEvent) => void) | null = null; + onerror: ((event: unknown) => void) | null = null; + + private inner: WebSocketLike | null = null; + private state: number = SOCKET_STATES.connecting; + private readonly abort = new AbortController(); + private readonly unsignedQueryString: string; + private opened = false; + private closeRequested = false; + private settled = false; + private readonly timeoutId: ReturnType; + + constructor( + url: string, + private readonly protocols?: string | string[], + ) { + // The base URL passed in by Phoenix contains some query params we want to preserve, including + // vsn (protocol version to send on connect), api_key, user_token, and branch_slug. This query + // string is considered "unsigned" because it lacks an auth_token query param. + this.unsignedQueryString = getQueryString(url); + + // Wait for the caller to finish setting up callbacks and properties. This pattern was copied + // from Phoenix's LongPoll transport. + this.timeoutId = setTimeout(() => this.start(), 0); + } + + get readyState(): number { + return this.inner ? this.inner.readyState : this.state; + } + + get bufferedAmount(): number { + return this.inner ? this.inner.bufferedAmount : 0; + } + + send(data: string | ArrayBufferLike | ArrayBufferView | Blob): void { + if (!this.inner) { + // Phoenix only sends when a connection is open, so prevent calls to send before a + // connection is established. + throw new Error( + "GatewayTransport: cannot send before the connection is open", + ); + } + this.inner.send(data); + } + + close(code?: number, reason?: string): void { + if (this.closeRequested || this.settled) { + return; + } + + this.closeRequested = true; + + // In case start() has not run yet, clear the timeout used to invoke it in the constructor + clearTimeout(this.timeoutId); + + if (this.inner) { + this.inner.close(code, reason); + return; + } + + // The socket is closing before the fetch completed. This is still considered a clean close. + this.abort.abort(); + this.state = SOCKET_STATES.closed; + this.emitClose(CLOSE_CODE_NORMAL, "", true); + } + + private async start(): Promise { + // Phoenix may connect() and disconnect() in the same tick; teardown then calls close() on us + // before this runs. This prevents a fetch that would be discarded. + if (this.closeRequested) { + return; + } + + let result: GatewayConnectResult; + try { + result = await fetchConnectUrl(this.abort.signal); + } catch (error) { + if (this.closeRequested) { + // Ignore this error; the socket was already closed cleanly. + return; + } + // An unexepected error occurred while fetching + this.fail(error); + return; + } + + // The socket was closed while a fetch was in-progress + if (this.closeRequested) { + return; + } + + const connectUrl = appendQueryString( + result.signedUrl, + this.unsignedQueryString, + ); + + try { + this.inner = this.initBaseTransport(connectUrl); + } catch (error) { + // The base transport's constructor threw an exception. We never even attempted a + // connection, so onclose won't fire. The connect URL must be invalid. + onConnectUrlInvalid?.(); + this.fail(error); + } + } + + private initBaseTransport(url: string): WebSocketLike { + const inner = new baseTransport(url, this.protocols); + + inner.binaryType = this.binaryType; + inner.onopen = (ev) => { + this.opened = true; + this.onopen?.(ev); + }; + inner.onmessage = (ev) => this.onmessage?.(ev); + inner.onerror = (ev) => this.onerror?.(ev); + inner.onclose = (ev) => { + if (!this.opened && !this.closeRequested) { + // A connection was never opened and this close event was unexpected. + // The signed connect URL is likely invalid. + onConnectUrlInvalid?.(); + } + this.settled = true; + this.onclose?.(ev); + }; + + return inner; + } + + private fail(error: unknown): void { + this.state = SOCKET_STATES.closed; + this.onerror?.(error); + this.emitClose(CLOSE_CODE_ABNORMAL, "gateway_connect_failed", false); + } + + private emitClose(code: number, reason: string, wasClean: boolean): void { + // Check this.settled to prevent triggering onclose more than once. For example, this could + // occur if the transport is closed manually inside of an onerror callback. + if (this.settled) { + return; + } + + this.settled = true; + + // Mirrors Phoenix's LongPoll.close: CloseEvent exists in browsers. + // Fall back to a plain object elsewhere (e.g. React Native). + const opts = { code, reason, wasClean }; + if (typeof CloseEvent !== "undefined") { + this.onclose?.(new CloseEvent("close", opts)); + } else { + this.onclose?.(opts as CloseEvent); + } + } + }; +} diff --git a/packages/client/src/helpers.ts b/packages/client/src/helpers.ts index ea6329b95..a5b09517e 100644 --- a/packages/client/src/helpers.ts +++ b/packages/client/src/helpers.ts @@ -1,3 +1,5 @@ +import { Store } from "@tanstack/store"; + const uuidRegex = /^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/; @@ -5,6 +7,35 @@ export function isValidUuid(uuid: string) { return uuidRegex.test(uuid); } +/** + * Returns the query string for a given URL, minus the leading `"?"` character. When the URL has no + * query string, returns the empty string. + * + * This function does not use the `URL` class due to known issues in React Native's implementation + * of `URL`. + * + * @param url a valid string URL. For example, `"https://example.com/test?hello=world"`. + */ +export function getQueryString(url: string): string { + const idx = url.indexOf("?"); + return idx === -1 ? "" : url.slice(idx + 1); +} + +/** + * Appends a query string to a given URL. Note that this does not dedupe query params. + * + * This function does not use the `URL` class due to known issues in React Native's implementation + * of `URL`. + * + * @param url a valid string URL. + * @param queryString a query string, minus any leading `"?"` character. + */ +export function appendQueryString(url: string, queryString: string): string { + return queryString + ? `${url}${url.includes("?") ? "&" : "?"}${queryString}` + : url; +} + /** * Exponential backoff with full jitter and a minimum delay floor. * @@ -43,3 +74,44 @@ export function exponentialBackoffFullJitter( const jitterRange = exponentialDelay - minDelayMs; return minDelayMs + Math.floor(Math.random() * jitterRange); } + +export type CachedFetch = { + /** + * Returns the cached value if still valid, otherwise calls `fetchFunc` with the same args and + * caches the result. + */ + get: (...args: Args) => Promise; + /** + * Drops any cached value so the next `get` call fetches again. + */ + invalidate: () => void; +}; + +export const CACHE_EXPIRY_PADDING_MS = 5_000; + +/** + * Wraps `fetchFunc` so its result is cached until `expiresAt` (epoch ms), minus + * `CACHE_EXPIRY_PADDING_MS`. + */ +export function cacheUntilExpired< + Args extends unknown[], + T extends { expiresAt: number }, +>(fetchFunc: (...args: Args) => Promise): CachedFetch { + const store = new Store(null); + + const get = async (...args: Args) => { + const cached = store.state; + if (cached && cached.expiresAt - CACHE_EXPIRY_PADDING_MS > Date.now()) { + return cached; + } + const value = await fetchFunc(...args); + store.setState(() => value); + return value; + }; + + const invalidate = () => { + store.setState(() => null); + }; + + return { get, invalidate }; +} diff --git a/packages/client/test/api.test.ts b/packages/client/test/api.test.ts index 757d48287..609b353f7 100644 --- a/packages/client/test/api.test.ts +++ b/packages/client/test/api.test.ts @@ -3,6 +3,9 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import packageJson from "../package.json"; import ApiClient from "../src/api"; +import type { TransportConstructor } from "../src/gatewayTransport"; + +import { createWebSocketMock } from "./test-utils/mocks"; const TEST_BRANCH_SLUG = "lorem-ipsum-dolor-branch"; @@ -586,6 +589,75 @@ describe("API Client", () => { expect(response.statusCode).toBe("error"); expect(fetchMock).toHaveBeenCalledTimes(1); }); + + test("does not retry when maxRetries is 0", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: undefined, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + createJsonResponse({ error: "Server Error" }, 500), + ) + .mockResolvedValue(createJsonResponse({ success: true })); + setFetchMock(apiClient, fetchMock); + skipRetryDelays(apiClient); + + const consoleSpy = vi + .spyOn(console, "error") + .mockImplementation(() => { }); + try { + const response = await apiClient.makeRequest({ + method: "GET", + url: "/test", + maxRetries: 0, + }); + + expect(response.statusCode).toBe("error"); + expect(response.status).toBe(500); + expect(fetchMock).toHaveBeenCalledTimes(1); + } finally { + consoleSpy.mockRestore(); + } + }); + + test("retries up to maxRetries times when overridden", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: undefined, + }); + // Return a fresh Response per call, like a real fetch — a Response body + // can only be consumed once, so each retry must get its own. + const fetchMock = vi + .fn() + .mockImplementation(() => + createJsonResponse({ error: "Server Error" }, 500), + ); + setFetchMock(apiClient, fetchMock); + skipRetryDelays(apiClient); + + const consoleSpy = vi + .spyOn(console, "error") + .mockImplementation(() => { }); + try { + const response = await apiClient.makeRequest({ + method: "GET", + url: "/test", + maxRetries: 1, + }); + + expect(response.statusCode).toBe("error"); + expect(response.status).toBe(500); + + // The initial attempt plus one retry + expect(fetchMock).toHaveBeenCalledTimes(2); + } finally { + consoleSpy.mockRestore(); + } + }); }); describe("Request Configuration", () => { @@ -665,6 +737,27 @@ describe("API Client", () => { expect(apiClient.socket).toBeDefined(); }); + test("configures socket with the gateway transport", () => { + vi.stubGlobal("window", {}); + + vi.mocked(Socket).mockClear(); + + new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: undefined, + }); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + expect(typeof socketOpts.transport).toBe("function"); + expect((socketOpts.transport as NewableFunction).name).toBe( + "GatewayTransport", + ); + }); + test("configures socket with reconnectAfterMs and rejoinAfterMs", () => { vi.stubGlobal("window", {}); @@ -1056,4 +1149,458 @@ describe("API Client", () => { expect(socket.disconnect).not.toHaveBeenCalled(); }); }); + + describe("WebSocket Connection via Gateway", () => { + const MockWebSocket = createWebSocketMock(); + + beforeEach(() => { + vi.useFakeTimers(); + vi.resetAllMocks(); + + vi.mocked(Socket).mockImplementation( + createSocketMock as unknown as () => Socket, + ); + + vi.stubGlobal("window", {}); + vi.stubGlobal("WebSocket", MockWebSocket); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + test("establishes a WebSocket connection via the gateway transport", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + // The gateway's signed connect URL, as returned by `POST /v1/sockets/connect` + const signedUrl = + "wss://ws1.knock.app/ws/v1/websocket?auth_token=signed-token"; + const fetchMock = vi.fn().mockResolvedValue( + createJsonResponse({ + url: signedUrl, + expires_at: new Date(Date.now() + 60_000).toISOString(), + }), + ); + setFetchMock(apiClient, fetchMock); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + // The URL Phoenix would hand the transport, carrying the unsigned params + // (vsn, api_key, user_token) that the gateway's signed URL lacks. + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + // Call GatewayTransport directly to simulate Phoenix's socket establishing a connection + new GatewayTransport(PHOENIX_URL); + + await vi.waitFor(() => { expect(MockWebSocket).toHaveBeenCalledTimes(1); }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledWith( + "https://api.knock.app/v1/sockets/connect", + expect.objectContaining({ method: "POST" }), + ); + + // Each underlying WebSocket is constructed with the signed URL merged + // with the original unsigned query params. + const expectedUrl = `${signedUrl}&vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456`; + expect(MockWebSocket).toHaveBeenLastCalledWith(expectedUrl, undefined); + }); + + test("caches the signed WebSocket connect URL until it expires", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + const CACHE_TTL_MS = 60_000; + const startDate = new Date(2024, 0, 1, 0, 0, 0); + vi.setSystemTime(startDate); + + // Each fetch returns a signed URL that expires `CACHE_TTL_MS` from + // whatever the (mocked) system time is at the time of the call. + const signedUrl = + "wss://ws1.knock.app/ws/v1/websocket?auth_token=signed-token"; + const fetchMock = vi.fn().mockImplementation(() => + createJsonResponse({ + url: signedUrl, + expires_at: new Date(Date.now() + CACHE_TTL_MS).toISOString(), + }), + ); + setFetchMock(apiClient, fetchMock); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + // The first connection fetches a signed connect URL. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(1); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // A second connection, made before the cached URL expires, reuses it + // rather than fetching again. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(2); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Advance the system clock past the cached URL's expiry. + vi.setSystemTime(new Date(startDate.getTime() + CACHE_TTL_MS + 1)); + + // A third connection, made after expiry, fetches a fresh signed URL. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(3); + }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + test("skips establishing a WebSocket connection when a signed URL fetch fails", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + // A plain `Error` (no `.response`) isn't retryable, so this fails on + // the first attempt rather than after exhausting retries. + const fetchError = new Error("network down"); + const fetchMock = vi.fn().mockRejectedValue(fetchError); + setFetchMock(apiClient, fetchMock); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + const transport = new GatewayTransport(PHOENIX_URL); + const onClose = vi.fn(); + transport.onclose = onClose; + + // The transport reports the failure via `onclose`, since it never + // got far enough to open a connection. + await vi.waitFor(() => { + expect(onClose).toHaveBeenCalledTimes(1); + }); + + // No WebSocket is ever constructed, since the connect URL fetch + // failed before a connection could be attempted. + expect(MockWebSocket).not.toHaveBeenCalled(); + expect(fetchMock).toHaveBeenCalledTimes(1); + + const closeEvent = onClose.mock.calls[0]![0] as CloseEvent; + expect(closeEvent.wasClean).toBe(false); + }); + + test("does not retry a retryable signed URL fetch failure, leaving retries to Phoenix", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + // A 5xx is normally retried by `makeRequest`, but the gateway transport's fetch sets + // maxRetries to 0 so that each Phoenix connection attempt makes a single request. + const fetchMock = vi + .fn() + .mockResolvedValue(createJsonResponse({ error: "Server Error" }, 503)); + setFetchMock(apiClient, fetchMock); + skipRetryDelays(apiClient); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + const consoleSpy = vi + .spyOn(console, "error") + .mockImplementation(() => { }); + try { + const transport = new GatewayTransport(PHOENIX_URL); + const onClose = vi.fn(); + transport.onclose = onClose; + + await vi.waitFor(() => { + expect(onClose).toHaveBeenCalledTimes(1); + }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(MockWebSocket).not.toHaveBeenCalled(); + + // An unclean close is what triggers Phoenix's reconnect backoff + const closeEvent = onClose.mock.calls[0]![0] as CloseEvent; + expect(closeEvent.wasClean).toBe(false); + } finally { + consoleSpy.mockRestore(); + } + }); + + test("invalidates the cached connect URL when the gateway rejects it, so the next connection refetches", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + // Each fetch returns a signed URL with a distinct token, so the two + // connections' URLs can be told apart, and an expiry far enough out + // that the cache wouldn't naturally expire between them. + let fetchCount = 0; + const fetchMock = vi.fn().mockImplementation(() => { + fetchCount += 1; + return createJsonResponse({ + url: `wss://ws1.knock.app/ws/v1/websocket?auth_token=signed-token-${fetchCount}`, + expires_at: new Date(Date.now() + 60_000).toISOString(), + }); + }); + setFetchMock(apiClient, fetchMock); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + // The first connection fetches a signed URL and connects. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(1); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // The underlying WebSocket closes before ever opening (`onopen` is + // never called), which looks like the signed URL is invalid + const ws = MockWebSocket.mock.results[0]!.value; + ws.onclose({ code: 1006, wasClean: false }); + + // A second connection, made while the cached URL would otherwise still + // be unexpired, refetches rather than reusing the invalidated URL. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(2); + }); + expect(fetchMock).toHaveBeenCalledTimes(2); + + const [firstUrl] = MockWebSocket.mock.calls[0]!; + const [secondUrl] = MockWebSocket.mock.calls[1]!; + expect(secondUrl).not.toBe(firstUrl); + }); + + test("keeps the cached connect URL when a connection opens and later drops", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + const signedUrl = + "wss://ws1.knock.app/ws/v1/websocket?auth_token=signed-token"; + const fetchMock = vi.fn().mockResolvedValue( + createJsonResponse({ + url: signedUrl, + expires_at: new Date(Date.now() + 60_000).toISOString(), + }), + ); + setFetchMock(apiClient, fetchMock); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(1); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // The connection opens successfully and only later drops uncleanly. + // Unlike a close that arrives before `onopen`, this isn't evidence the + // signed URL itself was bad, so the cache should survive it. + const ws = MockWebSocket.mock.results[0]!.value; + ws.onopen(); + ws.onclose({ wasClean: false }); + + // A second connection reuses the cached URL instead of refetching. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(2); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + test("does not invalidate the cache when Phoenix closes a still-connecting socket", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + const signedUrl = + "wss://ws1.knock.app/ws/v1/websocket?auth_token=signed-token"; + const fetchMock = vi.fn().mockResolvedValue( + createJsonResponse({ + url: signedUrl, + expires_at: new Date(Date.now() + 60_000).toISOString(), + }), + ); + setFetchMock(apiClient, fetchMock); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + const transport = new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(1); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // The mock WebSocket's `close` doesn't emit a close event on its own + // the way a real one would, so wire it to do so here. + const ws = MockWebSocket.mock.results[0]!.value; + ws.close.mockImplementation(() => { + ws.onclose({ code: 1000, wasClean: false }); + }); + + // Phoenix tears the transport down (e.g. the caller disconnected) + // while the socket is still connecting, before it ever opened. This is + // a deliberate close, not a rejected signed URL, so it shouldn't + // invalidate the cache. + transport.close(); + + // A second connection reuses the cached URL instead of refetching. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(MockWebSocket).toHaveBeenCalledTimes(2); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + test("aborts the in-flight connect request when closed before it resolves, and does not cache the aborted result", async () => { + const apiClient = new ApiClient({ + host: "https://api.knock.app", + apiKey: "pk_test_12345", + userToken: "user_token_456", + disconnectOnPageHidden: false, + }); + + // Mimics real fetch: rejects with an AbortError once the request's + // signal is aborted, rather than ever resolving on its own. + const fetchMock = vi.fn().mockImplementation( + (_url: string, init?: RequestInit) => + new Promise((_resolve, reject) => { + const abortError = new DOMException( + "The operation was aborted.", + "AbortError", + ); + const signal = init?.signal; + if (signal?.aborted) { + reject(abortError); + return; + } + signal?.addEventListener("abort", () => reject(abortError)); + }), + ); + setFetchMock(apiClient, fetchMock); + + const socketOpts = vi.mocked(Socket).mock.calls[0]![1] as Record< + string, + unknown + >; + const GatewayTransport = socketOpts.transport as TransportConstructor; + + const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?vsn=2.0.0&api_key=pk_test_12345&user_token=user_token_456"; + + const consoleSpy = vi + .spyOn(console, "error") + .mockImplementation(() => { }); + try { + const transport = new GatewayTransport(PHOENIX_URL); + const onClose = vi.fn(); + const onError = vi.fn(); + transport.onclose = onClose; + transport.onerror = onError; + + // Wait for the HTTP request to actually be in flight before closing. + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + transport.close(); + + await vi.waitFor(() => { + expect(onClose).toHaveBeenCalledTimes(1); + }); + + // The abort signal propagated all the way through the cache, + // fetchSocketConnectUrl, and makeRequest/buildRequestInit, down to + // the real fetch call. + const requestInit = fetchMock.mock.calls[0]![1] as RequestInit; + expect(requestInit.signal?.aborted).toBe(true); + + // The close was clean and happened before any WebSocket existed. + expect(MockWebSocket).not.toHaveBeenCalled(); + expect(onError).not.toHaveBeenCalled(); + + const closeEvent = onClose.mock.calls[0]![0] as CloseEvent; + expect(closeEvent.wasClean).toBe(true); + } finally { + consoleSpy.mockRestore(); + } + + // The aborted fetch's result isn't cached, so a later connection + // attempt fetches again rather than reusing a cached value. + new GatewayTransport(PHOENIX_URL); + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + }); + }); }); diff --git a/packages/client/test/gatewayTransport.test.ts b/packages/client/test/gatewayTransport.test.ts new file mode 100644 index 000000000..057f44004 --- /dev/null +++ b/packages/client/test/gatewayTransport.test.ts @@ -0,0 +1,634 @@ +import { + afterEach, + assert, + beforeEach, + describe, + expect, + test, + vi, +} from "vitest"; + +import type ApiClient from "../src/api"; +import { + GatewayConnectResult, + SOCKET_STATES, + TransportConstructor, + WebSocketLike, + createGatewayTransport, + fetchSocketConnectUrl, +} from "../src/gatewayTransport"; + +class MockTransport implements WebSocketLike { + static instance: MockTransport | null = null; + + // GatewayTransport should only construct a single instance of the base transport, so use a static + // integer to keep track of how many times MockTransport's constructor was called + static constructionCount = 0; + + readyState: number = SOCKET_STATES.connecting; + bufferedAmount = 0; + binaryType: BinaryType = "blob"; + + onopen: ((event: Event) => void) | null = null; + onclose: ((event: CloseEvent) => void) | null = null; + onmessage: ((event: MessageEvent) => void) | null = null; + onerror: ((event: Event) => void) | null = null; + + send = vi.fn(); + + close = vi.fn((code, reason) => { + this.readyState = SOCKET_STATES.closed; + this.onclose?.({ + code: code ?? 1000, + reason: reason ?? "", + wasClean: true, + } as CloseEvent); + }); + + constructor( + public url: string, + public protocols?: string | string[], + ) { + MockTransport.constructionCount++; + if (MockTransport.instance) { + assert.fail("MockTransport: constructed more than once"); + } + MockTransport.instance = this; + } + + open() { + this.readyState = SOCKET_STATES.open; + this.onopen?.({} as Event); + } + + serverClose(code: number, wasClean: boolean) { + this.readyState = SOCKET_STATES.closed; + this.onclose?.({ code, reason: "", wasClean } as CloseEvent); + } +} + +// GatewayTransport's start function is called inside a setTimeout within the constructor. This +// function advances timers so that start is invoked. +const flushStart = () => vi.advanceTimersByTimeAsync(0); + +const PHOENIX_URL = + "wss://api.knock.app/ws/v1/websocket?user_token=ut&api_key=pk&branch_slug=null&vsn=2.0.0"; +const SIGNED_URL = "wss://ws1.knock-dev.app/ws/v1/websocket?auth_token=jwt"; +const MERGED_URL = `${SIGNED_URL}&user_token=ut&api_key=pk&branch_slug=null&vsn=2.0.0`; +const PROTOCOLS = ["soap", "wamp"]; + +describe("fetchSocketConnectUrl", () => { + const makeRequest = vi.fn(); + const apiClient = { makeRequest } as unknown as ApiClient; + const abortController = new AbortController(); + + beforeEach(() => { + vi.clearAllMocks(); + }); + + test("returns valid signed URL and expiration", async () => { + const expiresAt = "2024-01-01T00:00:00.000Z"; + + makeRequest.mockResolvedValueOnce({ + statusCode: "ok", + status: 200, + body: { url: SIGNED_URL, expires_at: expiresAt }, + }); + + const result = await fetchSocketConnectUrl({ + apiClient, + signal: abortController.signal, + }); + + expect(makeRequest).toHaveBeenCalledWith({ + method: "POST", + url: "/v1/sockets/connect", + signal: abortController.signal, + maxRetries: 0, + }); + expect(result).toEqual({ + signedUrl: SIGNED_URL, + expiresAt: Date.parse(expiresAt), + }); + }); + + test("throws an error when the request fails", async () => { + const error = new Error("network down"); + + makeRequest.mockResolvedValueOnce({ + statusCode: "error", + status: 0, + error, + }); + + await expect( + fetchSocketConnectUrl({ apiClient, signal: abortController.signal }), + ).rejects.toBe(error); + }); + + test("throws an error when response status is not ok", async () => { + makeRequest.mockResolvedValueOnce({ + statusCode: "error", + status: 500, + }); + + await expect( + fetchSocketConnectUrl({ apiClient, signal: abortController.signal }), + ).rejects.toThrow("Socket connect request failed with status 500"); + }); + + test("throws an error when expiration is not a string", async () => { + makeRequest.mockResolvedValueOnce({ + statusCode: "ok", + status: 200, + body: { url: SIGNED_URL, expires_at: 12345 }, + }); + + await expect( + fetchSocketConnectUrl({ apiClient, signal: abortController.signal }), + ).rejects.toThrow("Invalid socket connect response"); + }); + + test("throws an error when expiration is not a valid date", async () => { + makeRequest.mockResolvedValueOnce({ + statusCode: "ok", + status: 200, + body: { url: SIGNED_URL, expires_at: "not-a-date" }, + }); + + await expect( + fetchSocketConnectUrl({ apiClient, signal: abortController.signal }), + ).rejects.toThrow("Invalid socket connect response"); + }); + + test("throws an error when URL is missing from response", async () => { + makeRequest.mockResolvedValueOnce({ + statusCode: "ok", + status: 200, + body: { expires_at: "2024-01-01T00:00:00.000Z" }, + }); + + await expect( + fetchSocketConnectUrl({ apiClient, signal: abortController.signal }), + ).rejects.toThrow("Invalid socket connect response"); + }); + + test("throws an error when URL is not a string", async () => { + makeRequest.mockResolvedValueOnce({ + statusCode: "ok", + status: 200, + body: { url: 12345, expires_at: "2024-01-01T00:00:00.000Z" }, + }); + + await expect( + fetchSocketConnectUrl({ apiClient, signal: abortController.signal }), + ).rejects.toThrow("Invalid socket connect response"); + }); +}); + +describe("createGatewayTransport", () => { + const fetchConnectUrl = vi.fn< + (signal: AbortSignal) => Promise + >(async () => ({ + signedUrl: SIGNED_URL, + expiresAt: Date.now() + 60_000, + })); + const onConnectUrlInvalid = vi.fn<() => void>(); + let GatewayTransport: TransportConstructor; + + beforeEach(() => { + vi.useFakeTimers(); + vi.clearAllMocks(); + + MockTransport.instance = null; + MockTransport.constructionCount = 0; + + GatewayTransport = createGatewayTransport({ + baseTransport: MockTransport, + fetchConnectUrl, + onConnectUrlInvalid, + }); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + test("fetches the connect URL, then constructs the base transport with merged URL and protocols", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + + expect(fetchConnectUrl).not.toHaveBeenCalled(); + expect(MockTransport.constructionCount).toBe(0); + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.connecting); + expect(gatewayTransport.bufferedAmount).toBe(0); + + await flushStart(); + + expect(fetchConnectUrl).toHaveBeenCalledWith(expect.any(AbortSignal)); + const inner = getMockTransportInstance(); + expect(inner.url).toBe(MERGED_URL); + expect(inner.protocols).toBe(PROTOCOLS); + expect(gatewayTransport.readyState).toBe(inner.readyState); + expect(gatewayTransport.bufferedAmount).toBe(0); + }); + + test("applies a binaryType assigned after construction to the inner transport", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + + expect(gatewayTransport.binaryType).toBe("blob"); + expect(MockTransport.constructionCount).toBe(0); + + gatewayTransport.binaryType = "arraybuffer"; + + await flushStart(); + + expect(getMockTransportInstance().binaryType).toBe("arraybuffer"); + }); + + test("forwards events from the inner transport and delegates readyState/bufferedAmount", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + + const onopen = vi.fn(); + const onmessage = vi.fn(); + const onerror = vi.fn(); + const onclose = vi.fn(); + gatewayTransport.onopen = onopen; + gatewayTransport.onmessage = onmessage; + gatewayTransport.onerror = onerror; + gatewayTransport.onclose = onclose; + + inner.open(); + expect(onopen).toHaveBeenCalledTimes(1); + + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.open); + + const messageEvent = { data: "x" } as MessageEvent; + inner.onmessage?.(messageEvent); + expect(onmessage).toHaveBeenCalledWith(messageEvent); + + const error = new Error("boom"); + inner.onerror?.(error as unknown as Event); + expect(onerror).toHaveBeenCalledWith(error); + + const bufferedAmount = 128; + inner.bufferedAmount = bufferedAmount; + expect(gatewayTransport.bufferedAmount).toBe(bufferedAmount); + + const closeEvent = { code: 1000, reason: "", wasClean: true } as CloseEvent; + inner.onclose?.(closeEvent); + expect(onclose).toHaveBeenCalledWith(closeEvent); + }); + + test("send throws before the inner transport exists and delegates once connected", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + + expect(() => gatewayTransport.send("hi")).toThrow( + "GatewayTransport: cannot send before the connection is open", + ); + + await flushStart(); + + const inner = getMockTransportInstance(); + gatewayTransport.send("hi"); + + expect(inner.send).toHaveBeenCalledWith("hi"); + }); + + test("close() while the fetch is pending aborts it and emits a single clean close", async () => { + fetchConnectUrl.mockImplementationOnce( + () => + new Promise((resolve) => { + setTimeout( + () => + resolve({ + signedUrl: SIGNED_URL, + expiresAt: Date.now() + 60_000, + }), + 1000, + ); + }), + ); + + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + + const onclose = vi.fn(); + gatewayTransport.onclose = onclose; + + gatewayTransport.close(); + + const capturedSignal = fetchConnectUrl.mock.lastCall?.[0]; + + expect(capturedSignal?.aborted).toBe(true); + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.closed); + expect(onclose).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ code: 1000, reason: "", wasClean: true }), + ); + + // Advance test timer so that fetchConnectUrl resolves + await vi.advanceTimersByTimeAsync(1000); + + // No inner transport was instantiated, as the fetch never completed + expect(MockTransport.constructionCount).toBe(0); + expect(onclose).toHaveBeenCalledTimes(1); + + // Closing the gateway transport manually does not trigger another onclose event + gatewayTransport.close(); + expect(onclose).toHaveBeenCalledTimes(1); + }); + + test("close() while the fetch is pending ignores a rejection caused by the abort", async () => { + fetchConnectUrl.mockImplementationOnce( + (signal) => + new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => + reject(new DOMException("Aborted", "AbortError")), + ); + }), + ); + + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + const onerror = vi.fn(); + const onclose = vi.fn(); + gatewayTransport.onerror = onerror; + gatewayTransport.onclose = onclose; + + await flushStart(); + + gatewayTransport.close(); + + // Run timers to trigger the rejection + await vi.runAllTimersAsync(); + + expect(onerror).not.toHaveBeenCalled(); + expect(onclose).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ code: 1000, reason: "", wasClean: true }), + ); + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.closed); + + // No inner transport was instantiated, as the fetch was aborted + expect(MockTransport.constructionCount).toBe(0); + }); + + test("close() called immediately after construction emits a clean close without fetching", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + + const onclose = vi.fn(); + gatewayTransport.onclose = onclose; + + gatewayTransport.close(); + + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.closed); + expect(onclose).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ code: 1000, reason: "", wasClean: true }), + ); + + await flushStart(); + + expect(fetchConnectUrl).not.toHaveBeenCalled(); + + // No inner transport was instantiated, as the fetch never occurred + expect(MockTransport.constructionCount).toBe(0); + expect(onclose).toHaveBeenCalledTimes(1); + }); + + test("close(code, reason) after connecting delegates the same arguments to the inner transport", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + inner.open(); + + gatewayTransport.close(1000, "bye"); + + expect(inner.close).toHaveBeenCalledWith(1000, "bye"); + }); + + test("close() called again before the inner transport's close event fires is a no-op", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + inner.open(); + + const onclose = vi.fn(); + gatewayTransport.onclose = onclose; + + // Like a native WebSocket, emit the close event asynchronously after close() is called + inner.close.mockImplementation((code, reason) => { + inner.readyState = SOCKET_STATES.closing; + setTimeout(() => { + inner.readyState = SOCKET_STATES.closed; + inner.onclose?.({ + code: code ?? 1000, + reason: reason ?? "", + wasClean: true, + } as CloseEvent); + }, 0); + }); + + gatewayTransport.close(1000, "bye"); + gatewayTransport.close(1000, "bye again"); + + expect(inner.close).toHaveBeenCalledExactlyOnceWith(1000, "bye"); + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.closing); + expect(onclose).not.toHaveBeenCalled(); + + await vi.runAllTimersAsync(); + + expect(onclose).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ code: 1000, reason: "bye", wasClean: true }), + ); + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.closed); + }); + + test("fetchConnectUrl rejecting fails the connection without creating an inner transport", async () => { + const error = new Error("network down"); + fetchConnectUrl.mockRejectedValueOnce(error); + + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + const onerror = vi.fn(); + const onclose = vi.fn(); + gatewayTransport.onerror = onerror; + gatewayTransport.onclose = onclose; + + await flushStart(); + + expect(onerror).toHaveBeenCalledWith(error); + expect(onclose).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + code: 1006, + reason: "gateway_connect_failed", + wasClean: false, + }), + ); + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.closed); + + // No inner transport was instantiated, as the fetch failed + expect(MockTransport.constructionCount).toBe(0); + + // Closing the failed gateway transport does not trigger another onclose event + gatewayTransport.close(); + expect(onclose).toHaveBeenCalledTimes(1); + }); + + test("close() called from onerror during a failed fetch emits a single clean close", async () => { + const error = new Error("network down"); + fetchConnectUrl.mockRejectedValueOnce(error); + + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + const onerror = vi.fn(() => gatewayTransport.close()); + const onclose = vi.fn(); + gatewayTransport.onerror = onerror; + gatewayTransport.onclose = onclose; + + await flushStart(); + + expect(onerror).toHaveBeenCalledWith(error); + + expect(onclose).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ code: 1000, reason: "", wasClean: true }), + ); + expect(gatewayTransport.readyState).toBe(SOCKET_STATES.closed); + + // No inner transport was instantiated, as the fetch failed + expect(MockTransport.constructionCount).toBe(0); + }); + + test("connection fails when the base transport cannot be instantiated", async () => { + const error = new Error("Failed to instantiate base transport"); + class ThrowingTransport { + constructor() { + throw error; + } + } + const ThrowingGatewayTransport = createGatewayTransport({ + baseTransport: ThrowingTransport as unknown as TransportConstructor, + fetchConnectUrl, + onConnectUrlInvalid, + }); + + const transport = new ThrowingGatewayTransport(PHOENIX_URL, PROTOCOLS); + const onerror = vi.fn(); + const onclose = vi.fn(); + transport.onerror = onerror; + transport.onclose = onclose; + + await flushStart(); + + expect(onerror).toHaveBeenCalledWith(error); + expect(onclose).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + code: 1006, + reason: "gateway_connect_failed", + wasClean: false, + }), + ); + expect(transport.readyState).toBe(SOCKET_STATES.closed); + + // Closing the failed gateway transport does not trigger another onclose event + transport.close(); + expect(onclose).toHaveBeenCalledTimes(1); + + expect(onConnectUrlInvalid).toHaveBeenCalledTimes(1); + }); + + test("close() after the inner transport has closed on its own is a no-op", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + + const onclose = vi.fn(); + gatewayTransport.onclose = onclose; + + inner.open(); + inner.serverClose(1006, false); + + expect(onclose).toHaveBeenCalledTimes(1); + + gatewayTransport.close(); + + expect(inner.close).not.toHaveBeenCalled(); + expect(onclose).toHaveBeenCalledTimes(1); + }); + + test("onclose is called with plain object event when CloseEvent is unsupported", () => { + vi.stubGlobal("CloseEvent", undefined); + + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + const onclose = vi.fn(); + gatewayTransport.onclose = onclose; + + gatewayTransport.close(); + + expect(onclose).toHaveBeenCalledTimes(1); + const event = onclose.mock.calls[0]?.[0]; + expect(Object.getPrototypeOf(event)).toBe(Object.prototype); + expect(event).toEqual({ code: 1000, reason: "", wasClean: true }); + }); + + test("onConnectUrlInvalid fires when the inner transport closes before ever opening", async () => { + new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + + inner.serverClose(1006, false); + + expect(onConnectUrlInvalid).toHaveBeenCalledTimes(1); + }); + + test("onConnectUrlInvalid does not fire once the inner transport has opened", async () => { + new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + + inner.open(); + inner.serverClose(1006, false); + + expect(onConnectUrlInvalid).not.toHaveBeenCalled(); + }); + + test("onConnectUrlInvalid does not fire for a close that Phoenix itself requested", async () => { + const gatewayTransport = new GatewayTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + + // Simulate Phoenix closing the socket while the inner transport is still + // connecting: the inner's own close() fires an unclean close event. + inner.close.mockImplementation((code, reason) => { + inner.readyState = SOCKET_STATES.closed; + inner.onclose?.({ + code: code ?? 1006, + reason: reason ?? "", + wasClean: false, + } as CloseEvent); + }); + + gatewayTransport.close(); + + expect(onConnectUrlInvalid).not.toHaveBeenCalled(); + }); + + test("omitting onConnectUrlInvalid does not throw when the url is invalid", async () => { + const NoCallbackTransport = createGatewayTransport({ + baseTransport: MockTransport, + fetchConnectUrl, + }); + new NoCallbackTransport(PHOENIX_URL, PROTOCOLS); + await flushStart(); + const inner = getMockTransportInstance(); + + expect(() => inner.serverClose(1006, false)).not.toThrow(); + }); + + test("the returned class is named GatewayTransport", () => { + expect(GatewayTransport.name).toBe("GatewayTransport"); + }); + + function getMockTransportInstance(): MockTransport { + expect(MockTransport.constructionCount).toBe(1); + expect(MockTransport.instance).not.toBeNull(); + return MockTransport.instance!; + } +}); diff --git a/packages/client/test/helpers.test.ts b/packages/client/test/helpers.test.ts index 46bf918a2..7e000666c 100644 --- a/packages/client/test/helpers.test.ts +++ b/packages/client/test/helpers.test.ts @@ -1,6 +1,13 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; -import { exponentialBackoffFullJitter, isValidUuid } from "../src/helpers"; +import { + appendQueryString, + CACHE_EXPIRY_PADDING_MS, + cacheUntilExpired, + exponentialBackoffFullJitter, + getQueryString, + isValidUuid, +} from "../src/helpers"; /** * Helper Functions Test Suite @@ -114,6 +121,54 @@ describe("Helper Functions", () => { }); }); + describe("getQueryString", () => { + test("returns everything after the first '?'", () => { + expect(getQueryString("https://example.com/test?hello=world")).toBe( + "hello=world", + ); + }); + + test("returns the empty string when there is no '?'", () => { + expect(getQueryString("https://example.com/test")).toBe(""); + }); + + test("returns the empty string when the '?' is the last character", () => { + expect(getQueryString("https://example.com/test?")).toBe(""); + }); + }); + + describe("appendQueryString", () => { + test("appends the query string to a URL with no existing query string", () => { + expect(appendQueryString("https://example.com/test", "a=1")).toBe( + "https://example.com/test?a=1", + ); + }); + + test("appends the query string with '&' when the URL already has one", () => { + expect( + appendQueryString("https://example.com/test?a=1", "b=2"), + ).toBe("https://example.com/test?a=1&b=2"); + }); + + test("returns the URL unchanged when the query string is empty", () => { + expect(appendQueryString("https://example.com/test", "")).toBe( + "https://example.com/test", + ); + }); + + test("returns the URL unchanged when the query string is empty, even if the URL already has one", () => { + expect(appendQueryString("https://example.com/test?a=1", "")).toBe( + "https://example.com/test?a=1", + ); + }); + + test("does not dedupe keys shared between the URL and the query string", () => { + expect( + appendQueryString("https://example.com/test?a=1", "a=2"), + ).toBe("https://example.com/test?a=1&a=2"); + }); + }); + describe("exponentialBackoffFullJitter", () => { beforeEach(() => { vi.spyOn(Math, "random"); @@ -193,5 +248,123 @@ describe("Helper Functions", () => { }), ).toBe(250); }); - }) + }); + + describe("cacheUntilExpired", () => { + const TEST_ARG = "Hello, world!"; + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(0); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + test("first get call invokes fetchFunc with the given argument", async () => { + const fetchFunc = vi.fn(async (testArg: string) => ({ + testArg, + expiresAt: 60_000, + })); + const { get } = cacheUntilExpired(fetchFunc); + + const result = await get(TEST_ARG); + + expect(fetchFunc).toHaveBeenCalledTimes(1); + expect(fetchFunc).toHaveBeenCalledWith(TEST_ARG); + expect(result).toEqual({ testArg: TEST_ARG, expiresAt: 60_000 }); + }); + + test("second call before expiry returns the cached value without refetching", async () => { + const fetchFunc = vi.fn(async (_testArg: string) => ({ + value: 1, + expiresAt: 60_000, + })); + const { get } = cacheUntilExpired(fetchFunc); + + const first = await get(TEST_ARG); + const second = await get(TEST_ARG); + + expect(fetchFunc).toHaveBeenCalledTimes(1); + expect(fetchFunc).toHaveBeenCalledWith(TEST_ARG); + expect(second).toBe(first); + }); + + test("refetches within the expiry padding window and caches the new value", async () => { + const fetchFunc = vi + .fn< + (testArg: string) => Promise<{ value: number; expiresAt: number }> + >() + .mockResolvedValueOnce({ value: 1, expiresAt: 60_000 }) + .mockResolvedValueOnce({ value: 2, expiresAt: 120_000 }); + const { get } = cacheUntilExpired(fetchFunc); + + await get(TEST_ARG); + vi.setSystemTime(60_000 - CACHE_EXPIRY_PADDING_MS); + const second = await get(TEST_ARG); + const third = await get(TEST_ARG); + + expect(fetchFunc).toHaveBeenCalledTimes(2); + expect(fetchFunc).toHaveBeenCalledWith(TEST_ARG); + expect(second).toEqual({ value: 2, expiresAt: 120_000 }); + expect(third).toBe(second); + }); + + test("refetches after expiry and caches the new value", async () => { + const fetchFunc = vi + .fn< + (testArg: string) => Promise<{ value: number; expiresAt: number }> + >() + .mockResolvedValueOnce({ value: 1, expiresAt: 60_000 }) + .mockResolvedValueOnce({ value: 2, expiresAt: 120_000 }); + const { get } = cacheUntilExpired(fetchFunc); + + await get(TEST_ARG); + vi.setSystemTime(60_001); + const second = await get(TEST_ARG); + const third = await get(TEST_ARG); + + expect(fetchFunc).toHaveBeenCalledTimes(2); + expect(fetchFunc).toHaveBeenCalledWith(TEST_ARG); + expect(second).toEqual({ value: 2, expiresAt: 120_000 }); + expect(third).toBe(second); + }); + + test("invalidate forces the next get call to refetch", async () => { + const fetchFunc = vi + .fn< + (testArg: string) => Promise<{ value: number; expiresAt: number }> + >() + .mockResolvedValueOnce({ value: 1, expiresAt: 60_000 }) + .mockResolvedValueOnce({ value: 2, expiresAt: 120_000 }); + const { get, invalidate } = cacheUntilExpired(fetchFunc); + + await get(TEST_ARG); + invalidate(); + const second = await get(TEST_ARG); + + expect(fetchFunc).toHaveBeenCalledTimes(2); + expect(fetchFunc).toHaveBeenCalledWith(TEST_ARG); + expect(second).toEqual({ value: 2, expiresAt: 120_000 }); + }); + + test("a rejected fetchFunc propagates and does not populate the cache", async () => { + const error = new Error("boom"); + const fetchFunc = vi + .fn< + (testArg: string) => Promise<{ value: number; expiresAt: number }> + >() + .mockRejectedValueOnce(error) + .mockResolvedValueOnce({ value: 1, expiresAt: 60_000 }); + const { get } = cacheUntilExpired(fetchFunc); + + await expect(get(TEST_ARG)).rejects.toThrow(error); + const second = await get(TEST_ARG); + + expect(fetchFunc).toHaveBeenCalledTimes(2); + expect(fetchFunc).toHaveBeenCalledWith(TEST_ARG); + expect(second).toEqual({ value: 1, expiresAt: 60_000 }); + }); + }); }); diff --git a/packages/client/test/test-utils/mocks.ts b/packages/client/test/test-utils/mocks.ts index f6e2f0ff3..6b1ff061b 100644 --- a/packages/client/test/test-utils/mocks.ts +++ b/packages/client/test/test-utils/mocks.ts @@ -46,7 +46,7 @@ export const createMockKnock = (apiKey = "pk_test_12345") => { vi.spyOn(knock, "client").mockReturnValue( mockApiClient as unknown as ReturnType, ); - vi.spyOn(knock, "log").mockImplementation(() => {}); + vi.spyOn(knock, "log").mockImplementation(() => { }); return { knock, mockApiClient }; }; @@ -212,21 +212,23 @@ export const setupBrowserEnvironment = () => { }; // Simple WebSocket mock -export const createWebSocketMock = () => { - const mockWebSocket = { - send: vi.fn(), - close: vi.fn(), - addEventListener: vi.fn(), - removeEventListener: vi.fn(), - readyState: 1, // OPEN - onopen: null, - onclose: null, - onmessage: null, - onerror: null, - }; - - return mockWebSocket; -}; +export const createWebSocketMock = () => + // Vitest 4 only lets a mock stand in as a constructor when its implementation is a `function` (or class), not an arrow function + vi.fn(function (url: string, protocols?: string | string[]) { + return { + url, + protocols, + readyState: 1, // OPEN + bufferedAmount: 0, + binaryType: "blob", + onopen: null, + onclose: null, + onmessage: null, + onerror: null, + send: vi.fn(), + close: vi.fn(), + }; + }); // Simple BroadcastChannel mock export const createBroadcastChannelMock = () => {