import type { WsClientConfig, WsSendOptions, NotifyFn, TranslateFn, ReactiveRef, CreateRefFn, } from './types' import { plainRef } from './types' // --------------------------------------------------------------------------- // Types // --------------------------------------------------------------------------- interface WsSocket extends WebSocket { socket_id?: string } /** Links a `sendTimeout` deadline to the send it bounds. */ interface SendControl { timedOut: boolean /** Drops the queued payload and removes the response listeners. */ cancel: (() => void) | null } /** A send written to a socket, tracked for `rejectOnDisconnect`. */ interface InflightSend { socket: WsSocket | null fail: ((err: { message: string; code: string }) => void) | null lost: { message: string; code: string } | null } export interface WsChannel { name: string is_established: boolean establish(): Promise send(event: string, data?: any): Promise unsubscribe(): Promise } export interface WsClient { socket: WsSocket | null channels: WsChannel[] is_opened: ReactiveRef is_setup: ReactiveRef is_connecting_socket: ReactiveRef is_after_lost_connection: ReactiveRef heartbeat: ReturnType | null last_reconnect_try: number send_queue: any[] connect(force_reset?: boolean): Promise ensureConnected(): Promise channel(channel_name?: string | null): Promise /** The 4th argument is a progress callback, or `{ progress, timeout }`. */ send( event: string, data?: object, channel_name?: string | null, progress?: ((data: any) => void) | WsSendOptions, ): Promise unsubscribe(channel_name?: string | null): Promise /** * Listen for a WS event. Returns an unsubscribe function. * Works with any framework's cleanup (React useEffect, Vue onUnmounted, etc.) */ listen(event: string, channel_name: string | null | undefined, callback: (data: any) => void): () => void /** Resolve once when the given event fires. */ listenOnce(event: string, channel_name?: string | null): Promise /** Signal that app initialization is complete. Unblocks gated send() calls. */ setAppReady(): void /** Force channels to re-subscribe with the current auth token on next send(). */ resetConnection(): void /** Update configuration at runtime. */ configure(partial: Partial): void /** Close the connection, clear intervals, and reset all state. */ destroy(): void } // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- const _isProtocolEvent = (event: string): boolean => /[.:](?:subscribe|unsubscribe|ping|pong)$/.test(event) // --------------------------------------------------------------------------- // SSR stub — safe no-op returned when isServer is true // --------------------------------------------------------------------------- function createSsrStub(createRef: CreateRefFn): WsClient { return { socket: null, channels: [], is_opened: createRef(false), is_setup: createRef(false), is_connecting_socket: createRef(false), is_after_lost_connection: createRef(false), heartbeat: null, last_reconnect_try: 0, send_queue: [], connect: () => Promise.resolve(), ensureConnected: () => Promise.resolve(), channel: () => Promise.resolve(undefined), send: () => Promise.resolve(null as any), unsubscribe: () => Promise.resolve(true), listen: () => () => {}, listenOnce: () => Promise.resolve(null), setAppReady: () => {}, resetConnection: () => {}, configure: () => {}, destroy: () => {}, } } // --------------------------------------------------------------------------- // Channel // --------------------------------------------------------------------------- class WebsocketChannel implements WsChannel { name: string is_established = false _establishPromise: Promise | null = null private _ws: WsClientImpl constructor(name: string, ws: WsClientImpl) { this.name = name this._ws = ws ws.channels.push(this) } async establish(): Promise { if (this.is_established && this._ws.is_setup.value) return this this._establishPromise ??= this._doEstablish() return this._establishPromise } private async _doEstablish(): Promise { try { await this._ws.ensureConnected() const authtoken = this._ws._getAuthToken() await this._ws.send( 'websocket.subscribe', { channel: this.name, authtoken: authtoken ?? undefined }, null, ) this.is_established = true this._ws.is_setup.value = true return this } catch (error) { this.is_established = false this._establishPromise = null throw error } } async send(event: string, data: any = {}): Promise { if (!_isProtocolEvent(event) && !this.is_established) await this.establish() return this._ws.send(event, data, _isProtocolEvent(event) ? null : this.name) } async unsubscribe(): Promise { if (this.is_established) { await this._ws .send('websocket.unsubscribe', { channel: this.name }, this.name) .catch(() => {}) } this._ws.channels = this._ws.channels.filter((c) => c !== this) return true } } // --------------------------------------------------------------------------- // Main WS client // --------------------------------------------------------------------------- class WsClientImpl extends EventTarget implements WsClient { private _config: WsClientConfig private _notify: NotifyFn | undefined private _translate: TranslateFn | undefined socket: WsSocket | null = null channels: WsChannel[] = [] is_opened: ReactiveRef is_setup: ReactiveRef is_connecting_socket: ReactiveRef is_after_lost_connection: ReactiveRef heartbeat: ReturnType | null = null last_reconnect_try = 0 send_queue: any[] = [] // App-readiness gate _appReady = false private _appReadyResolve: (() => void) | null = null private _appReadyPromise: Promise // Connection-ready promise private _connectedResolve: (() => void) | null = null private _connectedPromise: Promise | null = null // Connect promise coalescing private _connectPromise: Promise | null = null // A connect() that landed on the reconnect throttle is retried when the // throttle window opens (instead of being dropped). private _deferredConnectTimer: ReturnType | null = null // Auto-reconnect scheduled by a close; destroy() cancels it. private _reconnectTimer: ReturnType | null = null // rejectOnDisconnect: sends written to a socket, keyed by that socket. private _inflight = new Set() private _payloadSend = new WeakMap() // sendTimeout: queued payloads whose send already timed out; never written. private _timedOutPayloads = new WeakSet() constructor(config: WsClientConfig, createRef: CreateRefFn) { super() this._config = { defaultChannel: 'websocket', heartbeatInterval: 20_000, reconnectDelay: 3000, reconnectThrottle: 3000, autoReconnect: true, showConnectionNotifications: true, ...config, } this._notify = config.notify this._translate = config.translate this.is_opened = createRef(false) this.is_setup = createRef(false) this.is_connecting_socket = createRef(false) this.is_after_lost_connection = createRef(false) this._appReadyPromise = new Promise((r) => { this._appReadyResolve = r }) } // ------------------------------------------------------------------------- // Configuration // ------------------------------------------------------------------------- configure(partial: Partial): void { Object.assign(this._config, partial) if (partial.notify !== undefined) this._notify = partial.notify if (partial.translate !== undefined) this._translate = partial.translate } /** @internal — called by channels to obtain the current auth token */ _getAuthToken(): string | null | undefined { return this._config.getAuthToken?.() } private _isNativePlatform(): boolean { const v = this._config.isNativePlatform return typeof v === 'function' ? v() : (v ?? false) } private _getUrl(): string { const u = this._config.url return typeof u === 'function' ? u() : u } private _t(key: string, fallback: string): string { if (this._translate) { const result = this._translate(key) if (result) return result } return fallback } private _shouldNotify(): boolean { return ( this._config.showConnectionNotifications !== false && !this._isNativePlatform() && !!this._notify ) } // ------------------------------------------------------------------------- // Connection lifecycle // ------------------------------------------------------------------------- ensureConnected(): Promise { if (this.socket?.socket_id && this.is_opened.value) return Promise.resolve() // One waiter per outage. It is resolved by whichever connection attempt // succeeds next — a failed attempt must not orphan it (that left every // send queued behind a failed first connect hanging forever). this._connectedPromise ??= new Promise((r) => { this._connectedResolve = r }) this.connect().catch(() => {}) return this._connectedPromise } setAppReady(): void { this._appReady = true this._appReadyResolve?.() } resetConnection(): void { for (const ch of this.channels) { const c = ch as WebsocketChannel c.is_established = false c._establishPromise = null } this.is_setup.value = false } async connect(force_reset = false): Promise { if (force_reset && this.socket) { try { this.socket.close() } catch {} this.socket = null this._connectPromise = null } if (this.socket && this.socket.readyState !== WebSocket.CLOSED) return this.socket if (this._connectPromise) return this._connectPromise const throttle = this._config.reconnectThrottle ?? 3000 const sinceLast = Date.now() - this.last_reconnect_try if (this.last_reconnect_try && sinceLast < throttle) { console.log('[ws] Reconnect too fast, deferring') this._deferConnect(throttle - sinceLast) return } this.last_reconnect_try = Date.now() this._connectPromise = this._doConnect(force_reset) return this._connectPromise } private _doConnect(force_reset: boolean): Promise { const hbInterval = this._config.heartbeatInterval ?? 20_000 if (force_reset || !this.heartbeat) { if (this.heartbeat) clearInterval(this.heartbeat) this.heartbeat = setInterval(() => { if (this.socket?.readyState === WebSocket.OPEN) { this.socket.send('{"event":"websocket.ping","data":{}}') } }, hbInterval) } if (force_reset) this.channels = [] const url = this._getUrl() console.debug('[blax-networking] Connecting to', url) this.socket = new WebSocket(url) as WsSocket this.is_connecting_socket.value = true this._config.onConnectionStateChange?.('connecting') return new Promise((resolve, reject) => { const socket = this.socket! socket.addEventListener('error', () => { // A replaced (connect(true)) or destroyed socket must not touch the // live connection's state. if (this.socket === socket) { this.is_connecting_socket.value = false this._connectPromise = null } reject(new Error('WebSocket error')) }) socket.addEventListener('close', () => { this._failInflight(socket) // Stale socket (replaced by connect(true), or destroy()ed): its late // close must not wipe the live socket or schedule a reconnect. if (this.socket !== socket) { reject(new Error('Socket closed')) return } this.channels = [] this.is_connecting_socket.value = false this.is_opened.value = false this.is_setup.value = false this.socket = null this._connectPromise = null // Keep _connectedPromise / _connectedResolve: callers waiting in // ensureConnected() are resolved by the next successful connection. this.is_after_lost_connection.value = true if (this._shouldNotify()) { const text = this._t('websocket.connectionlost', 'Connection lost. Reconnecting…') this._notify!({ id: 'websocket-connection-state', type: 'info', text, timeout: 50_000 }) } this._config.onConnectionStateChange?.('disconnected') if (this._appReady && this._config.autoReconnect !== false) { const delay = this._config.reconnectDelay ?? 3000 this._config.onConnectionStateChange?.('reconnecting') if (this._reconnectTimer) clearTimeout(this._reconnectTimer) this._reconnectTimer = setTimeout(() => { this._reconnectTimer = null this.connect().catch(() => {}) }, delay) } reject(new Error('Socket closed')) }) socket.addEventListener('open', () => { if (this.is_after_lost_connection.value && this._shouldNotify()) { const text = this._t('websocket.connectionrestored', 'Connection restored') this._notify!({ id: 'websocket-connection-state', type: 'success', text, timeout: 1000 }) } this.is_opened.value = true this._config.onConnectionStateChange?.('connected') // Warmup ping socket.send('{"event":"websocket.ping","data":{}}') // Detect missing connection_established response (usually app key mismatch) setTimeout(() => { if (this.socket === socket && !socket.socket_id) { const wsUrl = this._getUrl() console.error( '[blax-networking] WebSocket opened but server did not send ' + '"websocket.connection_established" within 5s.\n' + 'Connected to: ' + wsUrl + '\n' + 'This usually means the app key in the URL does not match any app ' + 'configured on the backend (PUSHER_APP_KEY). Check that the path ' + 'segment after /app/ matches the server\'s PUSHER_APP_KEY.', ) } }, 5000) }) socket.addEventListener('message', (raw) => { const msg = JSON.parse(raw.data) if (msg?.event === 'websocket.connection_established') { const data = JSON.parse(msg.data) if (data?.socket_id && this.socket === socket) { this.socket.socket_id = data.socket_id this.is_connecting_socket.value = false resolve(this.socket) const waiting = this._connectedResolve this._connectedPromise = null this._connectedResolve = null waiting?.() // Proactively establish the default channel this.channel() this._workSendQueue() } return } // Parse stringified data payloads if (msg?.data && typeof msg.data === 'string') { try { msg.data = JSON.parse(msg.data) } catch {} } this.dispatchEvent( new CustomEvent(msg.event, { detail: { event: msg.event, data: msg.data, channel: msg.channel }, }), ) }) }) } // ------------------------------------------------------------------------- // Channel management // ------------------------------------------------------------------------- async channel(channel_name: string | null = null): Promise { channel_name ??= this._config.defaultChannel ?? 'websocket' const existing = this.channels.find((c) => c.name === channel_name) return (existing ?? new WebsocketChannel(channel_name, this)).establish() } private async _workSendQueue(): Promise { if (!this.send_queue.length) return const queue = this.send_queue this.send_queue = [] for (const payload of queue) { if (this._timedOutPayloads.has(payload)) continue await this.channel(payload.channel) if (this._timedOutPayloads.has(payload)) continue this.socket?.send(JSON.stringify(payload)) const tracked = this._payloadSend.get(payload) if (tracked) tracked.socket = this.socket } } // ------------------------------------------------------------------------- // Send / receive // ------------------------------------------------------------------------- async send( event: string, data: object = {}, channel_name: string | null = null, progress?: ((data: any) => void) | WsSendOptions, ): Promise { const opts: WsSendOptions = typeof progress === 'function' ? { progress } : (progress ?? {}) const timeout = opts.timeout ?? this._config.sendTimeout ?? 0 if (!(timeout > 0) || _isProtocolEvent(event)) { return this._send(event, data, channel_name, opts.progress, true, null) } const ctl: SendControl = { timedOut: false, cancel: null } let timer: ReturnType | undefined const deadline = new Promise((_, reject) => { timer = setTimeout(() => { ctl.timedOut = true ctl.cancel?.() reject({ message: this._t('websocket.requesttimeout', 'The server did not answer in time.'), code: 'timeout', }) }, timeout) }) return Promise.race([ this._send(event, data, channel_name, opts.progress, true, ctl), deadline, ]).finally(() => clearTimeout(timer)) } private async _send( event: string, data: object, channel_name: string | null, progress: ((data: any) => void) | undefined, _retryOnSubscriptionLost: boolean, ctl: SendControl | null, ): Promise { // Gate non-protocol events until app signals readiness if (!this._appReady && !_isProtocolEvent(event)) { await this._appReadyPromise } channel_name ??= this._config.defaultChannel ?? 'websocket' if (!this.socket) await this.connect() // Build unique event suffix so the server response can be matched to this call let sendingevent: string if (event === 'websocket.subscribe') { sendingevent = 'websocket.subscribe' channel_name = null } else { sendingevent = event + '[' + Math.random().toString(36).substring(7) + ']' } const payload = { event: sendingevent, data, channel: channel_name } // rejectOnDisconnect: remember which socket this send went out on, so a // close of that socket rejects it instead of leaving it pending forever. const tracked: InflightSend | null = this._config.rejectOnDisconnect === true && !_isProtocolEvent(event) ? { socket: null, fail: null, lost: null } : null if (tracked) { this._inflight.add(tracked) this._payloadSend.set(payload, tracked) } // sendTimeout: a payload still sitting in the send queue is never written. const dropQueued = () => { this._timedOutPayloads.add(payload) if (tracked) this._inflight.delete(tracked) } if (ctl) ctl.cancel = dropQueued try { // Ensure the target channel is subscribed if (channel_name && !_isProtocolEvent(event)) { await this.channel(channel_name) } // The caller gave up while the channel was being established. if (ctl?.timedOut) throw new Error('Send timed out before it was written') // Send or queue if (this.is_opened.value && this.socket?.readyState === WebSocket.OPEN) { this.socket.send(JSON.stringify(payload)) if (tracked) tracked.socket = this.socket } else { this.send_queue.push(payload) } await this.connect() if (!this.socket) throw new Error('Socket not connected') } catch (error) { if (tracked) this._inflight.delete(tracked) throw error } const startTime = typeof performance !== 'undefined' ? performance.now() : Date.now() return new Promise((resolve, reject) => { const cleanup = () => { this.removeEventListener(sendingevent + ':progress', handler) this.removeEventListener(sendingevent + ':error', handler) this.removeEventListener(sendingevent + ':response', handler) if (tracked) this._inflight.delete(tracked) } const handler = (m: any) => { const msg = m.detail const duration = Math.round( (typeof performance !== 'undefined' ? performance.now() : Date.now()) - startTime, ) // Success if ( (event === 'websocket.subscribe' && msg?.data?.channel === channel_name) || msg?.event?.includes(sendingevent + ':response') ) { cleanup() console.log(`[ws] ${sendingevent} ${duration}ms`) resolve(msg.data) return } // Error / timeout if ( msg?.event?.includes(sendingevent + ':error') || msg?.event?.includes(sendingevent + ':timeout') ) { cleanup() console.log(`[ws] ${sendingevent} failed ${duration}ms`) reject(msg.data) return } // Progress if (progress && msg?.event?.includes(sendingevent + ':progress')) { progress(msg.data) } } this.addEventListener(sendingevent + ':progress', handler) this.addEventListener(sendingevent + ':error', handler) this.addEventListener(sendingevent + ':response', handler) // sendTimeout: stop listening once the caller has given up. if (ctl) { ctl.cancel = () => { dropQueued() cleanup() } if (ctl.timedOut) ctl.cancel() } if (tracked) { tracked.fail = (err) => { cleanup() reject(err) } // The socket already closed between writing and waiting. if (tracked.lost) tracked.fail(tracked.lost) } }).catch((error: any) => { // If the server dropped our subscription, re-establish and retry once if ( _retryOnSubscriptionLost && error?.message === 'Subscription not established' ) { const ch = this.channels.find( (c) => c.name === channel_name, ) as WebsocketChannel | undefined if (ch) { ch.is_established = false ch._establishPromise = null } console.log('[ws] Re-establishing channel after subscription loss') return this._send(event, data, channel_name, progress, false, ctl) } throw error }) as Promise } /** Retry a throttled connect() once the throttle window has passed. */ private _deferConnect(ms: number): void { if (this._deferredConnectTimer) return this._deferredConnectTimer = setTimeout(() => { this._deferredConnectTimer = null if (!this.socket && !this._connectPromise) this.connect().catch(() => {}) }, Math.max(0, ms) + 5) } /** rejectOnDisconnect: reject every send that went out on `socket`. */ private _failInflight(socket: WsSocket): void { for (const entry of [...this._inflight]) { if (entry.socket !== socket) continue this._inflight.delete(entry) const err = { message: this._t('websocket.requestlost', 'The connection was lost before the server answered.'), code: 'connection_lost', } if (entry.fail) entry.fail(err) else entry.lost = err } } // ------------------------------------------------------------------------- // Event listeners (framework-agnostic — return cleanup function) // ------------------------------------------------------------------------- listen( event: string, channel_name: string | null | undefined, callback: (data: any) => void, ): () => void { channel_name ??= this._config.defaultChannel ?? 'websocket' const handler = (m: any) => { if (m.detail.channel === channel_name) callback(m.detail.data) } this.addEventListener(event, handler) return () => this.removeEventListener(event, handler) } listenOnce(event: string, channel_name?: string | null): Promise { channel_name ??= this._config.defaultChannel ?? 'websocket' return new Promise((resolve) => { const handler = (m: any) => { if (m.detail.channel === channel_name) { resolve(m.detail.data) this.removeEventListener(event, handler) } } this.addEventListener(event, handler) }) } async unsubscribe(channel_name?: string | null): Promise { channel_name ??= this._config.defaultChannel ?? 'websocket' const channel = this.channels.find((c) => c.name === channel_name) if (channel) await channel.unsubscribe() this.channels = this.channels.filter((c) => c.name !== channel_name) return true } // ------------------------------------------------------------------------- // Lifecycle // ------------------------------------------------------------------------- destroy(): void { if (this.heartbeat) clearInterval(this.heartbeat) this.heartbeat = null if (this._deferredConnectTimer) clearTimeout(this._deferredConnectTimer) this._deferredConnectTimer = null if (this._reconnectTimer) clearTimeout(this._reconnectTimer) this._reconnectTimer = null if (this.socket) { try { this.socket.close() } catch {} this.socket = null } this.channels = [] this.send_queue = [] this.is_opened.value = false this.is_setup.value = false this.is_connecting_socket.value = false } } // --------------------------------------------------------------------------- // Factory // --------------------------------------------------------------------------- /** * Create a WebSocket client. * * @param config - Connection and behavior configuration. * @param createRef - Reactive ref factory. Pass `ref` from Vue for reactive state, * or omit for plain `{ value }` objects. */ export function createWsClient( config: WsClientConfig, createRef: CreateRefFn = plainRef, ): WsClient { const isServer = typeof config.isServer === 'function' ? config.isServer() : (config.isServer ?? false) if (isServer) return createSsrStub(createRef) // Validate configuration and warn developers about common issues. // IMPORTANT: only resolve a STRING url here. A function url is lazy by design — // it may depend on a runtime context that does not exist yet at construction // time (e.g. Nuxt's `useRuntimeConfig()`, which throws when called outside a // plugin/setup/lifecycle). Calling it eagerly here crashed module-scope // singletons built with `createVueWsClient({ url: () => useRuntimeConfig()... })`. // Function urls are resolved lazily at connect time instead. if (typeof config.url === 'string') { const url = config.url if (!url || url === 'wss:///app/' || url === 'ws:///app/') { console.error( '[blax-networking] WebSocket URL is empty or malformed: ' + JSON.stringify(url) + '\n' + 'Ensure WEBS_URL and PUSHER_APP_KEY are configured. ' + 'Expected format: wss://your-ws-host/app/{appKey}', ) } else if (url.endsWith('/app/') || url.endsWith('/app')) { console.error( '[blax-networking] WebSocket URL is missing the app key: ' + url + '\n' + 'Ensure PUSHER_APP_KEY is set. The URL must end with /app/{appKey} ' + 'where {appKey} matches the PUSHER_APP_KEY on the backend.', ) } } return new WsClientImpl(config, createRef) }