feat(better-ws): added new package (#1989)
--------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by-agent: Codex
This commit is contained in:
Neko
@@ -41,11 +41,13 @@
|
||||
"dependencies": {
|
||||
"@guiiai/logg": "catalog:",
|
||||
"@moeru/std": "catalog:",
|
||||
"@proj-airi/better-ws": "workspace:^",
|
||||
"@proj-airi/server-shared": "workspace:^",
|
||||
"crossws": "catalog:",
|
||||
"h3": "catalog:",
|
||||
"nanoid": "catalog:",
|
||||
"srvx": "catalog:",
|
||||
"superjson": "catalog:"
|
||||
"superjson": "catalog:",
|
||||
"valibot": "catalog:"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,8 +1,12 @@
|
||||
import type { WebSocketBaseEvent, WebSocketEvents } from '@proj-airi/server-shared/types'
|
||||
|
||||
import type { ConsumerStickyAssignment } from './server-ws/airi/consumers'
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import { heartbeatFrameFrom, resolveEventDelivery, selectConsumerPeerId } from './index'
|
||||
import { heartbeatFrameFrom } from './server-ws/airi/codec'
|
||||
import { selectConsumerPeerId } from './server-ws/airi/consumers'
|
||||
import { resolveEventDelivery } from './server-ws/airi/routing'
|
||||
|
||||
function createInputTextEvent(
|
||||
overrides: Partial<WebSocketBaseEvent<'input:text', WebSocketEvents['input:text']>> = {},
|
||||
@@ -133,7 +137,7 @@ describe('selectConsumerPeerId', () => {
|
||||
})
|
||||
|
||||
it('keeps sticky delivery on the same consumer when available', () => {
|
||||
const stickyAssignments = new Map<string, string>()
|
||||
const stickyAssignments = new Map<string, ConsumerStickyAssignment>()
|
||||
|
||||
const firstSelectedPeerId = selectConsumerPeerId({
|
||||
eventType: 'input:text',
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
+35
-48
@@ -1,16 +1,16 @@
|
||||
import type { WebSocketEvent } from '@proj-airi/server-shared/types'
|
||||
|
||||
import { stringify } from 'superjson'
|
||||
import { stringify as stringifySuperJson } from 'superjson'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import {
|
||||
AiriWebSocketEventFormatError,
|
||||
createResponses,
|
||||
heartbeatFrameFrom,
|
||||
InvalidEventError,
|
||||
parseEvent,
|
||||
} from '.'
|
||||
stringifyEvent,
|
||||
} from './codec'
|
||||
|
||||
describe('airi websocket protocol codec', () => {
|
||||
describe('airi websocket codec', () => {
|
||||
it('parses superjson encoded events', () => {
|
||||
const event: WebSocketEvent = {
|
||||
type: 'module:authenticate',
|
||||
@@ -25,7 +25,7 @@ describe('airi websocket protocol codec', () => {
|
||||
},
|
||||
}
|
||||
|
||||
expect(parseEvent(stringify(event))).toEqual(event)
|
||||
expect(parseEvent(stringifySuperJson(event))).toEqual(event)
|
||||
})
|
||||
|
||||
it('falls back to plain JSON events', () => {
|
||||
@@ -45,61 +45,48 @@ describe('airi websocket protocol codec', () => {
|
||||
expect(parseEvent(JSON.stringify(event))).toEqual(event)
|
||||
})
|
||||
|
||||
it('rejects payloads without event type', () => {
|
||||
it('rejects invalid event envelopes', () => {
|
||||
expect(() => parseEvent('null'))
|
||||
.toThrow(AiriWebSocketEventFormatError)
|
||||
.toThrow(InvalidEventError)
|
||||
expect(() => parseEvent(JSON.stringify({ data: {} })))
|
||||
.toThrow(AiriWebSocketEventFormatError)
|
||||
})
|
||||
|
||||
it('rejects payloads with non-string event type', () => {
|
||||
.toThrow(InvalidEventError)
|
||||
expect(() => parseEvent(JSON.stringify({ type: 0, data: {} })))
|
||||
.toThrow(AiriWebSocketEventFormatError)
|
||||
})
|
||||
|
||||
it('rejects payloads without object event data', () => {
|
||||
.toThrow(InvalidEventError)
|
||||
expect(() => parseEvent(JSON.stringify({ type: 'module:authenticate' })))
|
||||
.toThrow(AiriWebSocketEventFormatError)
|
||||
.toThrow(InvalidEventError)
|
||||
expect(() => parseEvent(JSON.stringify({ type: 'module:authenticate', data: null })))
|
||||
.toThrow(AiriWebSocketEventFormatError)
|
||||
.toThrow(InvalidEventError)
|
||||
expect(() => parseEvent(JSON.stringify({ type: 'module:authenticate', data: 'secret' })))
|
||||
.toThrow(AiriWebSocketEventFormatError)
|
||||
})
|
||||
|
||||
it('rejects payloads with array event data', () => {
|
||||
.toThrow(InvalidEventError)
|
||||
expect(() => parseEvent(JSON.stringify({ type: 'module:authenticate', data: [] })))
|
||||
.toThrow(AiriWebSocketEventFormatError)
|
||||
.toThrow(InvalidEventError)
|
||||
})
|
||||
|
||||
it('classifies raw ping and pong control frames', () => {
|
||||
it('keeps validation cause and source on invalid event errors', () => {
|
||||
const source = { type: 'module:authenticate', data: 'secret' }
|
||||
|
||||
try {
|
||||
parseEvent(JSON.stringify(source))
|
||||
expect.unreachable('Expected invalid event parsing to throw.')
|
||||
}
|
||||
catch (error) {
|
||||
expect(error).toBeInstanceOf(InvalidEventError)
|
||||
expect(error).toMatchObject({ source })
|
||||
expect((error as InvalidEventError).cause).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
message: 'Expected event data to be a non-array object.',
|
||||
}),
|
||||
]))
|
||||
}
|
||||
})
|
||||
|
||||
it('detects raw ping and pong control frames', () => {
|
||||
expect(heartbeatFrameFrom('ping')).toBe('ping')
|
||||
expect(heartbeatFrameFrom('pong')).toBe('pong')
|
||||
expect(heartbeatFrameFrom('{"type":"ping"}')).toBeUndefined()
|
||||
})
|
||||
|
||||
/**
|
||||
* @example
|
||||
* expect(responses.peerAuthenticated('peer-1').type).toBe('peer:authenticated')
|
||||
* expect(responses.extensionAuthenticated({ id: 'airi-extension-chess' }).type).toBe('extension:authenticated')
|
||||
*/
|
||||
it('creates peer and extension authentication responses separately', () => {
|
||||
const responses = createResponses('server-1')
|
||||
|
||||
expect(responses.peerAuthenticated('peer-1')).toMatchObject({
|
||||
type: 'peer:authenticated',
|
||||
data: {
|
||||
authenticated: true,
|
||||
peerId: 'peer-1',
|
||||
},
|
||||
})
|
||||
expect(responses.extensionAuthenticated({ id: 'airi-extension-chess' })).toMatchObject({
|
||||
type: 'extension:authenticated',
|
||||
data: {
|
||||
authenticated: true,
|
||||
identity: {
|
||||
id: 'airi-extension-chess',
|
||||
},
|
||||
},
|
||||
})
|
||||
it('preserves raw string events when stringifying', () => {
|
||||
expect(stringifyEvent('raw-payload')).toBe('raw-payload')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,81 @@
|
||||
import type { WebSocketBaseEvent, WebSocketEvent } from '@proj-airi/server-shared/types'
|
||||
|
||||
import { MessageHeartbeatKind } from '@proj-airi/server-shared/types'
|
||||
import { parse, stringify } from 'superjson'
|
||||
import { check, objectWithRest, pipe, safeParse, string, unknown } from 'valibot'
|
||||
|
||||
const invalidAiriWebSocketEventFormatMessage = 'Invalid WebSocket event format.'
|
||||
|
||||
const eventDataSchema = pipe(
|
||||
unknown(),
|
||||
check(
|
||||
value => Boolean(value) && typeof value === 'object' && !Array.isArray(value),
|
||||
'Expected event data to be a non-array object.',
|
||||
),
|
||||
)
|
||||
|
||||
const eventEnvelopeSchema = objectWithRest({
|
||||
type: string(),
|
||||
data: eventDataSchema,
|
||||
}, unknown())
|
||||
|
||||
interface InvalidEventErrorOptions {
|
||||
cause?: unknown
|
||||
source?: unknown
|
||||
}
|
||||
|
||||
/** Error thrown when parsed websocket text is not an AIRI event envelope. */
|
||||
export class InvalidEventError extends Error {
|
||||
readonly source?: unknown
|
||||
|
||||
constructor(options: InvalidEventErrorOptions = {}) {
|
||||
super(invalidAiriWebSocketEventFormatMessage, { cause: options.cause })
|
||||
this.name = 'InvalidEventError'
|
||||
this.source = options.source
|
||||
}
|
||||
}
|
||||
|
||||
/** Checks whether an error came from AIRI websocket event envelope validation. */
|
||||
export function isInvalidEventError(error: unknown): error is InvalidEventError {
|
||||
return error instanceof InvalidEventError
|
||||
}
|
||||
|
||||
/** Detects raw ping/pong text frames that should not enter the event protocol. */
|
||||
export function heartbeatFrameFrom(text: string): MessageHeartbeatKind | undefined {
|
||||
if (text === MessageHeartbeatKind.Ping || text === MessageHeartbeatKind.Pong) {
|
||||
return text
|
||||
}
|
||||
}
|
||||
|
||||
/** Parses one AIRI websocket protocol event from SuperJSON or plain JSON text. */
|
||||
export function parseEvent(text: string): WebSocketEvent {
|
||||
// NOTICE:
|
||||
// SDK clients send events using superjson.stringify, so websocket runtime code must
|
||||
// use superjson.parse instead of message.json() or plain JSON.parse first.
|
||||
// JSON.parse on a superjson-encoded string returns the wrapper object
|
||||
// `{ json: {...}, meta: {...} }` with no protocol `type`, which breaks routing.
|
||||
// Keep this until all AIRI websocket clients share one non-wrapper wire format.
|
||||
let parsed: WebSocketEvent | undefined
|
||||
try {
|
||||
parsed = parse<WebSocketEvent>(text)
|
||||
}
|
||||
catch {
|
||||
parsed = undefined
|
||||
}
|
||||
|
||||
const potentialEvent = (parsed && typeof parsed === 'object' && 'type' in parsed)
|
||||
? parsed
|
||||
: JSON.parse(text)
|
||||
|
||||
const result = safeParse(eventEnvelopeSchema, potentialEvent)
|
||||
if (!result.success) {
|
||||
throw new InvalidEventError({ cause: result.issues, source: potentialEvent })
|
||||
}
|
||||
|
||||
return potentialEvent as WebSocketEvent
|
||||
}
|
||||
|
||||
/** Serializes one AIRI websocket protocol event with the existing SuperJSON wire format. */
|
||||
export function stringifyEvent(event: WebSocketBaseEvent<string, unknown> | string) {
|
||||
return typeof event === 'string' ? event : stringify(event)
|
||||
}
|
||||
+6
-6
@@ -1,13 +1,13 @@
|
||||
import type { ServerWsStickyAssignment } from '.'
|
||||
import type { ConsumerStickyAssignment } from './consumers'
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import {
|
||||
createConsumerOrchestrator,
|
||||
selectConsumerPeerId,
|
||||
} from '.'
|
||||
} from './consumers'
|
||||
|
||||
describe('server-ws consumer selection', () => {
|
||||
describe('airi websocket consumer selection', () => {
|
||||
it('selects highest priority then earliest registration', () => {
|
||||
expect(selectConsumerPeerId({
|
||||
eventType: 'event:test',
|
||||
@@ -36,7 +36,7 @@ describe('server-ws consumer selection', () => {
|
||||
})
|
||||
|
||||
it('preserves sticky assignment for the same sticky key', () => {
|
||||
const stickyAssignments = new Map<string, ServerWsStickyAssignment>()
|
||||
const stickyAssignments = new Map<string, ConsumerStickyAssignment>()
|
||||
const delivery = { mode: 'consumer-group' as const, group: 'workers', selection: 'sticky' as const, stickyKey: 'job-1' }
|
||||
const candidates = [
|
||||
{ peerId: 'a', priority: 0, registeredAt: 1, authenticated: true },
|
||||
@@ -73,7 +73,7 @@ describe('server-ws consumer selection', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('server-ws consumer registry', () => {
|
||||
describe('airi websocket consumer registry', () => {
|
||||
it('registers and unregisters consumers', () => {
|
||||
const registry = createConsumerOrchestrator()
|
||||
|
||||
@@ -106,7 +106,7 @@ describe('server-ws consumer registry', () => {
|
||||
})
|
||||
|
||||
it('keeps sticky assignments isolated for delimiter-like event and group names', () => {
|
||||
const stickyAssignments = new Map<string, ServerWsStickyAssignment>()
|
||||
const stickyAssignments = new Map<string, ConsumerStickyAssignment>()
|
||||
const candidates = [
|
||||
{ peerId: 'event::group-target', priority: 0, registeredAt: 1, authenticated: true },
|
||||
{ peerId: 'other-target', priority: 0, registeredAt: 2, authenticated: true },
|
||||
@@ -0,0 +1,301 @@
|
||||
import type { DeliveryConfig } from '@proj-airi/server-shared/types'
|
||||
|
||||
const DEFAULT_CONSUMER_GROUP = 'default'
|
||||
|
||||
interface ConsumerRegistryRef {
|
||||
event: string
|
||||
group: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Candidate peer metadata used for AIRI consumer selection.
|
||||
*/
|
||||
export interface ConsumerSelectionCandidate {
|
||||
/** Peer id available to receive the event. */
|
||||
peerId: string
|
||||
/** Higher values are selected before lower values. */
|
||||
priority: number
|
||||
/** Timestamp captured when the peer registered as a consumer. */
|
||||
registeredAt: number
|
||||
/** Whether the peer has completed protocol-level authentication. */
|
||||
authenticated: boolean
|
||||
/** Explicit `false` excludes the peer from selection. */
|
||||
healthy?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Stored AIRI consumer registration.
|
||||
*/
|
||||
export interface ConsumerRegistration {
|
||||
/** Protocol event type consumed by the peer. */
|
||||
event: string
|
||||
/** Normalized consumer group name. */
|
||||
group: string
|
||||
/** Peer id that registered for the event/group pair. */
|
||||
peerId: string
|
||||
/** Higher values are selected before lower values. */
|
||||
priority: number
|
||||
/** Timestamp captured when the peer registered as a consumer. */
|
||||
registeredAt: number
|
||||
}
|
||||
|
||||
/**
|
||||
* Sticky AIRI consumer assignment stored by the consumer selector.
|
||||
*/
|
||||
export interface ConsumerStickyAssignment {
|
||||
/** Protocol event type the sticky assignment belongs to. */
|
||||
event: string
|
||||
/** Normalized consumer group the sticky assignment belongs to. */
|
||||
group: string
|
||||
/** Peer selected for the sticky key. */
|
||||
peerId: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether a delivery mode targets the AIRI consumer registry.
|
||||
*/
|
||||
export function isConsumerDeliveryMode(mode: unknown): mode is 'consumer' | 'consumer-group' {
|
||||
return mode === 'consumer' || mode === 'consumer-group'
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalizes delivery mode for AIRI consumer registration.
|
||||
*
|
||||
* Before:
|
||||
* - undefined with group "workers"
|
||||
*
|
||||
* After:
|
||||
* - "consumer-group"
|
||||
*/
|
||||
export function normalizeConsumerMode(mode: unknown, group?: string): 'consumer' | 'consumer-group' {
|
||||
if (isConsumerDeliveryMode(mode)) {
|
||||
return mode
|
||||
}
|
||||
|
||||
return group ? 'consumer-group' : 'consumer'
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalizes AIRI consumer priority.
|
||||
*
|
||||
* Before:
|
||||
* - NaN
|
||||
*
|
||||
* After:
|
||||
* - 0
|
||||
*/
|
||||
export function normalizeConsumerPriority(priority: unknown) {
|
||||
return typeof priority === 'number' && Number.isFinite(priority)
|
||||
? priority
|
||||
: 0
|
||||
}
|
||||
|
||||
function normalizeConsumerGroup(mode: 'consumer' | 'consumer-group', group?: string) {
|
||||
if (mode === 'consumer') {
|
||||
return DEFAULT_CONSUMER_GROUP
|
||||
}
|
||||
|
||||
return group || DEFAULT_CONSUMER_GROUP
|
||||
}
|
||||
|
||||
function sortConsumers(entries: Array<Pick<ConsumerSelectionCandidate, 'peerId' | 'priority' | 'registeredAt'>>) {
|
||||
return [...entries].sort((left, right) => {
|
||||
if (right.priority !== left.priority) {
|
||||
return right.priority - left.priority
|
||||
}
|
||||
|
||||
return left.registeredAt - right.registeredAt
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Selects a concrete peer for AIRI consumer-style delivery modes.
|
||||
*
|
||||
* Sticky and round-robin state are keyed with structured JSON tuples so event,
|
||||
* group, and sticky key values may contain delimiter-like text safely.
|
||||
*/
|
||||
export function selectConsumerPeerId(options: {
|
||||
eventType: string
|
||||
fromPeerId: string
|
||||
delivery?: DeliveryConfig
|
||||
candidates: ConsumerSelectionCandidate[]
|
||||
roundRobinCursor?: Map<string, number>
|
||||
stickyAssignments?: Map<string, ConsumerStickyAssignment>
|
||||
}) {
|
||||
const { candidates, delivery, eventType, fromPeerId } = options
|
||||
if (!delivery || !isConsumerDeliveryMode(delivery.mode)) {
|
||||
return
|
||||
}
|
||||
|
||||
const normalizedGroup = normalizeConsumerGroup(delivery.mode, delivery.group)
|
||||
const registryKey = JSON.stringify([eventType, normalizedGroup])
|
||||
const availableEntries = sortConsumers(
|
||||
candidates
|
||||
.filter(entry => entry.peerId !== fromPeerId)
|
||||
.filter(entry => entry.authenticated && entry.healthy !== false),
|
||||
)
|
||||
|
||||
if (availableEntries.length === 0) {
|
||||
return
|
||||
}
|
||||
|
||||
const selection = delivery.selection ?? 'first'
|
||||
if (selection === 'sticky' && delivery.stickyKey) {
|
||||
const stickyRegistryKey = JSON.stringify([eventType, normalizedGroup, delivery.stickyKey])
|
||||
const stickyAssignment = options.stickyAssignments?.get(stickyRegistryKey)
|
||||
if (stickyAssignment && stickyAssignment.peerId !== fromPeerId) {
|
||||
const stickyCandidate = availableEntries.find(entry => entry.peerId === stickyAssignment.peerId)
|
||||
if (stickyCandidate) {
|
||||
return stickyAssignment.peerId
|
||||
}
|
||||
}
|
||||
|
||||
const selected = availableEntries[0]
|
||||
options.stickyAssignments?.set(stickyRegistryKey, { event: eventType, group: normalizedGroup, peerId: selected.peerId })
|
||||
return selected.peerId
|
||||
}
|
||||
|
||||
if (selection === 'round-robin') {
|
||||
const cursor = options.roundRobinCursor?.get(registryKey) ?? 0
|
||||
const selected = availableEntries[cursor % availableEntries.length]
|
||||
options.roundRobinCursor?.set(registryKey, (cursor + 1) % availableEntries.length)
|
||||
return selected.peerId
|
||||
}
|
||||
|
||||
return availableEntries[0].peerId
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the AIRI consumer delivery orchestrator for websocket peers.
|
||||
*
|
||||
* The orchestrator owns registration, unregister, listing, selection, and
|
||||
* sticky/round-robin cleanup state for AIRI consumer routing.
|
||||
*/
|
||||
export function createConsumerOrchestrator() {
|
||||
const consumerRegistry = new Map<string, Map<string, Map<string, ConsumerRegistration>>>()
|
||||
const consumerKeysByPeer = new Map<string, Map<string, ConsumerRegistryRef>>()
|
||||
const deliveryRoundRobinCursor = new Map<string, number>()
|
||||
const stickyAssignments = new Map<string, ConsumerStickyAssignment>()
|
||||
|
||||
function removeStickyAssignmentsFor(event: string, group: string, peerId?: string) {
|
||||
for (const [stickyKey, assignment] of stickyAssignments.entries()) {
|
||||
if (peerId && assignment.peerId !== peerId) {
|
||||
continue
|
||||
}
|
||||
|
||||
if (assignment.event === event && assignment.group === group) {
|
||||
stickyAssignments.delete(stickyKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
register(input: { peerId: string, event: string, mode: 'consumer' | 'consumer-group', group?: string, priority?: number }) {
|
||||
const normalizedGroup = normalizeConsumerGroup(input.mode, input.group)
|
||||
const registryKey = JSON.stringify([input.event, normalizedGroup])
|
||||
let groups = consumerRegistry.get(input.event)
|
||||
if (!groups) {
|
||||
groups = new Map()
|
||||
consumerRegistry.set(input.event, groups)
|
||||
}
|
||||
|
||||
let peersForGroup = groups.get(normalizedGroup)
|
||||
if (!peersForGroup) {
|
||||
peersForGroup = new Map()
|
||||
groups.set(normalizedGroup, peersForGroup)
|
||||
}
|
||||
|
||||
const didGrowMembership = !peersForGroup.has(input.peerId)
|
||||
peersForGroup.set(input.peerId, {
|
||||
event: input.event,
|
||||
group: normalizedGroup,
|
||||
peerId: input.peerId,
|
||||
priority: normalizeConsumerPriority(input.priority),
|
||||
registeredAt: Date.now(),
|
||||
})
|
||||
if (didGrowMembership) {
|
||||
deliveryRoundRobinCursor.delete(registryKey)
|
||||
}
|
||||
|
||||
let registrations = consumerKeysByPeer.get(input.peerId)
|
||||
if (!registrations) {
|
||||
registrations = new Map()
|
||||
consumerKeysByPeer.set(input.peerId, registrations)
|
||||
}
|
||||
registrations.set(registryKey, { event: input.event, group: normalizedGroup })
|
||||
},
|
||||
unregister(input: { peerId: string, event: string, mode: 'consumer' | 'consumer-group', group?: string }) {
|
||||
const normalizedGroup = normalizeConsumerGroup(input.mode, input.group)
|
||||
const registryKey = JSON.stringify([input.event, normalizedGroup])
|
||||
const groups = consumerRegistry.get(input.event)
|
||||
const peersForGroup = groups?.get(normalizedGroup)
|
||||
const didDelete = peersForGroup?.delete(input.peerId) ?? false
|
||||
|
||||
if (!didDelete) {
|
||||
return
|
||||
}
|
||||
|
||||
deliveryRoundRobinCursor.delete(registryKey)
|
||||
if (peersForGroup?.size === 0) {
|
||||
groups?.delete(normalizedGroup)
|
||||
}
|
||||
if (groups?.size === 0) {
|
||||
consumerRegistry.delete(input.event)
|
||||
}
|
||||
|
||||
const registrations = consumerKeysByPeer.get(input.peerId)
|
||||
registrations?.delete(registryKey)
|
||||
if (registrations?.size === 0) {
|
||||
consumerKeysByPeer.delete(input.peerId)
|
||||
}
|
||||
|
||||
removeStickyAssignmentsFor(input.event, normalizedGroup, input.peerId)
|
||||
},
|
||||
unregisterPeer(peerId: string) {
|
||||
const registrations = consumerKeysByPeer.get(peerId)
|
||||
if (!registrations?.size) {
|
||||
return
|
||||
}
|
||||
|
||||
for (const registration of registrations.values()) {
|
||||
const { event, group } = registration
|
||||
const groups = consumerRegistry.get(event)
|
||||
const peersForGroup = groups?.get(group)
|
||||
peersForGroup?.delete(peerId)
|
||||
deliveryRoundRobinCursor.delete(JSON.stringify([event, group]))
|
||||
if (peersForGroup?.size === 0) {
|
||||
groups?.delete(group)
|
||||
}
|
||||
if (groups?.size === 0) {
|
||||
consumerRegistry.delete(event)
|
||||
}
|
||||
|
||||
removeStickyAssignmentsFor(event, group, peerId)
|
||||
}
|
||||
|
||||
consumerKeysByPeer.delete(peerId)
|
||||
},
|
||||
listFor(input: { event: string, mode: 'consumer' | 'consumer-group', group?: string }) {
|
||||
const normalizedGroup = normalizeConsumerGroup(input.mode, input.group)
|
||||
return [...consumerRegistry.get(input.event)?.get(normalizedGroup)?.values() ?? []]
|
||||
},
|
||||
select(input: {
|
||||
eventType: string
|
||||
fromPeerId: string
|
||||
delivery?: DeliveryConfig
|
||||
candidates: ConsumerSelectionCandidate[]
|
||||
}) {
|
||||
return selectConsumerPeerId({
|
||||
...input,
|
||||
roundRobinCursor: deliveryRoundRobinCursor,
|
||||
stickyAssignments,
|
||||
})
|
||||
},
|
||||
clear() {
|
||||
consumerRegistry.clear()
|
||||
consumerKeysByPeer.clear()
|
||||
deliveryRoundRobinCursor.clear()
|
||||
stickyAssignments.clear()
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1,354 +1,27 @@
|
||||
import type { DeliveryConfig, ExtensionIdentity, MessageHeartbeat, MetadataEventSource, WebSocketBaseEvent, WebSocketEvent } from '@proj-airi/server-shared/types'
|
||||
|
||||
import type {
|
||||
RouteContext,
|
||||
RouteDecision,
|
||||
RouteMiddleware,
|
||||
} from '../../middlewares'
|
||||
import type { Peer } from '../../types'
|
||||
|
||||
import { ServerErrorMessages } from '@proj-airi/server-shared'
|
||||
import {
|
||||
getProtocolEventMetadata,
|
||||
MessageHeartbeatKind,
|
||||
WebSocketEventSource,
|
||||
} from '@proj-airi/server-shared/types'
|
||||
import { nanoid } from 'nanoid'
|
||||
import { parse, stringify } from 'superjson'
|
||||
|
||||
import packageJSON from '../../../package.json'
|
||||
|
||||
import { createEventCodec, createGatewayLifecycle } from '../core'
|
||||
|
||||
const invalidAiriWebSocketEventFormatMessage = 'Invalid WebSocket event format.'
|
||||
|
||||
/**
|
||||
* Close details surfaced by the websocket runtime for AIRI peer shutdown logging.
|
||||
*/
|
||||
export interface AiriServerWsCloseDetails {
|
||||
/** WebSocket close code when the runtime reports one. */
|
||||
code?: number
|
||||
/** WebSocket close reason when the runtime reports one. */
|
||||
reason?: string
|
||||
/** Whether the runtime considers the close clean. */
|
||||
wasClean?: unknown
|
||||
}
|
||||
|
||||
/**
|
||||
* Error thrown when a websocket message parses as JSON but is not an AIRI event envelope.
|
||||
*
|
||||
* Use when:
|
||||
* - The runtime must distinguish malformed event envelopes from invalid JSON text
|
||||
*
|
||||
* Expects:
|
||||
* - Callers convert this to the protocol `invalidEventFormat` response
|
||||
*
|
||||
* Returns:
|
||||
* - A typed error for invalid AIRI websocket event envelopes
|
||||
*/
|
||||
export class AiriWebSocketEventFormatError extends Error {
|
||||
constructor() {
|
||||
super(invalidAiriWebSocketEventFormatMessage)
|
||||
this.name = 'AiriWebSocketEventFormatError'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the AIRI websocket gateway wrapper.
|
||||
*
|
||||
* Use when:
|
||||
* - `setupApp(...)` needs a gateway object to mount on `/ws`
|
||||
*
|
||||
* Expects:
|
||||
* - `handler` preserves the existing AIRI websocket lifecycle behavior
|
||||
*
|
||||
* Returns:
|
||||
* - A gateway object compatible with H3 `defineWebSocketHandler(...)`
|
||||
*/
|
||||
export function createGateway(input: {
|
||||
handler: {
|
||||
open: (peer: Peer) => void
|
||||
message: (peer: Peer, message: { text: () => string }) => void
|
||||
error: (peer: Peer, error: unknown) => void
|
||||
close: (peer: Peer, details?: AiriServerWsCloseDetails) => void
|
||||
}
|
||||
dispose?: () => void
|
||||
}) {
|
||||
return createGatewayLifecycle({
|
||||
handler: input.handler,
|
||||
dispose: input.dispose,
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates metadata for events emitted by the AIRI websocket runtime.
|
||||
*
|
||||
* Use when:
|
||||
* - The server sends protocol events to connected peers
|
||||
* - Response events should preserve parent event correlation
|
||||
*
|
||||
* Expects:
|
||||
* - `serverInstanceId` identifies the active server runtime instance
|
||||
*
|
||||
* Returns:
|
||||
* - AIRI protocol metadata with server source and event id
|
||||
*/
|
||||
export function createEventMetadata(
|
||||
serverInstanceId: string,
|
||||
parentId?: string,
|
||||
): { source: MetadataEventSource, event: { id: string, parentId?: string } } {
|
||||
return {
|
||||
event: {
|
||||
id: nanoid(),
|
||||
parentId,
|
||||
},
|
||||
source: {
|
||||
kind: 'plugin',
|
||||
plugin: {
|
||||
id: WebSocketEventSource.Server,
|
||||
version: packageJSON.version,
|
||||
},
|
||||
id: serverInstanceId,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates AIRI server response event factories.
|
||||
*
|
||||
* Use when:
|
||||
* - WebSocket handlers need stable response event shapes
|
||||
*
|
||||
* Expects:
|
||||
* - `serverInstanceId` identifies the current server runtime
|
||||
*
|
||||
* Returns:
|
||||
* - Factory methods for protocol responses emitted by the server
|
||||
*/
|
||||
export function createResponses(serverInstanceId: string) {
|
||||
return {
|
||||
authenticated(parentId?: string) {
|
||||
return {
|
||||
type: 'module:authenticated',
|
||||
data: { authenticated: true },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
peerAuthenticated(peerId: string, parentId?: string) {
|
||||
return {
|
||||
type: 'peer:authenticated',
|
||||
data: { authenticated: true, peerId },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
extensionAuthenticated(identity: ExtensionIdentity, parentId?: string) {
|
||||
return {
|
||||
type: 'extension:authenticated',
|
||||
data: { identity, authenticated: true },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
notAuthenticated(parentId?: string) {
|
||||
return {
|
||||
type: 'error',
|
||||
data: { message: ServerErrorMessages.notAuthenticated },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
error(message: string, parentId?: string) {
|
||||
return {
|
||||
type: 'error',
|
||||
data: { message },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
heartbeat(kind: MessageHeartbeatKind, message: MessageHeartbeat | string, parentId?: string) {
|
||||
return {
|
||||
type: 'transport:connection:heartbeat',
|
||||
data: { kind, message, at: Date.now() },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether an error came from AIRI websocket event envelope validation.
|
||||
*
|
||||
* Use when:
|
||||
* - Message handlers need to map invalid envelopes to protocol errors
|
||||
*
|
||||
* Expects:
|
||||
* - Parser code throws {@link AiriWebSocketEventFormatError} for envelope failures
|
||||
*
|
||||
* Returns:
|
||||
* - `true` when the error should become `ServerErrorMessages.invalidEventFormat`
|
||||
*/
|
||||
export function isAiriWebSocketEventFormatError(error: unknown): error is AiriWebSocketEventFormatError {
|
||||
return error instanceof AiriWebSocketEventFormatError
|
||||
}
|
||||
|
||||
/**
|
||||
* Detects raw websocket heartbeat control frames surfaced as text payloads.
|
||||
*
|
||||
* Use when:
|
||||
* - A websocket runtime forwards ping/pong frames through the normal message callback
|
||||
* - The runtime should ignore transport heartbeats instead of treating them as protocol JSON
|
||||
*
|
||||
* Expects:
|
||||
* - Raw text payloads such as `ping` and `pong`
|
||||
*
|
||||
* Returns:
|
||||
* - The heartbeat kind when the text is a control frame, otherwise `undefined`
|
||||
*/
|
||||
export function heartbeatFrameFrom(text: string): MessageHeartbeatKind | undefined {
|
||||
if (text === MessageHeartbeatKind.Ping || text === MessageHeartbeatKind.Pong) {
|
||||
return text
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parses one AIRI websocket protocol event.
|
||||
*
|
||||
* Use when:
|
||||
* - Reading text messages from WebSocket peers
|
||||
*
|
||||
* Expects:
|
||||
* - SDK clients may send `superjson.stringify(...)`
|
||||
* - External clients may send plain JSON
|
||||
*
|
||||
* Returns:
|
||||
* - A WebSocket event with a string `type`
|
||||
*/
|
||||
export function parseEvent(text: string): WebSocketEvent {
|
||||
// NOTICE:
|
||||
// SDK clients send events using superjson.stringify, so websocket runtime code must
|
||||
// use superjson.parse instead of message.json() or plain JSON.parse first.
|
||||
// JSON.parse on a superjson-encoded string returns the wrapper object
|
||||
// `{ json: {...}, meta: {...} }` with no protocol `type`, which breaks routing.
|
||||
// Keep this until all AIRI websocket clients share one non-wrapper wire format.
|
||||
let parsed: WebSocketEvent | undefined
|
||||
try {
|
||||
parsed = parse<WebSocketEvent>(text)
|
||||
}
|
||||
catch {
|
||||
parsed = undefined
|
||||
}
|
||||
|
||||
const potentialEvent = (parsed && typeof parsed === 'object' && 'type' in parsed)
|
||||
? parsed
|
||||
: JSON.parse(text)
|
||||
|
||||
if (
|
||||
!potentialEvent
|
||||
|| typeof potentialEvent !== 'object'
|
||||
|| !('type' in potentialEvent)
|
||||
|| typeof potentialEvent.type !== 'string'
|
||||
|| !('data' in potentialEvent)
|
||||
|| !potentialEvent.data
|
||||
|| typeof potentialEvent.data !== 'object'
|
||||
|| Array.isArray(potentialEvent.data)
|
||||
) {
|
||||
throw new AiriWebSocketEventFormatError()
|
||||
}
|
||||
|
||||
return potentialEvent as WebSocketEvent
|
||||
}
|
||||
|
||||
/**
|
||||
* Serializes one AIRI websocket protocol event.
|
||||
*
|
||||
* Use when:
|
||||
* - Sending AIRI events through WebSocket peers
|
||||
*
|
||||
* Expects:
|
||||
* - `event` is already protocol-shaped
|
||||
*
|
||||
* Returns:
|
||||
* - SuperJSON text payload matching existing runtime behavior
|
||||
*/
|
||||
export function stringifyEvent(event: WebSocketBaseEvent<string, unknown> | string) {
|
||||
return typeof event === 'string' ? event : stringify(event)
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves the effective event delivery policy.
|
||||
*
|
||||
* Use when:
|
||||
* - Protocol defaults should be merged with route-level delivery overrides
|
||||
* - Routing needs to know whether the event should broadcast or target one consumer
|
||||
*
|
||||
* Expects:
|
||||
* - Route delivery to override protocol metadata field-by-field
|
||||
*
|
||||
* Returns:
|
||||
* - The merged broadcast/consumer delivery policy, or `undefined` when unrestricted
|
||||
*/
|
||||
export function resolveEventDelivery(event: WebSocketEvent): DeliveryConfig | undefined {
|
||||
const eventMetadata = getProtocolEventMetadata(event.type)
|
||||
const defaultDelivery = eventMetadata?.delivery
|
||||
const routeDelivery = event.route?.delivery
|
||||
|
||||
if (!defaultDelivery && !routeDelivery) {
|
||||
return undefined
|
||||
}
|
||||
|
||||
return {
|
||||
...defaultDelivery,
|
||||
...routeDelivery,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates event serializer hooks used by server websocket adapters.
|
||||
*
|
||||
* Use when:
|
||||
* - A gateway wants protocol-specific parsing, stringifying, and control-frame detection
|
||||
*
|
||||
* Expects:
|
||||
* - Callers route raw control frames before protocol events
|
||||
*
|
||||
* Returns:
|
||||
* - A reusable `server-ws/core` codec configured for AIRI events
|
||||
*/
|
||||
export function createEventSerializer() {
|
||||
return createEventCodec<WebSocketEvent>({
|
||||
parse: parseEvent,
|
||||
stringify: stringifyEvent,
|
||||
detectControlFrame: heartbeatFrameFrom,
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Iterates event middlewares in declaration order until one returns a decision.
|
||||
*
|
||||
* Use when:
|
||||
* - The websocket runtime needs the first route decision from configured middleware
|
||||
*
|
||||
* Expects:
|
||||
* - Middleware functions are ordered by caller policy
|
||||
*
|
||||
* Returns:
|
||||
* - The first route decision, or `undefined` when no middleware decided
|
||||
*/
|
||||
export function forEachEventMiddlewares(input: {
|
||||
event: WebSocketEvent
|
||||
fromPeer: RouteContext['fromPeer']
|
||||
peers: Map<string, RouteContext['fromPeer']>
|
||||
destinations?: RouteContext['destinations']
|
||||
middleware: RouteMiddleware[]
|
||||
}): RouteDecision | undefined {
|
||||
const context: RouteContext = {
|
||||
event: input.event,
|
||||
fromPeer: input.fromPeer,
|
||||
peers: input.peers,
|
||||
destinations: input.destinations,
|
||||
}
|
||||
|
||||
for (const middleware of input.middleware) {
|
||||
const result = middleware(context)
|
||||
if (result) {
|
||||
return result
|
||||
}
|
||||
}
|
||||
}
|
||||
export {
|
||||
heartbeatFrameFrom,
|
||||
InvalidEventError,
|
||||
isInvalidEventError,
|
||||
parseEvent,
|
||||
stringifyEvent,
|
||||
} from './codec'
|
||||
export {
|
||||
createConsumerOrchestrator,
|
||||
isConsumerDeliveryMode,
|
||||
normalizeConsumerMode,
|
||||
normalizeConsumerPriority,
|
||||
selectConsumerPeerId,
|
||||
} from './consumers'
|
||||
export type {
|
||||
ConsumerRegistration,
|
||||
ConsumerSelectionCandidate,
|
||||
ConsumerStickyAssignment,
|
||||
} from './consumers'
|
||||
export {
|
||||
resolveHealthCheckIntervalMs,
|
||||
serverWsDefaultHeartbeatTtlMs,
|
||||
serverWsHealthCheckIntervalDivisor,
|
||||
serverWsMinimumHealthCheckIntervalMs,
|
||||
} from './liveness'
|
||||
export { createEventMetadata, createResponses } from './responses'
|
||||
export { forEachEventMiddlewares, resolveEventDelivery } from './routing'
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import {
|
||||
resolveHealthCheckIntervalMs,
|
||||
serverWsDefaultHeartbeatTtlMs,
|
||||
serverWsMinimumHealthCheckIntervalMs,
|
||||
} from './liveness'
|
||||
|
||||
describe('airi websocket liveness policy', () => {
|
||||
it('uses the AIRI default heartbeat TTL', () => {
|
||||
expect(serverWsDefaultHeartbeatTtlMs).toBe(60_000)
|
||||
expect(resolveHealthCheckIntervalMs(serverWsDefaultHeartbeatTtlMs)).toBe(12_000)
|
||||
})
|
||||
|
||||
it('keeps health checks at least five seconds apart', () => {
|
||||
expect(serverWsMinimumHealthCheckIntervalMs).toBe(5_000)
|
||||
expect(resolveHealthCheckIntervalMs(1_000)).toBe(serverWsMinimumHealthCheckIntervalMs)
|
||||
expect(resolveHealthCheckIntervalMs(24_999)).toBe(serverWsMinimumHealthCheckIntervalMs)
|
||||
expect(resolveHealthCheckIntervalMs(25_000)).toBe(serverWsMinimumHealthCheckIntervalMs)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,13 @@
|
||||
/** Default heartbeat read timeout. */
|
||||
export const serverWsDefaultHeartbeatTtlMs = 60_000
|
||||
|
||||
/** Number of liveness checks scheduled within one heartbeat TTL. */
|
||||
export const serverWsHealthCheckIntervalDivisor = 5
|
||||
|
||||
/** Minimum interval to avoid busy liveness loops. */
|
||||
export const serverWsMinimumHealthCheckIntervalMs = 5_000
|
||||
|
||||
/** Resolves the AIRI heartbeat health-check interval in milliseconds. */
|
||||
export function resolveHealthCheckIntervalMs(heartbeatTtlMs: number) {
|
||||
return Math.max(serverWsMinimumHealthCheckIntervalMs, Math.floor(heartbeatTtlMs / serverWsHealthCheckIntervalDivisor))
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
import { WebSocketEventSource } from '@proj-airi/server-shared/types'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import packageJSON from '../../../package.json'
|
||||
|
||||
import { createEventMetadata, createResponses } from './responses'
|
||||
|
||||
describe('airi websocket responses', () => {
|
||||
it('creates server metadata with source and parent event ids', () => {
|
||||
const metadata = createEventMetadata('server-1', 'parent-event-1')
|
||||
|
||||
expect(metadata.source).toEqual({
|
||||
kind: 'plugin',
|
||||
plugin: {
|
||||
id: WebSocketEventSource.Server,
|
||||
version: packageJSON.version,
|
||||
},
|
||||
id: 'server-1',
|
||||
})
|
||||
expect(metadata.event).toEqual({
|
||||
id: expect.any(String),
|
||||
parentId: 'parent-event-1',
|
||||
})
|
||||
})
|
||||
|
||||
it('creates peer and extension authentication response shapes', () => {
|
||||
const responses = createResponses('server-1')
|
||||
|
||||
expect(responses.peerAuthenticated('peer-1', 'event-1')).toMatchObject({
|
||||
type: 'peer:authenticated',
|
||||
data: {
|
||||
authenticated: true,
|
||||
peerId: 'peer-1',
|
||||
},
|
||||
metadata: {
|
||||
source: {
|
||||
id: 'server-1',
|
||||
plugin: {
|
||||
id: WebSocketEventSource.Server,
|
||||
version: packageJSON.version,
|
||||
},
|
||||
},
|
||||
event: {
|
||||
parentId: 'event-1',
|
||||
},
|
||||
},
|
||||
})
|
||||
expect(responses.extensionAuthenticated({ id: 'airi-extension-chess' }, 'event-2')).toMatchObject({
|
||||
type: 'extension:authenticated',
|
||||
data: {
|
||||
authenticated: true,
|
||||
identity: {
|
||||
id: 'airi-extension-chess',
|
||||
},
|
||||
},
|
||||
metadata: {
|
||||
source: {
|
||||
id: 'server-1',
|
||||
plugin: {
|
||||
id: WebSocketEventSource.Server,
|
||||
version: packageJSON.version,
|
||||
},
|
||||
},
|
||||
event: {
|
||||
parentId: 'event-2',
|
||||
},
|
||||
},
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,76 @@
|
||||
import type { ExtensionIdentity, MessageHeartbeat, MessageHeartbeatKind, MetadataEventSource, WebSocketEvent } from '@proj-airi/server-shared/types'
|
||||
|
||||
import { ServerErrorMessages } from '@proj-airi/server-shared'
|
||||
import { WebSocketEventSource } from '@proj-airi/server-shared/types'
|
||||
import { nanoid } from 'nanoid'
|
||||
|
||||
import packageJSON from '../../../package.json'
|
||||
|
||||
/** Creates AIRI server event metadata and preserves optional parent correlation. */
|
||||
export function createEventMetadata(
|
||||
serverInstanceId: string,
|
||||
parentId?: string,
|
||||
): { source: MetadataEventSource, event: { id: string, parentId?: string } } {
|
||||
return {
|
||||
event: {
|
||||
id: nanoid(),
|
||||
parentId,
|
||||
},
|
||||
source: {
|
||||
kind: 'plugin',
|
||||
plugin: {
|
||||
id: WebSocketEventSource.Server,
|
||||
version: packageJSON.version,
|
||||
},
|
||||
id: serverInstanceId,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/** Creates AIRI server response event factories. */
|
||||
export function createResponses(serverInstanceId: string) {
|
||||
return {
|
||||
authenticated(parentId?: string) {
|
||||
return {
|
||||
type: 'module:authenticated',
|
||||
data: { authenticated: true },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
peerAuthenticated(peerId: string, parentId?: string) {
|
||||
return {
|
||||
type: 'peer:authenticated',
|
||||
data: { authenticated: true, peerId },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
extensionAuthenticated(identity: ExtensionIdentity, parentId?: string) {
|
||||
return {
|
||||
type: 'extension:authenticated',
|
||||
data: { identity, authenticated: true },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
notAuthenticated(parentId?: string) {
|
||||
return {
|
||||
type: 'error',
|
||||
data: { message: ServerErrorMessages.notAuthenticated },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
error(message: string, parentId?: string) {
|
||||
return {
|
||||
type: 'error',
|
||||
data: { message },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
heartbeat(kind: MessageHeartbeatKind, message: MessageHeartbeat | string, parentId?: string) {
|
||||
return {
|
||||
type: 'transport:connection:heartbeat',
|
||||
data: { kind, message, at: Date.now() },
|
||||
metadata: createEventMetadata(serverInstanceId, parentId),
|
||||
} satisfies WebSocketEvent<Record<string, unknown>>
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
import type { DeliveryConfig, WebSocketEvent } from '@proj-airi/server-shared/types'
|
||||
|
||||
import type { RouteContext, RouteDecision, RouteMiddleware } from '../../middlewares'
|
||||
|
||||
import { getProtocolEventMetadata } from '@proj-airi/server-shared/types'
|
||||
|
||||
/**
|
||||
* Resolves the effective event delivery policy.
|
||||
*
|
||||
* Use when:
|
||||
* - Protocol defaults should be merged with route-level delivery overrides
|
||||
* - Routing needs to know whether the event should broadcast or target one consumer
|
||||
*
|
||||
* Expects:
|
||||
* - Route delivery to override protocol metadata field-by-field
|
||||
*
|
||||
* Returns:
|
||||
* - The merged broadcast/consumer delivery policy, or `undefined` when unrestricted
|
||||
*/
|
||||
export function resolveEventDelivery(event: WebSocketEvent): DeliveryConfig | undefined {
|
||||
const eventMetadata = getProtocolEventMetadata(event.type)
|
||||
const defaultDelivery = eventMetadata?.delivery
|
||||
const routeDelivery = event.route?.delivery
|
||||
|
||||
if (!defaultDelivery && !routeDelivery) {
|
||||
return undefined
|
||||
}
|
||||
|
||||
return {
|
||||
...defaultDelivery,
|
||||
...routeDelivery,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Iterates event middlewares in declaration order until one returns a decision.
|
||||
*
|
||||
* Use when:
|
||||
* - The websocket runtime needs the first route decision from configured middleware
|
||||
*
|
||||
* Expects:
|
||||
* - Middleware functions are ordered by caller policy
|
||||
*
|
||||
* Returns:
|
||||
* - The first route decision, or `undefined` when no middleware decided
|
||||
*/
|
||||
export function forEachEventMiddlewares(input: {
|
||||
event: WebSocketEvent
|
||||
fromPeer: RouteContext['fromPeer']
|
||||
peers: Map<string, RouteContext['fromPeer']>
|
||||
destinations?: RouteContext['destinations']
|
||||
middleware: RouteMiddleware[]
|
||||
}): RouteDecision | undefined {
|
||||
const context: RouteContext = {
|
||||
event: input.event,
|
||||
fromPeer: input.fromPeer,
|
||||
peers: input.peers,
|
||||
destinations: input.destinations,
|
||||
}
|
||||
|
||||
for (const middleware of input.middleware) {
|
||||
const result = middleware(context)
|
||||
if (result) {
|
||||
return result
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,543 +0,0 @@
|
||||
/**
|
||||
* Delivery settings used by the reusable websocket gateway.
|
||||
*
|
||||
* @param TMode - Delivery mode literals accepted by the adapter.
|
||||
*/
|
||||
export interface ServerWsDeliveryConfig<TMode extends string = 'broadcast' | 'consumer' | 'consumer-group'> {
|
||||
/**
|
||||
* Delivery mode selected by the protocol adapter.
|
||||
*
|
||||
* @default undefined
|
||||
*/
|
||||
mode?: TMode
|
||||
/**
|
||||
* Optional consumer group.
|
||||
*
|
||||
* @default "default" for consumer delivery modes.
|
||||
*/
|
||||
group?: string
|
||||
/**
|
||||
* Selection strategy within the target consumer set.
|
||||
*
|
||||
* @default "first"
|
||||
*/
|
||||
selection?: 'first' | 'priority' | 'sticky' | 'round-robin'
|
||||
/**
|
||||
* Sticky routing key used when `selection` is `sticky`.
|
||||
*
|
||||
* @default undefined
|
||||
*/
|
||||
stickyKey?: string
|
||||
/**
|
||||
* Whether missing consumers should be surfaced as an error by the adapter.
|
||||
*
|
||||
* @default false
|
||||
*/
|
||||
required?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Delivery settings accepted by the reusable consumer registry.
|
||||
*
|
||||
* @param TMode - Consumer delivery mode literals accepted by the adapter.
|
||||
*/
|
||||
export type ServerWsConsumerDeliveryConfig<TMode extends string = 'consumer' | 'consumer-group'> = ServerWsDeliveryConfig<TMode>
|
||||
|
||||
/**
|
||||
* Candidate peer metadata used for consumer selection.
|
||||
*/
|
||||
export interface ServerWsConsumerSelectionCandidate {
|
||||
/** Peer id available to receive the event. */
|
||||
peerId: string
|
||||
/** Higher values are selected before lower values. */
|
||||
priority: number
|
||||
/** Timestamp captured when the peer registered as a consumer. */
|
||||
registeredAt: number
|
||||
/** Whether the peer has completed protocol-level authentication. */
|
||||
authenticated: boolean
|
||||
/** Explicit `false` excludes the peer from selection. */
|
||||
healthy?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Stored consumer registration.
|
||||
*/
|
||||
export interface ServerWsConsumerRegistration {
|
||||
/** Protocol event type consumed by the peer. */
|
||||
event: string
|
||||
/** Normalized consumer group name. */
|
||||
group: string
|
||||
/** Peer id that registered for the event/group pair. */
|
||||
peerId: string
|
||||
/** Higher values are selected before lower values. */
|
||||
priority: number
|
||||
/** Timestamp captured when the peer registered as a consumer. */
|
||||
registeredAt: number
|
||||
}
|
||||
|
||||
/**
|
||||
* Describes protocol-agnostic text encoding and decoding for websocket events.
|
||||
*
|
||||
* @param TEvent - Event envelope shape owned by the protocol adapter.
|
||||
*/
|
||||
export interface ServerWsEventCodec<TEvent> {
|
||||
/** Parses one text payload into a protocol event. */
|
||||
parse: (text: string) => TEvent
|
||||
/** Serializes one protocol event or pre-serialized payload for peer sending. */
|
||||
stringify: (event: TEvent | string) => string
|
||||
/** Detects raw transport control payloads that should not enter protocol routing. */
|
||||
detectControlFrame?: (text: string) => string | undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Describes a websocket handler object accepted by H3 `defineWebSocketHandler`.
|
||||
*
|
||||
* @param TPeer - Runtime peer object accepted by lifecycle callbacks.
|
||||
* @param TMessage - Runtime message object accepted by the message callback.
|
||||
* @param TCloseDetails - Runtime close details object accepted by the close callback.
|
||||
*/
|
||||
export interface ServerWsGatewayHandler<TPeer = unknown, TMessage = unknown, TCloseDetails = unknown> {
|
||||
/** Called when a peer opens a websocket connection. */
|
||||
open?: (peer: TPeer) => void
|
||||
/** Called when a peer sends one websocket message. */
|
||||
message?: (peer: TPeer, message: TMessage) => void
|
||||
/** Called when the websocket runtime reports an error. */
|
||||
error?: (peer: TPeer, error: unknown) => void
|
||||
/** Called when a peer closes a websocket connection. */
|
||||
close?: (peer: TPeer, details?: TCloseDetails) => void
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimal websocket peer shape used by the reusable gateway.
|
||||
*/
|
||||
export interface ServerWsPeer {
|
||||
/** Stable peer id assigned by the websocket runtime. */
|
||||
get id(): string
|
||||
/** Sends one payload to the peer. */
|
||||
send: (data: unknown, options?: { compress?: boolean }) => number | void | undefined
|
||||
/** Closes the peer connection when the runtime exposes an explicit close hook. */
|
||||
close?: () => void
|
||||
/** WebSocket ready state when exposed by the runtime. */
|
||||
readyState?: number
|
||||
/** Request metadata associated with the websocket upgrade. */
|
||||
request?: {
|
||||
/** Request URL associated with the websocket upgrade. */
|
||||
url?: string
|
||||
/** Request headers associated with the websocket upgrade. */
|
||||
headers?: Headers
|
||||
}
|
||||
/** Remote peer address when exposed by the runtime. */
|
||||
remoteAddress?: string
|
||||
}
|
||||
|
||||
/** Default heartbeat read timeout used by the websocket gateway. */
|
||||
export const serverWsDefaultHeartbeatTtlMs = 60_000
|
||||
|
||||
/** Miss count where a peer becomes unhealthy but remains connected. */
|
||||
export const serverWsHealthCheckMissesUnhealthy = 5
|
||||
|
||||
/** Miss count where a peer is considered dead and should be closed. */
|
||||
export const serverWsHealthCheckMissesDead = serverWsHealthCheckMissesUnhealthy * 2
|
||||
|
||||
const DEFAULT_CONSUMER_GROUP = 'default'
|
||||
|
||||
interface ServerWsConsumerRegistryRef {
|
||||
event: string
|
||||
group: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Sticky consumer assignment stored by the reusable consumer selector.
|
||||
*/
|
||||
export interface ServerWsStickyAssignment {
|
||||
/** Protocol event type the sticky assignment belongs to. */
|
||||
event: string
|
||||
/** Normalized consumer group the sticky assignment belongs to. */
|
||||
group: string
|
||||
/** Peer selected for the sticky key. */
|
||||
peerId: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a websocket event codec from explicit parser and serializer callbacks.
|
||||
*
|
||||
* Use when:
|
||||
* - A protocol adapter wants to plug its own event envelope into `server-ws/core`
|
||||
*
|
||||
* Expects:
|
||||
* - Parser and serializer preserve the adapter's current wire format
|
||||
*
|
||||
* Returns:
|
||||
* - A protocol-agnostic codec object consumed by gateway code
|
||||
*/
|
||||
export function createEventCodec<TEvent>(codec: ServerWsEventCodec<TEvent>) {
|
||||
return codec
|
||||
}
|
||||
|
||||
/**
|
||||
* Wraps websocket lifecycle callbacks and disposal as a reusable mount object.
|
||||
*
|
||||
* Use when:
|
||||
* - Adapters need one stable lifecycle shape for server mounting
|
||||
*
|
||||
* Expects:
|
||||
* - `handler` contains already-bound protocol behavior
|
||||
*
|
||||
* Returns:
|
||||
* - A handler plus idempotent disposal hook
|
||||
*/
|
||||
export function createGatewayLifecycle<TPeer, TMessage, TCloseDetails = unknown>(input: {
|
||||
handler: ServerWsGatewayHandler<TPeer, TMessage, TCloseDetails>
|
||||
dispose?: () => void
|
||||
}) {
|
||||
let disposed = false
|
||||
|
||||
return {
|
||||
handler: input.handler,
|
||||
dispose: () => {
|
||||
if (disposed) {
|
||||
return
|
||||
}
|
||||
|
||||
disposed = true
|
||||
input.dispose?.()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves the interval used for heartbeat health checks.
|
||||
*
|
||||
* Use when:
|
||||
* - Gateway code needs to convert heartbeat TTL into periodic miss checks
|
||||
*
|
||||
* Expects:
|
||||
* - Very small TTL values should still avoid busy intervals
|
||||
*
|
||||
* Returns:
|
||||
* - Interval in milliseconds
|
||||
*/
|
||||
export function resolveServerWsHealthCheckIntervalMs(heartbeatTtlMs: number) {
|
||||
return Math.max(5_000, Math.floor(heartbeatTtlMs / serverWsHealthCheckMissesUnhealthy))
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a typed peer store around websocket peer state.
|
||||
*
|
||||
* Use when:
|
||||
* - A gateway needs stable peer lookup, iteration, and cleanup
|
||||
*
|
||||
* Expects:
|
||||
* - `TState` contains protocol-specific peer state
|
||||
*
|
||||
* Returns:
|
||||
* - A small registry over peers keyed by peer id
|
||||
*/
|
||||
export function createServerWsPeerStore<TState extends { peer: ServerWsPeer }>() {
|
||||
const peers = new Map<string, TState>()
|
||||
|
||||
return {
|
||||
peers,
|
||||
get(peerId: string) {
|
||||
return peers.get(peerId)
|
||||
},
|
||||
set(peerId: string, state: TState) {
|
||||
peers.set(peerId, state)
|
||||
return state
|
||||
},
|
||||
delete(peerId: string) {
|
||||
return peers.delete(peerId)
|
||||
},
|
||||
clear() {
|
||||
peers.clear()
|
||||
},
|
||||
values() {
|
||||
return peers.values()
|
||||
},
|
||||
entries() {
|
||||
return peers.entries()
|
||||
},
|
||||
size() {
|
||||
return peers.size
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether a delivery mode targets the consumer registry.
|
||||
*
|
||||
* Use when:
|
||||
* - A protocol adapter receives broad delivery modes but must call consumer-only APIs
|
||||
*
|
||||
* Expects:
|
||||
* - Non-consumer modes such as `broadcast` should remain outside the consumer registry
|
||||
*
|
||||
* Returns:
|
||||
* - `true` for `consumer` and `consumer-group`
|
||||
*/
|
||||
export function isConsumerDeliveryMode(mode: unknown): mode is ServerWsConsumerDeliveryConfig['mode'] {
|
||||
return mode === 'consumer' || mode === 'consumer-group'
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalizes delivery mode for consumer registration.
|
||||
*
|
||||
* Before:
|
||||
* - undefined with group "workers"
|
||||
*
|
||||
* After:
|
||||
* - "consumer-group"
|
||||
*/
|
||||
export function normalizeConsumerMode(mode: unknown, group?: string): 'consumer' | 'consumer-group' {
|
||||
if (isConsumerDeliveryMode(mode)) {
|
||||
return mode!
|
||||
}
|
||||
|
||||
return group ? 'consumer-group' : 'consumer'
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalizes consumer priority.
|
||||
*
|
||||
* Before:
|
||||
* - NaN
|
||||
*
|
||||
* After:
|
||||
* - 0
|
||||
*/
|
||||
export function normalizeConsumerPriority(priority: unknown) {
|
||||
return typeof priority === 'number' && Number.isFinite(priority)
|
||||
? priority
|
||||
: 0
|
||||
}
|
||||
|
||||
function normalizeConsumerGroup(mode: ServerWsConsumerDeliveryConfig['mode'], group?: string) {
|
||||
if (mode === 'consumer') {
|
||||
return DEFAULT_CONSUMER_GROUP
|
||||
}
|
||||
|
||||
return group || DEFAULT_CONSUMER_GROUP
|
||||
}
|
||||
|
||||
function getConsumerRegistryKey(event: string, group: string) {
|
||||
return JSON.stringify([event, group])
|
||||
}
|
||||
|
||||
function getStickyRegistryKey(event: string, group: string, stickyKey: string) {
|
||||
return JSON.stringify([event, group, stickyKey])
|
||||
}
|
||||
|
||||
function sortConsumers(entries: Array<Pick<ServerWsConsumerSelectionCandidate, 'peerId' | 'priority' | 'registeredAt'>>) {
|
||||
return [...entries].sort((left, right) => {
|
||||
if (right.priority !== left.priority) {
|
||||
return right.priority - left.priority
|
||||
}
|
||||
|
||||
return left.registeredAt - right.registeredAt
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Selects a concrete consumer peer for consumer-style delivery modes.
|
||||
*
|
||||
* Use when:
|
||||
* - An event should be sent to exactly one registered consumer
|
||||
* - Sticky or round-robin routing needs to be resolved against live peer metadata
|
||||
*
|
||||
* Expects:
|
||||
* - Candidates already describe authenticated and health state
|
||||
*
|
||||
* Returns:
|
||||
* - The selected peer id, or `undefined` when no eligible consumer is available
|
||||
*/
|
||||
export function selectConsumerPeerId(options: {
|
||||
eventType: string
|
||||
fromPeerId: string
|
||||
delivery?: ServerWsDeliveryConfig
|
||||
candidates: ServerWsConsumerSelectionCandidate[]
|
||||
roundRobinCursor?: Map<string, number>
|
||||
stickyAssignments?: Map<string, ServerWsStickyAssignment>
|
||||
}) {
|
||||
const { candidates, delivery, eventType, fromPeerId } = options
|
||||
if (!delivery || (delivery.mode !== 'consumer' && delivery.mode !== 'consumer-group')) {
|
||||
return
|
||||
}
|
||||
|
||||
const normalizedGroup = normalizeConsumerGroup(delivery.mode, delivery.group)
|
||||
const registryKey = getConsumerRegistryKey(eventType, normalizedGroup)
|
||||
const availableEntries = sortConsumers(
|
||||
candidates
|
||||
.filter(entry => entry.peerId !== fromPeerId)
|
||||
.filter(entry => entry.authenticated && entry.healthy !== false),
|
||||
)
|
||||
|
||||
if (availableEntries.length === 0) {
|
||||
return
|
||||
}
|
||||
|
||||
const selection = delivery.selection ?? 'first'
|
||||
if (selection === 'sticky' && delivery.stickyKey) {
|
||||
const stickyRegistryKey = getStickyRegistryKey(eventType, normalizedGroup, delivery.stickyKey)
|
||||
const stickyAssignment = options.stickyAssignments?.get(stickyRegistryKey)
|
||||
if (stickyAssignment && stickyAssignment.peerId !== fromPeerId) {
|
||||
const stickyCandidate = availableEntries.find(entry => entry.peerId === stickyAssignment.peerId)
|
||||
if (stickyCandidate) {
|
||||
return stickyAssignment.peerId
|
||||
}
|
||||
}
|
||||
|
||||
const selected = availableEntries[0]
|
||||
options.stickyAssignments?.set(stickyRegistryKey, { event: eventType, group: normalizedGroup, peerId: selected.peerId })
|
||||
return selected.peerId
|
||||
}
|
||||
|
||||
if (selection === 'round-robin') {
|
||||
const cursor = options.roundRobinCursor?.get(registryKey) ?? 0
|
||||
const selected = availableEntries[cursor % availableEntries.length]
|
||||
options.roundRobinCursor?.set(registryKey, (cursor + 1) % availableEntries.length)
|
||||
return selected.peerId
|
||||
}
|
||||
|
||||
return availableEntries[0].peerId
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a reusable consumer delivery orchestrator for websocket peers.
|
||||
*
|
||||
* Use when:
|
||||
* - A protocol adapter supports one-consumer delivery or consumer groups
|
||||
*
|
||||
* Expects:
|
||||
* - Peer liveness is checked by the caller before delivery
|
||||
*
|
||||
* Returns:
|
||||
* - Registration, unregister, listing, selection, and cleanup helpers
|
||||
*/
|
||||
export function createConsumerOrchestrator() {
|
||||
const consumerRegistry = new Map<string, Map<string, Map<string, ServerWsConsumerRegistration>>>()
|
||||
const consumerKeysByPeer = new Map<string, Map<string, ServerWsConsumerRegistryRef>>()
|
||||
const deliveryRoundRobinCursor = new Map<string, number>()
|
||||
const stickyAssignments = new Map<string, ServerWsStickyAssignment>()
|
||||
|
||||
function removeStickyAssignmentsFor(event: string, group: string, peerId?: string) {
|
||||
for (const [stickyKey, assignment] of stickyAssignments.entries()) {
|
||||
if (peerId && assignment.peerId !== peerId) {
|
||||
continue
|
||||
}
|
||||
|
||||
if (assignment.event === event && assignment.group === group) {
|
||||
stickyAssignments.delete(stickyKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
register(input: { peerId: string, event: string, mode: ServerWsConsumerDeliveryConfig['mode'], group?: string, priority?: number }) {
|
||||
const normalizedGroup = normalizeConsumerGroup(input.mode, input.group)
|
||||
const registryKey = getConsumerRegistryKey(input.event, normalizedGroup)
|
||||
let groups = consumerRegistry.get(input.event)
|
||||
if (!groups) {
|
||||
groups = new Map()
|
||||
consumerRegistry.set(input.event, groups)
|
||||
}
|
||||
|
||||
let peersForGroup = groups.get(normalizedGroup)
|
||||
if (!peersForGroup) {
|
||||
peersForGroup = new Map()
|
||||
groups.set(normalizedGroup, peersForGroup)
|
||||
}
|
||||
|
||||
const didGrowMembership = !peersForGroup.has(input.peerId)
|
||||
peersForGroup.set(input.peerId, {
|
||||
event: input.event,
|
||||
group: normalizedGroup,
|
||||
peerId: input.peerId,
|
||||
priority: normalizeConsumerPriority(input.priority),
|
||||
registeredAt: Date.now(),
|
||||
})
|
||||
if (didGrowMembership) {
|
||||
deliveryRoundRobinCursor.delete(registryKey)
|
||||
}
|
||||
|
||||
let registrations = consumerKeysByPeer.get(input.peerId)
|
||||
if (!registrations) {
|
||||
registrations = new Map()
|
||||
consumerKeysByPeer.set(input.peerId, registrations)
|
||||
}
|
||||
registrations.set(registryKey, { event: input.event, group: normalizedGroup })
|
||||
},
|
||||
unregister(input: { peerId: string, event: string, mode: ServerWsConsumerDeliveryConfig['mode'], group?: string }) {
|
||||
const normalizedGroup = normalizeConsumerGroup(input.mode, input.group)
|
||||
const registryKey = getConsumerRegistryKey(input.event, normalizedGroup)
|
||||
const groups = consumerRegistry.get(input.event)
|
||||
const peersForGroup = groups?.get(normalizedGroup)
|
||||
const didDelete = peersForGroup?.delete(input.peerId) ?? false
|
||||
|
||||
if (!didDelete) {
|
||||
return
|
||||
}
|
||||
|
||||
deliveryRoundRobinCursor.delete(registryKey)
|
||||
if (peersForGroup?.size === 0) {
|
||||
groups?.delete(normalizedGroup)
|
||||
}
|
||||
if (groups?.size === 0) {
|
||||
consumerRegistry.delete(input.event)
|
||||
}
|
||||
|
||||
const registrations = consumerKeysByPeer.get(input.peerId)
|
||||
registrations?.delete(registryKey)
|
||||
if (registrations?.size === 0) {
|
||||
consumerKeysByPeer.delete(input.peerId)
|
||||
}
|
||||
|
||||
removeStickyAssignmentsFor(input.event, normalizedGroup, input.peerId)
|
||||
},
|
||||
unregisterPeer(peerId: string) {
|
||||
const registrations = consumerKeysByPeer.get(peerId)
|
||||
if (!registrations?.size) {
|
||||
return
|
||||
}
|
||||
|
||||
for (const registration of registrations.values()) {
|
||||
const { event, group } = registration
|
||||
const groups = consumerRegistry.get(event)
|
||||
const peersForGroup = groups?.get(group)
|
||||
peersForGroup?.delete(peerId)
|
||||
deliveryRoundRobinCursor.delete(getConsumerRegistryKey(event, group))
|
||||
if (peersForGroup?.size === 0) {
|
||||
groups?.delete(group)
|
||||
}
|
||||
if (groups?.size === 0) {
|
||||
consumerRegistry.delete(event)
|
||||
}
|
||||
|
||||
removeStickyAssignmentsFor(event, group, peerId)
|
||||
}
|
||||
|
||||
consumerKeysByPeer.delete(peerId)
|
||||
},
|
||||
listFor(input: { event: string, mode: ServerWsConsumerDeliveryConfig['mode'], group?: string }) {
|
||||
const normalizedGroup = normalizeConsumerGroup(input.mode, input.group)
|
||||
return [...consumerRegistry.get(input.event)?.get(normalizedGroup)?.values() ?? []]
|
||||
},
|
||||
select(input: {
|
||||
eventType: string
|
||||
fromPeerId: string
|
||||
delivery?: ServerWsDeliveryConfig
|
||||
candidates: ServerWsConsumerSelectionCandidate[]
|
||||
}) {
|
||||
return selectConsumerPeerId({
|
||||
...input,
|
||||
roundRobinCursor: deliveryRoundRobinCursor,
|
||||
stickyAssignments,
|
||||
})
|
||||
},
|
||||
clear() {
|
||||
consumerRegistry.clear()
|
||||
consumerKeysByPeer.clear()
|
||||
deliveryRoundRobinCursor.clear()
|
||||
stickyAssignments.clear()
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -12,6 +12,7 @@ const serveMocks = vi.hoisted(() => {
|
||||
|
||||
const closeCall = vi.fn(async () => {})
|
||||
const disposeCall = vi.fn(() => {})
|
||||
const createH3CrossWsPluginCall = vi.fn(() => ({ name: 'better-ws-h3-plugin' }))
|
||||
const setupAppCall = vi.fn(() => ({
|
||||
app: {
|
||||
fetch: vi.fn(async () => ({ crossws: {} })),
|
||||
@@ -22,6 +23,7 @@ const serveMocks = vi.hoisted(() => {
|
||||
|
||||
return {
|
||||
closeCall,
|
||||
createH3CrossWsPluginCall,
|
||||
disposeCall,
|
||||
rejectServe: (error: Error) => rejectServe?.(error),
|
||||
resolveServe: () => resolveServe?.(),
|
||||
@@ -34,15 +36,14 @@ vi.mock('h3', () => ({
|
||||
H3: class {
|
||||
get = vi.fn()
|
||||
},
|
||||
defineWebSocketHandler: vi.fn(handler => handler),
|
||||
serve: vi.fn(() => ({
|
||||
serve: serveMocks.serveCall,
|
||||
close: serveMocks.closeCall,
|
||||
})),
|
||||
}))
|
||||
|
||||
vi.mock('crossws/server', () => ({
|
||||
plugin: vi.fn(() => ({})),
|
||||
vi.mock('@proj-airi/better-ws/server/h3', () => ({
|
||||
createH3CrossWsPlugin: serveMocks.createH3CrossWsPluginCall,
|
||||
}))
|
||||
|
||||
vi.mock('./index', () => ({
|
||||
@@ -72,6 +73,9 @@ describe('createServer', async () => {
|
||||
|
||||
await Promise.all([firstStart, secondStart])
|
||||
expect(serveMocks.serveCall).toHaveBeenCalledTimes(1)
|
||||
expect(serveMocks.createH3CrossWsPluginCall).toHaveBeenCalledWith(expect.objectContaining({
|
||||
fetch: expect.any(Function),
|
||||
}))
|
||||
})
|
||||
|
||||
it('clears the single-flight state when start fails', async () => {
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import type { H3CrossWsApp, H3CrossWsResponse } from '@proj-airi/better-ws/server/h3'
|
||||
|
||||
import type { AppOptions } from '..'
|
||||
|
||||
import { isIP } from 'node:net'
|
||||
@@ -5,7 +7,7 @@ import { networkInterfaces } from 'node:os'
|
||||
|
||||
import { useLogg } from '@guiiai/logg'
|
||||
import { merge } from '@moeru/std'
|
||||
import { plugin as ws } from 'crossws/server'
|
||||
import { createH3CrossWsPlugin } from '@proj-airi/better-ws/server/h3'
|
||||
import { serve } from 'h3'
|
||||
|
||||
import { normalizeLoggerConfig, setupApp } from '..'
|
||||
@@ -151,13 +153,15 @@ export function createServer(opts?: ServerOptions): Server {
|
||||
startTask = (async () => {
|
||||
const secureEnabled = options?.tlsConfig != null
|
||||
const h3App = setupApp(options)
|
||||
const crossWsApp = {
|
||||
fetch: async request => await h3App.app.fetch(request) as H3CrossWsResponse,
|
||||
} satisfies H3CrossWsApp
|
||||
|
||||
const port = options.port
|
||||
const hostname = options.hostname
|
||||
|
||||
const instance = serve(h3App.app, {
|
||||
// @ts-expect-error - the .crossws property wasn't extended in types
|
||||
plugins: [ws({ resolve: async req => (await h3App.app.fetch(req)).crossws })],
|
||||
plugins: [createH3CrossWsPlugin(crossWsApp)],
|
||||
port,
|
||||
hostname,
|
||||
tls: options?.tlsConfig || undefined,
|
||||
|
||||
@@ -0,0 +1,235 @@
|
||||
import type { WebSocketEvent } from '@proj-airi/server-shared/types'
|
||||
|
||||
import type { Peer } from './types'
|
||||
|
||||
import { parse, stringify } from 'superjson'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { setupApp } from './index'
|
||||
|
||||
interface TestWebSocketHandler {
|
||||
open?: (peer: Peer) => void
|
||||
message?: (peer: Peer, message: { text: () => string }) => void
|
||||
close?: (peer: Peer, details?: { code?: number, reason?: string, wasClean?: unknown }) => void
|
||||
}
|
||||
|
||||
interface TestWsServer {
|
||||
accept: (
|
||||
adapter: { id: string, send: (message: { text: () => string }) => void | number, close?: () => void },
|
||||
options: { state: { rawPeer: Peer } },
|
||||
) => void
|
||||
peers: {
|
||||
get: (peerId: string) => { receive: (message: { text: () => string }) => void } | undefined
|
||||
}
|
||||
remove: (peerId: string, details?: { code?: number, reason?: string, wasClean?: unknown }) => void
|
||||
}
|
||||
|
||||
const h3Mocks = vi.hoisted(() => ({
|
||||
handlers: new Map<string, unknown>(),
|
||||
}))
|
||||
|
||||
vi.mock('h3', () => ({
|
||||
H3: class {
|
||||
get(path: string, handler: unknown) {
|
||||
h3Mocks.handlers.set(path, handler)
|
||||
}
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@proj-airi/better-ws/server/h3', () => ({
|
||||
toH3Handler: vi.fn((server: TestWsServer, options: { state: (peer: Peer) => { rawPeer: Peer } }) => ({
|
||||
open(peer: Peer) {
|
||||
server.accept({
|
||||
id: peer.id,
|
||||
send: message => peer.send(message.text()),
|
||||
close: () => peer.close?.(),
|
||||
}, {
|
||||
state: options.state(peer),
|
||||
})
|
||||
},
|
||||
message(peer: Peer, message: { text: () => string }) {
|
||||
server.peers.get(peer.id)?.receive(message)
|
||||
},
|
||||
close(peer: Peer, details?: { code?: number, reason?: string, wasClean?: unknown }) {
|
||||
server.remove(peer.id, details)
|
||||
},
|
||||
})),
|
||||
}))
|
||||
|
||||
function createPeer(id: string) {
|
||||
const sent: string[] = []
|
||||
const send: Peer['send'] = (data) => {
|
||||
sent.push(String(data))
|
||||
}
|
||||
|
||||
return {
|
||||
peer: {
|
||||
id,
|
||||
send: vi.fn(send),
|
||||
close: vi.fn(),
|
||||
request: { url: `/ws?id=${id}` },
|
||||
remoteAddress: '127.0.0.1',
|
||||
} satisfies Peer,
|
||||
sent,
|
||||
}
|
||||
}
|
||||
|
||||
function wsHandler() {
|
||||
const handler = h3Mocks.handlers.get('/ws') as TestWebSocketHandler | undefined
|
||||
if (!handler) {
|
||||
throw new Error('Expected setupApp to register a /ws websocket handler.')
|
||||
}
|
||||
|
||||
return handler
|
||||
}
|
||||
|
||||
function sendEvent(
|
||||
handler: TestWebSocketHandler,
|
||||
peer: Peer,
|
||||
event: WebSocketEvent,
|
||||
) {
|
||||
handler.message?.(peer, { text: () => stringify(event) })
|
||||
}
|
||||
|
||||
function decodeEvents(sent: string[]) {
|
||||
return sent.map(message => parse<WebSocketEvent>(message))
|
||||
}
|
||||
|
||||
function createExtensionModuleAnnounceEvent(): WebSocketEvent {
|
||||
return {
|
||||
type: 'extension:module:announce',
|
||||
data: {
|
||||
name: 'memory',
|
||||
possibleEvents: [],
|
||||
identity: {
|
||||
id: 'memory-module-1',
|
||||
extension: {
|
||||
id: 'extension-1',
|
||||
},
|
||||
},
|
||||
},
|
||||
metadata: {
|
||||
source: {
|
||||
kind: 'plugin',
|
||||
id: 'extension-1',
|
||||
plugin: {
|
||||
id: 'extension-1',
|
||||
},
|
||||
},
|
||||
event: {
|
||||
id: 'announce-1',
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
describe('setupApp websocket liveness', () => {
|
||||
beforeEach(() => {
|
||||
h3Mocks.handlers.clear()
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('broadcasts extension module unhealthy events from better-ws liveness checks', () => {
|
||||
const runtime = setupApp({ heartbeat: { readTimeout: 20_000 } })
|
||||
const handler = wsHandler()
|
||||
const observer = createPeer('observer')
|
||||
const modulePeer = createPeer('module-peer')
|
||||
|
||||
handler.open?.(observer.peer)
|
||||
handler.open?.(modulePeer.peer)
|
||||
sendEvent(handler, modulePeer.peer, createExtensionModuleAnnounceEvent())
|
||||
observer.sent.length = 0
|
||||
|
||||
vi.advanceTimersByTime(25_000)
|
||||
|
||||
expect(decodeEvents(observer.sent)).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
type: 'registry:modules:health:unhealthy',
|
||||
data: {
|
||||
name: 'memory',
|
||||
identity: {
|
||||
id: 'memory-module-1',
|
||||
extension: {
|
||||
id: 'extension-1',
|
||||
},
|
||||
},
|
||||
reason: 'heartbeat late',
|
||||
},
|
||||
}),
|
||||
]))
|
||||
|
||||
runtime.dispose()
|
||||
})
|
||||
|
||||
it('de-announces expired extension modules when better-ws removes stale peers', () => {
|
||||
const runtime = setupApp({ heartbeat: { readTimeout: 20_000 } })
|
||||
const handler = wsHandler()
|
||||
const observer = createPeer('observer')
|
||||
const modulePeer = createPeer('module-peer')
|
||||
|
||||
handler.open?.(observer.peer)
|
||||
handler.open?.(modulePeer.peer)
|
||||
sendEvent(handler, modulePeer.peer, createExtensionModuleAnnounceEvent())
|
||||
observer.sent.length = 0
|
||||
|
||||
vi.advanceTimersByTime(25_000)
|
||||
handler.message?.(observer.peer, { text: () => 'pong' })
|
||||
observer.sent.length = 0
|
||||
vi.advanceTimersByTime(25_000)
|
||||
|
||||
expect(modulePeer.peer.close).toHaveBeenCalledOnce()
|
||||
expect(decodeEvents(observer.sent)).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
type: 'extension:module:de-announced',
|
||||
data: expect.objectContaining({
|
||||
name: 'memory',
|
||||
reason: 'heartbeat expired',
|
||||
}),
|
||||
}),
|
||||
]))
|
||||
|
||||
runtime.dispose()
|
||||
})
|
||||
|
||||
it('de-announces extension modules before accepting a same-id reconnect', () => {
|
||||
const runtime = setupApp({ heartbeat: { readTimeout: 20_000 } })
|
||||
const handler = wsHandler()
|
||||
const observer = createPeer('observer')
|
||||
const firstModulePeer = createPeer('module-peer')
|
||||
const secondModulePeer = createPeer('module-peer')
|
||||
|
||||
handler.open?.(observer.peer)
|
||||
handler.open?.(firstModulePeer.peer)
|
||||
sendEvent(handler, firstModulePeer.peer, createExtensionModuleAnnounceEvent())
|
||||
observer.sent.length = 0
|
||||
|
||||
handler.open?.(secondModulePeer.peer)
|
||||
|
||||
expect(decodeEvents(observer.sent)).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
type: 'extension:module:de-announced',
|
||||
data: expect.objectContaining({
|
||||
name: 'memory',
|
||||
reason: 'connection closed',
|
||||
}),
|
||||
}),
|
||||
]))
|
||||
|
||||
runtime.dispose()
|
||||
})
|
||||
|
||||
it('closes each raw peer once during runtime disposal', () => {
|
||||
const runtime = setupApp({ heartbeat: { readTimeout: 20_000 } })
|
||||
const handler = wsHandler()
|
||||
const peer = createPeer('peer-1')
|
||||
|
||||
handler.open?.(peer.peer)
|
||||
runtime.dispose()
|
||||
|
||||
expect(peer.peer.close).toHaveBeenCalledOnce()
|
||||
})
|
||||
})
|
||||
@@ -52,5 +52,10 @@ export interface AuthenticatedPeer extends NamedPeer {
|
||||
extensionModules?: Map<string, RegisteredExtensionModule>
|
||||
lastHeartbeatAt?: number
|
||||
healthy?: boolean
|
||||
/**
|
||||
* REVIEW: Legacy field name kept during the better-ws migration.
|
||||
* The value now stores peer silence duration in milliseconds, not a miss count.
|
||||
* Rename this with the server-runtime peer state cleanup.
|
||||
*/
|
||||
missedHeartbeats?: number
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user