File size: 3,456 Bytes
6aa76ed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | /**
* `/api/events` subscriber for the dashboard chat sidebar.
*
* A notification-only view of the shared JSON-RPC client: the server never
* answers requests on this socket (subscribers don't speak), so replay is
* off and `request()` is not part of the surface. One instance lives for
* the component's whole life — the owner calls `connect()` again after a
* drop instead of building a new client — so the shared per-generation
* guards (stale-socket close, connect timeout, `error`+`close` collapsed
* into one `closed` transition) apply to every attempt.
*/
import { JsonRpcGatewayClient, type ConnectionState, type GatewayEvent, type GatewayEventName } from '@hermes/shared'
import { buildWsUrl } from '@/lib/api'
import { maybeReloadForLoopbackWsAuthFailure } from '@/lib/dashboard-auth-reload'
import { EVENTS_CONNECT_TIMEOUT_MS } from '@/lib/events-reconnect'
export type { ConnectionState, GatewayEvent, GatewayEventName }
type CloseHandler = (code: number | undefined) => void
export class EventsFeedClient extends JsonRpcGatewayClient {
private readonly closeHandlers = new Set<CloseHandler>()
private closeCode: number | null | undefined = undefined
constructor() {
super({
closedErrorMessage: 'events feed closed',
connectErrorMessage: 'events feed connection failed',
connectTimeoutMs: EVENTS_CONNECT_TIMEOUT_MS,
heartbeatDeadlineMs: 0,
heartbeatIntervalMs: 0,
onSocketClose: event => {
this.closeCode = event.code
// Loopback + stale token: the page reloads; nobody should retry.
return maybeReloadForLoopbackWsAuthFailure(event.code)
},
replay: false,
requestIdPrefix: 'e'
})
this.onState(state => {
if (state === 'closed' || state === 'error') {
const code = this.closeCode ?? undefined
this.closeCode = undefined
for (const handler of this.closeHandlers) {
handler(code)
}
}
})
}
/** Close code of the most recent drop; `null` before any socket was dialed. */
get lastCloseCode(): number | null | undefined {
return this.closeCode
}
/**
* Called once per generation that ends without opening or after opening:
* `code` is the WebSocket close code, or `undefined` when the handshake
* timed out / errored without a close frame.
*/
onClose(handler: CloseHandler): () => void {
this.closeHandlers.add(handler)
return () => this.closeHandlers.delete(handler)
}
/** Mint a fresh single-use ticket and dial `/api/events` for `channel`. */
async connect(channel: string): Promise<void> {
this.closeCode = null
// Cover ticket minting with the same deadline as the handshake: a stalled
// pre-socket request otherwise emits no close event and strands the retry
// loop at its last "reconnecting in …" banner.
const url = await withTimeout(buildWsUrl('/api/events', { channel }), EVENTS_CONNECT_TIMEOUT_MS)
await super.connect(url)
}
}
function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => reject(new Error('events feed ticket request timed out')), ms)
promise.then(
value => {
clearTimeout(timer)
resolve(value)
},
(error: unknown) => {
clearTimeout(timer)
reject(error instanceof Error ? error : new Error(String(error)))
}
)
})
}
|