24597013493

Committed 18 Apr 2026 04:35AM UTC coverage: 47.213% (-1.6%) from 48.812%

Build # 24597013493

Build Type

push

github

Committed by

web-flow

Commit Message

feat: add opt-in event retention purge (#359) (#412)

Co-authored-by: Ricardo Cabral <me@ricardocabral.io>

Coverage Stats

461 of 1107 branches covered (41.64%)

Branch coverage included in aggregate %.

55 of 83 new or added lines in 5 files covered. (66.27%)

71 existing lines in 8 files now uncovered.

1343 of 2714 relevant lines covered (49.48%)

3.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

17.11

/src/adapters/web-socket-adapter.ts

import cluster from 'cluster'
import { EventEmitter } from 'stream'
import { IncomingMessage as IncomingHttpMessage } from 'http'
import { WebSocket } from 'ws'

import { ContextMetadata, Factory } from '../@types/base'
import { createNoticeMessage, createOutgoingEventMessage } from '../utils/messages'
import { IAbortable, IMessageHandler } from '../@types/message-handlers'
import { IncomingMessage, OutgoingMessage } from '../@types/messages'
import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters'
import { SubscriptionFilter, SubscriptionId } from '../@types/subscription'
import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter'
import { attemptValidation } from '../utils/validation'
import { ContextMetadataKey } from '../constants/base'
import { createLogger } from '../factories/logger-factory'
import { Event } from '../@types/event'
import { getRemoteAddress } from '../utils/http'
import { IRateLimiter } from '../@types/utils'
import { isEventMatchingFilter } from '../utils/event'
import { messageSchema } from '../schemas/message-schema'
import { Settings } from '../@types/settings'
import { SocketAddress } from 'net'


const debug = createLogger('web-socket-adapter')
const debugHeartbeat = debug.extend('heartbeat')

const abortableMessageHandlers: WeakMap<WebSocket, IAbortable[]> = new WeakMap()

export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter {
  public clientId: string
  private clientAddress: SocketAddress
  private alive: boolean
  private subscriptions: Map<SubscriptionId, SubscriptionFilter[]>

  public constructor(
    private readonly client: WebSocket,
    private readonly request: IncomingHttpMessage,
    private readonly webSocketServer: IWebSocketServerAdapter,
    private readonly createMessageHandler: Factory<IMessageHandler, [IncomingMessage, IWebSocketAdapter]>,
    private readonly slidingWindowRateLimiter: Factory<IRateLimiter>,
    private readonly settings: Factory<Settings>,
  ) {
    super()
    this.alive = true
    this.subscriptions = new Map()

    this.clientId = Buffer.from(this.request.headers['sec-websocket-key'] as string, 'base64').toString('hex')

    const address = getRemoteAddress(this.request, this.settings())

    this.clientAddress = new SocketAddress({
      address: address,
      family: address.indexOf(':') >= 0 ? 'ipv6' : 'ipv4',
    })

    this.client
      .on('error', (error) => {
        if (error.name === 'RangeError' && error.message === 'Max payload size exceeded') {
          console.error(`web-socket-adapter: client ${this.clientId} (${this.getClientAddress()}) sent payload too large`)
        } else if (error.name === 'RangeError' && error.message === 'Invalid WebSocket frame: RSV1 must be clear') {
          debug(`client ${this.clientId} (${this.getClientAddress()}) enabled compression`)
        } else {
          console.error(`web-socket-adapter: client error ${this.clientId} (${this.getClientAddress()}):`, error)
        }

        this.client.close()
      })
      .on('message', this.onClientMessage.bind(this))
      .on('close', this.onClientClose.bind(this))
      .on('pong', this.onClientPong.bind(this))
      .on('ping', this.onClientPing.bind(this))

    this
      .on(WebSocketAdapterEvent.Heartbeat, this.onHeartbeat.bind(this))
      .on(WebSocketAdapterEvent.Subscribe, this.onSubscribed.bind(this))
      .on(WebSocketAdapterEvent.Unsubscribe, this.onUnsubscribed.bind(this))
      .on(WebSocketAdapterEvent.Event, this.onSendEvent.bind(this))
      .on(WebSocketAdapterEvent.Broadcast, this.onBroadcast.bind(this))
      .on(WebSocketAdapterEvent.Message, this.sendMessage.bind(this))

    debug('client %s connected from %s', this.clientId, this.clientAddress.address)
  }

  public getClientId(): string {
    return this.clientId
  }

  public getClientAddress(): string {
    return this.clientAddress.address
  }

  public onUnsubscribed(subscriptionId: string): void {
    debug('client %s unsubscribed %s', this.clientId, subscriptionId)
    this.subscriptions.delete(subscriptionId)
  }

  public onSubscribed(subscriptionId: string, filters: SubscriptionFilter[]): void {
    debug('client %s subscribed %s to %o', this.clientId, subscriptionId, filters)
    this.subscriptions.set(subscriptionId, filters)
  }

  public onBroadcast(event: Event): void {
    this.webSocketServer.emit(WebSocketServerAdapterEvent.Broadcast, event)
    if (cluster.isWorker && typeof process.send === 'function') {
      process.send({
        eventName: WebSocketServerAdapterEvent.Broadcast,
        event,
      })
    }
  }

  public onSendEvent(event: Event): void {
    this.subscriptions.forEach((filters, subscriptionId) => {
      if (
        filters.map(isEventMatchingFilter).some((isMatch) => isMatch(event))
      ) {
        debug('sending event to client %s: %o', this.clientId, event)
        this.sendMessage(createOutgoingEventMessage(subscriptionId, event))
      }
    })
  }

  private sendMessage(message: OutgoingMessage): void {
    if (this.client.readyState !== WebSocket.OPEN) {
      return
    }
    this.client.send(JSON.stringify(message))
  }

  public onHeartbeat(): void {
    if (!this.alive && !this.subscriptions.size) {
      console.error(`web-socket-adapter: pong timeout for client ${this.clientId} (${this.getClientAddress()})`)
      this.client.close()
      return
    }

    this.alive = false
    this.client.ping()
    debugHeartbeat('client %s ping', this.clientId)
  }

  public getSubscriptions(): Map<string, SubscriptionFilter[]> {
    return new Map(this.subscriptions)
  }

  private async onClientMessage(raw: Buffer) {
    this.alive = true
    let abortable = false
    let messageHandler: IMessageHandler & IAbortable | undefined = undefined
    try {
      if (await this.isRateLimited(this.clientAddress.address)) {
        this.sendMessage(createNoticeMessage('rate limited'))
        return
      }

      const message = attemptValidation(messageSchema)(JSON.parse(raw.toString('utf8')))

      message[ContextMetadataKey] = {
        remoteAddress: this.clientAddress,
      } as ContextMetadata

      messageHandler = this.createMessageHandler([message, this]) as IMessageHandler & IAbortable
      if (!messageHandler) {
        console.error('web-socket-adapter: unhandled message: no handler found:', message)
        return
      }

      abortable = typeof messageHandler.abort === 'function'

      if (abortable) {
        const handlers = abortableMessageHandlers.get(this.client) ?? []
        handlers.push(messageHandler)
        abortableMessageHandlers.set(this.client, handlers)
      }

      await messageHandler.handleMessage(message)
    } catch (error) {
      if (error instanceof Error) {
        if (error.name === 'AbortError') {
          console.error(`web-socket-adapter: abort from client ${this.clientId} (${this.getClientAddress()})`)
        } else if (error.name === 'SyntaxError' || error.name === 'ValidationError') {
          if (typeof (error as any).annotate === 'function') {
            debug('invalid message client %s (%s): %o', this.clientId, this.getClientAddress(), (error as any).annotate())
          } else {
            console.error(`web-socket-adapter: malformed message from client ${this.clientId} (${this.getClientAddress()}):`, error.message)
          }
          this.sendMessage(createNoticeMessage(`invalid: ${error.message}`))
        } else {
          console.error('web-socket-adapter: unable to handle message:', error)
        }
      } else {
        console.error('web-socket-adapter: unable to handle message:', error)
      }
    } finally {
      if (abortable && messageHandler) {
        const handlers = abortableMessageHandlers.get(this.client)
        if (handlers) {
          const index = handlers.indexOf(messageHandler)
          if (index >= 0) {
            handlers.splice(index, 1)
          }
        }
      }
    }
  }

  private async isRateLimited(client: string): Promise<boolean> {
    const {
      rateLimits,
      ipWhitelist = [],
    } = this.settings().limits?.message ?? {}

    if (!Array.isArray(rateLimits) || !rateLimits.length || ipWhitelist.includes(client)) {
      return false
    }

    const rateLimiter = this.slidingWindowRateLimiter()

    const hit = (period: number, rate: number) =>
      rateLimiter.hit(
        `${client}:message:${period}`,
        1,
        { period, rate },
      )

    let limited = false
    for (const { rate, period } of rateLimits) {
      const isRateLimited = await hit(period, rate)


      if (isRateLimited) {
        debug('rate limited %s: %d messages / %d ms exceeded', client, rate, period)

        limited = true
      }
    }

    return limited
  }

  private onClientPong() {
    debugHeartbeat('client %s pong', this.clientId)
    this.alive = true
  }

  private onClientPing(data: any) {
    debugHeartbeat('client %s ping', this.clientId)
    this.client.pong(data)
    this.alive = true
  }

  private onClientClose() {
    this.alive = false
    this.subscriptions.clear()

    const handlers = abortableMessageHandlers.get(this.client)
    if (Array.isArray(handlers) && handlers.length) {
      for (const handler of handlers) {
        try {
          handler.abort()
        } catch (error) {
          console.error('Unable to abort message handler', error)
        }
      }
    }

    this.removeAllListeners()
    this.client.removeAllListeners()
  }
}

1	import cluster from 'cluster'	1✔
2	import { EventEmitter } from 'stream'	1✔
3	import { IncomingMessage as IncomingHttpMessage } from 'http'
4	import { WebSocket } from 'ws'	1✔
5
6	import { ContextMetadata, Factory } from '../@types/base'
7	import { createNoticeMessage, createOutgoingEventMessage } from '../utils/messages'	1✔
8	import { IAbortable, IMessageHandler } from '../@types/message-handlers'
9	import { IncomingMessage, OutgoingMessage } from '../@types/messages'
10	import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters'
11	import { SubscriptionFilter, SubscriptionId } from '../@types/subscription'
12	import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter'	1✔
13	import { attemptValidation } from '../utils/validation'	1✔
14	import { ContextMetadataKey } from '../constants/base'	1✔
15	import { createLogger } from '../factories/logger-factory'	1✔
16	import { Event } from '../@types/event'
17	import { getRemoteAddress } from '../utils/http'	1✔
18	import { IRateLimiter } from '../@types/utils'
19	import { isEventMatchingFilter } from '../utils/event'	1✔
20	import { messageSchema } from '../schemas/message-schema'	1✔
21	import { Settings } from '../@types/settings'
22	import { SocketAddress } from 'net'	1✔
23
24
25	const debug = createLogger('web-socket-adapter')	1✔
26	const debugHeartbeat = debug.extend('heartbeat')	1✔
27
28	const abortableMessageHandlers: WeakMap<WebSocket, IAbortable[]> = new WeakMap()	1✔
29
30	export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter {	1✔
31	public clientId: string
32	private clientAddress: SocketAddress
33	private alive: boolean
34	private subscriptions: Map<SubscriptionId, SubscriptionFilter[]>
35
36	public constructor(
37	private readonly client: WebSocket,	1✔
38	private readonly request: IncomingHttpMessage,	1✔
39	private readonly webSocketServer: IWebSocketServerAdapter,	1✔
40	private readonly createMessageHandler: Factory<IMessageHandler, [IncomingMessage, IWebSocketAdapter]>,	1✔
41	private readonly slidingWindowRateLimiter: Factory<IRateLimiter>,	1✔
42	private readonly settings: Factory<Settings>,	1✔
43	) {
44	super()	1✔
45	this.alive = true	1✔
46	this.subscriptions = new Map()	1✔
47
48	this.clientId = Buffer.from(this.request.headers['sec-websocket-key'] as string, 'base64').toString('hex')	1✔
49
50	const address = getRemoteAddress(this.request, this.settings())	1✔
51
52	this.clientAddress = new SocketAddress({	1✔
53	address: address,
54	family: address.indexOf(':') >= 0 ? 'ipv6' : 'ipv4',	1!
55	})
56
57	this.client	1✔
58	.on('error', (error) => {
59	if (error.name === 'RangeError' && error.message === 'Max payload size exceeded') {	×
60	console.error(`web-socket-adapter: client ${this.clientId} (${this.getClientAddress()}) sent payload too large`)	×
61	} else if (error.name === 'RangeError' && error.message === 'Invalid WebSocket frame: RSV1 must be clear') {	×
62	debug(`client ${this.clientId} (${this.getClientAddress()}) enabled compression`)	×
63	} else {
64	console.error(`web-socket-adapter: client error ${this.clientId} (${this.getClientAddress()}):`, error)	×
65	}
66
67	this.client.close()	×
68	})
69	.on('message', this.onClientMessage.bind(this))
70	.on('close', this.onClientClose.bind(this))
71	.on('pong', this.onClientPong.bind(this))
72	.on('ping', this.onClientPing.bind(this))
73
74	this	1✔
75	.on(WebSocketAdapterEvent.Heartbeat, this.onHeartbeat.bind(this))
76	.on(WebSocketAdapterEvent.Subscribe, this.onSubscribed.bind(this))
77	.on(WebSocketAdapterEvent.Unsubscribe, this.onUnsubscribed.bind(this))
78	.on(WebSocketAdapterEvent.Event, this.onSendEvent.bind(this))
79	.on(WebSocketAdapterEvent.Broadcast, this.onBroadcast.bind(this))
80	.on(WebSocketAdapterEvent.Message, this.sendMessage.bind(this))
81
82	debug('client %s connected from %s', this.clientId, this.clientAddress.address)	1✔
83	}
84
85	public getClientId(): string {
86	return this.clientId	×
87	}
88
89	public getClientAddress(): string {
90	return this.clientAddress.address	×
91	}
92
93	public onUnsubscribed(subscriptionId: string): void {
UNCOV 94	debug('client %s unsubscribed %s', this.clientId, subscriptionId)	×
UNCOV 95	this.subscriptions.delete(subscriptionId)	×
96	}
97
98	public onSubscribed(subscriptionId: string, filters: SubscriptionFilter[]): void {
UNCOV 99	debug('client %s subscribed %s to %o', this.clientId, subscriptionId, filters)	×
UNCOV 100	this.subscriptions.set(subscriptionId, filters)	×
101	}
102
103	public onBroadcast(event: Event): void {
UNCOV 104	this.webSocketServer.emit(WebSocketServerAdapterEvent.Broadcast, event)	×
UNCOV 105	if (cluster.isWorker && typeof process.send === 'function') {	×
106	process.send({	×
107	eventName: WebSocketServerAdapterEvent.Broadcast,
108	event,
109	})
110	}
111	}
112
113	public onSendEvent(event: Event): void {
UNCOV 114	this.subscriptions.forEach((filters, subscriptionId) => {	×
UNCOV 115	if (	×
UNCOV 116	filters.map(isEventMatchingFilter).some((isMatch) => isMatch(event))	×
117	) {
UNCOV 118	debug('sending event to client %s: %o', this.clientId, event)	×
UNCOV 119	this.sendMessage(createOutgoingEventMessage(subscriptionId, event))	×
120	}
121	})
122	}
123
124	private sendMessage(message: OutgoingMessage): void {
UNCOV 125	if (this.client.readyState !== WebSocket.OPEN) {	×
126	return	×
127	}
UNCOV 128	this.client.send(JSON.stringify(message))	×
129	}
130
131	public onHeartbeat(): void {
132	if (!this.alive && !this.subscriptions.size) {	×
133	console.error(`web-socket-adapter: pong timeout for client ${this.clientId} (${this.getClientAddress()})`)	×
134	this.client.close()	×
135	return	×
136	}
137
138	this.alive = false	×
139	this.client.ping()	×
140	debugHeartbeat('client %s ping', this.clientId)	×
141	}
142
143	public getSubscriptions(): Map<string, SubscriptionFilter[]> {
UNCOV 144	return new Map(this.subscriptions)	×
145	}
146
147	private async onClientMessage(raw: Buffer) {
UNCOV 148	this.alive = true	×
UNCOV 149	let abortable = false	×
UNCOV 150	let messageHandler: IMessageHandler & IAbortable \| undefined = undefined	×
UNCOV 151	try {	×
UNCOV 152	if (await this.isRateLimited(this.clientAddress.address)) {	×
153	this.sendMessage(createNoticeMessage('rate limited'))	×
154	return	×
155	}
156
UNCOV 157	const message = attemptValidation(messageSchema)(JSON.parse(raw.toString('utf8')))	×
158
UNCOV 159	message[ContextMetadataKey] = {	×
160	remoteAddress: this.clientAddress,
161	} as ContextMetadata
162
UNCOV 163	messageHandler = this.createMessageHandler([message, this]) as IMessageHandler & IAbortable	×
UNCOV 164	if (!messageHandler) {	×
165	console.error('web-socket-adapter: unhandled message: no handler found:', message)	×
166	return	×
167	}
168
UNCOV 169	abortable = typeof messageHandler.abort === 'function'	×
170
UNCOV 171	if (abortable) {	×
UNCOV 172	const handlers = abortableMessageHandlers.get(this.client) ?? []	×
UNCOV 173	handlers.push(messageHandler)	×
UNCOV 174	abortableMessageHandlers.set(this.client, handlers)	×
175	}
176
UNCOV 177	await messageHandler.handleMessage(message)	×
178	} catch (error) {
179	if (error instanceof Error) {	×
180	if (error.name === 'AbortError') {	×
181	console.error(`web-socket-adapter: abort from client ${this.clientId} (${this.getClientAddress()})`)	×
182	} else if (error.name === 'SyntaxError' \|\| error.name === 'ValidationError') {	×
183	if (typeof (error as any).annotate === 'function') {	×
184	debug('invalid message client %s (%s): %o', this.clientId, this.getClientAddress(), (error as any).annotate())	×
185	} else {
186	console.error(`web-socket-adapter: malformed message from client ${this.clientId} (${this.getClientAddress()}):`, error.message)	×
187	}
188	this.sendMessage(createNoticeMessage(`invalid: ${error.message}`))	×
189	} else {
190	console.error('web-socket-adapter: unable to handle message:', error)	×
191	}
192	} else {
193	console.error('web-socket-adapter: unable to handle message:', error)	×
194	}
195	} finally {
UNCOV 196	if (abortable && messageHandler) {	×
UNCOV 197	const handlers = abortableMessageHandlers.get(this.client)	×
UNCOV 198	if (handlers) {	×
UNCOV 199	const index = handlers.indexOf(messageHandler)	×
UNCOV 200	if (index >= 0) {	×
UNCOV 201	handlers.splice(index, 1)	×
202	}
203	}
204	}
205	}
206	}
207
208	private async isRateLimited(client: string): Promise<boolean> {
209	const {
210	rateLimits,
211	ipWhitelist = [],	×
UNCOV 212	} = this.settings().limits?.message ?? {}	×
213
UNCOV 214	if (!Array.isArray(rateLimits) \|\| !rateLimits.length \|\| ipWhitelist.includes(client)) {	×
UNCOV 215	return false	×
216	}
217
218	const rateLimiter = this.slidingWindowRateLimiter()	×
219
220	const hit = (period: number, rate: number) =>	×
221	rateLimiter.hit(	×
222	`${client}:message:${period}`,
223	1,
224	{ period, rate },
225	)
226
227	let limited = false	×
228	for (const { rate, period } of rateLimits) {	×
229	const isRateLimited = await hit(period, rate)	×
230
231
232	if (isRateLimited) {	×
233	debug('rate limited %s: %d messages / %d ms exceeded', client, rate, period)	×
234
235	limited = true	×
236	}
237	}
238
239	return limited	×
240	}
241
242	private onClientPong() {
243	debugHeartbeat('client %s pong', this.clientId)	×
244	this.alive = true	×
245	}
246
247	private onClientPing(data: any) {
248	debugHeartbeat('client %s ping', this.clientId)	×
249	this.client.pong(data)	×
250	this.alive = true	×
251	}
252
253	private onClientClose() {
UNCOV 254	this.alive = false	×
UNCOV 255	this.subscriptions.clear()	×
256
UNCOV 257	const handlers = abortableMessageHandlers.get(this.client)	×
UNCOV 258	if (Array.isArray(handlers) && handlers.length) {	×
259	for (const handler of handlers) {	×
260	try {	×
261	handler.abort()	×
262	} catch (error) {
263	console.error('Unable to abort message handler', error)	×
264	}
265	}
266	}
267
UNCOV 268	this.removeAllListeners()	×
UNCOV 269	this.client.removeAllListeners()	×
270	}
271	}

cameri / nostream / 24597013493

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous