main
/**
* App-wide realtime hub: exactly one `ChatWsClient` socket for the whole
* session, shared by every room view and any future consumer (notifications,
* presence, reminders).
*
* Why a module singleton instead of a React provider:
* - the socket's lifetime is the *session*, not a component's, so unmounting a
* chat column must not tear down another panel's subscription;
* - StrictMode's double mount would otherwise open and immediately kill one
* upgrade per mount, and every handshake JWT is single-use on the server;
* - keeping it framework-free makes the reconnect/gap logic testable without a
* DOM (see the React bindings in `./react`).
*
* Room intents are reference counted: the socket joins a room when the first
* viewer arrives and leaves it when the last one goes, which is what keeps a
* click through twenty channels inside the server's `max_subscriptions`.
*/
import { ChatWsClient, type WsClientIssue } from './client'
import { parseServerFrame, type ServerFrame, type UserEvent, type WsStatus } from './protocol'
type StatusListener = (status: WsStatus) => void
type FrameListener = (frame: ServerFrame) => void
type IssueListener = (issue: WsClientIssue) => void
type UserEventListener = (event: UserEvent) => void
export class ChatRealtimeHub {
private client: ChatWsClient | null = null
private status: WsStatus = 'idle'
private userId: string | undefined
private readonly viewers = new Map<string, number>()
private readonly statusListeners = new Set<StatusListener>()
private readonly frameListeners = new Set<FrameListener>()
private readonly issueListeners = new Set<IssueListener>()
private readonly userEventListeners = new Set<UserEventListener>()
/** `ready.user_id`, undefined until the first handshake — the viewer's own id. */
get selfId(): string | undefined {
return this.userId
}
get connectionStatus(): WsStatus {
return this.status
}
/** Rooms with at least one live viewer. */
watchedRooms(): string[] {
return [...this.viewers.entries()].filter(([, count]) => count > 0).map(([id]) => id)
}
/**
* Create the socket on first use. Nothing touches the network before this:
* the jwt endpoint must not be hit on the auth screens.
*/
ensureClient(): ChatWsClient {
if (this.client) return this.client
const client = new ChatWsClient()
client.onStatus((status) => {
this.status = status
const ready = client.userId
if (ready) this.userId = ready
for (const listener of [...this.statusListeners]) listener(status)
})
client.onIssue((issue) => {
for (const listener of [...this.issueListeners]) listener(issue)
})
client.onFrame((frame) => {
if (frame.type === 'user_event') {
if (frame.event) for (const listener of [...this.userEventListeners]) listener(frame.event)
return
}
for (const listener of [...this.frameListeners]) listener(frame)
})
this.client = client
return client
}
/** Add a viewer for a room; starts the socket on the first one. */
watchRoom(roomId: string): () => void {
const next = (this.viewers.get(roomId) ?? 0) + 1
this.viewers.set(roomId, next)
if (next === 1) {
const client = this.ensureClient()
client.subscribe(roomId)
// `start()` is idempotent while connecting and is also the only way back
// out of a terminal `auth_required` / `failed`, which is exactly what
// needing a room again means after a session was refreshed.
client.start()
}
let released = false
return () => {
if (released) return
released = true
const count = (this.viewers.get(roomId) ?? 1) - 1
if (count > 0) {
this.viewers.set(roomId, count)
return
}
this.viewers.delete(roomId)
this.client?.unsubscribe(roomId)
}
}
/** Force a reconnect attempt after a terminal `auth_required` / `failed`. */
retry(): void {
const client = this.ensureClient()
client.stop()
client.start()
}
sendTyping(roomId: string, threadId?: string | null): void {
this.client?.sendTyping(roomId, threadId)
}
/** Live typing indicators, own echo excluded by the client. */
typingUsers(roomId: string, threadId?: string | null) {
return this.client?.typingUsers(roomId, { threadId }) ?? []
}
/** Fires with the room id whenever its live typing set changes. */
onTypingChange(listener: (roomId: string) => void): () => void {
return this.ensureClient().onTypingChange(listener)
}
lastKnownSeq(roomId: string): number | undefined {
return this.client?.lastKnownSeq(roomId)
}
onStatus(listener: StatusListener): () => void {
this.statusListeners.add(listener)
return () => this.statusListeners.delete(listener)
}
onFrame(listener: FrameListener): () => void {
this.frameListeners.add(listener)
return () => this.frameListeners.delete(listener)
}
onIssue(listener: IssueListener): () => void {
this.issueListeners.add(listener)
return () => this.issueListeners.delete(listener)
}
onUserEvent(listener: UserEventListener): () => void {
this.userEventListeners.add(listener)
return () => this.userEventListeners.delete(listener)
}
/** Drop everything — tests and signed-out teardown only. */
reset(): void {
this.client?.dispose()
this.client = null
this.status = 'idle'
this.userId = undefined
this.viewers.clear()
}
}
let hub: ChatRealtimeHub | null = null
export function getRealtimeHub(): ChatRealtimeHub {
hub ??= new ChatRealtimeHub()
return hub
}
/** Exported for tests; production code never resets the session socket. */
export function resetRealtimeHub(): void {
hub?.reset()
hub = null
}
/** Re-exported so consumers never deep-import the protocol module. */
export { parseServerFrame, type ServerFrame, type UserEvent, type WsClientIssue, type WsStatus }