main
/**
* Browser client for the realtime gateway — **KenEvent over Socket.IO**.
*
* 历史：这一层以前实现的是 `GET /ws/v1`（`gitdataai/app/im`，裸 WebSocket +
* `gitdata.jwt.*` 子协议 + subscribe/history/event 帧）。im 网关已从 KenApi 删除，
* 实时通道整体换成 KenEvent（RocketMQ → Socket.IO），本文件是迁移后的唯一入口。
*
* 保留的对外契约（`hub.ts` / `react.ts` / 页面都用它，迁移时零改动）：
* - `onStatus`：`WsStatus` 生命周期；
* - `onFrame`：`ServerFrame`（`event` / `user_event` / `resync_required`）；
* - `onIssue`：协议层无法自愈的情况（订阅被拒、鉴权失败、连接失败）；
* - `subscribe` / `unsubscribe`：房间意图，refcount 由 `ChatRealtimeHub` 维护。
*
* 与旧实现的三处语义差别，都来自 KenEvent 的契约：
* 1. **没有 history 快照**：订阅不会回放消息，页面一律用 REST 回拉（
* `use-room-stream` 本来就有 HTTP re-seed 路径）。
* 2. **没有 typing 广播**：`docs/events.json` 里 chat 没有 typing tag，
* `sendTyping` 保留为本地 no-op，接口不变以免页面改动。
* 3. **重连由 Socket.IO 自带**：指数退避 + 连接恢复都在库内，本层只在
* `connect` 时重放订阅（服务端订阅是 per-socket 的）。
*/
import { unwrapApi } from '@app/lib'
import type { MessageDto } from '@app/services/api'
import Apis from '@app/services/api'
import { type KenEventEnvelope, KenEventSocket, prop } from './kenEvent'
import { frameSeq, toServerFrame } from './kenEventFrames'
import {
classifyWsError,
type RoomEvent,
type ServerFrame,
type UserEvent,
WS_SERVER_DEFAULTS,
type WsServerErrorKind,
type WsStatus,
type WsTypingIndicator,
} from './protocol'
import { TypingRegistry } from './typing'
export interface ChatWsClientOptions {
/** 覆盖 realtime 入口地址；默认同源（vite dev 与网关都反代 `/socket.io`）。 */
url?: string
/** 保留字段：KenEvent 的鉴权走会话 cookie，不再需要单独换 JWT。 */
fetchJwt?: () => Promise<unknown>
/** 保留字段：心跳由 Socket.IO 自己完成（30s ping/pong）。 */
pingIntervalMs?: number
minReconnectDelayMs?: number
maxReconnectDelayMs?: number
maxConnectFailures?: number
pongTimeoutMs?: number
/** 保留字段：KenEvent 的订阅上限由服务端强制。 */
maxSubscriptions?: number
historyLimitHint?: number
maxResyncPages?: number
typingThrottleMs?: number
correlateSubscribeErrors?: boolean
}
/**
* 协议层无法自愈的情况。页面只对其中少数几种做反应（forbidden → 收回权限，
* resync / cursor_lost → HTTP 重新播种），其余用于观测。
*/
export type WsClientIssue =
| { kind: 'server_error'; error: WsServerErrorKind; message: string; room_id?: string }
| { kind: 'subscription_limit'; room_id: string; limit: number }
| { kind: 'gap'; room_id: string; from: number; to: number }
| { kind: 'resync_required'; room_id: string; reason: string; dropped: boolean }
| { kind: 'cursor_lost'; room_id: string; reason: string }
| { kind: 'history_truncated'; room_id: string; pages: number; page_size: number }
| { kind: 'server_bye'; message: string }
| { kind: 'socket_closed'; code: number; ready: boolean }
| { kind: 'pong_timeout'; outstanding_pings: number }
| { kind: 'unrecognized_frame'; snippet: string }
| { kind: 'binary_frame' }
| { kind: 'unparseable_payload'; channel: 'room' | 'user'; room_id?: string }
| { kind: 'connect_failed'; phase: 'jwt' | 'handshake'; message: string; failures: number }
| { kind: 'auth_required'; message: string }
| { kind: 'gave_up'; failures: number }
export class ChatWsClient {
private transport: KenEventSocket | null = null
private status: WsStatus = 'idle'
private selfIdValue: string | undefined
private readonly rooms = new Set<string>()
private readonly sequences = new Map<string, number>()
private readonly typing = new TypingRegistry()
private readonly statusListeners = new Set<(status: WsStatus) => void>()
private readonly frameListeners = new Set<(frame: ServerFrame) => void>()
private readonly issueListeners = new Set<(issue: WsClientIssue) => void>()
private readonly typingListeners = new Set<(roomId: string) => void>()
private readonly seenNotifications = new Set<string>()
private failures = 0
private typingNoticeShown = false
private readonly options: ChatWsClientOptions
constructor(options: ChatWsClientOptions = {}) {
this.options = options
this.typing.onTyping((roomId) => {
for (const listener of [...this.typingListeners]) listener(roomId)
})
}
/** `ready.user_id`：KenEvent 不在帧里回传，保留接口供页面消费（Undefined 即未知）。 */
get userId(): string | undefined {
return this.selfIdValue
}
get connectionStatus(): WsStatus {
return this.status
}
onStatus(listener: (status: WsStatus) => void): () => void {
this.statusListeners.add(listener)
return () => this.statusListeners.delete(listener)
}
onFrame(listener: (frame: ServerFrame) => void): () => void {
this.frameListeners.add(listener)
return () => this.frameListeners.delete(listener)
}
onIssue(listener: (issue: WsClientIssue) => void): () => void {
this.issueListeners.add(listener)
return () => this.issueListeners.delete(listener)
}
onTypingChange(listener: (roomId: string) => void): () => void {
this.typingListeners.add(listener)
return () => this.typingListeners.delete(listener)
}
/** 幂等：已连接或连接中时直接返回。 */
start(): void {
if (this.transport && this.status !== 'closed' && this.status !== 'failed') return
this.failures = 0
this.setStatus('connecting')
const transport = new KenEventSocket({
url: this.options.url,
onOpen: ({ sid }) => {
this.failures = 0
void sid
this.setStatus('open')
// 个人通道（通知/提及/已读）每个连接订阅一次。
transport.subscribePersonal()
for (const roomId of this.rooms) transport.subscribeRoom(roomId)
},
onClose: (_reason, wasOpen) => {
if (this.status === 'auth_required' || this.status === 'failed') return
// Socket.IO 会自己重连；这里只反映状态，不主动重建。
this.setStatus(wasOpen ? 'reconnecting' : 'closed')
this.emitIssue({ kind: 'socket_closed', code: 1006, ready: wasOpen })
},
onAuthRequired: (message) => {
// 终态：等页面在登录后显式 `start()`（hub 的 watchRoom 会这么做）。
this.setStatus('auth_required')
this.emitIssue({ kind: 'auth_required', message })
transport.stop()
},
onConnectError: (message) => {
this.failures += 1
this.emitIssue({ kind: 'connect_failed', phase: 'handshake', message, failures: this.failures })
this.setStatus('reconnecting')
},
onEnvelope: (envelope) => this.handleEnvelope(envelope),
onResync: (envelope) => {
const roomId = prop(envelope, 'room_id')
if (!roomId) return
this.emitFrame({ type: 'resync_required', room_id: roomId, reason: 'lagged' })
this.emitIssue({ kind: 'resync_required', room_id: roomId, reason: 'lagged', dropped: true })
},
onServerError: (message) => {
const error = classifyServerError(message)
this.emitIssue({ kind: 'server_error', error, message })
},
onSubscribeError: (subscription, message) => {
const roomId = subscription.split(':')[0]
const error = classifyServerError(message)
this.emitIssue({ kind: 'server_error', error, message, room_id: roomId })
},
})
this.transport = transport
transport.start()
}
stop(): void {
this.transport?.stop()
this.transport = null
this.setStatus('closed')
}
dispose(): void {
this.transport?.dispose()
this.transport = null
this.rooms.clear()
this.sequences.clear()
this.typing.clear()
this.statusListeners.clear()
this.frameListeners.clear()
this.issueListeners.clear()
this.typingListeners.clear()
this.setStatus('idle')
}
subscribe(roomId: string): void {
if (this.rooms.has(roomId)) return
this.rooms.add(roomId)
this.start()
this.transport?.subscribeRoom(roomId)
}
unsubscribe(roomId: string): void {
this.rooms.delete(roomId)
this.transport?.unsubscribeRoom(roomId)
this.typing.clear(roomId)
}
/**
* typing 广播在 KenEvent 里没有对应 tag（`docs/events.json` 的 `gitdata-chat`
* 不含 typing），所以这里保持 no-op：接口不变，页面调用无害，等服务端补上
* 事件后只需在这里 `emit('typing', …)`。
*/
sendTyping(roomId: string, threadId?: string | null): void {
void roomId
void threadId
if (!this.typingNoticeShown && import.meta.env.DEV) {
this.typingNoticeShown = true
console.info('[ws] typing 事件在 KenEvent 侧尚未定义，已忽略 sendTyping()')
}
}
typingUsers(roomId: string, options: { threadId?: string | null } = {}): WsTypingIndicator[] {
return this.typing.active(roomId, {
excludeUserId: this.selfIdValue,
threadId: options.threadId ?? null,
})
}
lastKnownSeq(roomId: string): number | undefined {
return this.sequences.get(roomId)
}
private handleEnvelope(envelope: KenEventEnvelope): void {
void this.pump(envelope)
}
/**
* 信封 → 帧。KenApi 的事件体只是摘要（`summary_body` = id/kind/occurred_at/
* summary），消息正文必须回拉 `GET /chat/rooms/{room}/messages/{id}` 才能
* 交给 transcript reducer，这也是 KenEvent 文档里"详情走 REST"的含义。
*/
private async pump(envelope: KenEventEnvelope): Promise<void> {
const hydrated = await this.hydrate(envelope)
let frame = toServerFrame(hydrated)
if (!frame) {
// 已知的 chat 事件映射不出来（比如 reaction-update 只带摘要）时，
// 退化成一次房间级软刷新，页面自己回拉最新状态。
const roomId = prop(envelope, 'room_id')
if (roomId && envelope.topic === 'gitdata-chat') {
frame = { type: 'event', room_id: roomId, event: { kind: 'room_updated', room_id: roomId } }
} else {
// 未消费的事件（issue/pull/kenshell/oauth…）静默丢弃：它们有各自的页面。
return
}
}
const seq = frameSeq(frame)
if (seq) {
const current = this.sequences.get(seq.roomId) ?? 0
if (seq.seq > current) this.sequences.set(seq.roomId, seq.seq)
}
this.emitFrame(frame)
}
/** 消息类事件回拉完整 DTO；失败时原样返回信封，由上层降级处理。 */
private async hydrate(envelope: KenEventEnvelope): Promise<KenEventEnvelope> {
const hydrated = await this.hydrateNotification(envelope)
if (hydrated) return hydrated
return this.hydrateMessage(envelope)
}
/**
* 通知事件同样只带摘要，而 `WsNotification` 要渲染标题/正文，所以回拉最新一条
* （`GET /me/notifications?limit=1`）。同一条只处理一次，避免重连补发时重复弹窗。
*/
private async hydrateNotification(envelope: KenEventEnvelope): Promise<KenEventEnvelope | null> {
if (envelope.topic !== 'gitdata-notify' || envelope.tag !== 'notification-create') return null
if (isNotificationDto(envelope.body)) return envelope
try {
const page = await unwrapApi<{ items?: unknown[] }>(
Apis.notifications.list_notifications({ params: { limit: 1, offset: 0 } }),
)
const latest = page?.items?.[0]
if (!isNotificationDto(latest)) return null
if (this.seenNotifications.has(latest.id)) return null
this.seenNotifications.add(latest.id)
return { ...envelope, body: latest }
} catch {
return null
}
}
private async hydrateMessage(envelope: KenEventEnvelope): Promise<KenEventEnvelope> {
const needsMessage = envelope.tag === 'message-create' || envelope.tag === 'message-update'
if (!needsMessage) return envelope
if (isMessageDto(envelope.body)) return envelope
const roomId = prop(envelope, 'room_id')
const messageId = prop(envelope, 'message_id')
if (!roomId || !messageId) return envelope
try {
const message = await unwrapApi<MessageDto>(
Apis.chat.get_message({ pathParams: { room_id: roomId, message_id: messageId } }),
)
return { ...envelope, body: message }
} catch {
return envelope
}
}
private emitFrame(frame: ServerFrame): void {
for (const listener of [...this.frameListeners]) listener(frame)
}
private emitIssue(issue: WsClientIssue): void {
for (const listener of [...this.issueListeners]) listener(issue)
}
private setStatus(status: WsStatus): void {
if (this.status === status) return
this.status = status
for (const listener of [...this.statusListeners]) listener(status)
}
}
/** 可渲染的通知 DTO（摘要体只有 `{id, kind, occurred_at, summary}`）。 */
function isNotificationDto(value: unknown): value is { id: string; notification_type: string } {
if (typeof value !== 'object' || value === null) return false
const body = value as Record<string, unknown>
return typeof body.id === 'string' && typeof body.notification_type === 'string'
}
/** 真正可渲染的消息 DTO（摘要体只有 `{id, kind, occurred_at, summary}`）。 */
function isMessageDto(value: unknown): boolean {
if (typeof value !== 'object' || value === null) return false
const body = value as Record<string, unknown>
return typeof body.id === 'string' && typeof body.room_id === 'string' && typeof body.author_id === 'number'
}
function classifyServerError(message: string): WsServerErrorKind {
// 先按旧协议的四条固定文案匹配，再按 KenEvent 的措辞兜底
// （`events/subscriptions.rs:validate` 会回 "unsupported subscription topic or tag"）。
const known = classifyWsError(message)
if (known !== 'unknown') return known
const text = message.toLowerCase()
if (text.includes('limit')) return 'subscription_limit'
if (text.includes('forbidden') || text.includes('permission') || text.includes('not allowed')) {
return 'forbidden'
}
if (text.includes('subscri')) return 'subscribe_failed'
if (text.includes('invalid') || text.includes('unsupported') || text.includes('unrecognized')) {
return 'invalid_frame'
}
return 'unknown'
}
/** Re-exported for consumers that still type against `RoomEvent`/`UserEvent`. */
export type { RoomEvent, ServerFrame, UserEvent }
export { WS_SERVER_DEFAULTS }