fad06fd
import { WS_SERVER_DEFAULTS, type RoomEvent, type WsTypingIndicator } from './protocol'
/**
* TTL bookkeeping for `event{kind:'typing'}` broadcasts.
*
* The backend is stateless about typing: it re-broadcasts whatever a socket
* sends, throttled to one frame per 3s per socket, and stamps the event with
* `ttl_secs = max(1, chat.typing_ttl_secs)` (gitdataai/app/im/src/socket.rs,
* share/realtime.rs:53-58). There is no "stopped typing" frame, so the *only*
* correct way to drop an indicator is to honour `ttl_secs` client-side.
*/
/** `event{kind:'typing'}` narrowed to the variant the registry consumes. */
export type WsTypingEvent = Extract<RoomEvent, { kind: 'typing' }>
type WsTypingListener = (room_id: string) => void
export class TypingRegistry {
/** room_id -> user_id -> indicator (one slot per user: refreshes overwrite). */
private readonly rooms = new Map<string, Map<string, WsTypingIndicator>>()
private readonly listeners = new Set<WsTypingListener>()
private sweepTimer: ReturnType<typeof setTimeout> | undefined
/** Record a typing broadcast. Re-broadcasts refresh the expiry, as on the wire. */
record(event: WsTypingEvent, now: number = Date.now()): void {
// Defence against a payload without a usable ttl_secs: an unbounded
// indicator would hang around forever, so fall back to the server default.
const raw = event.ttl_secs
const ttlMs =
typeof raw === 'number' && Number.isFinite(raw) && raw > 0
? Math.max(raw, 1) * 1000
: WS_SERVER_DEFAULTS.typingTtlSecs * 1000
let byUser = this.rooms.get(event.room_id)
if (!byUser) {
byUser = new Map()
this.rooms.set(event.room_id, byUser)
}
byUser.set(event.user_id, {
room_id: event.room_id,
user_id: event.user_id,
thread_id: event.thread_id ?? null,
expires_at: now + ttlMs,
updated_at: now,
})
this.scheduleSweep(now)
this.notify(event.room_id)
}
/**
* Live indicators for a room. This is a pure read (no pruning, no listener
* callback) so it is safe as a `useSyncExternalStore` snapshot: expiry is
* evaluated against `now`, and the sweep timer does the bookkeeping.
* `excludeUserId` hides your own echo, because the room channel also reaches
* the socket that sent the typing frame.
*/
active(
roomId: string,
options: { excludeUserId?: string; threadId?: string | null } = {},
now: number = Date.now(),
): WsTypingIndicator[] {
const byUser = this.rooms.get(roomId)
if (!byUser) return []
const threadId = options.threadId
return [...byUser.values()]
.filter(indicator => indicator.expires_at > now)
.filter(indicator => indicator.user_id !== options.excludeUserId)
.filter(indicator =>
threadId === undefined ? true : (indicator.thread_id ?? null) === (threadId ?? null),
)
.sort((a, b) => a.updated_at - b.updated_at)
}
/** Rooms that currently have at least one live indicator. */
activeRoomIds(now: number = Date.now()): string[] {
const roomIds: string[] = []
for (const [roomId, byUser] of this.rooms) {
for (const indicator of byUser.values()) {
if (indicator.expires_at > now) {
roomIds.push(roomId)
break
}
}
}
return roomIds
}
/** Drop indicators, e.g. when a room intent is removed or the socket dies. */
clear(roomId?: string): void {
const changed = roomId === undefined ? [...this.rooms.keys()] : this.rooms.has(roomId) ? [roomId] : []
if (roomId === undefined) this.rooms.clear()
else this.rooms.delete(roomId)
if (this.rooms.size === 0) this.clearSweepTimer()
for (const id of changed) this.notify(id)
}
/** Fires with the room id whenever its live indicator set changes. */
onTyping(listener: WsTypingListener): () => void {
this.listeners.add(listener)
return () => this.listeners.delete(listener)
}
dispose(): void {
this.clearSweepTimer()
this.rooms.clear()
this.listeners.clear()
}
private prune(roomId: string, now: number): Map<string, WsTypingIndicator> | undefined {
const byUser = this.rooms.get(roomId)
if (!byUser) return undefined
let removed = false
for (const [userId, indicator] of byUser) {
if (indicator.expires_at <= now) {
byUser.delete(userId)
removed = true
}
}
if (byUser.size === 0) {
this.rooms.delete(roomId)
if (this.rooms.size === 0) this.clearSweepTimer()
removed = true
}
if (removed) this.notify(roomId)
return this.rooms.get(roomId)
}
private scheduleSweep(now: number): void {
let earliest = Number.POSITIVE_INFINITY
for (const byUser of this.rooms.values()) {
for (const indicator of byUser.values()) {
if (indicator.expires_at < earliest) earliest = indicator.expires_at
}
}
if (!Number.isFinite(earliest)) {
this.clearSweepTimer()
return
}
this.clearSweepTimer()
this.sweepTimer = setTimeout(() => {
this.sweepTimer = undefined
for (const roomId of [...this.rooms.keys()]) this.prune(roomId, Date.now())
}, Math.max(earliest - now, 0))
}
private clearSweepTimer(): void {
if (this.sweepTimer !== undefined) {
clearTimeout(this.sweepTimer)
this.sweepTimer = undefined
}
}
private notify(roomId: string): void {
for (const listener of [...this.listeners]) listener(roomId)
}
}