chore(eventa): use upstream hono adapter
Upgrade @moeru/eventa to v1.0.0-beta.8 through the pnpm catalog and remove the local Hono adapter copy. Keep AIRI on business-level Eventa usage while adapter contract tests live upstream in @moeru/eventa. Signed-off-by: RainbowBird <git@luoling.moe> Commit-Message-Assisted-by: Claude (via Claude Code)
This commit is contained in:
RainbowBird
@@ -1,208 +0,0 @@
|
||||
import type { EventContext, InvocableEventContext } from '@moeru/eventa'
|
||||
import type { WSContext, WSEvents } from 'hono/ws'
|
||||
|
||||
import { and, createContext, defineEventa, defineInboundEventa, defineOutboundEventa, EventaFlowDirection, matchBy } from '@moeru/eventa'
|
||||
|
||||
// Re-implement the internal websocket payload helpers since they are not exported
|
||||
// from @moeru/eventa's public API. These match the wire format used by the H3 and
|
||||
// native adapters so that clients using any eventa adapter remain interoperable.
|
||||
|
||||
interface WebsocketPayload {
|
||||
id: string
|
||||
type: string
|
||||
payload: Record<string, unknown>
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
function generateWebsocketPayload(type: string, payload: Record<string, unknown>): WebsocketPayload {
|
||||
return {
|
||||
id: crypto.randomUUID(),
|
||||
type,
|
||||
payload,
|
||||
timestamp: Date.now(),
|
||||
}
|
||||
}
|
||||
|
||||
function parseWebsocketPayload(data: string): WebsocketPayload {
|
||||
return JSON.parse(data)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Lifecycle events
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export const wsConnectedEvent = defineEventa('eventa:adapters:hono-ws:connected')
|
||||
export const wsDisconnectedEvent = defineEventa('eventa:adapters:hono-ws:disconnected')
|
||||
export const wsErrorEvent = defineEventa('eventa:adapters:hono-ws:error')
|
||||
|
||||
interface HonoWsRawEventOptions {
|
||||
raw?: {
|
||||
error?: Event
|
||||
message?: HonoWsMessageEvent
|
||||
}
|
||||
}
|
||||
|
||||
type HonoWsMessageEvent = Parameters<NonNullable<WSEvents['onMessage']>>[0]
|
||||
|
||||
export type HonoWsEventContext = EventContext<any, HonoWsRawEventOptions>
|
||||
export type HonoWsInvocableEventContext = InvocableEventContext<any, HonoWsRawEventOptions>
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Per-peer adapter
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface CreatePeerHooksOptions {
|
||||
/** Called when a new peer connects and its EventContext is ready. */
|
||||
onContext?: (ctx: HonoWsInvocableEventContext) => void
|
||||
}
|
||||
|
||||
export interface PeerHooksResult {
|
||||
hooks: WSEvents
|
||||
}
|
||||
|
||||
/**
|
||||
* Create Hono WSEvents hooks that manage one eventa EventContext per peer.
|
||||
*
|
||||
* This is the Hono equivalent of the H3 adapter's `createPeerContext` /
|
||||
* `createPeerHooks`. Each time `onOpen` fires a fresh EventContext is created,
|
||||
* outbound events are serialised to `ws.send()`, and incoming messages are
|
||||
* routed into the context as inbound events.
|
||||
*/
|
||||
export function createPeerHooks(options: CreatePeerHooksOptions = {}): PeerHooksResult {
|
||||
let ctx: HonoWsInvocableEventContext | undefined
|
||||
let cleanup: (() => void) | undefined
|
||||
|
||||
const hooks: WSEvents = {
|
||||
onOpen(_event, ws) {
|
||||
ctx = createContext<any, HonoWsRawEventOptions>()
|
||||
|
||||
// Intercept outbound events and forward them over the WebSocket.
|
||||
// This mirrors the H3 adapter's pattern exactly.
|
||||
const offOutbound = ctx.on(
|
||||
and(
|
||||
matchBy((e: any) => e._flowDirection === EventaFlowDirection.Outbound || !e._flowDirection),
|
||||
matchBy('*'),
|
||||
),
|
||||
(event: any) => {
|
||||
const data = JSON.stringify(
|
||||
generateWebsocketPayload(event.id, {
|
||||
...defineOutboundEventa(event.type),
|
||||
...event,
|
||||
}),
|
||||
)
|
||||
ws.send(data)
|
||||
},
|
||||
)
|
||||
|
||||
cleanup = offOutbound
|
||||
|
||||
// Emit lifecycle event
|
||||
ctx.emit(wsConnectedEvent, {}, { raw: {} })
|
||||
|
||||
// Notify caller
|
||||
options.onContext?.(ctx)
|
||||
},
|
||||
|
||||
onMessage(message) {
|
||||
if (!ctx)
|
||||
return
|
||||
|
||||
try {
|
||||
const raw = typeof message.data === 'string' ? message.data : String(message.data)
|
||||
const { type, payload } = parseWebsocketPayload(raw)
|
||||
ctx.emit(defineInboundEventa(type), (payload as any).body, { raw: { message } })
|
||||
}
|
||||
catch (error) {
|
||||
console.error('Failed to parse WebSocket message:', error)
|
||||
ctx.emit(wsErrorEvent, { error }, { raw: { message } })
|
||||
}
|
||||
},
|
||||
|
||||
onClose() {
|
||||
if (!ctx)
|
||||
return
|
||||
|
||||
ctx.emit(wsDisconnectedEvent, {}, { raw: {} })
|
||||
cleanup?.()
|
||||
ctx = undefined
|
||||
cleanup = undefined
|
||||
},
|
||||
|
||||
onError(event, _ws) {
|
||||
if (!ctx)
|
||||
return
|
||||
|
||||
ctx.emit(wsErrorEvent, { error: event }, { raw: { error: event } })
|
||||
},
|
||||
}
|
||||
|
||||
return { hooks }
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Global (broadcast) adapter
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface GlobalHooksResult {
|
||||
hooks: WSEvents
|
||||
context: HonoWsEventContext
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a single shared EventContext that broadcasts outbound events to every
|
||||
* connected peer — the Hono equivalent of the H3 adapter's
|
||||
* `createGlobalContext`.
|
||||
*/
|
||||
export function createGlobalHooks(): GlobalHooksResult {
|
||||
const ctx = createContext<any, HonoWsRawEventOptions>()
|
||||
const peers = new Set<WSContext>()
|
||||
|
||||
// Broadcast outbound events to all connected peers.
|
||||
ctx.on(
|
||||
and(
|
||||
matchBy((e: any) => e._flowDirection === EventaFlowDirection.Outbound || !e._flowDirection),
|
||||
matchBy('*'),
|
||||
),
|
||||
(event: any) => {
|
||||
const data = JSON.stringify(
|
||||
generateWebsocketPayload(event.id, {
|
||||
...defineOutboundEventa(event.type),
|
||||
...event,
|
||||
}),
|
||||
)
|
||||
for (const peer of peers) {
|
||||
peer.send(data)
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
const hooks: WSEvents = {
|
||||
onOpen(_event, ws) {
|
||||
peers.add(ws)
|
||||
ctx.emit(wsConnectedEvent, {}, { raw: {} })
|
||||
},
|
||||
|
||||
onMessage(message) {
|
||||
try {
|
||||
const raw = typeof message.data === 'string' ? message.data : String(message.data)
|
||||
const { type, payload } = parseWebsocketPayload(raw)
|
||||
ctx.emit(defineInboundEventa(type), (payload as any).body, { raw: { message } })
|
||||
}
|
||||
catch (error) {
|
||||
console.error('Failed to parse WebSocket message:', error)
|
||||
ctx.emit(wsErrorEvent, { error }, { raw: { message } })
|
||||
}
|
||||
},
|
||||
|
||||
onClose(_event, ws) {
|
||||
peers.delete(ws)
|
||||
ctx.emit(wsDisconnectedEvent, {}, { raw: {} })
|
||||
},
|
||||
|
||||
onError(event, _ws) {
|
||||
ctx.emit(wsErrorEvent, { error: event }, { raw: { error: event } })
|
||||
},
|
||||
}
|
||||
|
||||
return { hooks, context: ctx }
|
||||
}
|
||||
@@ -1,104 +0,0 @@
|
||||
import type { WSContext } from 'hono/ws'
|
||||
|
||||
import { defineEventa, defineInvokeEventa, defineInvokeHandler } from '@moeru/eventa'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { createPeerHooks } from '../eventa-hono-adapter'
|
||||
|
||||
function createMockWSContext(): WSContext & { sentMessages: string[] } {
|
||||
const sentMessages: string[] = []
|
||||
return {
|
||||
send: vi.fn((data: string) => sentMessages.push(data)),
|
||||
close: vi.fn(),
|
||||
readyState: 1,
|
||||
raw: {},
|
||||
url: null,
|
||||
protocol: null,
|
||||
sentMessages,
|
||||
} as any
|
||||
}
|
||||
|
||||
type HonoWsMessageEvent = Parameters<NonNullable<import('hono/ws').WSEvents['onMessage']>>[0]
|
||||
|
||||
describe('eventa Hono adapter', () => {
|
||||
it('creates peer context on open and cleans up on close', () => {
|
||||
let contextReceived = false
|
||||
const { hooks } = createPeerHooks({
|
||||
onContext: () => { contextReceived = true },
|
||||
})
|
||||
const ws = createMockWSContext()
|
||||
hooks.onOpen!({} as any, ws)
|
||||
expect(contextReceived).toBe(true)
|
||||
hooks.onClose!({} as any, ws)
|
||||
})
|
||||
|
||||
it('routes inbound messages to eventa context and invokes handler', async () => {
|
||||
const echo = defineInvokeEventa<{ out: string }, { in: string }>('test:echo')
|
||||
|
||||
const { hooks } = createPeerHooks({
|
||||
onContext: (ctx) => {
|
||||
defineInvokeHandler(ctx, echo, (req) => {
|
||||
return { out: req.in.toUpperCase() }
|
||||
})
|
||||
},
|
||||
})
|
||||
|
||||
const ws = createMockWSContext()
|
||||
hooks.onOpen!({} as any, ws)
|
||||
|
||||
// The invoke wire format: type must match sendEvent id, payload.body
|
||||
// carries the invoke envelope with invokeId + content.
|
||||
const invokeId = 'invoke-1'
|
||||
const payload = JSON.stringify({
|
||||
id: 'msg-1',
|
||||
type: echo.sendEvent.id,
|
||||
payload: {
|
||||
body: {
|
||||
invokeId,
|
||||
content: { in: 'hello' },
|
||||
},
|
||||
},
|
||||
timestamp: Date.now(),
|
||||
})
|
||||
|
||||
const messageEvent = { data: payload } as HonoWsMessageEvent
|
||||
hooks.onMessage!(messageEvent, ws)
|
||||
|
||||
await new Promise(r => setTimeout(r, 100))
|
||||
|
||||
// Filter out lifecycle event messages (wsConnectedEvent)
|
||||
const responses = ws.sentMessages
|
||||
.map(m => JSON.parse(m))
|
||||
.filter((m: any) => !m.type.startsWith('eventa:adapters:'))
|
||||
|
||||
expect(responses.length).toBeGreaterThan(0)
|
||||
|
||||
// The invoke response payload contains the full event object spread into it.
|
||||
// The actual response content is nested under body.content.
|
||||
const response = responses[0]
|
||||
expect(response.payload.body.invokeId).toBe(invokeId)
|
||||
expect(response.payload.body.content).toEqual({ out: 'HELLO' })
|
||||
})
|
||||
|
||||
it('emits simple events and forwards them over the wire', () => {
|
||||
const ping = defineEventa<{ msg: string }>('test:ping')
|
||||
|
||||
let capturedCtx: any
|
||||
const { hooks } = createPeerHooks({
|
||||
onContext: (ctx) => { capturedCtx = ctx },
|
||||
})
|
||||
|
||||
const ws = createMockWSContext()
|
||||
hooks.onOpen!({} as any, ws)
|
||||
|
||||
// Clear the connected lifecycle message
|
||||
ws.sentMessages.length = 0
|
||||
|
||||
// Emit an outbound event from the context
|
||||
capturedCtx.emit(ping, { msg: 'pong' })
|
||||
|
||||
expect(ws.sentMessages.length).toBe(1)
|
||||
const sent = JSON.parse(ws.sentMessages[0])
|
||||
expect(sent.type).toBe('test:ping')
|
||||
})
|
||||
})
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { HonoWsInvocableEventContext } from '../../libs/eventa-hono-adapter'
|
||||
import type { HonoWsInvocableEventContext } from '@moeru/eventa/adapters/websocket/hono'
|
||||
|
||||
import type { ChatBroadcastPayload } from '../../utils/chat-broadcast'
|
||||
|
||||
import { newMessages } from '@proj-airi/server-sdk-shared'
|
||||
|
||||
@@ -4,8 +4,8 @@ import type { EngagementMetrics } from '../../otel'
|
||||
import type { ChatService } from '../../services/domain/chats'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { createPeerHooks, wsDisconnectedEvent } from '@moeru/eventa/adapters/websocket/hono'
|
||||
|
||||
import { createPeerHooks, wsDisconnectedEvent } from '../../libs/eventa-hono-adapter'
|
||||
import { createChatBroadcastCoordinator } from './broadcast'
|
||||
import { createChatConnectionRegistry } from './connection-registry'
|
||||
import { registerChatRpcHandlers } from './rpc'
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { HonoWsInvocableEventContext } from '../../libs/eventa-hono-adapter'
|
||||
import type { HonoWsInvocableEventContext } from '@moeru/eventa/adapters/websocket/hono'
|
||||
|
||||
import type { EngagementMetrics } from '../../otel'
|
||||
import type { ChatService } from '../../services/domain/chats'
|
||||
import type { ChatBroadcastCoordinator } from './broadcast'
|
||||
|
||||
Reference in New Issue
Block a user