feat(chat-ws): add versioned Eventa protocols (#2308)

Signed-off-by: RainbowBird <git@luoling.moe>
This commit is contained in:
RainbowBird
2026-08-20 12:24:47 +08:00
committed by GitHub
parent 702b4c862d
commit d22daf6b4c
29 changed files with 510 additions and 252 deletions
-20
View File
@@ -1,20 +0,0 @@
# @proj-airi/server-sdk-shared
Shared event contracts and types for AIRI server SDK consumers and server-side handlers.
## Usage
```shell
ni @proj-airi/server-sdk-shared -D
pnpm i @proj-airi/server-sdk-shared -D
```
```typescript
import type { WireMessage } from '@proj-airi/server-sdk-shared'
import { newMessages, pullMessages, sendMessages } from '@proj-airi/server-sdk-shared'
```
## License
[MIT](../../LICENSE)
-45
View File
@@ -1,45 +0,0 @@
import { defineInvokeEventa, defineOutboundEventa } from '@moeru/eventa'
export interface WireMessage {
id: string
chatId: string
senderId: string | null
role: 'system' | 'user' | 'assistant' | 'tool' | 'error'
content: string
seq: number
createdAt: number
updatedAt: number
}
export type MessageRole = WireMessage['role']
export interface SendMessagesRequest {
chatId: string
messages: { id: string, role: string, content: string }[]
}
export interface SendMessagesResponse {
seq: number
}
export interface PullMessagesRequest {
chatId: string
afterSeq: number
limit?: number
}
export interface PullMessagesResponse {
messages: WireMessage[]
seq: number
}
export interface NewMessagesPayload {
chatId: string
messages: WireMessage[]
fromSeq: number
toSeq: number
}
export const sendMessages = defineInvokeEventa<SendMessagesResponse, SendMessagesRequest>('chat:send-messages')
export const pullMessages = defineInvokeEventa<PullMessagesResponse, PullMessagesRequest>('chat:pull-messages')
export const newMessages = defineOutboundEventa<NewMessagesPayload>('chat:new-messages')
@@ -6,18 +6,18 @@ import { buildChatWsUrl, computeReconnectDelay, createChatWsUrlRef, mapStatus, W
describe('buildChatWsUrl', () => {
/**
* @example
* "https://api.example.com" + "abc" → "wss://api.example.com/ws/chat?token=abc"
* "https://api.example.com" + "abc" → "wss://api.example.com/ws/v2/chat?token=abc"
*/
it('upgrades https → wss and appends /ws/chat with token query', () => {
expect(buildChatWsUrl('https://api.example.com', 'abc')).toBe('wss://api.example.com/ws/chat?token=abc')
it('upgrades https → wss and appends the version-two chat path with a token', () => {
expect(buildChatWsUrl('https://api.example.com', 'abc')).toBe('wss://api.example.com/ws/v2/chat?token=abc')
})
/**
* @example
* "http://localhost:3000" + "tok" → "ws://localhost:3000/ws/chat?token=tok"
* "http://localhost:3000" + "tok" → "ws://localhost:3000/ws/v2/chat?token=tok"
*/
it('upgrades http → ws on plain origins', () => {
expect(buildChatWsUrl('http://localhost:3000', 'tok')).toBe('ws://localhost:3000/ws/chat?token=tok')
expect(buildChatWsUrl('http://localhost:3000', 'tok')).toBe('ws://localhost:3000/ws/v2/chat?token=tok')
})
/**
@@ -25,8 +25,8 @@ describe('buildChatWsUrl', () => {
* Trailing slashes on the server URL must not double up the path.
*/
it('normalizes trailing slashes', () => {
expect(buildChatWsUrl('https://api.example.com/', 'a')).toBe('wss://api.example.com/ws/chat?token=a')
expect(buildChatWsUrl('https://api.example.com//', 'a')).toBe('wss://api.example.com/ws/chat?token=a')
expect(buildChatWsUrl('https://api.example.com/', 'a')).toBe('wss://api.example.com/ws/v2/chat?token=a')
expect(buildChatWsUrl('https://api.example.com//', 'a')).toBe('wss://api.example.com/ws/v2/chat?token=a')
})
/**
@@ -34,7 +34,7 @@ describe('buildChatWsUrl', () => {
* URL-unsafe token characters get percent-encoded by URLSearchParams.
*/
it('encodes tokens safely', () => {
expect(buildChatWsUrl('https://api.example.com', 'a b+c=')).toBe('wss://api.example.com/ws/chat?token=a+b%2Bc%3D')
expect(buildChatWsUrl('https://api.example.com', 'a b+c=')).toBe('wss://api.example.com/ws/v2/chat?token=a+b%2Bc%3D')
})
})
@@ -141,26 +141,16 @@ describe('createChatWsUrlRef', () => {
// ROOT CAUSE:
//
// Production wired `getToken: () => localStorage.getItem('auth/v1/token')`.
// The Vue `computed` cannot track non-reactive reads (DOM storage,
// module-level let, etc.), so the URL froze at first evaluation. After an
// OIDC `oauth2/token` refresh wrote a new access token into localStorage,
// `useWebSocket` kept reconnecting with the stale token in the query
// string, producing an infinite `/ws/chat?token=<old>` → 401 loop until
// the user reloaded the tab.
//
// Fix: callers MUST pass a closure that reads from a reactive source
// (Pinia store ref / Vue ref / computed). The two cases below pin the
// contract: reactive source rebuilds the URL on rotation; non-reactive
// source intentionally does NOT (so future regressions show up here).
it('rebuilds url when getToken reads a reactive ref (token rotation)', () => {
// The client must read a reactive token source so the client can react to
// token rotation. The URL carries the current token for the next upgrade.
it('rebuilds the URL when getToken reads a reactive ref (token rotation)', () => {
const enabled = ref(true)
const tokenRef = ref<string | null>('old-token')
const url = createChatWsUrlRef(enabled, () => tokenRef.value, 'https://api.example.com')
expect(url.value).toBe('wss://api.example.com/ws/chat?token=old-token')
expect(url.value).toBe('wss://api.example.com/ws/v2/chat?token=old-token')
tokenRef.value = 'new-token'
expect(url.value).toBe('wss://api.example.com/ws/chat?token=new-token')
expect(url.value).toBe('wss://api.example.com/ws/v2/chat?token=new-token')
})
it('freezes ws URL when getToken is non-reactive (regression guard)', () => {
@@ -170,12 +160,10 @@ describe('createChatWsUrlRef', () => {
let storage: string | null = 'frozen-token'
const url = createChatWsUrlRef(enabled, () => storage, 'https://api.example.com')
expect(url.value).toBe('wss://api.example.com/ws/chat?token=frozen-token')
expect(url.value).toBe('wss://api.example.com/ws/v2/chat?token=frozen-token')
storage = 'rotated-token'
// Still the old value — this is what broke production. If this ever
// starts returning 'rotated-token' Vue's reactivity model changed and
// the contract comment on createChatWsUrlRef can be relaxed.
expect(url.value).toBe('wss://api.example.com/ws/chat?token=frozen-token')
// Still the old value. This is the stale-token reconnect regression guard.
expect(url.value).toBe('wss://api.example.com/ws/v2/chat?token=frozen-token')
})
})
@@ -1,10 +1,10 @@
import type { NewMessagesPayload, PullMessagesRequest, PullMessagesResponse, SendMessagesRequest, SendMessagesResponse } from '@proj-airi/server-sdk-shared'
import type { NewMessagesPayload, PullMessagesRequest, PullMessagesResponse, SendMessagesRequest, SendMessagesResponse } from '@proj-airi/server-sdk-shared/v2'
import type { ComputedRef, Ref } from 'vue'
import { defineInvoke } from '@moeru/eventa'
import { createContext as createWsContext, wsErrorEvent } from '@moeru/eventa/adapters/websocket/native'
import { errorMessageFrom } from '@moeru/std'
import { newMessages, pullMessages, sendMessages } from '@proj-airi/server-sdk-shared'
import { newMessages, pullMessages, sendMessages } from '@proj-airi/server-sdk-shared/v2'
import { useWebSocket } from '@vueuse/core'
import { computed, ref, shallowRef, watch } from 'vue'
@@ -28,7 +28,7 @@ export const WS_CLOSE_UNAUTHORIZED = 4001
// The native ws adapter's context type is not directly exported from
// `@moeru/eventa/adapters/websocket/native`; use the inferred return type so
// `ctx.on` / `ctx.emit` overloads stay accurate.
// Source: @moeru/eventa@0.3.0 — adapter exports only `createContext` and the
// Source: @moeru/eventa@1.0.0-beta.15 — adapter exports only `createContext` and the
// event constants.
// Removal condition: the adapter exports a public `EventContext` type.
type WsEventContext = ReturnType<typeof createWsContext>['context']
@@ -69,7 +69,7 @@ export type ChatWsUnsubscribe = () => void
export interface CreateChatWsClientOptions {
/**
* Base server URL, e.g. `https://api.airi.build`. The client appends
* `/ws/chat?token=<jwt>` to build the WebSocket URL.
* `/ws/v2/chat` to build the WebSocket URL.
*/
serverUrl: string
/**
@@ -103,13 +103,13 @@ export interface ChatWsClient {
}
/**
* Build the `/ws/chat?token=<jwt>` URL from a base server URL.
* Build the `/ws/v2/chat` URL from a base server URL.
*
* Before:
* - "https://api.airi.build", token="abc"
* - "https://api.airi.build"
*
* After:
* - "wss://api.airi.build/ws/chat?token=abc"
* - "wss://api.airi.build/ws/v2/chat?token=abc"
*
* @internal
*/
@@ -118,7 +118,7 @@ export function buildChatWsUrl(serverUrl: string, token: string): string {
// serverUrl are normalized cleanly.
const url = new URL(serverUrl)
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'
url.pathname = `${url.pathname.replace(/\/+$/, '')}/ws/chat`
url.pathname = `${url.pathname.replace(/\/+$/, '')}/ws/v2/chat`
url.searchParams.set('token', token)
return url.toString()
}
@@ -166,27 +166,23 @@ export function mapStatus(vue: 'OPEN' | 'CONNECTING' | 'CLOSED', enabled: boolea
* - The user is signed in and the chat store wants real-time sync.
*
* Expects:
* - `serverUrl` includes scheme (https/http). Token must be a valid JWT;
* 401s during the WebSocket upgrade close the socket immediately and the
* auto-reconnect loop will keep retrying with whatever `getToken()`
* returns next.
* - `serverUrl` includes scheme (https/http). `getToken()` returns a valid JWT
* when the socket opens. The token is sent as a query parameter during the
* WebSocket upgrade.
*
* Returns:
* - A handle exposing connect/disconnect/destroy, RPC functions, and event
* hooks. RPC closures resolve the live `EventContext` per invocation so a
* reconnect-induced context swap is transparent. In-flight RPCs reject on
* disconnect with `chat-ws: rpc cancelled` so callers do not hang
* indefinitely (eventa@0.3.0 does not flush its internal pending maps when
* the underlying context is disposed; we wrap each invoke in a race).
* indefinitely. Eventa `1.0.0-beta.15` aborts pending invokes when the
* native WebSocket closes.
*/
/**
* Build the reactive ws URL ref `useWebSocket` watches.
*
* `getToken` MUST read from a reactive source (Pinia store ref, Vue ref,
* computed). A non-reactive read (e.g. `localStorage.getItem`) freezes the
* URL at first evaluation and `useWebSocket` will reconnect forever with
* the stale token after the next OIDC refresh — verified by
* `freezes ws URL when getToken is non-reactive` in ws-client.test.ts.
* computed). The reactive dependency rebuilds the URL when the token changes.
*/
export function createChatWsUrlRef(
enabled: Ref<boolean>,
@@ -274,9 +270,7 @@ export function createChatWsClient(options: CreateChatWsClientOptions): ChatWsCl
}))
}
// The url-as-ref form lets useWebSocket reconnect when `urlRef` changes
// (token rotation, disconnect intent). VueUse internally compares the
// value and reopens; passing `undefined` cleanly closes any open socket.
// The URL ref controls the user connection intent and token presence.
const ws = useWebSocket<string>(urlRef, {
immediate: false,
autoClose: true,
@@ -300,9 +294,8 @@ export function createChatWsClient(options: CreateChatWsClientOptions): ChatWsCl
// protocol signal for "this token will never succeed without
// rotation"; calling `ws.close()` here sets
// useWebSocket's internal `explicitlyClosed` flag so the next
// onclose path skips the reconnect schedule. The next time
// `urlRef` changes (token refresh), `watch(urlRef, open)` calls
// `open()` which resets `explicitlyClosed` to false and re-inits.
// onclose path skips the reconnect schedule. A token change below
// closes the old context and starts a new connection with the new URL.
if (ev.code === WS_CLOSE_UNAUTHORIZED) {
console.warn('[chat-ws] server rejected auth (4001), pausing reconnect until token rotates')
ws.close()
+11 -8
View File
@@ -3926,12 +3926,6 @@ importers:
specifier: 'catalog:'
version: 1.4.2(typescript@5.9.3)
packages/server-sdk-shared:
dependencies:
'@moeru/eventa':
specifier: 'catalog:'
version: 1.0.0-beta.15(electron@41.2.1)(h3@2.0.1-rc.25)(hono@4.11.3)(web-worker@1.5.0)
packages/server-shared:
dependencies:
'@proj-airi/plugin-protocol':
@@ -4290,7 +4284,7 @@ importers:
version: link:../server-sdk
'@proj-airi/server-sdk-shared':
specifier: workspace:^
version: link:../server-sdk-shared
version: link:../../server/packages/server-sdk-shared
'@proj-airi/stage-shared':
specifier: workspace:^
version: link:../stage-shared
@@ -5427,7 +5421,7 @@ importers:
version: link:../../packages/auth-shared
'@proj-airi/server-sdk-shared':
specifier: workspace:*
version: link:../../../packages/server-sdk-shared
version: link:../../packages/server-sdk-shared
drizzle-orm:
specifier: 'catalog:'
version: 0.45.2(@electric-sql/pglite@0.4.4)(@opentelemetry/api@1.9.1)(@prisma/client@5.22.0)(@types/pg@8.20.0)(better-sqlite3@12.5.0)(kysely@0.29.4)(pg@8.20.0)(postgres@3.4.9)
@@ -5629,6 +5623,15 @@ importers:
specifier: 'catalog:'
version: 0.45.2(@electric-sql/pglite@0.4.4)(@opentelemetry/api@1.9.1)(@prisma/client@5.22.0)(@types/pg@8.20.0)(better-sqlite3@12.5.0)(kysely@0.29.4)(pg@8.20.0)(postgres@3.4.9)
server/packages/server-sdk-shared:
dependencies:
'@moeru/eventa':
specifier: 'catalog:'
version: 1.0.0-beta.15(electron@41.2.1)(h3@2.0.1-rc.25)(hono@4.11.3)(web-worker@1.5.0)
valibot:
specifier: 'catalog:'
version: 1.4.2(typescript@5.9.3)
services/computer-use-mcp:
dependencies:
'@modelcontextprotocol/sdk':
+8 -4
View File
@@ -9,6 +9,7 @@ while production deployment configuration remains in `proj-airi/airi-railway`.
- `apps/api`: resource API, business domains, database migrations, and API runtime.
- `apps/auth`: standalone Better Auth and OIDC service.
- `packages/auth-shared`: Auth-owned database schema and principal contracts.
- `packages/server-sdk-shared`: Eventa contracts for the hosted chat WebSocket.
- `dev/caddy`: local-only public edge routing for the shared Auth/API origin.
- `docker-compose.yaml`: complete local API + Auth + PostgreSQL + Redis + Caddy stack.
@@ -63,11 +64,14 @@ migrations. After either service deploys, Railway must receive `200` from that
service's `/readyz`; deployment success alone is not sufficient evidence that
the service can reach its required dependencies.
## Not included
## Package boundaries
Frontend applications remain under `apps/`. Cross-runtime server SDK and
protocol packages remain under `packages/` because Web, Electron, plugins,
and independent services consume them.
Frontend applications remain under `apps/`. Hosted-backend packages that
define a resource API protocol can live under `server/packages/`, even when a
frontend consumes their generated contract.
Cross-runtime server SDK and protocol packages remain under `packages/`
because Web, Electron, plugins, and independent services consume them.
Production Caddy routing, OpenTelemetry Collector configuration, observability
storage, and Grafana dashboards live in `proj-airi/airi-railway` so deployment
+1 -1
View File
@@ -10,7 +10,7 @@ COPY pnpm-lock.yaml pnpm-workspace.yaml package.json tsconfig.json ./
COPY patches/ ./patches/
COPY server/apps/api server/apps/api
COPY server/packages/auth-shared server/packages/auth-shared
COPY packages/server-sdk-shared packages/server-sdk-shared
COPY server/packages/server-sdk-shared server/packages/server-sdk-shared
RUN --mount=type=cache,id=pnpm-store,target=/root/.pnpm-store \
pnpm install --frozen-lockfile --ignore-scripts
@@ -10,7 +10,7 @@ COPY pnpm-lock.yaml pnpm-workspace.yaml package.json tsconfig.json ./
COPY patches/ ./patches/
COPY server/apps/api server/apps/api
COPY server/packages/auth-shared server/packages/auth-shared
COPY packages/server-sdk-shared packages/server-sdk-shared
COPY server/packages/server-sdk-shared server/packages/server-sdk-shared
RUN pnpm install --frozen-lockfile --ignore-scripts
+1 -1
View File
@@ -4,7 +4,7 @@ dockerfilePath = "/server/apps/api/production/railway/Dockerfile"
watchPatterns = [
"server/apps/api/**",
"server/packages/auth-shared/**",
"packages/server-sdk-shared/**",
"server/packages/server-sdk-shared/**",
"package.json",
"pnpm-lock.yaml",
"pnpm-workspace.yaml",
+26 -3
View File
@@ -47,7 +47,9 @@ import { registerWsOnlineUsersGauge } from './otel/gauges/ws-online-users'
import { createAudioSpeechWsHandlers } from './routes/audio-speech-ws'
import { createAudioTranscriptionStreamHandler } from './routes/audio-transcription-stream/route'
import { createCharacterRoutes } from './routes/characters'
import { createChatWsHandlers } from './routes/chat-ws'
import { createChatWsRuntime } from './routes/chat-ws/runtime'
import { createChatWsV1Handlers } from './routes/chat-ws/v1'
import { createChatWsV2Handlers } from './routes/chat-ws/v2'
import { createChatRoutes } from './routes/chats'
import { createFluxRoutes } from './routes/flux'
import { createInternalAuthRoutes } from './routes/internal-auth'
@@ -148,8 +150,13 @@ export async function buildApp(deps: AppDeps) {
// SERVER_INSTANCE_ID, which is fine because we only need uniqueness across
// simultaneously-running api instances, not across restarts.
const instanceId = process.env.SERVER_INSTANCE_ID || nanoid()
const chatWsSetup = createChatWsHandlers(deps.chatService, deps.redis, instanceId, deps.otel?.engagement ?? null)
const chatWsRuntime = createChatWsRuntime(deps.redis, instanceId, deps.otel?.engagement ?? null)
const chatWsV2Setup = createChatWsV2Handlers(deps.chatService, deps.redis, instanceId, deps.otel?.engagement ?? null, chatWsRuntime)
const chatWsV1Setup = createChatWsV1Handlers(deps.chatService, deps.redis, instanceId, deps.otel?.engagement ?? null, chatWsRuntime)
// `/ws/chat` keeps query-token authentication for deployed clients. The
// Eventa beta.15 adapter accepts their beta.13 envelopes. `/ws/v2/chat`
// keeps the versioned endpoint for its updated authentication flow.
app.get('/ws/chat', upgradeWebSocket(async (c) => {
const token = c.req.query('token')
if (!token)
@@ -163,7 +170,23 @@ export async function buildApp(deps: AppDeps) {
if (!session?.user)
return createUnauthorizedWsEvents()
return chatWsSetup(session.user.id)
return chatWsV1Setup(session.user.id)
}))
app.get('/ws/v2/chat', upgradeWebSocket(async (c) => {
const token = c.req.query('token')
if (!token)
return createUnauthorizedWsEvents()
const session = await resolveRequestAuth(
deps.db,
deps.env,
new Headers({ Authorization: `Bearer ${token}` }),
)
if (!session?.user)
return createUnauthorizedWsEvents()
return chatWsV2Setup(session.user.id)
}))
// Bidirectional streaming TTS proxy. The handler factory builds one ws-to-ws
+1 -1
View File
@@ -91,7 +91,7 @@ export interface EngagementMetrics {
* scrape instead of leaking forever.
*
* Expects:
* - Caller (`createChatWsHandlers`) registers exactly one callback via
* - Caller (`createChatWsRuntime`) registers exactly one callback via
* `addCallback`. Multiple callbacks would double-count.
*/
wsConnectionsActive: ObservableGauge
@@ -17,8 +17,8 @@ export type { AudioSpeechWsHandlersOptions } from './types'
* Use when:
* - Wiring `/api/v1/audio/speech/ws` in {@link app.ts}. The factory returns a
* curried `setupPeer(userId)` that produces hono `WSEvents`, mirroring the
* shape of {@link createChatWsHandlers} so app.ts wires both routes the
* same way.
* shape of the chat websocket version handlers so app.ts wires both routes
* with the same curried setup pattern.
*
* Expects:
* - The route handler has already resolved auth via the `?token=` query
@@ -1,23 +1,19 @@
import type { HonoWsInvocableEventContext } from '@moeru/eventa/adapters/websocket/hono'
import type { ChatBroadcastPayload } from '../../utils/chat-broadcast'
import { newMessages } from '@proj-airi/server-sdk-shared'
/**
* In-process websocket connection registry keyed by authenticated user id.
*/
export interface ChatConnectionRegistry {
/** Adds one websocket Eventa context for the user. */
add: (userId: string, ctx: HonoWsInvocableEventContext) => void
/** Removes one websocket Eventa context and deletes the user bucket when empty. */
remove: (userId: string, ctx: HonoWsInvocableEventContext) => void
/** Adds one version-specific websocket emitter for the user. */
add: (userId: string, connectionId: string, emit: (payload: ChatBroadcastPayload) => void) => void
/** Removes one websocket emitter and deletes the user bucket when empty. */
remove: (userId: string, connectionId: string) => void
/** Returns whether this process still has local connections for the user. */
hasUser: (userId: string) => boolean
/** Counts all local websocket connections across users for metrics export. */
activeCount: () => number
/** Emits `chat:new-messages` to all local user devices except an optional sender context. */
emitNewMessages: (userId: string, excludeCtx: HonoWsInvocableEventContext | null, payload: ChatBroadcastPayload) => void
emitNewMessages: (userId: string, excludeConnectionId: string | null, payload: ChatBroadcastPayload) => void
}
/**
@@ -34,23 +30,23 @@ export interface ChatConnectionRegistry {
* - A mutable registry scoped to one chat websocket runtime.
*/
export function createChatConnectionRegistry(): ChatConnectionRegistry {
const userConnections = new Map<string, Set<HonoWsInvocableEventContext>>()
const userConnections = new Map<string, Map<string, (payload: ChatBroadcastPayload) => void>>()
return {
add(userId, ctx) {
add(userId, connectionId, emit) {
let conns = userConnections.get(userId)
if (!conns) {
conns = new Set()
conns = new Map()
userConnections.set(userId, conns)
}
conns.add(ctx)
conns.set(connectionId, emit)
},
remove(userId, ctx) {
remove(userId, connectionId) {
const conns = userConnections.get(userId)
if (!conns)
return
conns.delete(ctx)
conns.delete(connectionId)
if (conns.size === 0)
userConnections.delete(userId)
},
@@ -66,13 +62,13 @@ export function createChatConnectionRegistry(): ChatConnectionRegistry {
return total
},
emitNewMessages(userId, excludeCtx, payload) {
emitNewMessages(userId, excludeConnectionId, payload) {
const conns = userConnections.get(userId)
if (!conns)
return
for (const ctx of conns) {
if (ctx !== excludeCtx)
ctx.emit(newMessages, payload)
for (const [connectionId, emit] of conns) {
if (connectionId !== excludeConnectionId)
emit(payload)
}
},
}
+3 -70
View File
@@ -1,70 +1,3 @@
import type Redis from 'ioredis'
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 { createChatBroadcastCoordinator } from './broadcast'
import { createChatConnectionRegistry } from './connection-registry'
import { registerChatRpcHandlers } from './rpc'
const log = useLogger('chat-ws').useGlobalConfig()
/**
* Creates websocket handlers for chat sync RPC and message fanout.
*
* Use when:
* - Mounting `/ws/chat` after bearer-token auth has resolved a user id.
*
* Expects:
* - `instanceId` is stable for this process so Redis echo suppression works.
* - Redis Pub/Sub is used only for best-effort cross-instance notification.
*
* Returns:
* - A per-user Hono websocket setup function.
*/
export function createChatWsHandlers(
chatService: ChatService,
redis: Redis,
instanceId: string,
metrics?: EngagementMetrics | null,
) {
const registry = createChatConnectionRegistry()
const broadcast = createChatBroadcastCoordinator({ redis, registry, instanceId })
// Pull-based active-connection gauge: walk the local registry on each
// export interval and report the actual live count. Registered exactly
// once per process here (factory runs once via injeca); duplicate
// registration would double-count.
metrics?.wsConnectionsActive.addCallback((result) => {
result.observe(registry.activeCount())
})
return function setupPeer(userId: string) {
const { hooks } = createPeerHooks({
onContext: (ctx) => {
registry.add(userId, ctx)
broadcast.ensureSubscribed(userId)
log.withFields({ userId }).log('WS connected')
ctx.on(wsDisconnectedEvent, () => {
registry.remove(userId, ctx)
broadcast.maybeUnsubscribe(userId)
log.withFields({ userId }).log('WS disconnected')
})
registerChatRpcHandlers({
ctx,
userId,
chatService,
registry,
broadcast,
metrics,
})
},
})
return hooks
}
}
export { createChatWsRuntime } from './runtime'
export { createChatWsV1Handlers } from './v1'
export { createChatWsV2Handlers } from './v2'
@@ -0,0 +1,59 @@
import type { HonoWsInvocableEventContext } from '@moeru/eventa/adapters/websocket/hono'
import type { EngagementMetrics } from '../../otel'
import type { ChatService } from '../../services/domain/chats'
import type { ChatWsRuntime } from './runtime'
import { useLogger } from '@guiiai/logg'
import { wsDisconnectedEvent } from '@moeru/eventa/adapters/websocket/hono'
import { newMessages } from '@proj-airi/server-sdk-shared'
import { nanoid } from '../../utils/id'
import { registerChatRpcHandlers } from './rpc'
const log = useLogger('chat-ws').useGlobalConfig()
export interface RegisterChatWsPeerOptions {
/** Eventa websocket context for one authenticated peer. */
ctx: HonoWsInvocableEventContext
/** User that owns the authenticated peer. */
userId: string
/** Domain service that persists and reads chat messages. */
chatService: ChatService
/** Shared local registry and Redis broadcast runtime. */
runtime: ChatWsRuntime
/** Optional engagement metrics. */
metrics?: EngagementMetrics | null
}
/**
* Registers one authenticated chat peer with the shared Eventa beta.15 runtime.
*
* Both `/ws/chat` and `/ws/v2/chat` call this function after their own
* authentication step. The beta.15 adapter accepts the beta.13 wire envelope.
*/
export function registerChatWsPeer(options: RegisterChatWsPeerOptions): void {
const { ctx, userId, chatService, runtime, metrics } = options
const connectionId = nanoid()
runtime.registry.add(userId, connectionId, (payload) => {
void ctx.emit(newMessages, payload)
})
runtime.broadcast.ensureSubscribed(userId)
log.withFields({ userId }).log('WS connected')
ctx.on(wsDisconnectedEvent, () => {
runtime.registry.remove(userId, connectionId)
runtime.broadcast.maybeUnsubscribe(userId)
log.withFields({ userId }).log('WS disconnected')
})
registerChatRpcHandlers({
ctx,
connectionId,
userId,
chatService,
registry: runtime.registry,
broadcast: runtime.broadcast,
metrics,
})
}
+18 -20
View File
@@ -7,7 +7,7 @@ import type { ChatConnectionRegistry } from './connection-registry'
import { useLogger } from '@guiiai/logg'
import { defineInvokeHandler } from '@moeru/eventa'
import { pullMessages, sendMessages } from '@proj-airi/server-sdk-shared'
import { parsePullMessagesRequest, parseSendMessagesRequest, pullMessages, sendMessages } from '@proj-airi/server-sdk-shared'
const log = useLogger('chat-ws').useGlobalConfig()
@@ -20,6 +20,8 @@ export interface RegisterChatRpcHandlersOptions {
chatService: ChatService
/** Local websocket registry for same-instance fanout. */
registry: ChatConnectionRegistry
/** Stable id for this connection in the shared registry. */
connectionId: string
/** Redis coordinator for cross-instance fanout. */
broadcast: ChatBroadcastCoordinator
/** Optional engagement metrics. */
@@ -27,40 +29,35 @@ export interface RegisterChatRpcHandlersOptions {
}
/**
* Registers chat Eventa RPC handlers on one websocket context.
* Registers chat RPC handlers that both WebSocket URL versions share.
*
* Use when:
* - A peer context has just been created by the Hono Eventa adapter.
*
* Expects:
* - `chatService` enforces membership and message sequencing.
*
* Returns:
* - Nothing; handlers are attached to the provided context.
* The Eventa beta.15 adapter accepts beta.13 envelopes. Parse each request
* before the handler reads its fields or calls the chat service.
*/
export function registerChatRpcHandlers(options: RegisterChatRpcHandlersOptions): void {
const { ctx, userId, chatService, registry, broadcast, metrics } = options
const { ctx, userId, chatService, registry, connectionId, broadcast, metrics } = options
defineInvokeHandler(ctx, sendMessages, async (req) => {
log.withFields({ userId, chatId: req!.chatId, count: req!.messages.length }).log('sendMessages')
const result = await chatService.pushMessages(userId, req!.chatId, req!.messages)
const request = parseSendMessagesRequest(req)
log.withFields({ userId, chatId: request.chatId, count: request.messages.length }).log('sendMessages')
const result = await chatService.pushMessages(userId, request.chatId, request.messages)
const wireMessages = await chatService.pullMessages(userId, req!.chatId, result.fromSeq - 1, result.toSeq - result.fromSeq + 1)
const wireMessages = await chatService.pullMessages(userId, request.chatId, result.fromSeq - 1, result.toSeq - result.fromSeq + 1)
const broadcastPayload = {
chatId: req!.chatId,
chatId: request.chatId,
messages: wireMessages.messages,
fromSeq: result.fromSeq,
toSeq: result.toSeq,
}
const members = await chatService.getMembers(req!.chatId)
const members = await chatService.getMembers(request.chatId)
const memberUserIds = members
.filter(m => m.memberType === 'user' && m.userId != null)
.map(m => m.userId!)
for (const memberUserId of memberUserIds) {
const excludeCtx = memberUserId === userId ? ctx : null
registry.emitNewMessages(memberUserId, excludeCtx, broadcastPayload)
const excludeConnectionId = memberUserId === userId ? connectionId : null
registry.emitNewMessages(memberUserId, excludeConnectionId, broadcastPayload)
broadcast.publish(memberUserId, broadcastPayload)
}
@@ -69,7 +66,8 @@ export function registerChatRpcHandlers(options: RegisterChatRpcHandlersOptions)
})
defineInvokeHandler(ctx, pullMessages, async (req) => {
log.withFields({ userId, chatId: req!.chatId, afterSeq: req!.afterSeq }).log('pullMessages')
return chatService.pullMessages(userId, req!.chatId, req!.afterSeq, req!.limit)
const request = parsePullMessagesRequest(req)
log.withFields({ userId, chatId: request.chatId, afterSeq: request.afterSeq }).log('pullMessages')
return chatService.pullMessages(userId, request.chatId, request.afterSeq, request.limit)
})
}
@@ -0,0 +1,29 @@
import type Redis from 'ioredis'
import type { EngagementMetrics } from '../../otel'
import type { ChatBroadcastCoordinator } from './broadcast'
import type { ChatConnectionRegistry } from './connection-registry'
import { createChatBroadcastCoordinator } from './broadcast'
import { createChatConnectionRegistry } from './connection-registry'
export interface ChatWsRuntime {
registry: ChatConnectionRegistry
broadcast: ChatBroadcastCoordinator
}
/** Creates the shared fanout runtime used by both chat websocket versions. */
export function createChatWsRuntime(
redis: Redis,
instanceId: string,
metrics?: EngagementMetrics | null,
): ChatWsRuntime {
const registry = createChatConnectionRegistry()
const broadcast = createChatBroadcastCoordinator({ redis, registry, instanceId })
metrics?.wsConnectionsActive.addCallback((result) => {
result.observe(registry.activeCount())
})
return { registry, broadcast }
}
@@ -0,0 +1,35 @@
import type Redis from 'ioredis'
import type { EngagementMetrics } from '../../../otel'
import type { ChatService } from '../../../services/domain/chats'
import type { ChatWsRuntime } from '../runtime'
import { createPeerHooks } from '@moeru/eventa/adapters/websocket/hono'
import { registerChatWsPeer } from '../peer'
import { createChatWsRuntime } from '../runtime'
/**
* Creates the version-one `/ws/chat` handlers.
*
* The route keeps query-token authentication for deployed clients. Eventa
* `1.0.0-beta.15` accepts their beta.13 wire envelopes.
*/
export function createChatWsV1Handlers(
chatService: ChatService,
redis: Redis,
instanceId: string,
metrics?: EngagementMetrics | null,
runtime?: ChatWsRuntime,
) {
const chatRuntime = runtime ?? createChatWsRuntime(redis, instanceId, metrics)
return function setupPeer(userId: string) {
const { hooks } = createPeerHooks({
onContext: (ctx) => {
registerChatWsPeer({ ctx, userId, chatService, runtime: chatRuntime, metrics })
},
})
return hooks
}
}
@@ -0,0 +1,78 @@
import type { HonoWsEventContext } from '@moeru/eventa/adapters/websocket/hono'
import { defineInboundEventa, defineOutboundEventa } from '@moeru/eventa'
import { createPeerHooks } from '@moeru/eventa/adapters/websocket/hono'
import { WSContext } from 'hono/ws'
import { describe, expect, it, vi } from 'vitest'
const legacyPing = defineInboundEventa<{ value: string }>('chat-ws:legacy-ping')
const serverPong = defineOutboundEventa<{ value: string }>('chat-ws:server-pong')
function createPeer(sent: string[]): WSContext {
return new WSContext({
send(data) {
sent.push(String(data))
},
close() {},
readyState: 1,
})
}
describe('v1 chat WebSocket protocol', () => {
// https://github.com/moeru-ai/airi/pull/2308
// ROOT CAUSE:
//
// Eventa beta.14 changed the adapter envelope. The chat server then added a
// second Eventa package to parse beta.13 clients.
//
// Eventa beta.15 restores dual-shape transport support. This test sends an
// actual beta.13 envelope and checks that the server also writes its legacy
// fields. Therefore the server needs one Eventa package.
it('reads and writes beta.13 envelopes through the beta.15 adapter', async () => {
const received: Array<{ value: string }> = []
const sent: string[] = []
let context: HonoWsEventContext | undefined
const { hooks } = createPeerHooks({
onContext(created) {
context = created
created.on(legacyPing, (event) => {
if (!event.body)
throw new Error('Legacy Eventa envelope did not include a body')
received.push(event.body)
})
},
})
const peer = createPeer(sent)
hooks.onOpen?.(new Event('open'), peer)
hooks.onMessage?.(new MessageEvent('message', {
data: JSON.stringify({
id: 'beta13-delivery',
type: 'chat-ws:legacy-ping',
payload: {
id: 'chat-ws:legacy-ping',
body: { value: 'from-beta13' },
},
}),
}), peer)
await vi.waitFor(() => {
expect(received).toEqual([{ value: 'from-beta13' }])
})
if (!context)
throw new Error('WebSocket context was not created')
await context.emit(serverPong, { value: 'from-beta15' })
expect(sent).toHaveLength(2)
expect(JSON.parse(sent.at(-1)!)).toMatchObject({
deliveryId: expect.any(String),
hopsRemaining: expect.any(Number),
eventa: { id: 'chat-ws:server-pong', body: { value: 'from-beta15' } },
id: expect.any(String),
type: 'chat-ws:server-pong',
payload: { id: 'chat-ws:server-pong', body: { value: 'from-beta15' } },
})
})
})
@@ -0,0 +1,23 @@
import { parsePullMessagesRequest, parseSendMessagesRequest } from '@proj-airi/server-sdk-shared'
import { describe, expect, it } from 'vitest'
describe('v1 chat WebSocket request contracts', () => {
// https://github.com/moeru-ai/airi/pull/2308#discussion_r3796624651
// ROOT CAUSE:
//
// Eventa decodes the invoke envelope but does not validate its body. The
// handlers used body fields directly, so malformed authenticated requests
// reached ChatService.
//
// The shared protocol now owns the request schemas. The v1 handlers parse
// every invoke body before logging or calling ChatService.
it('rejects malformed send-messages requests', () => {
expect(() => parseSendMessagesRequest({ chatId: 'chat-1', messages: [{ id: 'message-1', content: 'hello' }] }))
.toThrow()
})
it('rejects malformed pull-messages requests', () => {
expect(() => parsePullMessagesRequest({ chatId: 'chat-1', afterSeq: -1 }))
.toThrow()
})
})
@@ -0,0 +1,42 @@
import type Redis from 'ioredis'
import type { EngagementMetrics } from '../../../otel'
import type { ChatService } from '../../../services/domain/chats'
import type { ChatWsRuntime } from '../runtime'
import { createPeerHooks } from '@moeru/eventa/adapters/websocket/hono'
import { registerChatWsPeer } from '../peer'
import { createChatWsRuntime } from '../runtime'
/**
* Creates websocket handlers for chat sync RPC and message fanout.
*
* Use when:
* - Mounting an already authenticated chat peer.
*
* Expects:
* - `instanceId` is stable for this process so Redis echo suppression works.
* - Redis Pub/Sub is used only for best-effort cross-instance notification.
*
* Returns:
* - A per-user Hono websocket setup function.
*/
export function createChatWsV2Handlers(
chatService: ChatService,
redis: Redis,
instanceId: string,
metrics?: EngagementMetrics | null,
runtime?: ChatWsRuntime,
) {
const chatRuntime = runtime ?? createChatWsRuntime(redis, instanceId, metrics)
return function setupPeer(userId: string) {
const { hooks } = createPeerHooks({
onContext: (ctx) => {
registerChatWsPeer({ ctx, userId, chatService, runtime: chatRuntime, metrics })
},
})
return hooks
}
}
+2 -2
View File
@@ -1,6 +1,6 @@
/**
* Local integration smoke for the `ws.connections.active` ObservableGauge
* pattern used in `server/apps/api/src/routes/chat-ws/index.ts`.
* pattern used in `server/apps/api/src/routes/chat-ws/runtime.ts`.
*
* Reproduces the exact pattern (Map<userId, Set<ctx>> registry +
* `addCallback` walking it) inside a real Hono + @hono/node-ws server, then
@@ -50,7 +50,7 @@ const wsConnectionsActive = meter.createObservableGauge('ws.connections.active',
description: 'Active WS connections (live registry size)',
})
// Identical structure to chat-ws/index.ts: Map<userId, Set<connectionKey>>.
// Identical structure to chat-ws/connection-registry.ts: Map<userId, Set<connectionKey>>.
// Multi-tab support requires Set (not just Map.size).
const userConnections = new Map()
@@ -0,0 +1,26 @@
# @proj-airi/server-sdk-shared
Eventa contracts for the hosted chat WebSocket.
## Usage
```shell
ni @proj-airi/server-sdk-shared -D
pnpm i @proj-airi/server-sdk-shared -D
```
```typescript
import type { WireMessage } from '@proj-airi/server-sdk-shared'
import { newMessages, pullMessages, sendMessages } from '@proj-airi/server-sdk-shared'
```
The package uses Eventa `1.0.0-beta.15`. Its WebSocket adapter accepts beta.13
`id/type/payload` envelopes and sends these fields with current envelopes.
`/ws/chat` keeps query-token authentication for deployed clients. `/ws/v2/chat`
authenticates after the WebSocket opens.
## License
[MIT](../../../LICENSE)
@@ -3,7 +3,7 @@
"type": "module",
"version": "0.11.3",
"private": true,
"description": "Shared event contracts and types consumed by AIRI server-sdk clients and server-side handlers",
"description": "Eventa contracts for the hosted chat WebSocket",
"author": {
"name": "Moeru AI Project AIRI Team",
"email": "airi@moeru.ai",
@@ -13,12 +13,16 @@
"repository": {
"type": "git",
"url": "https://github.com/moeru-ai/airi.git",
"directory": "packages/server-sdk-shared"
"directory": "server/packages/server-sdk-shared"
},
"exports": {
".": {
"types": "./dist/index.d.mts",
"default": "./dist/index.mjs"
},
"./v2": {
"types": "./dist/v2.d.mts",
"default": "./dist/v2.mjs"
}
},
"main": "./dist/index.mjs",
@@ -34,6 +38,7 @@
"typecheck": "tsc --noEmit"
},
"dependencies": {
"@moeru/eventa": "catalog:"
"@moeru/eventa": "catalog:",
"valibot": "catalog:"
}
}
@@ -0,0 +1,63 @@
import * as v from 'valibot'
const NonEmptyStringSchema = v.pipe(v.string(), v.minLength(1))
const SendMessageSchema = v.object({
id: NonEmptyStringSchema,
role: v.string(),
content: v.string(),
})
export const SendMessagesRequestSchema = v.object({
chatId: NonEmptyStringSchema,
messages: v.array(SendMessageSchema),
})
export const PullMessagesRequestSchema = v.object({
chatId: NonEmptyStringSchema,
afterSeq: v.pipe(v.number(), v.integer(), v.minValue(0)),
limit: v.optional(v.pipe(v.number(), v.integer(), v.minValue(1))),
})
export interface WireMessage {
id: string
chatId: string
senderId: string | null
role: 'system' | 'user' | 'assistant' | 'tool' | 'error'
content: string
seq: number
createdAt: number
updatedAt: number
}
export type MessageRole = WireMessage['role']
export type SendMessagesRequest = v.InferOutput<typeof SendMessagesRequestSchema>
export interface SendMessagesResponse {
seq: number
}
export type PullMessagesRequest = v.InferOutput<typeof PullMessagesRequestSchema>
export interface PullMessagesResponse {
messages: WireMessage[]
seq: number
}
export interface NewMessagesPayload {
chatId: string
messages: WireMessage[]
fromSeq: number
toSeq: number
}
/** Parses a `chat:send-messages` payload at the WebSocket boundary. */
export function parseSendMessagesRequest(request: unknown): SendMessagesRequest {
return v.parse(SendMessagesRequestSchema, request)
}
/** Parses a `chat:pull-messages` payload at the WebSocket boundary. */
export function parsePullMessagesRequest(request: unknown): PullMessagesRequest {
return v.parse(PullMessagesRequestSchema, request)
}
@@ -0,0 +1 @@
export * from './v2'
@@ -0,0 +1,23 @@
import type { NewMessagesPayload, PullMessagesRequest, PullMessagesResponse, SendMessagesRequest, SendMessagesResponse } from './chat'
import { defineInvokeEventa, defineOutboundEventa } from '@moeru/eventa'
export type {
MessageRole,
NewMessagesPayload,
PullMessagesRequest,
PullMessagesResponse,
SendMessagesRequest,
SendMessagesResponse,
WireMessage,
} from './chat'
export {
parsePullMessagesRequest,
parseSendMessagesRequest,
PullMessagesRequestSchema,
SendMessagesRequestSchema,
} from './chat'
export const sendMessages = defineInvokeEventa<SendMessagesResponse, SendMessagesRequest>('chat:send-messages')
export const pullMessages = defineInvokeEventa<PullMessagesResponse, PullMessagesRequest>('chat:pull-messages')
export const newMessages = defineOutboundEventa<NewMessagesPayload>('chat:new-messages')
@@ -3,6 +3,7 @@ import { defineConfig } from 'tsdown'
export default defineConfig({
entry: {
index: 'src/index.ts',
v2: 'src/v2.ts',
},
sourcemap: true,
unused: true,