aaaaf4c
import {
classifyWsError,
isSubscribeError,
parseServerFrame,
roomEventSeq,
WS_SERVER_DEFAULTS,
asSeq,
type ClientFrame,
type RoomEvent,
type ServerFrame,
type WsServerErrorKind,
type WsStatus,
type WsTypingIndicator,
} from './protocol'
import { requestWsJwt, isWsJwtError, type WsJwt } from './jwt'
import { TypingRegistry } from './typing'
/**
* Browser client for the chat WebSocket (`GET /ws/v1`, served by the im gateway).
*
* Protocol facts it encodes, all traceable to gitdataai/app/im/src:
* - Auth is a single-use JWT offered as the `gitdata.jwt.<token>` subprotocol
* (minted by `GET /auth/jwt-request?appid=chat`, consumed once by the im
* gateway via redis `SETNX jti`), so every connect mints a fresh one.
* - The server sends `ready` first, then `history` / `event` / `user_event`,
* `gap` / `resync_required` when stream order breaks, `pong` for app-level
* `ping`, and `bye` right before a server-initiated close (:279-336, :550-552,
* :577-587).
* - Idle sockets are reaped after `heartbeat_secs * 3` (:220-224) and the server
* never pings on its own, so proving liveness is the client's job.
* - Subscriptions are per socket (:390-504), so room intents are replayed after a
* reconnect and missed messages are pulled with `resync`.
* - `resync` answers with at most `history_limit` messages ordered **ascending**
* from the cursor (:505-531 + services/messages.rs:677-687) and `history`
* carries no `has_more`, so a big backlog is walked page by page here.
* - `error` frames carry no `room_id`; a rejected subscribe is attributed through
* the server's in-order frame handling (`handle_client_frame(..).await` inside
* the read loop, :242-255) using one ping barrier per subscribe.
*/
const DEFAULT_PATH = '/ws/v1'
const JWT_PROTOCOL_PREFIX = 'gitdata.jwt.'
const DEFAULT_PING_INTERVAL_MS = 25_000 // heartbeat is 30s, the server says bye at 3x
const DEFAULT_MIN_RECONNECT_DELAY_MS = 500
const DEFAULT_MAX_RECONNECT_DELAY_MS = 30_000
const DEFAULT_MAX_CONNECT_FAILURES = 12
/** Snippet size reported for payloads this build does not recognise. */
const UNRECOGNIZED_SNIPPET_CHARS = 256
export interface ChatWsClientOptions {
/** Full ws:// or wss:// endpoint. Defaults to same-origin `/ws/v1`. */
url?: string
/** Provides the single-use handshake JWT. Defaults to `GET /auth/jwt-request?appid=chat`. */
fetchJwt?: () => Promise<WsJwt>
/** Interval between app-level `{op:'ping'}` frames; clamped to >= 5s (server floor). */
pingIntervalMs?: number
minReconnectDelayMs?: number
maxReconnectDelayMs?: number
/**
* Consecutive failed attempts (jwt, handshake, or a close that never
* produced `ready`) tolerated before going terminal `failed`. 0 disables.
*/
maxConnectFailures?: number
/** How long an unanswered ping may stay outstanding before the socket counts as half-open. */
pongTimeoutMs?: number
/** Mirrors `[chat.ws] max_subscriptions`; the client refuses to exceed it locally. */
maxSubscriptions?: number
/** Mirrors `[chat.ws] history_limit`; used to detect truncated `history` pages. */
historyLimitHint?: number
/** Safety bound for the forward resync walk (pages per gap). */
maxResyncPages?: number
/** Client-side typing throttle; the server drops anything inside its own 3s window. */
typingThrottleMs?: number
/**
* Follow each `subscribe` with a `{op:'ping'}` barrier so `error` frames can be
* attributed to the room that caused them (default true). Turn it off to send
* the minimum frame set; attribution then degrades to "no room_id".
*/
correlateSubscribeErrors?: boolean
}
/**
* Conditions the protocol cannot resolve on its own. Subscribe to these instead
* of parsing `error` frames: each entry already carries the room when the backend
* gives enough information to identify it.
*/
export type WsClientIssue =
/** The server sent `{type:'error'}`; `room_id` is attributed best-effort. */
| { kind: 'server_error'; error: WsServerErrorKind; message: string; room_id?: string }
| { kind: 'subscription_limit'; room_id: string; limit: number }
| { kind: 'gap'; room_id: string; from: number; to: number }
| { kind: 'resync_required'; room_id: string; reason: string; dropped: boolean }
/** No local cursor to resume from: re-seed the transcript over HTTP. */
| { kind: 'cursor_lost'; room_id: string; reason: string }
/** The forward resync walk hit `maxResyncPages`; older messages are still missing. */
| { kind: 'history_truncated'; room_id: string; pages: number; page_size: number }
| { kind: 'server_bye'; message: string }
/** The socket dropped; `ready: false` means the upgrade never completed. */
| { kind: 'socket_closed'; code: number; ready: boolean }
| { kind: 'pong_timeout'; outstanding_pings: number }
/** Inbound payload that cannot be mapped onto `ServerFrame` (protocol drift). */
| { kind: 'unrecognized_frame'; snippet: string }
/** The backend only writes text frames, so binary data means a broken hop. */
| { kind: 'binary_frame' }
/** A broadcast whose payload even the server could not parse (frames.rs:36-41). */
| { kind: 'unparseable_payload'; channel: 'room' | 'user'; room_id?: string }
| { kind: 'connect_failed'; phase: 'jwt' | 'handshake'; message: string; failures: number }
| { kind: 'auth_required'; message: string }
| { kind: 'gave_up'; failures: number }
interface SubscribeBarrier {
sent_at: number
/** room_id -> subscribe epoch collected since the previous barrier. */
rooms: Map<string, number>
}
export class ChatWsClient {
private readonly url: string
private readonly fetchJwt: () => Promise<WsJwt>
private readonly pingIntervalMs: number
private readonly minReconnectDelayMs: number
private readonly maxReconnectDelayMs: number
private readonly maxConnectFailures: number
private readonly pongTimeoutMs: number
private readonly maxSubscriptions: number
private readonly historyLimitHint: number
private readonly maxResyncPages: number
private readonly typingThrottleMs: number
private readonly correlateSubscribeErrors: boolean
private ws: WebSocket | null = null
private connectionStatus: WsStatus = 'idle'
private started = false
private generation = 0
private reconnectAttempts = 0
private connectFailures = 0
private attemptReachedReady = false
private reconnectTimer: ReturnType<typeof setTimeout> | undefined
private pingTimer: ReturnType<typeof setInterval> | undefined
private lastTypingSentAt = 0
/** Room subscription intents; replayed on every `ready`. */
private readonly rooms = new Set<string>()
/** Resync requests that must survive a reconnect. */
private readonly forcedResync = new Set<string>()
/** Rooms with an outstanding forward resync walk. */
private readonly awaitingResync = new Set<string>()
private readonly lastSeq = new Map<string, number>()
/** Pages consumed by the current resync walk, per room. */
private readonly resyncPages = new Map<string, number>()
private barriers: SubscribeBarrier[] = []
private barrierRooms = new Map<string, number>()
private readonly subscribeEpoch = new Map<string, number>()
private currentUserId: string | undefined
private currentSessionId: string | undefined
private currentJwtTtlSecs: number | undefined
private readonly typingRegistry = new TypingRegistry()
private readonly frameHandlers = new Set<(frame: ServerFrame) => void>()
private readonly statusHandlers = new Set<(status: WsStatus) => void>()
private readonly issueHandlers = new Set<(issue: WsClientIssue) => void>()
constructor(options: ChatWsClientOptions = {}) {
this.url = options.url ?? sameOriginWsUrl()
this.fetchJwt = options.fetchJwt ?? requestWsJwt
this.pingIntervalMs = Math.max(options.pingIntervalMs ?? DEFAULT_PING_INTERVAL_MS, 5_000)
this.minReconnectDelayMs = options.minReconnectDelayMs ?? DEFAULT_MIN_RECONNECT_DELAY_MS
this.maxReconnectDelayMs = options.maxReconnectDelayMs ?? DEFAULT_MAX_RECONNECT_DELAY_MS
this.maxConnectFailures = options.maxConnectFailures ?? DEFAULT_MAX_CONNECT_FAILURES
this.pongTimeoutMs = options.pongTimeoutMs ?? this.pingIntervalMs * 2 + 5_000
this.maxSubscriptions = options.maxSubscriptions ?? WS_SERVER_DEFAULTS.maxSubscriptions
this.historyLimitHint = options.historyLimitHint ?? WS_SERVER_DEFAULTS.historyLimit
this.maxResyncPages = options.maxResyncPages ?? 10
this.typingThrottleMs = options.typingThrottleMs ?? WS_SERVER_DEFAULTS.typingThrottleSecs * 1000
this.correlateSubscribeErrors = options.correlateSubscribeErrors ?? true
}
/* ------------------------------------------------------------------ state */
get status(): WsStatus {
return this.connectionStatus
}
/** `ready.user_id`; undefined until the first successful handshake. */
get userId(): string | undefined {
return this.currentUserId
}
/** `ready.session_id` (ws/mod.rs:135), a fresh v7 uuid per socket: correlate with server logs. */
get sessionId(): string | undefined {
return this.currentSessionId
}
/** `expires_in_secs` of the JWT used for the current/last handshake (services/jwt.rs). */
get jwtTtlSecs(): number | undefined {
return this.currentJwtTtlSecs
}
/** Highest `seq` observed for a room (from `history` / `event`), if any. */
lastKnownSeq(roomId: string): number | undefined {
return this.lastSeq.get(roomId)
}
isSubscribed(roomId: string): boolean {
return this.rooms.has(roomId)
}
subscribedRooms(): string[] {
return [...this.rooms]
}
/** Live typing indicators for a room, with this client's own echo filtered out. */
typingUsers(roomId: string, options: { threadId?: string | null } = {}): WsTypingIndicator[] {
return this.typingRegistry.active(roomId, { ...options, excludeUserId: this.currentUserId })
}
/* ---------------------------------------------------------- subscriptions */
/** Observe every server frame. Handlers run after the client reconciled its own state. */
onFrame(handler: (frame: ServerFrame) => void): () => void {
this.frameHandlers.add(handler)
return () => this.frameHandlers.delete(handler)
}
/** Observe connection lifecycle. Returns an unsubscribe function. */
onStatus(handler: (status: WsStatus) => void): () => void {
this.statusHandlers.add(handler)
handler(this.connectionStatus)
return () => this.statusHandlers.delete(handler)
}
/** Observe conditions the protocol cannot resolve on its own. */
onIssue(handler: (issue: WsClientIssue) => void): () => void {
this.issueHandlers.add(handler)
return () => this.issueHandlers.delete(handler)
}
/** Fires with the room id whenever its live typing set changes. */
onTypingChange(handler: (roomId: string) => void): () => void {
return this.typingRegistry.onTyping(handler)
}
/**
* Joins a room stream. Unless `resync: false`, the server answers with a
* `history` frame holding the newest `history_limit` messages (ws/mod.rs:470-495).
* The intent survives reconnects; history can repeat messages already seen, so
* dedupe by `id`.
*/
subscribe(roomId: string, options: { resync?: boolean } = {}): void {
if (this.rooms.has(roomId)) return
if (this.rooms.size >= this.maxSubscriptions) {
// The server's limit error carries no room id, so enforce the cap while the
// offending room is still known.
this.emitIssue({ kind: 'subscription_limit', room_id: roomId, limit: this.maxSubscriptions })
return
}
this.rooms.add(roomId)
this.sendSubscribe(roomId, options.resync ?? true)
}
unsubscribe(roomId: string): void {
if (!this.rooms.delete(roomId)) return
this.forgetRoom(roomId)
this.write({ op: 'unsubscribe', room_id: roomId })
}
/**
* Asks for messages with `seq > afterSeq`; the reply is a `history` frame.
* Forward-only and ascending, so pass the cursor you already hold (or 0 to walk
* the room from its first message). Full pages are followed automatically up to
* `maxResyncPages`. If the socket is down the pull is replayed after `ready`,
* but only for rooms that still have a subscription intent.
*/
resync(roomId: string, afterSeq: number): void {
this.resyncPages.delete(roomId)
this.awaitingResync.add(roomId)
this.startResync(roomId, afterSeq)
}
/**
* Broadcasts a typing signal. The throttle is per socket — mirroring the
* server's own 3s per-socket window (ws/mod.rs:533-549) — because a frame sent
* inside that window is dropped regardless of which room it names.
*/
sendTyping(roomId: string, threadId?: string | null): void {
const now = Date.now()
if (now - this.lastTypingSentAt < this.typingThrottleMs) return
this.lastTypingSentAt = now
this.write({ op: 'typing', room_id: roomId, thread_id: threadId ?? null })
}
/* -------------------------------------------------------------- lifecycle */
/** Opens the socket (and keeps it open until {@link stop}). Idempotent. */
start(): void {
if (this.started) return
this.started = true
this.reconnectAttempts = 0
this.connectFailures = 0
void this.connect()
}
/**
* Closes the socket and stops reconnecting. Room intents are kept, so a later
* {@link start} rejoins them. Terminal states (`auth_required`, `failed`) also
* need an explicit `start()` to resume.
*/
stop(): void {
if (!this.started) {
if (this.connectionStatus === 'auth_required' || this.connectionStatus === 'failed') {
this.teardownSocket()
this.setStatus('closed')
}
return
}
this.started = false
this.teardownSocket()
this.setStatus('closed')
}
/** Release timers and handlers. Only for teardown paths and tests. */
dispose(): void {
this.stop()
this.typingRegistry.dispose()
this.frameHandlers.clear()
this.statusHandlers.clear()
this.issueHandlers.clear()
}
private teardownSocket(): void {
this.generation += 1
this.clearReconnectTimer()
this.clearPingTimer()
const ws = this.ws
this.ws = null
this.resetBarrierState()
this.typingRegistry.clear()
try {
ws?.close(1000)
} catch {
// already closing/closed
}
}
/* ---------------------------------------------------------------- connect */
private async connect(): Promise<void> {
if (!this.started) return
this.clearReconnectTimer()
this.setStatus(this.reconnectAttempts === 0 ? 'connecting' : 'reconnecting')
this.attemptReachedReady = false
const generation = ++this.generation
let jwt: WsJwt
try {
jwt = await this.fetchJwt()
} catch (error) {
if (generation !== this.generation || !this.started) return
if (isWsJwtError(error) && error.kind === 'auth') {
// 401/403 means the session is gone; the upgrade would be rejected too, so
// a backoff loop only burns the jwt endpoint.
this.started = false
this.connectFailures += 1
this.emitIssue({ kind: 'auth_required', message: error.message })
this.setStatus('auth_required')
return
}
this.registerFailure('jwt', error)
this.scheduleReconnect()
return
}
if (generation !== this.generation || !this.started) return
this.currentJwtTtlSecs = jwt.expiresInSecs
let ws: WebSocket
try {
// The upgrade response re-offers the same protocol name (ws/mod.rs) and
// JWTs only contain base64url / '.' characters, so they are valid
// subprotocol tokens.
ws = new WebSocket(this.url, [`${JWT_PROTOCOL_PREFIX}${jwt.jwt}`])
} catch (error) {
if (generation !== this.generation) return
this.registerFailure('handshake', error)
this.scheduleReconnect()
return
}
this.ws = ws
this.lastTypingSentAt = 0 // the server throttle is per connection
this.resetBarrierState()
ws.onopen = () => {
if (generation !== this.generation || this.ws !== ws) return
this.startPingLoop()
}
ws.onmessage = event => {
if (generation !== this.generation || this.ws !== ws) return
this.handleMessage(event.data)
}
ws.onclose = event => {
if (generation !== this.generation || this.ws !== ws) return
this.ws = null
this.clearPingTimer()
this.resetBarrierState()
this.typingRegistry.clear() // peers may have stopped typing while we were away
if (!this.started) return
this.emitIssue({ kind: 'socket_closed', code: event.code, ready: this.attemptReachedReady })
// A socket that never produced `ready` is a failed attempt: every rejection
// in ws/mod.rs:99-132 is a plain HTTP 400, which browsers hide from JS.
if (!this.attemptReachedReady) {
this.registerFailure('handshake', { message: `handshake closed (${event.code})` })
}
this.scheduleReconnect()
}
}
private registerFailure(phase: 'jwt' | 'handshake', error: unknown): void {
this.connectFailures += 1
this.emitIssue({
kind: 'connect_failed',
phase,
message: error instanceof Error ? error.message : String(error),
failures: this.connectFailures,
})
if (this.maxConnectFailures > 0 && this.connectFailures >= this.maxConnectFailures) {
this.started = false
this.clearReconnectTimer()
this.emitIssue({ kind: 'gave_up', failures: this.connectFailures })
this.setStatus('failed')
}
}
/* --------------------------------------------------------------- incoming */
private handleMessage(data: unknown): void {
if (typeof data !== 'string') {
this.emitIssue({ kind: 'binary_frame' })
return
}
const frame = parseServerFrame(data)
if (!frame) {
this.emitIssue({
kind: 'unrecognized_frame',
snippet: data.slice(0, UNRECOGNIZED_SNIPPET_CHARS),
})
return
}
this.applyFrameEffects(frame)
for (const handler of [...this.frameHandlers]) handler(frame)
}
/** Keeps seq cursors, typing state and room intents consistent with the server. */
private applyFrameEffects(frame: ServerFrame): void {
switch (frame.type) {
case 'ready':
// Subscriptions are per socket, so re-offer every room intent.
this.attemptReachedReady = true
this.connectFailures = 0
this.reconnectAttempts = 0
this.currentUserId = frame.user_id
this.currentSessionId = frame.session_id
this.resetBarrierState()
this.setStatus('open')
this.resubscribeRooms()
break
case 'history':
this.applyHistory(frame.room_id, frame.messages)
break
case 'event':
this.applyRoomEvent(frame.room_id, frame.event)
break
case 'user_event':
if (frame.event === null) this.emitIssue({ kind: 'unparseable_payload', channel: 'user' })
break
case 'gap':
// The server noticed dropped events: backfill from our own cursor and let
// the app dedupe by id/seq.
this.emitIssue({ kind: 'gap', room_id: frame.room_id, from: frame.from, to: frame.to })
if (this.rooms.has(frame.room_id)) {
this.resync(
frame.room_id,
Number.isFinite(frame.from) ? frame.from - 1 : 0,
)
}
break
case 'resync_required': {
// `access revoked` means the server already dropped the subscription
// (ws/mod.rs:318-341); anything else (currently `lagged`, :289-298) means
// it also forgot its cursor for this room, so the client must re-pull.
const dropped = frame.reason === 'access revoked'
if (dropped) {
this.dropRoomIntent(frame.room_id)
} else if (!this.lastSeq.has(frame.room_id)) {
this.emitIssue({ kind: 'cursor_lost', room_id: frame.room_id, reason: frame.reason })
}
this.emitIssue({
kind: 'resync_required',
room_id: frame.room_id,
reason: frame.reason,
dropped,
})
if (!dropped && this.rooms.has(frame.room_id)) {
this.resync(frame.room_id, this.lastSeq.get(frame.room_id) ?? 0)
}
break
}
case 'error':
this.applyServerError(frame.message)
break
case 'pong':
// Pongs arrive in ping order, so the head barrier is fully settled: every
// response (including errors) for its frames has already been delivered.
this.barriers.shift()
break
case 'bye':
// Only `idle timeout` / `internal error` are produced, both retriable, so
// `onclose` keeps driving the reconnect while `started`.
this.emitIssue({ kind: 'server_bye', message: frame.message })
break
}
}
private applyHistory(roomId: string, messages: readonly { seq: number }[]): void {
if (messages.length > 0) {
let max = this.lastSeq.get(roomId) ?? 0
for (const message of messages) {
const seq = asSeq(message.seq)
if (seq !== undefined && seq > max) max = seq
}
this.lastSeq.set(roomId, max)
}
// `history` only ever follows an authorised read, so any subscribe still
// pending for this room cannot have failed on permissions.
this.settleRoom(roomId)
if (!this.awaitingResync.has(roomId)) return
const pages = (this.resyncPages.get(roomId) ?? 0) + 1
if (messages.length < this.historyLimitHint) {
this.endResync(roomId)
return
}
if (pages > this.maxResyncPages) {
this.endResync(roomId)
this.emitIssue({
kind: 'history_truncated',
room_id: roomId,
pages: pages - 1,
page_size: this.historyLimitHint,
})
return
}
// A full page means more may be ahead: continue the forward walk.
this.resyncPages.set(roomId, pages)
this.write({ op: 'resync', room_id: roomId, after_seq: this.lastSeq.get(roomId) ?? 0 })
}
private applyRoomEvent(roomId: string, event: RoomEvent | null): void {
if (event === null) {
this.emitIssue({ kind: 'unparseable_payload', channel: 'room', room_id: roomId })
return
}
const seq = roomEventSeq(event)
if (seq !== undefined && seq > (this.lastSeq.get(roomId) ?? 0)) {
this.lastSeq.set(roomId, seq)
}
if (event.kind === 'typing') this.typingRegistry.record(event)
this.settleRoom(roomId)
}
private applyServerError(message: string): void {
const error = classifyWsError(message)
const roomId = isSubscribeError(error) ? this.attributeSubscribeError() : undefined
if (roomId !== undefined) this.dropRoomIntent(roomId)
this.emitIssue({ kind: 'server_error', error, message, room_id: roomId })
}
/* ------------------------------------------------------ intent bookkeeping */
private dropRoomIntent(roomId: string): void {
this.rooms.delete(roomId)
this.forgetRoom(roomId)
}
private forgetRoom(roomId: string): void {
this.forcedResync.delete(roomId)
this.awaitingResync.delete(roomId)
this.resyncPages.delete(roomId)
this.lastSeq.delete(roomId)
this.typingRegistry.clear(roomId)
// Keep the newest epoch: it may still sit in an un-acked barrier, and the
// server's `error` reply carries no room_id, so forgetting it would make
// attribution impossible.
const keep = this.subscribeEpoch.get(roomId)
this.subscribeEpoch.clear()
if (keep !== undefined) this.subscribeEpoch.set(roomId, keep)
for (const barrier of this.barriers) {
if (keep !== undefined && barrier.rooms.get(roomId) === keep) continue
barrier.rooms.delete(roomId)
}
if (keep !== undefined && this.barrierRooms.get(roomId) === keep) {
this.barrierRooms.delete(roomId)
}
}
private endResync(roomId: string): void {
this.awaitingResync.delete(roomId)
this.resyncPages.delete(roomId)
this.forcedResync.delete(roomId)
}
/** Remember a resync that could not be written, so `ready` can replay it. */
private startResync(roomId: string, afterSeq: number): void {
if (this.write({ op: 'resync', room_id: roomId, after_seq: afterSeq })) return
this.forcedResync.add(roomId)
}
/* ------------------------------------------------------- subscribe barriers */
private resubscribeRooms(): void {
for (const roomId of this.rooms) {
const afterSeq = this.lastSeq.get(roomId)
const forced = this.forcedResync.has(roomId)
// Skip the snapshot when we already know the cursor; the explicit resync
// below fetches exactly the missed range.
this.sendSubscribe(roomId, afterSeq === undefined)
if (afterSeq !== undefined || forced) {
this.resyncPages.delete(roomId)
this.awaitingResync.add(roomId)
this.write({ op: 'resync', room_id: roomId, after_seq: afterSeq ?? 0 })
}
}
}
private sendSubscribe(roomId: string, resync: boolean): void {
if (!this.rooms.has(roomId)) return
const epoch = (this.subscribeEpoch.get(roomId) ?? 0) + 1
this.subscribeEpoch.set(roomId, epoch)
if (!this.write({ op: 'subscribe', room_id: roomId, resync })) return
if (!this.correlateSubscribeErrors) return
// One barrier per subscribe keeps error attribution unambiguous.
this.barrierRooms.set(roomId, epoch)
this.armPingBarrier()
}
/** Close the current barrier group with a ping; its `pong` settles the group. */
private armPingBarrier(): void {
const rooms = this.barrierRooms
this.barrierRooms = new Map()
if (this.write({ op: 'ping' })) {
this.barriers.push({ sent_at: Date.now(), rooms })
return
}
// Nothing reached the server, so keep the intents for the next round.
for (const entry of rooms) this.barrierRooms.set(entry[0], entry[1])
}
private settleRoom(roomId: string): void {
const epoch = this.subscribeEpoch.get(roomId)
if (epoch === undefined) return
this.subscribeEpoch.delete(roomId)
for (const barrier of this.barriers) {
if (barrier.rooms.get(roomId) === epoch) barrier.rooms.delete(roomId)
}
if (this.barrierRooms.get(roomId) === epoch) this.barrierRooms.delete(roomId)
}
/**
* The server answers client frames strictly in order, so an `error` belongs to
* the oldest un-acked barrier. Attribute only when that barrier holds exactly
* one subscribe; otherwise nothing is guessed.
*/
private attributeSubscribeError(): string | undefined {
const head = this.barriers[0]
if (!head || head.rooms.size !== 1) return undefined
const entry = [...head.rooms.entries()][0]
if (!entry) return undefined
const roomId = entry[0]
const epoch = entry[1]
head.rooms.delete(roomId)
if (this.subscribeEpoch.get(roomId) !== epoch) return undefined
return roomId
}
private resetBarrierState(): void {
this.barriers = []
this.barrierRooms = new Map()
}
/* ------------------------------------------------------------------ timers */
private startPingLoop(): void {
this.clearPingTimer()
this.pingTimer = setInterval(() => {
const ws = this.ws
if (!ws || ws.readyState !== WebSocket.OPEN) return
const head = this.barriers[0]
if (head && Date.now() - head.sent_at > this.pongTimeoutMs) {
// A ping never came back, so the socket is half-open. Room traffic is no
// proof of liveness: only an outstanding barrier can trip this.
const outstanding = this.barriers.length
this.barriers = []
this.setStatus('reconnecting')
ws.close(1000)
this.emitIssue({ kind: 'pong_timeout', outstanding_pings: outstanding })
return
}
this.armPingBarrier()
}, this.pingIntervalMs)
}
private scheduleReconnect(): void {
if (!this.started || this.reconnectTimer !== undefined) return
this.setStatus('reconnecting')
const exponential = this.minReconnectDelayMs * 2 ** Math.min(this.reconnectAttempts, 16)
this.reconnectAttempts += 1
const delay = Math.min(exponential, this.maxReconnectDelayMs) * (0.8 + Math.random() * 0.4)
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = undefined
void this.connect()
}, delay)
}
/* ------------------------------------------------------------------ write */
private canWrite(): boolean {
const ws = this.ws
return ws !== null && ws.readyState === WebSocket.OPEN
}
/**
* Write one frame. There is deliberately no send queue: `subscribe` intents are
* replayed from `ready`, `resync` is replayed through `forcedResync`, and
* `typing` / `ping` are ephemeral. Queuing them used to mean both duplicate
* history bursts and silent eviction of a queued subscribe.
*/
private write(frame: ClientFrame): boolean {
if (!this.canWrite()) return false
const ws = this.ws
if (!ws) return false
try {
ws.send(JSON.stringify(frame))
return true
} catch {
return false
}
}
/* ----------------------------------------------------------------- status */
private setStatus(status: WsStatus): void {
if (this.connectionStatus === status) return
this.connectionStatus = status
for (const handler of [...this.statusHandlers]) handler(status)
}
private emitIssue(issue: WsClientIssue): void {
for (const handler of [...this.issueHandlers]) handler(issue)
}
private clearReconnectTimer(): void {
if (this.reconnectTimer !== undefined) {
clearTimeout(this.reconnectTimer)
this.reconnectTimer = undefined
}
}
private clearPingTimer(): void {
if (this.pingTimer !== undefined) {
clearInterval(this.pingTimer)
this.pingTimer = undefined
}
}
}
function sameOriginWsUrl(): string {
const proto = window.location.protocol === 'https:' ? 'wss' : 'ws'
return `${proto}://${window.location.host}${DEFAULT_PATH}`
}