diff --git a/apps/mobile/modules/t3-websocket/expo-module.config.json b/apps/mobile/modules/t3-websocket/expo-module.config.json new file mode 100644 index 00000000000..01aee17c36a --- /dev/null +++ b/apps/mobile/modules/t3-websocket/expo-module.config.json @@ -0,0 +1,6 @@ +{ + "platforms": ["apple"], + "apple": { + "modules": ["T3WebSocketModule"] + } +} diff --git a/apps/mobile/modules/t3-websocket/ios/T3WebSocket.podspec b/apps/mobile/modules/t3-websocket/ios/T3WebSocket.podspec new file mode 100644 index 00000000000..c595e8f7bae --- /dev/null +++ b/apps/mobile/modules/t3-websocket/ios/T3WebSocket.podspec @@ -0,0 +1,19 @@ +Pod::Spec.new do |s| + s.name = 'T3WebSocket' + s.version = '1.0.0' + s.summary = 'URLSession-backed WebSocket for T3 Code mobile.' + s.description = 'WebSocket client backed by URLSessionWebSocketTask so iOS negotiates permessage-deflate, which SocketRocket (React Native default) cannot.' + s.author = 'T3 Tools' + s.homepage = 'https://t3tools.com' + s.platforms = { + :ios => '18.0', + } + s.source = { :path => '.' } + s.static_framework = true + + s.dependency 'ExpoModulesCore' + s.pod_target_xcconfig = { + 'DEFINES_MODULE' => 'YES', + } + s.source_files = '**/*.{h,m,mm,swift,hpp,cpp}' +end diff --git a/apps/mobile/modules/t3-websocket/ios/T3WebSocketModule.swift b/apps/mobile/modules/t3-websocket/ios/T3WebSocketModule.swift new file mode 100644 index 00000000000..bdbf3fe85cf --- /dev/null +++ b/apps/mobile/modules/t3-websocket/ios/T3WebSocketModule.swift @@ -0,0 +1,210 @@ +import ExpoModulesCore +import Foundation + +// WebSocket client backed by URLSessionWebSocketTask. React Native's built-in +// iOS WebSocket (SocketRocket) never offers permessage-deflate, so frames from +// the T3 server arrive uncompressed. URLSessionWebSocketTask offers the +// extension in its handshake by default, letting the server (which negotiates +// permessage-deflate since #4705) compress the JSON RPC stream. +// +// Only the surface Effect's `Socket.fromWebSocket` needs is exposed: connect, +// send text/binary, close, and open/message/close/error events routed by a +// JS-assigned connection id. +public final class T3WebSocketModule: Module { + private let connections = T3WebSocketConnections() + + public func definition() -> ModuleDefinition { + Name("T3WebSocket") + + Events("onOpen", "onMessage", "onClose", "onError") + + Function("connect") { (id: Int, url: String, protocols: [String]) throws in + guard let parsedUrl = URL(string: url) else { + throw Exception(name: "InvalidUrl", description: "Invalid WebSocket URL: \(url)") + } + self.connections.open(id: id, url: parsedUrl, protocols: protocols) { [weak self] event, payload in + self?.sendEvent(event, payload) + } + } + + Function("sendText") { (id: Int, text: String) in + self.connections.send(id: id, message: .string(text)) + } + + Function("sendBinary") { (id: Int, base64: String) in + guard let data = Data(base64Encoded: base64) else { return } + self.connections.send(id: id, message: .data(data)) + } + + Function("close") { (id: Int, code: Int, reason: String) in + self.connections.close(id: id, code: code, reason: reason) + } + + OnDestroy { + self.connections.cancelAll() + } + } +} + +private typealias T3WebSocketEmitter = (_ event: String, _ payload: [String: Any]) -> Void + +private final class T3WebSocketConnections: NSObject, URLSessionWebSocketDelegate { + // Snapshot frames on the socket-fallback path can reach tens of MB; the + // URLSession default receive cap is 1 MB, which would abort the connection. + private static let maximumMessageSize = 128 * 1024 * 1024 + + private let queue = DispatchQueue(label: "com.t3tools.t3code.websocket") + private var tasksById: [Int: URLSessionWebSocketTask] = [:] + private var idsByTaskIdentifier: [Int: Int] = [:] + private var emittersById: [Int: T3WebSocketEmitter] = [:] + + // Held so teardown can invalidate it: URLSession retains its delegate (self) + // until invalidated, so skipping that leaks this object and the session on + // every module destroy/recreate cycle. Only touched from `queue`. + private var activeSession: URLSession? + + private var session: URLSession { + if let activeSession { return activeSession } + let configuration = URLSessionConfiguration.default + // Dev clients (and network debuggers) register custom URLProtocol classes + // that intercept URLSession traffic. They do not understand the WebSocket + // upgrade and drop the connection right after the 101 (NSURLErrorDomain + // -1005). WebSocket traffic has no reason to go through them. + configuration.protocolClasses = [] + let session = URLSession(configuration: configuration, delegate: self, delegateQueue: nil) + activeSession = session + return session + } + + func open(id: Int, url: URL, protocols: [String], emitter: @escaping T3WebSocketEmitter) { + queue.async { + let task = protocols.isEmpty + ? self.session.webSocketTask(with: url) + : self.session.webSocketTask(with: url, protocols: protocols) + task.maximumMessageSize = Self.maximumMessageSize + self.tasksById[id] = task + self.idsByTaskIdentifier[task.taskIdentifier] = id + self.emittersById[id] = emitter + task.resume() + self.receiveLoop(id: id, task: task) + } + } + + func send(id: Int, message: URLSessionWebSocketTask.Message) { + queue.async { + guard let task = self.tasksById[id] else { return } + task.send(message) { [weak self] error in + guard let self, let error else { return } + self.queue.async { + self.emitLocked(id: id, event: "onError", payload: ["message": error.localizedDescription]) + } + } + } + } + + func close(id: Int, code: Int, reason: String) { + queue.async { + guard let task = self.tasksById[id] else { return } + let closeCode = URLSessionWebSocketTask.CloseCode(rawValue: code) ?? .normalClosure + task.cancel(with: closeCode, reason: reason.data(using: .utf8)) + } + } + + func cancelAll() { + queue.async { + for task in self.tasksById.values { + task.cancel(with: .goingAway, reason: nil) + } + self.tasksById.removeAll() + self.idsByTaskIdentifier.removeAll() + self.emittersById.removeAll() + // Releases the session's reference to this delegate. A later `open` call + // builds a fresh session rather than using an invalidated one. + self.activeSession?.invalidateAndCancel() + self.activeSession = nil + } + } + + private func receiveLoop(id: Int, task: URLSessionWebSocketTask) { + task.receive { [weak self] result in + guard let self else { return } + self.queue.async { + switch result { + case .success(let message): + switch message { + case .string(let text): + self.emitLocked(id: id, event: "onMessage", payload: ["data": text, "binary": false]) + case .data(let data): + self.emitLocked(id: id, event: "onMessage", payload: ["data": data.base64EncodedString(), "binary": true]) + @unknown default: + break + } + guard self.tasksById[id] != nil else { return } + self.receiveLoop(id: id, task: task) + case .failure: + // The delegate close/error callbacks own failure reporting; the loop + // just stops so it does not spin on a dead task. + break + } + } + } + } + + // Must be called on `queue`. + private func emitLocked(id: Int, event: String, payload: [String: Any]) { + guard let emitter = emittersById[id] else { return } + var enriched = payload + enriched["id"] = id + emitter(event, enriched) + } + + private func remove(id: Int, taskIdentifier: Int) { + tasksById.removeValue(forKey: id) + idsByTaskIdentifier.removeValue(forKey: taskIdentifier) + emittersById.removeValue(forKey: id) + } + + // MARK: - URLSessionWebSocketDelegate + + func urlSession( + _ session: URLSession, + webSocketTask: URLSessionWebSocketTask, + didOpenWithProtocol protocol: String? + ) { + queue.async { + guard let id = self.idsByTaskIdentifier[webSocketTask.taskIdentifier] else { return } + self.emitLocked(id: id, event: "onOpen", payload: ["protocol": `protocol` ?? ""]) + } + } + + func urlSession( + _ session: URLSession, + webSocketTask: URLSessionWebSocketTask, + didCloseWith closeCode: URLSessionWebSocketTask.CloseCode, + reason: Data? + ) { + queue.async { + guard let id = self.idsByTaskIdentifier[webSocketTask.taskIdentifier] else { return } + let reasonText = reason.flatMap { String(data: $0, encoding: .utf8) } ?? "" + self.emitLocked(id: id, event: "onClose", payload: ["code": closeCode.rawValue, "reason": reasonText]) + self.remove(id: id, taskIdentifier: webSocketTask.taskIdentifier) + } + } + + func urlSession( + _ session: URLSession, + task: URLSessionTask, + didCompleteWithError error: Error? + ) { + queue.async { + guard let id = self.idsByTaskIdentifier[task.taskIdentifier] else { return } + if let error { + self.emitLocked(id: id, event: "onError", payload: ["message": error.localizedDescription]) + // Abnormal termination: no close frame arrived, mirror the browser's + // 1006 so Effect's close-code classification treats it as an error. + self.emitLocked(id: id, event: "onClose", payload: ["code": 1006, "reason": error.localizedDescription]) + } + self.remove(id: id, taskIdentifier: task.taskIdentifier) + } + } +} diff --git a/apps/mobile/src/lib/nativeWebSocket.test.ts b/apps/mobile/src/lib/nativeWebSocket.test.ts new file mode 100644 index 00000000000..f4fa9a0724a --- /dev/null +++ b/apps/mobile/src/lib/nativeWebSocket.test.ts @@ -0,0 +1,181 @@ +import { beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +const mocks = vi.hoisted(() => { + type Listener = (payload: Record) => void; + const nativeListeners = new Map(); + const calls: Array<{ method: string; args: unknown[] }> = []; + let platformOS = "ios"; + let moduleAvailable = true; + + const nativeModule = { + connect: (...args: unknown[]) => calls.push({ method: "connect", args }), + sendText: (...args: unknown[]) => calls.push({ method: "sendText", args }), + sendBinary: (...args: unknown[]) => calls.push({ method: "sendBinary", args }), + close: (...args: unknown[]) => calls.push({ method: "close", args }), + addListener: (eventName: string, listener: Listener) => { + nativeListeners.set(eventName, listener); + return { remove: () => nativeListeners.delete(eventName) }; + }, + }; + + return { + calls, + nativeListeners, + emit: (eventName: string, payload: Record) => { + nativeListeners.get(eventName)?.(payload); + }, + setPlatform: (os: string) => { + platformOS = os; + }, + setModuleAvailable: (available: boolean) => { + moduleAvailable = available; + }, + getPlatform: () => platformOS, + getModule: () => (moduleAvailable ? nativeModule : null), + // Native listeners intentionally survive reset: the wrapper attaches them + // once per app lifetime, so clearing them here would not match production. + reset: () => { + calls.length = 0; + platformOS = "ios"; + moduleAvailable = true; + }, + }; +}); + +vi.mock("react-native", () => ({ + Platform: { + get OS() { + return mocks.getPlatform(); + }, + }, +})); + +vi.mock("expo", () => ({ + requireOptionalNativeModule: () => mocks.getModule(), +})); + +import { loadNativeWebSocketConstructor } from "./nativeWebSocket"; + +async function openSocket() { + const construct = await loadNativeWebSocketConstructor(); + expect(construct).not.toBeNull(); + return construct!("wss://example.test/ws"); +} + +function lastCall(method: string) { + const entries = mocks.calls.filter((call) => call.method === method); + return entries[entries.length - 1]; +} + +function connectionIdOf(socket: WebSocket): number { + void socket; + return lastCall("connect")!.args[0] as number; +} + +describe("nativeWebSocketConstructor", () => { + beforeEach(() => { + mocks.reset(); + }); + + it("resolves null on Android and when the native module is missing", async () => { + mocks.setPlatform("android"); + expect(await loadNativeWebSocketConstructor()).toBeNull(); + + mocks.setPlatform("ios"); + mocks.setModuleAvailable(false); + expect(await loadNativeWebSocketConstructor()).toBeNull(); + }); + + it("connects and transitions readyState on open", async () => { + const socket = await openSocket(); + const id = connectionIdOf(socket); + expect(lastCall("connect")!.args).toEqual([id, "wss://example.test/ws", []]); + expect(socket.readyState).toBe(0); + + const opened = vi.fn(); + socket.addEventListener("open", opened, { once: true }); + mocks.emit("onOpen", { id }); + expect(socket.readyState).toBe(1); + expect(opened).toHaveBeenCalledTimes(1); + + // `once` listeners must not fire again. + mocks.emit("onOpen", { id }); + expect(opened).toHaveBeenCalledTimes(1); + }); + + it("delivers text messages as strings and binary messages as bytes", async () => { + const socket = await openSocket(); + const id = connectionIdOf(socket); + const received: unknown[] = []; + socket.addEventListener("message", (event) => { + received.push((event as { data: unknown }).data); + }); + + mocks.emit("onMessage", { id, data: "hello", binary: false }); + mocks.emit("onMessage", { id, data: btoa("\x01\x02\x03"), binary: true }); + + expect(received[0]).toBe("hello"); + expect(received[1]).toBeInstanceOf(Uint8Array); + expect([...(received[1] as Uint8Array)]).toEqual([1, 2, 3]); + }); + + it("sends binary data base64-encoded and strings as text", async () => { + const socket = await openSocket(); + const id = connectionIdOf(socket); + + socket.send("ping"); + expect(lastCall("sendText")!.args).toEqual([id, "ping"]); + + socket.send(new Uint8Array([1, 2, 3])); + expect(lastCall("sendBinary")!.args).toEqual([id, btoa("\x01\x02\x03")]); + }); + + it("dispatches close once and routes messages by connection id", async () => { + const first = await openSocket(); + const firstId = connectionIdOf(first); + const second = await openSocket(); + const secondId = connectionIdOf(second); + expect(firstId).not.toBe(secondId); + + const firstClosed = vi.fn(); + const secondClosed = vi.fn(); + first.addEventListener("close", firstClosed); + second.addEventListener("close", secondClosed); + + mocks.emit("onClose", { id: firstId, code: 1000, reason: "done" }); + mocks.emit("onClose", { id: firstId, code: 1000, reason: "done" }); + + expect(firstClosed).toHaveBeenCalledTimes(1); + expect(first.readyState).toBe(3); + expect(secondClosed).not.toHaveBeenCalled(); + expect(second.readyState).toBe(0); + }); + + it("close() is idempotent and forwards code and reason", async () => { + const socket = await openSocket(); + const id = connectionIdOf(socket); + + socket.close(4000, "bye"); + socket.close(4000, "bye"); + + const closeCalls = mocks.calls.filter((call) => call.method === "close"); + expect(closeCalls).toHaveLength(1); + expect(closeCalls[0]!.args).toEqual([id, 4000, "bye"]); + expect(socket.readyState).toBe(2); + }); + + it("ignores a native open that lands after close", async () => { + const socket = await openSocket(); + const id = connectionIdOf(socket); + const opened = vi.fn(); + socket.addEventListener("open", opened); + + // Effect closes while still CONNECTING (open timeout or scope teardown); + // the in-flight native connect can still report success afterwards. + socket.close(1000, ""); + mocks.emit("onOpen", { id }); + + expect(opened).not.toHaveBeenCalled(); + expect(socket.readyState).toBe(2); + }); +}); diff --git a/apps/mobile/src/lib/nativeWebSocket.ts b/apps/mobile/src/lib/nativeWebSocket.ts new file mode 100644 index 00000000000..e6d7509a485 --- /dev/null +++ b/apps/mobile/src/lib/nativeWebSocket.ts @@ -0,0 +1,226 @@ +import { Platform } from "react-native"; + +// URLSession-backed WebSocket for iOS. React Native's built-in WebSocket uses +// SocketRocket on iOS, which never offers permessage-deflate, so server frames +// arrive uncompressed. URLSessionWebSocketTask offers the extension by default. +// Android needs none of this: React Native's OkHttp-backed WebSocket already +// negotiates permessage-deflate on its own. +// +// Only the surface Effect's `Socket.fromWebSocket` drives is implemented: +// addEventListener/removeEventListener for open/message/close/error (with +// `once` support), `readyState`, `send`, and `close`. + +interface NativeEventSubscription { + remove(): void; +} + +interface T3WebSocketNativeModule { + connect(id: number, url: string, protocols: string[]): void; + sendText(id: number, text: string): void; + sendBinary(id: number, base64: string): void; + close(id: number, code: number, reason: string): void; + addListener( + eventName: "onOpen" | "onMessage" | "onClose" | "onError", + listener: (payload: { + id: number; + data?: string; + binary?: boolean; + code?: number; + reason?: string; + message?: string; + }) => void, + ): NativeEventSubscription; +} + +type NativeWebSocketEventType = "open" | "message" | "close" | "error"; + +interface ListenerEntry { + readonly listener: (event: unknown) => void; + readonly once: boolean; +} + +const activeSockets = new Map(); +let nextSocketId = 1; +let nativeListenersAttached = false; + +function attachNativeListeners(module: T3WebSocketNativeModule) { + if (nativeListenersAttached) { + return; + } + nativeListenersAttached = true; + module.addListener("onOpen", (payload) => { + activeSockets.get(payload.id)?.handleOpen(); + }); + module.addListener("onMessage", (payload) => { + activeSockets.get(payload.id)?.handleMessage(payload.data ?? "", payload.binary === true); + }); + module.addListener("onClose", (payload) => { + activeSockets.get(payload.id)?.handleClose(payload.code ?? 1006, payload.reason ?? ""); + }); + module.addListener("onError", (payload) => { + activeSockets.get(payload.id)?.handleError(payload.message ?? "WebSocket error"); + }); +} + +function base64ToUint8Array(base64: string): Uint8Array { + const binary = atob(base64); + const bytes = new Uint8Array(binary.length); + for (let index = 0; index < binary.length; index += 1) { + bytes[index] = binary.charCodeAt(index); + } + return bytes; +} + +function bytesToBase64(bytes: Uint8Array): string { + let binary = ""; + const chunkSize = 8192; + for (let index = 0; index < bytes.length; index += chunkSize) { + binary += String.fromCharCode(...bytes.subarray(index, index + chunkSize)); + } + return btoa(binary); +} + +class NativeWebSocket { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + + readyState: number = NativeWebSocket.CONNECTING; + + private readonly id: number; + private readonly module: T3WebSocketNativeModule; + private readonly listeners = new Map>(); + + constructor(module: T3WebSocketNativeModule, url: string, protocols?: string | Array) { + this.module = module; + this.id = nextSocketId++; + activeSockets.set(this.id, this); + attachNativeListeners(module); + const protocolList = + typeof protocols === "string" ? [protocols] : protocols ? [...protocols] : []; + try { + module.connect(this.id, url, protocolList); + } catch (error) { + activeSockets.delete(this.id); + throw error; + } + } + + addEventListener( + type: NativeWebSocketEventType, + listener: (event: unknown) => void, + options?: { once?: boolean }, + ): void { + let entries = this.listeners.get(type); + if (!entries) { + entries = new Set(); + this.listeners.set(type, entries); + } + entries.add({ listener, once: options?.once === true }); + } + + removeEventListener(type: NativeWebSocketEventType, listener: (event: unknown) => void): void { + const entries = this.listeners.get(type); + if (!entries) { + return; + } + for (const entry of entries) { + if (entry.listener === listener) { + entries.delete(entry); + } + } + } + + send(data: string | Uint8Array | ArrayBuffer): void { + if (typeof data === "string") { + this.module.sendText(this.id, data); + return; + } + const bytes = data instanceof Uint8Array ? data : new Uint8Array(data); + this.module.sendBinary(this.id, bytesToBase64(bytes)); + } + + close(code?: number, reason?: string): void { + if (this.readyState === NativeWebSocket.CLOSING || this.readyState === NativeWebSocket.CLOSED) { + return; + } + this.readyState = NativeWebSocket.CLOSING; + this.module.close(this.id, code ?? 1000, reason ?? ""); + } + + handleOpen(): void { + // `close()` during CONNECTING (an Effect open-timeout or scope teardown) + // leaves a native connect in flight, so its `didOpen` can still land. Only + // a socket that is still connecting may transition to OPEN; otherwise the + // close would be silently undone and open handlers would fire on a socket + // the caller already gave up on. + if (this.readyState !== NativeWebSocket.CONNECTING) { + return; + } + this.readyState = NativeWebSocket.OPEN; + this.dispatch("open", { type: "open" }); + } + + handleMessage(data: string, binary: boolean): void { + this.dispatch("message", { + type: "message", + data: binary ? base64ToUint8Array(data) : data, + }); + } + + handleClose(code: number, reason: string): void { + if (this.readyState === NativeWebSocket.CLOSED) { + return; + } + this.readyState = NativeWebSocket.CLOSED; + activeSockets.delete(this.id); + this.dispatch("close", { type: "close", code, reason }); + } + + handleError(message: string): void { + this.dispatch("error", new Error(message)); + } + + private dispatch(type: NativeWebSocketEventType, event: unknown): void { + const entries = this.listeners.get(type); + if (!entries) { + return; + } + for (const entry of entries) { + if (entry.once) { + entries.delete(entry); + } + entry.listener(event); + } + } +} + +/** + * Loads the WebSocket constructor backed by URLSessionWebSocketTask, resolving + * `null` when unavailable (Android, or an iOS binary predating the native + * module). Callers fall back to the global WebSocket. + * + * Async because "expo" is imported lazily: its module-scope setup code assumes + * a React Native environment, and this file's importer (runtime.ts) sits in + * many test import chains. + */ +export async function loadNativeWebSocketConstructor(): Promise< + ((url: string, protocols?: string | Array) => globalThis.WebSocket) | null +> { + if (Platform.OS !== "ios") { + return null; + } + let module: T3WebSocketNativeModule | null; + try { + const expo = await import("expo"); + module = expo.requireOptionalNativeModule("T3WebSocket"); + } catch { + return null; + } + if (!module) { + return null; + } + return (url, protocols) => + new NativeWebSocket(module, url, protocols) as unknown as globalThis.WebSocket; +} diff --git a/apps/mobile/src/lib/runtime.ts b/apps/mobile/src/lib/runtime.ts index 98730edfbfc..7349bab0909 100644 --- a/apps/mobile/src/lib/runtime.ts +++ b/apps/mobile/src/lib/runtime.ts @@ -1,3 +1,4 @@ +import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as ManagedRuntime from "effect/ManagedRuntime"; import * as Socket from "effect/unstable/socket/Socket"; @@ -9,6 +10,7 @@ import { managedRelayClientLayer } from "../features/cloud/managedRelayLayer"; import { resolveCloudPublicConfig } from "../features/cloud/publicConfig"; import { tracingLayer } from "../features/observability/tracing"; import * as Persistence from "../persistence/layer"; +import { loadNativeWebSocketConstructor } from "./nativeWebSocket"; function configuredRelayUrl(): string { return resolveCloudPublicConfig().relay.url ?? "http://relay.invalid"; @@ -16,6 +18,19 @@ function configuredRelayUrl(): string { const httpClientLayer = remoteHttpClientLayer(fetch); +// iOS swaps in a URLSessionWebSocketTask-backed constructor so the socket +// negotiates permessage-deflate (SocketRocket, behind the global WebSocket, +// cannot). Android's global WebSocket (OkHttp) already negotiates it. +const webSocketConstructorLayer: Layer.Layer = Layer.unwrap( + Effect.promise(loadNativeWebSocketConstructor).pipe( + Effect.map((nativeConstructor) => + nativeConstructor === null + ? Socket.layerWebSocketConstructorGlobal + : Layer.succeed(Socket.WebSocketConstructor, nativeConstructor), + ), + ), +); + type RuntimeLayerSource = | ReturnType | typeof Socket.layerWebSocketConstructorGlobal @@ -26,7 +41,7 @@ type RuntimeLayerSource = const runtimeLayer = Layer.merge( managedRelayClientLayer(configuredRelayUrl()), - Socket.layerWebSocketConstructorGlobal, + webSocketConstructorLayer, ).pipe( Layer.provideMerge(cryptoLayer), Layer.provideMerge(httpClientLayer),