feat(web): expose the realtime channel as React context
Replace the module-singleton consumption path with a provider that owns one
Socket.IO connection, reports connection status and hands out ref-counted
subscriptions. The transport gains generic subscribeTopic({topic, tag, filter})
so issue, pull, org and kenshell resource events can be subscribed, with the
intent replayed after every reconnect.
ZhenYi committed
Added view/app/src/services/realtime/client.ts view/app/src/services/realtime/client.ts +372 −0 View view/app/src/services/realtime/client.ts
@@ -0,0 +1,372 @@
1/**
2 * Browser client for the realtime gateway — **KenEvent over Socket.IO**.
3 *
4 * 历史：这一层以前实现的是 `GET /ws/v1`（`gitdataai/app/im`，裸 WebSocket +
5 * `gitdata.jwt.*` 子协议 + subscribe/history/event 帧）。im 网关已从 KenApi 删除，
6 * 实时通道整体换成 KenEvent（RocketMQ → Socket.IO），本文件是迁移后的唯一入口。
7 *
8 * 保留的对外契约（`hub.ts` / `react.ts` / 页面都用它，迁移时零改动）：
9 * - `onStatus`：`WsStatus` 生命周期；
10 * - `onFrame`：`ServerFrame`（`event` / `user_event` / `resync_required`）；
11 * - `onIssue`：协议层无法自愈的情况（订阅被拒、鉴权失败、连接失败）；
12 * - `subscribe` / `unsubscribe`：房间意图，refcount 由 `ChatRealtimeHub` 维护。
13 *
14 * 与旧实现的三处语义差别，都来自 KenEvent 的契约：
15 * 1. **没有 history 快照**：订阅不会回放消息，页面一律用 REST 回拉（
16 * `use-room-stream` 本来就有 HTTP re-seed 路径）。
17 * 2. **没有 typing 广播**：`docs/events.json` 里 chat 没有 typing tag，
18 * `sendTyping` 保留为本地 no-op，接口不变以免页面改动。
19 * 3. **重连由 Socket.IO 自带**：指数退避 + 连接恢复都在库内，本层只在
20 * `connect` 时重放订阅（服务端订阅是 per-socket 的）。
21 */
22
23import { unwrapApi } from '@app/lib'
24import type { MessageDto } from '@app/services/api'
25import Apis from '@app/services/api'
26import { type KenEventEnvelope, KenEventSocket, prop } from './kenEvent'
27import { frameSeq, toServerFrame } from './kenEventFrames'
28import {
29 classifyWsError,
30 type RoomEvent,
31 type ServerFrame,
32 type UserEvent,
33 WS_SERVER_DEFAULTS,
34 type WsServerErrorKind,
35 type WsStatus,
36 type WsTypingIndicator,
37} from './protocol'
38import { TypingRegistry } from './typing'
39
40export interface ChatWsClientOptions {
41 /** 覆盖 realtime 入口地址；默认同源（vite dev 与网关都反代 `/socket.io`）。 */
42 url?: string
43 /** 保留字段：KenEvent 的鉴权走会话 cookie，不再需要单独换 JWT。 */
44 fetchJwt?: () => Promise<unknown>
45 /** 保留字段：心跳由 Socket.IO 自己完成（30s ping/pong）。 */
46 pingIntervalMs?: number
47 minReconnectDelayMs?: number
48 maxReconnectDelayMs?: number
49 maxConnectFailures?: number
50 pongTimeoutMs?: number
51 /** 保留字段：KenEvent 的订阅上限由服务端强制。 */
52 maxSubscriptions?: number
53 historyLimitHint?: number
54 maxResyncPages?: number
55 typingThrottleMs?: number
56 correlateSubscribeErrors?: boolean
57}
58
59/**
60 * 协议层无法自愈的情况。页面只对其中少数几种做反应（forbidden → 收回权限，
61 * resync / cursor_lost → HTTP 重新播种），其余用于观测。
62 */
63export type WsClientIssue =
64 | { kind: 'server_error'; error: WsServerErrorKind; message: string; room_id?: string }
65 | { kind: 'subscription_limit'; room_id: string; limit: number }
66 | { kind: 'gap'; room_id: string; from: number; to: number }
67 | { kind: 'resync_required'; room_id: string; reason: string; dropped: boolean }
68 | { kind: 'cursor_lost'; room_id: string; reason: string }
69 | { kind: 'history_truncated'; room_id: string; pages: number; page_size: number }
70 | { kind: 'server_bye'; message: string }
71 | { kind: 'socket_closed'; code: number; ready: boolean }
72 | { kind: 'pong_timeout'; outstanding_pings: number }
73 | { kind: 'unrecognized_frame'; snippet: string }
74 | { kind: 'binary_frame' }
75 | { kind: 'unparseable_payload'; channel: 'room' | 'user'; room_id?: string }
76 | { kind: 'connect_failed'; phase: 'jwt' | 'handshake'; message: string; failures: number }
77 | { kind: 'auth_required'; message: string }
78 | { kind: 'gave_up'; failures: number }
79
80export class ChatWsClient {
81 private transport: KenEventSocket | null = null
82 private status: WsStatus = 'idle'
83 private selfIdValue: string | undefined
84 private readonly rooms = new Set<string>()
85 private readonly sequences = new Map<string, number>()
86 private readonly typing = new TypingRegistry()
87 private readonly statusListeners = new Set<(status: WsStatus) => void>()
88 private readonly frameListeners = new Set<(frame: ServerFrame) => void>()
89 private readonly issueListeners = new Set<(issue: WsClientIssue) => void>()
90 private readonly typingListeners = new Set<(roomId: string) => void>()
91 private readonly seenNotifications = new Set<string>()
92 private failures = 0
93 private typingNoticeShown = false
94
95 private readonly options: ChatWsClientOptions
96
97 constructor(options: ChatWsClientOptions = {}) {
98 this.options = options
99 this.typing.onTyping((roomId) => {
100 for (const listener of [...this.typingListeners]) listener(roomId)
101 })
102 }
103
104 /** `ready.user_id`：KenEvent 不在帧里回传，保留接口供页面消费（Undefined 即未知）。 */
105 get userId(): string | undefined {
106 return this.selfIdValue
107 }
108
109 get connectionStatus(): WsStatus {
110 return this.status
111 }
112
113 onStatus(listener: (status: WsStatus) => void): () => void {
114 this.statusListeners.add(listener)
115 return () => this.statusListeners.delete(listener)
116 }
117
118 onFrame(listener: (frame: ServerFrame) => void): () => void {
119 this.frameListeners.add(listener)
120 return () => this.frameListeners.delete(listener)
121 }
122
123 onIssue(listener: (issue: WsClientIssue) => void): () => void {
124 this.issueListeners.add(listener)
125 return () => this.issueListeners.delete(listener)
126 }
127
128 onTypingChange(listener: (roomId: string) => void): () => void {
129 this.typingListeners.add(listener)
130 return () => this.typingListeners.delete(listener)
131 }
132
133 /** 幂等：已连接或连接中时直接返回。 */
134 start(): void {
135 if (this.transport && this.status !== 'closed' && this.status !== 'failed') return
136 this.failures = 0
137 this.setStatus('connecting')
138
139 const transport = new KenEventSocket({
140 url: this.options.url,
141 onOpen: ({ sid }) => {
142 this.failures = 0
143 void sid
144 this.setStatus('open')
145 // 个人通道（通知/提及/已读）每个连接订阅一次。
146 transport.subscribePersonal()
147 for (const roomId of this.rooms) transport.subscribeRoom(roomId)
148 },
149 onClose: (_reason, wasOpen) => {
150 if (this.status === 'auth_required' || this.status === 'failed') return
151 // Socket.IO 会自己重连；这里只反映状态，不主动重建。
152 this.setStatus(wasOpen ? 'reconnecting' : 'closed')
153 this.emitIssue({ kind: 'socket_closed', code: 1006, ready: wasOpen })
154 },
155 onAuthRequired: (message) => {
156 // 终态：等页面在登录后显式 `start()`（hub 的 watchRoom 会这么做）。
157 this.setStatus('auth_required')
158 this.emitIssue({ kind: 'auth_required', message })
159 transport.stop()
160 },
161 onConnectError: (message) => {
162 this.failures += 1
163 this.emitIssue({ kind: 'connect_failed', phase: 'handshake', message, failures: this.failures })
164 this.setStatus('reconnecting')
165 },
166 onEnvelope: (envelope) => this.handleEnvelope(envelope),
167 onResync: (envelope) => {
168 const roomId = prop(envelope, 'room_id')
169 if (!roomId) return
170 this.emitFrame({ type: 'resync_required', room_id: roomId, reason: 'lagged' })
171 this.emitIssue({ kind: 'resync_required', room_id: roomId, reason: 'lagged', dropped: true })
172 },
173 onServerError: (message) => {
174 const error = classifyServerError(message)
175 this.emitIssue({ kind: 'server_error', error, message })
176 },
177 onSubscribeError: (subscription, message) => {
178 const roomId = subscription.split(':')[0]
179 const error = classifyServerError(message)
180 this.emitIssue({ kind: 'server_error', error, message, room_id: roomId })
181 },
182 })
183
184 this.transport = transport
185 transport.start()
186 }
187
188 stop(): void {
189 this.transport?.stop()
190 this.transport = null
191 this.setStatus('closed')
192 }
193
194 dispose(): void {
195 this.transport?.dispose()
196 this.transport = null
197 this.rooms.clear()
198 this.sequences.clear()
199 this.typing.clear()
200 this.statusListeners.clear()
201 this.frameListeners.clear()
202 this.issueListeners.clear()
203 this.typingListeners.clear()
204 this.setStatus('idle')
205 }
206
207 subscribe(roomId: string): void {
208 if (this.rooms.has(roomId)) return
209 this.rooms.add(roomId)
210 this.start()
211 this.transport?.subscribeRoom(roomId)
212 }
213
214 unsubscribe(roomId: string): void {
215 this.rooms.delete(roomId)
216 this.transport?.unsubscribeRoom(roomId)
217 this.typing.clear(roomId)
218 }
219
220 /**
221 * typing 广播在 KenEvent 里没有对应 tag（`docs/events.json` 的 `gitdata-chat`
222 * 不含 typing），所以这里保持 no-op：接口不变，页面调用无害，等服务端补上
223 * 事件后只需在这里 `emit('typing', …)`。
224 */
225 sendTyping(roomId: string, threadId?: string | null): void {
226 void roomId
227 void threadId
228 if (!this.typingNoticeShown && import.meta.env.DEV) {
229 this.typingNoticeShown = true
230 console.info('[ws] typing 事件在 KenEvent 侧尚未定义，已忽略 sendTyping()')
231 }
232 }
233
234 typingUsers(roomId: string, options: { threadId?: string | null } = {}): WsTypingIndicator[] {
235 return this.typing.active(roomId, {
236 excludeUserId: this.selfIdValue,
237 threadId: options.threadId ?? null,
238 })
239 }
240
241 lastKnownSeq(roomId: string): number | undefined {
242 return this.sequences.get(roomId)
243 }
244
245 private handleEnvelope(envelope: KenEventEnvelope): void {
246 void this.pump(envelope)
247 }
248
249 /**
250 * 信封 → 帧。KenApi 的事件体只是摘要（`summary_body` = id/kind/occurred_at/
251 * summary），消息正文必须回拉 `GET /chat/rooms/{room}/messages/{id}` 才能
252 * 交给 transcript reducer，这也是 KenEvent 文档里"详情走 REST"的含义。
253 */
254 private async pump(envelope: KenEventEnvelope): Promise<void> {
255 const hydrated = await this.hydrate(envelope)
256 let frame = toServerFrame(hydrated)
257
258 if (!frame) {
259 // 已知的 chat 事件映射不出来（比如 reaction-update 只带摘要）时，
260 // 退化成一次房间级软刷新，页面自己回拉最新状态。
261 const roomId = prop(envelope, 'room_id')
262 if (roomId && envelope.topic === 'gitdata-chat') {
263 frame = { type: 'event', room_id: roomId, event: { kind: 'room_updated', room_id: roomId } }
264 } else {
265 // 未消费的事件（issue/pull/kenshell/oauth…）静默丢弃：它们有各自的页面。
266 return
267 }
268 }
269
270 const seq = frameSeq(frame)
271 if (seq) {
272 const current = this.sequences.get(seq.roomId) ?? 0
273 if (seq.seq > current) this.sequences.set(seq.roomId, seq.seq)
274 }
275 this.emitFrame(frame)
276 }
277
278 /** 消息类事件回拉完整 DTO；失败时原样返回信封，由上层降级处理。 */
279 private async hydrate(envelope: KenEventEnvelope): Promise<KenEventEnvelope> {
280 const hydrated = await this.hydrateNotification(envelope)
281 if (hydrated) return hydrated
282 return this.hydrateMessage(envelope)
283 }
284
285 /**
286 * 通知事件同样只带摘要，而 `WsNotification` 要渲染标题/正文，所以回拉最新一条
287 * （`GET /me/notifications?limit=1`）。同一条只处理一次，避免重连补发时重复弹窗。
288 */
289 private async hydrateNotification(envelope: KenEventEnvelope): Promise<KenEventEnvelope | null> {
290 if (envelope.topic !== 'gitdata-notify' || envelope.tag !== 'notification-create') return null
291 if (isNotificationDto(envelope.body)) return envelope
292 try {
293 const page = await unwrapApi<{ items?: unknown[] }>(
294 Apis.notifications.list_notifications({ params: { limit: 1, offset: 0 } }),
295 )
296 const latest = page?.items?.[0]
297 if (!isNotificationDto(latest)) return null
298 if (this.seenNotifications.has(latest.id)) return null
299 this.seenNotifications.add(latest.id)
300 return { ...envelope, body: latest }
301 } catch {
302 return null
303 }
304 }
305
306 private async hydrateMessage(envelope: KenEventEnvelope): Promise<KenEventEnvelope> {
307 const needsMessage = envelope.tag === 'message-create' || envelope.tag === 'message-update'
308 if (!needsMessage) return envelope
309 if (isMessageDto(envelope.body)) return envelope
310
311 const roomId = prop(envelope, 'room_id')
312 const messageId = prop(envelope, 'message_id')
313 if (!roomId || !messageId) return envelope
314 try {
315 const message = await unwrapApi<MessageDto>(
316 Apis.chat.get_message({ pathParams: { room_id: roomId, message_id: messageId } }),
317 )
318 return { ...envelope, body: message }
319 } catch {
320 return envelope
321 }
322 }
323
324 private emitFrame(frame: ServerFrame): void {
325 for (const listener of [...this.frameListeners]) listener(frame)
326 }
327
328 private emitIssue(issue: WsClientIssue): void {
329 for (const listener of [...this.issueListeners]) listener(issue)
330 }
331
332 private setStatus(status: WsStatus): void {
333 if (this.status === status) return
334 this.status = status
335 for (const listener of [...this.statusListeners]) listener(status)
336 }
337}
338
339/** 可渲染的通知 DTO（摘要体只有 `{id, kind, occurred_at, summary}`）。 */
340function isNotificationDto(value: unknown): value is { id: string; notification_type: string } {
341 if (typeof value !== 'object' || value === null) return false
342 const body = value as Record<string, unknown>
343 return typeof body.id === 'string' && typeof body.notification_type === 'string'
344}
345
346/** 真正可渲染的消息 DTO（摘要体只有 `{id, kind, occurred_at, summary}`）。 */
347function isMessageDto(value: unknown): boolean {
348 if (typeof value !== 'object' || value === null) return false
349 const body = value as Record<string, unknown>
350 return typeof body.id === 'string' && typeof body.room_id === 'string' && typeof body.author_id === 'number'
351}
352
353function classifyServerError(message: string): WsServerErrorKind {
354 // 先按旧协议的四条固定文案匹配，再按 KenEvent 的措辞兜底
355 // （`events/subscriptions.rs:validate` 会回 "unsupported subscription topic or tag"）。
356 const known = classifyWsError(message)
357 if (known !== 'unknown') return known
358 const text = message.toLowerCase()
359 if (text.includes('limit')) return 'subscription_limit'
360 if (text.includes('forbidden') || text.includes('permission') || text.includes('not allowed')) {
361 return 'forbidden'
362 }
363 if (text.includes('subscri')) return 'subscribe_failed'
364 if (text.includes('invalid') || text.includes('unsupported') || text.includes('unrecognized')) {
365 return 'invalid_frame'
366 }
367 return 'unknown'
368}
369
370/** Re-exported for consumers that still type against `RoomEvent`/`UserEvent`. */
371export type { RoomEvent, ServerFrame, UserEvent }
372export { WS_SERVER_DEFAULTS }
Added view/app/src/services/realtime/context.tsx view/app/src/services/realtime/context.tsx +136 −0 View view/app/src/services/realtime/context.tsx
@@ -0,0 +1,136 @@
1import { createContext, type ReactNode, useCallback, useContext, useEffect, useMemo, useRef, useState } from 'react'
2import { type KenEventEnvelope, KenEventSocket, prop } from './kenEvent'
3
4export type RealtimeStatus = 'disabled' | 'connecting' | 'online' | 'offline' | 'auth_required' | 'failed'
5
6export interface TopicSubscription {
7 topic: string
8 tag: string
9 filter?: Record<string, string>
10 key?: string
11}
12
13export interface RealtimeApi {
14 readonly enabled: boolean
15 readonly status: RealtimeStatus
16 readonly selfId?: number
17 subscribe(spec: TopicSubscription): () => void
18 onEnvelope(handler: (envelope: KenEventEnvelope) => void): () => void
19 retry(): void
20 lastError?: string
21}
22
23const DISABLED: RealtimeApi = {
24 enabled: false,
25 status: 'disabled',
26 subscribe: () => () => {},
27 onEnvelope: () => () => {},
28 retry: () => {},
29}
30
31const RealtimeContext = createContext<RealtimeApi | null>(null)
32
33function endpoint(): string | null {
34 const configured = import.meta.env.VITE_EVENT_URL as string | undefined
35 if (configured?.trim()) return configured.trim()
36 return null
37}
38
39export function RealtimeProvider({ children }: { children: ReactNode }) {
40 const url = endpoint()
41 const socketRef = useRef<KenEventSocket | null>(null)
42 const listeners = useRef(new Set<(envelope: KenEventEnvelope) => void>())
43 const pending = useRef(new Map<string, () => void>())
44 const [status, setStatus] = useState<RealtimeStatus>(url ? 'connecting' : 'disabled')
45 const [selfId, setSelfId] = useState<number | undefined>(undefined)
46 const [lastError, setLastError] = useState<string | undefined>(undefined)
47
48 useEffect(() => {
49 if (!url) return
50 const socket = new KenEventSocket({
51 url,
52 onOpen: () => {
53 setStatus('online')
54 setLastError(undefined)
55 },
56 onClose: (_reason, wasOpen) => {
57 setStatus(wasOpen ? 'offline' : 'failed')
58 },
59 onAuthRequired: (message) => {
60 setStatus('auth_required')
61 setLastError(message)
62 },
63 onConnectError: (message) => {
64 setStatus('failed')
65 setLastError(message)
66 },
67 onEnvelope: (envelope) => {
68 const recipient = envelope.recipient_user_id
69 if (typeof recipient === 'number') setSelfId((current) => current ?? recipient)
70 for (const listener of [...listeners.current]) listener(envelope)
71 },
72 onResync: (envelope) => {
73 for (const listener of [...listeners.current]) listener({ ...envelope, resync_required: true })
74 },
75 onServerError: (message) => setLastError(message),
76 onSubscribeError: (subscription, message) => setLastError(`${subscription}: ${message}`),
77 })
78 socketRef.current = socket
79 socket.subscribePersonal()
80 return () => {
81 socket.dispose()
82 socketRef.current = null
83 pending.current.clear()
84 }
85 }, [url])
86
87 const subscribe = useCallback((spec: TopicSubscription) => {
88 const socket = socketRef.current
89 if (!socket) return () => {}
90 const key = spec.key ?? `${spec.topic}|${spec.tag}|${JSON.stringify(spec.filter ?? {})}`
91 const existing = pending.current.get(key)
92 if (existing) return existing
93 const off = socket.subscribeTopic({ key, topic: spec.topic, tag: spec.tag, filter: spec.filter })
94 const release = () => {
95 pending.current.delete(key)
96 off()
97 }
98 pending.current.set(key, release)
99 return release
100 }, [])
101
102 const onEnvelope = useCallback((handler: (envelope: KenEventEnvelope) => void) => {
103 listeners.current.add(handler)
104 return () => {
105 listeners.current.delete(handler)
106 }
107 }, [])
108
109 const retry = useCallback(() => {
110 socketRef.current?.stop()
111 socketRef.current?.start()
112 }, [])
113
114 const value = useMemo<RealtimeApi>(
115 () => ({ enabled: Boolean(url), status, selfId, subscribe, onEnvelope, retry, lastError }),
116 [url, status, selfId, subscribe, onEnvelope, retry, lastError],
117 )
118
119 return <RealtimeContext.Provider value={value}>{children}</RealtimeContext.Provider>
120}
121
122export function useRealtime(): RealtimeApi {
123 return useContext(RealtimeContext) ?? DISABLED
124}
125
126export function useRealtimeEnvelope(handler: (envelope: KenEventEnvelope) => void): void {
127 const realtime = useRealtime()
128 const ref = useRef(handler)
129 useEffect(() => {
130 ref.current = handler
131 })
132 useEffect(() => realtime.onEnvelope((envelope) => ref.current(envelope)), [realtime])
133}
134
135/** 读取信封里的资源属性（RocketMQ 用户属性都是字符串）。 */
136export const envelopeProperty = prop
Added view/app/src/services/realtime/hub.ts view/app/src/services/realtime/hub.ts +173 −0 View view/app/src/services/realtime/hub.ts
@@ -0,0 +1,173 @@
1/**
2 * App-wide realtime hub: exactly one `ChatWsClient` socket for the whole
3 * session, shared by every room view and any future consumer (notifications,
4 * presence, reminders).
5 *
6 * Why a module singleton instead of a React provider:
7 * - the socket's lifetime is the *session*, not a component's, so unmounting a
8 * chat column must not tear down another panel's subscription;
9 * - StrictMode's double mount would otherwise open and immediately kill one
10 * upgrade per mount, and every handshake JWT is single-use on the server;
11 * - keeping it framework-free makes the reconnect/gap logic testable without a
12 * DOM (see the React bindings in `./react`).
13 *
14 * Room intents are reference counted: the socket joins a room when the first
15 * viewer arrives and leaves it when the last one goes, which is what keeps a
16 * click through twenty channels inside the server's `max_subscriptions`.
17 */
18import { ChatWsClient, type WsClientIssue } from './client'
19import { parseServerFrame, type ServerFrame, type UserEvent, type WsStatus } from './protocol'
20
21type StatusListener = (status: WsStatus) => void
22type FrameListener = (frame: ServerFrame) => void
23type IssueListener = (issue: WsClientIssue) => void
24type UserEventListener = (event: UserEvent) => void
25
26export class ChatRealtimeHub {
27 private client: ChatWsClient | null = null
28 private status: WsStatus = 'idle'
29 private userId: string | undefined
30
31 private readonly viewers = new Map<string, number>()
32 private readonly statusListeners = new Set<StatusListener>()
33 private readonly frameListeners = new Set<FrameListener>()
34 private readonly issueListeners = new Set<IssueListener>()
35 private readonly userEventListeners = new Set<UserEventListener>()
36
37 /** `ready.user_id`, undefined until the first handshake — the viewer's own id. */
38 get selfId(): string | undefined {
39 return this.userId
40 }
41
42 get connectionStatus(): WsStatus {
43 return this.status
44 }
45
46 /** Rooms with at least one live viewer. */
47 watchedRooms(): string[] {
48 return [...this.viewers.entries()].filter(([, count]) => count > 0).map(([id]) => id)
49 }
50
51 /**
52 * Create the socket on first use. Nothing touches the network before this:
53 * the jwt endpoint must not be hit on the auth screens.
54 */
55 ensureClient(): ChatWsClient {
56 if (this.client) return this.client
57 const client = new ChatWsClient()
58 client.onStatus((status) => {
59 this.status = status
60 const ready = client.userId
61 if (ready) this.userId = ready
62 for (const listener of [...this.statusListeners]) listener(status)
63 })
64 client.onIssue((issue) => {
65 for (const listener of [...this.issueListeners]) listener(issue)
66 })
67 client.onFrame((frame) => {
68 if (frame.type === 'user_event') {
69 if (frame.event) for (const listener of [...this.userEventListeners]) listener(frame.event)
70 return
71 }
72 for (const listener of [...this.frameListeners]) listener(frame)
73 })
74 this.client = client
75 return client
76 }
77
78 /** Add a viewer for a room; starts the socket on the first one. */
79 watchRoom(roomId: string): () => void {
80 const next = (this.viewers.get(roomId) ?? 0) + 1
81 this.viewers.set(roomId, next)
82 if (next === 1) {
83 const client = this.ensureClient()
84 client.subscribe(roomId)
85 // `start()` is idempotent while connecting and is also the only way back
86 // out of a terminal `auth_required` / `failed`, which is exactly what
87 // needing a room again means after a session was refreshed.
88 client.start()
89 }
90 let released = false
91 return () => {
92 if (released) return
93 released = true
94 const count = (this.viewers.get(roomId) ?? 1) - 1
95 if (count > 0) {
96 this.viewers.set(roomId, count)
97 return
98 }
99 this.viewers.delete(roomId)
100 this.client?.unsubscribe(roomId)
101 }
102 }
103
104 /** Force a reconnect attempt after a terminal `auth_required` / `failed`. */
105 retry(): void {
106 const client = this.ensureClient()
107 client.stop()
108 client.start()
109 }
110
111 sendTyping(roomId: string, threadId?: string | null): void {
112 this.client?.sendTyping(roomId, threadId)
113 }
114
115 /** Live typing indicators, own echo excluded by the client. */
116 typingUsers(roomId: string, threadId?: string | null) {
117 return this.client?.typingUsers(roomId, { threadId }) ?? []
118 }
119
120 /** Fires with the room id whenever its live typing set changes. */
121 onTypingChange(listener: (roomId: string) => void): () => void {
122 return this.ensureClient().onTypingChange(listener)
123 }
124
125 lastKnownSeq(roomId: string): number | undefined {
126 return this.client?.lastKnownSeq(roomId)
127 }
128
129 onStatus(listener: StatusListener): () => void {
130 this.statusListeners.add(listener)
131 return () => this.statusListeners.delete(listener)
132 }
133
134 onFrame(listener: FrameListener): () => void {
135 this.frameListeners.add(listener)
136 return () => this.frameListeners.delete(listener)
137 }
138
139 onIssue(listener: IssueListener): () => void {
140 this.issueListeners.add(listener)
141 return () => this.issueListeners.delete(listener)
142 }
143
144 onUserEvent(listener: UserEventListener): () => void {
145 this.userEventListeners.add(listener)
146 return () => this.userEventListeners.delete(listener)
147 }
148
149 /** Drop everything — tests and signed-out teardown only. */
150 reset(): void {
151 this.client?.dispose()
152 this.client = null
153 this.status = 'idle'
154 this.userId = undefined
155 this.viewers.clear()
156 }
157}
158
159let hub: ChatRealtimeHub | null = null
160
161export function getRealtimeHub(): ChatRealtimeHub {
162 hub ??= new ChatRealtimeHub()
163 return hub
164}
165
166/** Exported for tests; production code never resets the session socket. */
167export function resetRealtimeHub(): void {
168 hub?.reset()
169 hub = null
170}
171
172/** Re-exported so consumers never deep-import the protocol module. */
173export { parseServerFrame, type ServerFrame, type UserEvent, type WsClientIssue, type WsStatus }
Added view/app/src/services/realtime/index.ts view/app/src/services/realtime/index.ts +67 −0 View view/app/src/services/realtime/index.ts
@@ -0,0 +1,67 @@
1// Public API of the shared/ws segment — the app-wide realtime channel.
2//
3// 传输层是 **KenEvent（RocketMQ 事件 → Socket.IO, `/socket.io`）**：浏览器用
4// 会话 cookie（`gitdata_session`）或 CONNECT 包的 `auth.token` 完成鉴权，
5// 客户端只发 `subscribe`/`unsubscribe`，服务端推 `event`/`resync` 信封
6// `{topic, tag, properties, body}`，详情一律通过 REST 回拉。
7//
8// 旧的 `GET /ws/v1`（`app/im` 网关，裸 WebSocket + `gitdata.jwt.*` 子协议）
9// 已随 KenApi 重构删除，相关实现（含一次性握手 JWT）一并移除。
10//
11// 连接属于整个会话而不是某个聊天窗口，所以放在 shared 而不是 chat feature 内。
12// 房间意图按引用计数：第一个观看者进入时订阅，最后一个离开时退订。
13//
14// Typical usage:
15// const ws = new ChatWsClient()
16// ws.onFrame(frame => { if (frame.type === 'event') ... })
17// ws.onIssue(issue => ...) // 订阅被拒、鉴权失败、断流补齐
18// ws.onStatus(status => ...) // 含终态 'auth_required'
19// ws.start()
20// ws.subscribe(roomId) // 订阅房间；历史用 REST 拉
21export { ChatWsClient, type ChatWsClientOptions, type WsClientIssue } from './client'
22export { ChatRealtimeHub, getRealtimeHub, resetRealtimeHub } from './hub'
23export {
24 asServerFrame,
25 type ClientFrame,
26 classifyWsError,
27 isNotificationUserEvent,
28 isSubscribeError,
29 parseServerFrame,
30 type RoomEvent,
31 roomEventSeq,
32 type ServerFrame,
33 type UserEvent,
34 WS_BYE_MESSAGES,
35 WS_CLIENT_OPS,
36 WS_ERROR_MESSAGES,
37 WS_RESYNC_REASONS,
38 WS_ROOM_EVENT_KINDS,
39 WS_SERVER_DEFAULTS,
40 WS_SERVER_FRAME_TYPES,
41 WS_USER_EVENT_KINDS,
42 type WsByeMessage,
43 type WsErrorMessage,
44 type WsMessage,
45 type WsNotification,
46 type WsPingFrame,
47 type WsReactionSummary,
48 type WsResyncFrame,
49 type WsResyncReason,
50 type WsServerErrorKind,
51 type WsStatus,
52 type WsSubscribeFrame,
53 type WsTypingFrame,
54 type WsTypingIndicator,
55 type WsUnsubscribeFrame,
56} from './protocol'
57export {
58 sendTypingSignal,
59 useRoomWatch,
60 useTypingIndicators,
61 useWsFrames,
62 useWsIssues,
63 useWsStatus,
64 useWsUserEvents,
65 useWsUserId,
66} from './react'
67export { TypingRegistry, type WsTypingEvent } from './typing'
Added view/app/src/services/realtime/kenEvent.ts view/app/src/services/realtime/kenEvent.ts +337 −0 View view/app/src/services/realtime/kenEvent.ts
@@ -0,0 +1,337 @@
1/**
2 * KenEvent transport: the Socket.IO wire contract of the realtime gateway
3 * (`KenEvent/src/socket`, Engine.IO v4, path `/socket.io`).
4 *
5 * 与旧 im 网关（`GET /ws/v1`，裸 WebSocket + `gitdata.jwt.*` 子协议）的区别：
6 * - **连接**：Socket.IO，默认路径 `/socket.io`；浏览器握手带不了自定义头，
7 * 所以 token 由会话 cookie（`gitdata_session`）或 CONNECT 包的 `auth.token`
8 * 提供，KenEvent 侧见 `socket/response.rs` 与 `socket/session/state.rs`。
9 * - **订阅**：客户端发 `subscribe {topic, tag, filter, id}`，服务端按
10 * RocketMQ 的 topic/tag 精确匹配（`events/subscriptions.rs:validate`），
11 * 没有通配，所以一个房间要按关心的 tag 各订阅一条。
12 * - **推送**：服务端只发两种事件名——`event` 与 `resync`，载荷是
13 * `{topic, tag, user_id, recipient_user_id, properties, body}` 信封
14 * （`events/action/common.rs:payload`）。**详情一律走 REST 回拉**。
15 * - **个人事件**：`gitdata-notify`/`notification-create` 与 chat 的
16 * mention/read_state/presence 等由服务端强制绑定 `filter.user_id` 自己。
17 */
18import { io, type Socket } from 'socket.io-client'
19
20export const KEN_EVENT_PATH = '/socket.io'
21export const CHAT_TOPIC = 'gitdata-chat'
22export const NOTIFY_TOPIC = 'gitdata-notify'
23
24/** 房间级事件：一条订阅一个 tag，`filter` 固定为 `{room_id}`。 */
25export const ROOM_TAGS = [
26 'message-create',
27 'message-update',
28 'message-delete',
29 'messages-delete',
30 'pin-create',
31 'pin-delete',
32 'reaction-update',
33 'room-update',
34 'member-create',
35 'member-delete',
36 'thread-update',
37 'voice-presence',
38 'voice-hand',
39 'voice-layout',
40 'voice-moderate',
41] as const
42
43/** 用户级事件：不带 room 过滤，服务端会补上 `filter.user_id = 自己`。 */
44export const PERSONAL_CHAT_TAGS = [
45 'mention-create',
46 'reminder-create',
47 'membership-create',
48 'read_state-update',
49 'presence-update',
50] as const
51
52export const PERSONAL_NOTIFY_TAGS = ['notification-create'] as const
53
54/** 服务端推送的信封（`events/action/common.rs:80`）。 */
55export interface KenEventEnvelope {
56 topic: string
57 tag: string
58 user_id?: number | null
59 recipient_user_id?: number
60 properties?: Record<string, unknown> | null
61 body?: unknown
62 /** 断流补齐时 KenEvent 会把事件名换成 `resync` 并置这个标记。 */
63 resync_required?: boolean
64}
65
66export interface SubscriptionAck {
67 ok?: boolean
68 data?: unknown
69 error?: string
70}
71
72export interface KenEventSocketOptions {
73 /** 默认同源（`window.location.origin`）；vite dev 与网关都由 `/socket.io` 反代。 */
74 url?: string
75 onOpen: (info: { sid: string }) => void
76 onClose: (reason: string, wasOpen: boolean) => void
77 onAuthRequired: (message: string) => void
78 onConnectError: (message: string) => void
79 onEnvelope: (envelope: KenEventEnvelope) => void
80 onResync: (envelope: KenEventEnvelope) => void
81 onServerError: (message: string) => void
82 onSubscribeError: (subscription: string, message: string) => void
83}
84
85const DEVICE_KEY = 'gitdata.ws.device_id'
86
87/** 设备标识：KenEvent 用它做多端会话隔离（`session/state.rs`）。 */
88export function deviceId(): string {
89 try {
90 const stored = window.localStorage.getItem(DEVICE_KEY)
91 if (stored) return stored
92 const created = crypto.randomUUID()
93 window.localStorage.setItem(DEVICE_KEY, created)
94 return created
95 } catch {
96 // 隐私模式 / SSR：退化成会话级 id，重连后仍能复用（同一模块生命周期内）。
97 return crypto.randomUUID()
98 }
99}
100
101function asRecord(value: unknown): Record<string, unknown> | null {
102 return typeof value === 'object' && value !== null ? (value as Record<string, unknown>) : null
103}
104
105export function asEnvelope(value: unknown): KenEventEnvelope | null {
106 const record = asRecord(value)
107 if (!record) return null
108 const topic = record.topic
109 const tag = record.tag
110 if (typeof topic !== 'string' || typeof tag !== 'string') return null
111 return record as unknown as KenEventEnvelope
112}
113
114/** `properties` 里的字符串字段（RocketMQ 用户属性全是字符串）。 */
115export function prop(envelope: KenEventEnvelope, key: string): string | null {
116 const value = envelope.properties?.[key]
117 if (typeof value === 'string' && value.length > 0) return value
118 const fromBody = asRecord(envelope.body)?.[key]
119 return typeof fromBody === 'string' && fromBody.length > 0 ? fromBody : null
120}
121
122/**
123 * 订阅管理器：把「打开了哪些房间」这份意图翻译成 KenEvent 的逐 tag 订阅。
124 *
125 * 重连后服务端的订阅是空的（per-socket 状态），所以每次 `connect` 都要重放；
126 * 房间的 refcount 由上层 hub 维护，这里只负责 idempotent 增删。
127 */
128export class KenEventSocket {
129 private socket: Socket | null = null
130 /** roomId -> tag -> 订阅 id（KenEvent 要求 id 是 UUID，由客户端生成）。 */
131 private readonly rooms = new Map<string, Map<string, string>>()
132 /** `${topic}|${tag}` -> 个人订阅 id。 */
133 private readonly personalIds = new Map<string, string>()
134 /** 通用资源订阅：key -> 订阅意图（issue / pull / org / kenshell 等）。 */
135 private readonly topics = new Map<
136 string,
137 { topic: string; tag: string; filter: Record<string, string>; refs: number; id: string }
138 >()
139 private personal = false
140 private disposed = false
141
142 private readonly options: KenEventSocketOptions
143
144 constructor(options: KenEventSocketOptions) {
145 this.options = options
146 }
147
148 get connected(): boolean {
149 return this.socket?.connected ?? false
150 }
151
152 /** 幂等：连接中/已连接时什么也不做。 */
153 start(): void {
154 if (this.disposed || this.socket) return
155 const socket = io(this.options.url, {
156 path: KEN_EVENT_PATH,
157 // 先 polling 再升级 websocket：网关（KenGateWay）不代理 Upgrade，会回 501，
158 // 从 websocket 起步会直接连不上；polling 起步在两种部署下都能连上，
159 // 拿到连接后 socket.io 会自己尝试升级到 WebSocket（nginx 环境即可升级成功）。
160 transports: ['polling', 'websocket'],
161 // 兜底：某个传输被拒时继续尝试下一个，而不是直接失败。
162 tryAllTransports: true,
163 withCredentials: true,
164 auth: (cb) => cb({ device_id: deviceId() }),
165 reconnection: true,
166 reconnectionDelay: 500,
167 reconnectionDelayMax: 30_000,
168 timeout: 20_000,
169 })
170 this.socket = socket
171
172 socket.on('connect', () => {
173 this.options.onOpen({ sid: socket.id ?? '' })
174 this.replay()
175 })
176 socket.on('disconnect', (reason) => {
177 this.options.onClose(reason, true)
178 })
179 socket.on('connect_error', (error) => {
180 const message = error instanceof Error ? error.message : String(error)
181 // KenEvent 在鉴权失败时发 `connect_error('authentication failed')`
182 // （`socket/session/state.rs:connect`）。
183 if (/auth|unauthor|missing access token/i.test(message)) {
184 this.options.onAuthRequired(message)
185 return
186 }
187 this.options.onConnectError(message)
188 })
189 socket.on('event', (payload: unknown) => {
190 const envelope = asEnvelope(payload)
191 if (!envelope) {
192 this.options.onServerError('unrecognized event envelope')
193 return
194 }
195 if (envelope.resync_required) {
196 this.options.onResync(envelope)
197 return
198 }
199 this.options.onEnvelope(envelope)
200 })
201 socket.on('resync', (payload: unknown) => {
202 const envelope = asEnvelope(payload)
203 if (envelope) this.options.onResync(envelope)
204 })
205 socket.on('error', (payload: unknown) => {
206 const message = (asRecord(payload)?.message as string | undefined) ?? 'realtime gateway error'
207 this.options.onServerError(message)
208 })
209 }
210
211 stop(): void {
212 const socket = this.socket
213 this.socket = null
214 if (!socket) return
215 socket.removeAllListeners()
216 socket.disconnect()
217 }
218
219 dispose(): void {
220 this.disposed = true
221 this.rooms.clear()
222 this.personal = false
223 this.stop()
224 }
225
226 /** 加入一个房间：按 `ROOM_TAGS` 逐条订阅。 */
227 subscribeRoom(roomId: string): void {
228 if (this.rooms.has(roomId)) return
229 this.rooms.set(roomId, new Map())
230 this.start()
231 if (this.connected) this.emitRoom(roomId)
232 }
233
234 unsubscribeRoom(roomId: string): void {
235 const ids = this.rooms.get(roomId)
236 if (!ids) return
237 this.rooms.delete(roomId)
238 if (!this.connected) return
239 for (const [tag, id] of ids) this.emitUnsubscribe(CHAT_TOPIC, tag, id)
240 }
241
242 /**
243 * 订阅任意 topic/tag（带资源过滤）。同一 `key` 只发一次订阅，引用计数归零时退订；
244 * 返回的函数可安全重复调用。
245 */
246 subscribeTopic(spec: { key: string; topic: string; tag: string; filter?: Record<string, string> }): () => void {
247 const existing = this.topics.get(spec.key)
248 if (existing) {
249 existing.refs += 1
250 return () => this.releaseTopic(spec.key)
251 }
252 const entry = {
253 topic: spec.topic,
254 tag: spec.tag,
255 filter: spec.filter ?? {},
256 refs: 1,
257 id: crypto.randomUUID(),
258 }
259 this.topics.set(spec.key, entry)
260 this.start()
261 if (this.connected) this.emit(entry.topic, entry.tag, { id: entry.id, filter: entry.filter })
262 return () => this.releaseTopic(spec.key)
263 }
264
265 private releaseTopic(key: string): void {
266 const entry = this.topics.get(key)
267 if (!entry) return
268 entry.refs -= 1
269 if (entry.refs > 0) return
270 this.topics.delete(key)
271 if (this.connected) this.emitUnsubscribe(entry.topic, entry.tag, entry.id)
272 }
273
274 /** 个人通道（通知、提及、已读游标）：一次连接订阅一次。 */
275 subscribePersonal(): void {
276 if (this.personal) return
277 this.personal = true
278 this.start()
279 if (this.connected) this.emitPersonal()
280 }
281
282 /** 重放全部订阅意图；每次 `connect` 都会调用。 */
283 private replay(): void {
284 this.personalIds.clear()
285 for (const ids of this.rooms.values()) ids.clear()
286 if (this.personal) this.emitPersonal()
287 for (const roomId of this.rooms.keys()) this.emitRoom(roomId)
288 for (const entry of this.topics.values()) {
289 entry.id = crypto.randomUUID()
290 this.emit(entry.topic, entry.tag, { id: entry.id, filter: entry.filter })
291 }
292 }
293
294 private emitRoom(roomId: string): void {
295 const ids = this.rooms.get(roomId)
296 if (!ids) return
297 for (const tag of ROOM_TAGS) {
298 if (ids.has(tag)) continue
299 const id = crypto.randomUUID()
300 ids.set(tag, id)
301 this.emit(CHAT_TOPIC, tag, { id, filter: { room_id: roomId } })
302 }
303 }
304
305 private emitPersonal(): void {
306 for (const tag of PERSONAL_CHAT_TAGS) this.emitPersonalOne(CHAT_TOPIC, tag)
307 for (const tag of PERSONAL_NOTIFY_TAGS) this.emitPersonalOne(NOTIFY_TOPIC, tag)
308 }
309
310 private emitPersonalOne(topic: string, tag: string): void {
311 const key = `${topic}|${tag}`
312 if (this.personalIds.has(key)) return
313 const id = crypto.randomUUID()
314 this.personalIds.set(key, id)
315 this.emit(topic, tag, { id })
316 }
317
318 private emit(topic: string, tag: string, extra: { id: string; filter?: unknown }): void {
319 const socket = this.socket
320 if (!socket) return
321 const payload = { topic, tag, ...extra }
322 // ack 里的 `{ok:false,error}` 是唯一能归因到具体订阅的失败信号。
323 socket.timeout(10_000).emit('subscribe', payload, (error: Error | null, ack?: SubscriptionAck) => {
324 if (error) {
325 this.options.onSubscribeError(`${topic}/${tag}`, 'subscribe ack timeout')
326 return
327 }
328 if (ack && ack.ok === false) {
329 this.options.onSubscribeError(`${topic}/${tag}`, ack.error ?? 'subscribe rejected')
330 }
331 })
332 }
333
334 private emitUnsubscribe(topic: string, tag: string, id: string): void {
335 this.socket?.emit('unsubscribe', { topic, tag, id })
336 }
337}
Added view/app/src/services/realtime/kenEventFrames.ts view/app/src/services/realtime/kenEventFrames.ts +205 −0 View view/app/src/services/realtime/kenEventFrames.ts
@@ -0,0 +1,205 @@
1/**
2 * KenEvent 信封 → 前端 `ServerFrame` 的映射。
3 *
4 * 应用层（transcript reducer、typing registry、通知面板）消费的是 `RoomEvent` /
5 * `UserEvent` 联合类型，而 KenEvent 只发 RocketMQ 的 `{topic, tag, properties,
6 * body}`。这一层是唯一的翻译点：新增事件时只改这里，页面代码不动。
7 *
8 * 约定（来自 `docs/events.json` 与 KenApi 的发布点）：
9 * - `body` 对 chat 消息类事件就是 `MessageDto` 本体，`seq` 也在里面
10 * （所以 `message_published` 的游标可直接用 `body.seq`）。
11 * - `properties` 里的资源 id 全是字符串（RocketMQ 用户属性）。
12 * - 详情永远以 REST 为准，这里只负责"告诉页面什么变了"。
13 */
14import type { MessageDto } from '@app/services/api'
15
16import { CHAT_TOPIC, type KenEventEnvelope, NOTIFY_TOPIC, prop } from './kenEvent'
17import type { RoomEvent, ServerFrame, UserEvent, WsNotification, WsReactionSummary } from './protocol'
18
19function record(value: unknown): Record<string, unknown> | null {
20 return typeof value === 'object' && value !== null ? (value as Record<string, unknown>) : null
21}
22
23function str(value: unknown): string | null {
24 return typeof value === 'string' && value.length > 0 ? value : null
25}
26
27function num(value: unknown): number | undefined {
28 return typeof value === 'number' && Number.isFinite(value) ? value : undefined
29}
30
31function messageOf(envelope: KenEventEnvelope): MessageDto | null {
32 const body = record(envelope.body)
33 if (!body) return null
34 return typeof body.id === 'string' && typeof body.room_id === 'string' ? (body as unknown as MessageDto) : null
35}
36
37function reactionsOf(envelope: KenEventEnvelope): WsReactionSummary[] {
38 const body = record(envelope.body)
39 const raw = body?.reactions
40 if (!Array.isArray(raw)) return []
41 return raw.flatMap((entry) => {
42 const item = record(entry)
43 const emoji = str(item?.emoji)
44 const count = num(item?.count)
45 return emoji && count !== undefined ? [{ emoji, count }] : []
46 })
47}
48
49function deletedIds(envelope: KenEventEnvelope): string[] {
50 const body = record(envelope.body)
51 const candidates = [body?.message_ids, body?.ids, body?.messages]
52 for (const candidate of candidates) {
53 if (Array.isArray(candidate)) {
54 const ids = candidate.map((item) => str(item) ?? str(record(item)?.id)).filter(Boolean)
55 if (ids.length > 0) return ids as string[]
56 }
57 }
58 const single = prop(envelope, 'message_id') ?? str(body?.id)
59 return single ? [single] : []
60}
61
62function roomEvent(envelope: KenEventEnvelope): RoomEvent | null {
63 const roomId = prop(envelope, 'room_id')
64 const message = messageOf(envelope)
65 const messageId = prop(envelope, 'message_id') ?? str(record(envelope.body)?.id)
66 const userId = envelope.user_id != null ? String(envelope.user_id) : prop(envelope, 'user_id')
67 const seq = num(record(envelope.body)?.seq) ?? 0
68
69 switch (envelope.tag) {
70 case 'message-create':
71 case 'message-update': {
72 if (!roomId || !message) return null
73 return {
74 kind: envelope.tag === 'message-create' ? 'message_published' : 'message_edited',
75 room_id: roomId,
76 seq,
77 message,
78 }
79 }
80 case 'message-delete':
81 case 'messages-delete': {
82 if (!roomId) return null
83 const ids = deletedIds(envelope)
84 if (ids.length === 0) return null
85 return envelope.tag === 'messages-delete' && ids.length > 1
86 ? { kind: 'messages_deleted', room_id: roomId, seq, message_ids: ids }
87 : { kind: 'message_deleted', room_id: roomId, seq, message_id: ids[0] }
88 }
89 case 'pin-create':
90 case 'pin-delete': {
91 if (!roomId || !messageId) return null
92 return {
93 kind: 'message_pinned',
94 room_id: roomId,
95 seq,
96 message_id: messageId,
97 pinned: envelope.tag === 'pin-create',
98 }
99 }
100 case 'reaction-update': {
101 if (!roomId || !messageId) return null
102 return {
103 kind: 'reaction_updated',
104 room_id: roomId,
105 message_id: messageId,
106 reactions: reactionsOf(envelope),
107 }
108 }
109 case 'room-update':
110 return roomId ? { kind: 'room_updated', room_id: roomId } : null
111 case 'member-create':
112 return roomId && userId ? { kind: 'member_joined', room_id: roomId, user_id: userId } : null
113 case 'member-delete':
114 return roomId && userId ? { kind: 'member_left', room_id: roomId, user_id: userId } : null
115 case 'thread-update': {
116 const threadId = prop(envelope, 'thread_id')
117 return roomId && threadId ? { kind: 'thread_updated', room_id: roomId, thread_id: threadId } : null
118 }
119 case 'voice-presence':
120 return roomId ? { kind: 'voice_presence', room_id: roomId } : null
121 case 'voice-hand':
122 return roomId ? { kind: 'voice_hands', room_id: roomId } : null
123 case 'voice-layout':
124 return roomId ? { kind: 'voice_layout', room_id: roomId } : null
125 case 'voice-moderate':
126 return roomId && userId
127 ? {
128 kind: 'voice_moderation',
129 room_id: roomId,
130 user_id: userId,
131 action: str(record(envelope.body)?.action) ?? 'update',
132 }
133 : null
134 default:
135 return null
136 }
137}
138
139function userEvent(envelope: KenEventEnvelope): UserEvent | null {
140 const roomId = prop(envelope, 'room_id')
141 const messageId = prop(envelope, 'message_id') ?? str(record(envelope.body)?.id)
142
143 if (envelope.topic === NOTIFY_TOPIC && envelope.tag === 'notification-create') {
144 const body = record(envelope.body)
145 if (!body || typeof body.id !== 'string') return null
146 return { kind: 'notification', notification: body as unknown as WsNotification }
147 }
148
149 if (envelope.topic !== CHAT_TOPIC) return null
150
151 switch (envelope.tag) {
152 case 'mention-create':
153 return roomId && messageId
154 ? {
155 kind: 'mention',
156 room_id: roomId,
157 message_id: messageId,
158 everyone: record(envelope.body)?.everyone === true ? true : undefined,
159 }
160 : null
161 case 'reminder-create':
162 return roomId && messageId ? { kind: 'reminder', room_id: roomId, message_id: messageId } : null
163 case 'membership-create':
164 return roomId ? { kind: 'room_added', room_id: roomId } : null
165 case 'read_state-update': {
166 const body = record(envelope.body)
167 const lastRead = str(body?.last_read_message_id)
168 const lastSeq = num(body?.last_read_seq) ?? 0
169 return roomId
170 ? {
171 kind: 'read_marker',
172 room_id: roomId,
173 last_read_message_id: lastRead,
174 last_read_seq: lastSeq,
175 }
176 : null
177 }
178 case 'presence-update': {
179 const status = str(record(envelope.body)?.status)
180 return status ? { kind: 'presence', status: status as never } : null
181 }
182 default:
183 return null
184 }
185}
186
187/** 单个信封 → 页面帧；无法映射（未消费的事件）返回 null，由调用方静默忽略。 */
188export function toServerFrame(envelope: KenEventEnvelope): ServerFrame | null {
189 const event = userEvent(envelope)
190 if (event) return { type: 'user_event', event }
191
192 const room = roomEvent(envelope)
193 if (!room) return null
194 if ('room_id' in room) return { type: 'event', room_id: room.room_id, event: room }
195 return null
196}
197
198/** 从消息类帧里取房间游标，供 `lastKnownSeq` 使用。 */
199export function frameSeq(frame: ServerFrame): { roomId: string; seq: number } | null {
200 if (frame.type !== 'event' || !frame.event) return null
201 if ('seq' in frame.event && typeof frame.event.seq === 'number') {
202 return { roomId: frame.room_id, seq: frame.event.seq }
203 }
204 return null
205}
Added view/app/src/services/realtime/protocol.ts view/app/src/services/realtime/protocol.ts +431 −0 View view/app/src/services/realtime/protocol.ts
@@ -0,0 +1,431 @@
1/**
2 * Wire protocol of the (removed) chat WebSocket endpoint (`GET /ws/v1`, im gateway).
3 *
4 * 已过时：`app/im` 已从 KenApi 删除，实时通道归 `KenEvent`（Socket.IO）。
5 * 文件保留为旧协议的对照说明；按 KenEvent 重写时以
6 * `KenEvent/src/events/action/common.rs::payload` 为准（信封 = topic/tag/
7 * properties/body，没有 `subscribe` 帧）。
8 *
9 * This file is a 1:1 mirror of the Rust definitions in `gitdataai/app/im/src/frames.rs` and `lib/chat/src/share/realtime.rs`:
10 * src/ws/frames.rs -> ClientFrame / ServerFrame
11 * src/share/realtime.rs -> RoomEvent (`event` frame payloads)
12 * src/share/notify.rs + src/services/* -> the ad-hoc JSON on the user channel
13 * (`user_event` frame payloads)
14 *
15 * Serde rules this file encodes:
16 * - Client frames are internally tagged by `op` with `rename_all = "snake_case"`
17 * (frames.rs:6-7); server frames by `type` (frames.rs:29-30); the payloads
18 * carried by `event` / `user_event` by `kind` (realtime.rs:19-20).
19 * - Rust `Option<T>` fields are *always* serialised (`null` when absent) because
20 * the backend never uses `skip_serializing_if`, so those fields are
21 * `T | null` (required) here rather than `?: T`. The one exception is
22 * `mention.everyone`, which is built with `json!` and only appears on the
23 * @everyone fanout (notify.rs:231-237).
24 * - `Uuid` crosses as a string, `i64`/`u64` as a JSON number, `OffsetDateTime`
25 * as an RFC 3339 string.
26 * - `PresenceStatus` carries **no** `rename_all` (models/chat/presence_states.rs:7-31),
27 * so it is PascalCase on the wire, unlike every other chat enum. Reuse the
28 * generated type instead of hand-writing the variants.
29 */
30
31import type { MessageDto, NotificationType, PresenceStatus } from '@app/services/api'
32
33/* --------------------------------------------------------------------------
34 * Connection lifecycle (client-side only; never on the wire)
35 * ------------------------------------------------------------------------ */
36
37/**
38 * Connection lifecycle reported by `ChatWsClient.onStatus`.
39 * - `auth_required` / `failed` are terminal: the client stops reconnecting and
40 * needs an explicit `start()` (after re-login / after a manual retry).
41 */
42export type WsStatus = 'idle' | 'connecting' | 'open' | 'reconnecting' | 'closed' | 'auth_required' | 'failed'
43
44/* --------------------------------------------------------------------------
45 * Server knobs that leak into the protocol
46 * (config/default.toml:143-163 + lib/config/src/chat.rs:1-23,60-101)
47 * ------------------------------------------------------------------------ */
48
49/**
50 * Defaults of the `[chat]` / `[chat.ws]` config sections. They are *advisory*:
51 * a deployment can change them, so the client uses them as pre-flight limits and
52 * still treats the server's `error` / `resync_required` frames as authoritative.
53 */
54export const WS_SERVER_DEFAULTS = {
55 /** Lifetime of a `GET /auth/jwt-request` JWT; single-use (im gateway SETNX jti). */
56 jwtTtlSecs: 60,
57 /** Server tick interval; an idle socket gets `bye` after 3x (ws/mod.rs:220-224,310-315). */
58 heartbeatSecs: 30,
59 /** Max room subscriptions per socket (ws/mod.rs:228,410-419). */
60 maxSubscriptions: 64,
61 /** Inbound frame limit set on the socket (ws/mod.rs:141). */
62 maxFrameBytes: 262_144,
63 /** `history` / `resync` page size (ws/mod.rs:470,516). */
64 historyLimit: 200,
65 /** TTL carried by `typing` events, floor of 1s (ws/mod.rs:545). */
66 typingTtlSecs: 5,
67 /** Hard-coded per-socket typing throttle, not configurable (ws/mod.rs:534-538). */
68 typingThrottleSecs: 3,
69} as const
70
71/* --------------------------------------------------------------------------
72 * Client -> server frames (ws/frames.rs:6-27)
73 * ------------------------------------------------------------------------ */
74
75export const WS_CLIENT_OPS = ['subscribe', 'unsubscribe', 'resync', 'typing', 'ping'] as const
76
77/**
78 * `Subscribe { room_id, resync }`. `resync` is `Option<bool>` and the server
79 * does `resync.unwrap_or(true)` (frames.rs:9-13, ws/mod.rs:485), so omitting it
80 * asks for a `history` snapshot of the newest `history_limit` messages.
81 */
82export interface WsSubscribeFrame {
83 op: 'subscribe'
84 room_id: string
85 resync?: boolean
86}
87
88/** `Unsubscribe { room_id }` (frames.rs:14-16). */
89export interface WsUnsubscribeFrame {
90 op: 'unsubscribe'
91 room_id: string
92}
93
94/**
95 * `Resync { room_id, after_seq }`. The reply is a `history` frame built with
96 * `seq > after_seq` ordered **ascending** (ws/mod.rs:505-531 +
97 * services/messages.rs:677-687), i.e. it walks forward from the cursor and is
98 * truncated at `history_limit`; `resync(room, 0)` therefore returns the *oldest*
99 * messages of the room, not the newest.
100 */
101export interface WsResyncFrame {
102 op: 'resync'
103 room_id: string
104 after_seq: number
105}
106
107/**
108 * `Typing { room_id, thread_id }`. Requires read access, is throttled to one
109 * broadcast per 3s per socket, and is re-broadcast to the room as
110 * `event{kind:'typing'}` (ws/mod.rs:533-549).
111 */
112export interface WsTypingFrame {
113 op: 'typing'
114 room_id: string
115 thread_id?: string | null
116}
117
118/** `Ping` -> the server answers `{type:'pong'}` (frames.rs:26, ws/mod.rs:550-552). */
119export interface WsPingFrame {
120 op: 'ping'
121}
122
123export type ClientFrame = WsSubscribeFrame | WsUnsubscribeFrame | WsResyncFrame | WsTypingFrame | WsPingFrame
124
125/* --------------------------------------------------------------------------
126 * Message / reaction payloads
127 * ------------------------------------------------------------------------ */
128
129/**
130 * `MessageDto` (frames.rs:45 for `history`, realtime.rs:25/30 for events).
131 * Aliased to the generated type so it stays in sync with `bun run gen:api`.
132 *
133 * IMPORTANT: the backend serialises this DTO from the **actor's** point of view
134 * (`build_message_dto(.., caller)` — messages.rs:557-561 for publishes,
135 * :864-870 for edits, unfurl.rs:319-332 even uses `Uuid::nil()`), so per-viewer
136 * fields such as `reactions[].me` must not be trusted from a WS frame. The
137 * backend proves it knows this by stripping `me` from `reaction_updated`
138 * (services/reactions.rs:148-156).
139 */
140export type WsMessage = MessageDto
141
142/** Viewer-neutral reaction aggregate sent by `reaction_updated`. */
143export interface WsReactionSummary {
144 emoji: string
145 count: number
146}
147
148/* --------------------------------------------------------------------------
149 * Room broadcasts (`event` frames) (share/realtime.rs:19-74)
150 * ------------------------------------------------------------------------ */
151
152export const WS_ROOM_EVENT_KINDS = [
153 'message_published',
154 'message_edited',
155 'message_deleted',
156 'messages_deleted',
157 'message_pinned',
158 'reaction_updated',
159 'typing',
160 'voice_presence',
161 'voice_hands',
162 'voice_layout',
163 'voice_moderation',
164 'room_updated',
165 'member_joined',
166 'member_left',
167 'thread_updated',
168] as const
169
170/**
171 * The 11 `RoomEvent` variants. `seq` exists on exactly the five `message*`
172 * variants (realtime.rs:103-110): `message_published` allocates a fresh room seq
173 * (`next_room_seq`, notify.rs:17-33) while edits/pins/deletes reuse the *current
174 * tip* (`current_room_seq`, notify.rs:37-45) so they never inflate unread
175 * counters or reorder listings. `reaction_updated` has no seq at all.
176 */
177export type RoomEvent =
178 | { kind: 'message_published'; room_id: string; seq: number; message: WsMessage }
179 | { kind: 'message_edited'; room_id: string; seq: number; message: WsMessage }
180 | { kind: 'message_deleted'; room_id: string; seq: number; message_id: string }
181 | { kind: 'messages_deleted'; room_id: string; seq: number; message_ids: string[] }
182 | {
183 kind: 'message_pinned'
184 room_id: string
185 seq: number
186 message_id: string
187 pinned: boolean
188 }
189 | {
190 kind: 'reaction_updated'
191 room_id: string
192 message_id: string
193 reactions: WsReactionSummary[]
194 }
195 | {
196 kind: 'typing'
197 room_id: string
198 user_id: string
199 thread_id: string | null
200 /** Discard the indicator this many seconds after it was seen. */
201 ttl_secs: number
202 }
203 | { kind: 'voice_presence'; room_id: string }
204 | { kind: 'voice_hands'; room_id: string }
205 | { kind: 'voice_layout'; room_id: string }
206 | { kind: 'voice_moderation'; room_id: string; user_id: string; action: string }
207 | { kind: 'room_updated'; room_id: string }
208 | { kind: 'member_joined'; room_id: string; user_id: string }
209 | { kind: 'member_left'; room_id: string; user_id: string }
210 | { kind: 'thread_updated'; room_id: string; thread_id: string }
211
212/**
213 * Finite-number read for seq comparisons: a `seq` that is not a JSON number must
214 * never enter a cursor, because `NaN` would break every later comparison.
215 */
216export function asSeq(value: unknown): number | undefined {
217 return typeof value === 'number' && Number.isFinite(value) ? value : undefined
218}
219
220/** Typing payload for the client-side registry (TTL already resolved to ms). */
221export interface WsTypingIndicator {
222 room_id: string
223 user_id: string
224 thread_id: string | null
225 /** `Date.now()` based expiry stamp. */
226 expires_at: number
227 /** `Date.now()` stamp of the last refresh (indicators are refreshed, not re-added). */
228 updated_at: number
229}
230
231/** `RoomEvent::seq` when the variant carries one (realtime.rs:103-110). */
232export function roomEventSeq(event: RoomEvent | null | undefined): number | undefined {
233 if (!event) return undefined
234 if (!('seq' in event)) return undefined
235 // Guard against drift: a non-numeric seq would poison every cursor comparison.
236 return asSeq(event.seq)
237}
238
239/* --------------------------------------------------------------------------
240 * Per-user pushes (`user_event` frames)
241 * ------------------------------------------------------------------------ */
242
243export const WS_USER_EVENT_KINDS = [
244 'mention',
245 'presence',
246 'read_marker',
247 'reminder',
248 'room_added',
249 'notification',
250] as const
251
252/**
253 * The six payloads published on `chat:user:<uid>` (realtime.rs:15-17). They are
254 * hand-built `json!` objects, not a Rust enum, so this union is the only typed
255 * description of them:
256 * - `mention`: one-arg form per direct mention (messages.rs:591-596 and the edit
257 * path messages.rs:881-887); `everyone: true` only on the @everyone fanout
258 * (notify.rs:231-237).
259 * - `presence`: published to the acting user only (presence.rs:94-99), so it
260 * mirrors *your* status across tabs — a teammate's presence still has to come
261 * from `GET /chat/spaces/{space_name}/presences`.
262 * - `read_marker`: only when the marker actually advanced, and
263 * `last_read_message_id` is `null` when room-level state was advanced without a
264 * message id (read_state.rs:86-98 + notify.rs:301-326).
265 * - `reminder`: fired by the reminder worker (reminders.rs:251-258).
266 * - `room_added`: pushed to the user who just joined a room, next to the room's
267 * `member_joined` broadcast (rooms.rs:656-666).
268 * - `notification`: one row of `user_notifications` written by any crate
269 * (chat mentions/reminders/invites, issue replies/assignments, space invites);
270 * `export::notify::notification_event_json` wraps `Model::to_event_value()`,
271 * which mirrors the `GET /me/notifications` DTO field-for-field.
272 */
273export type UserEvent =
274 | { kind: 'mention'; room_id: string; message_id: string; everyone?: boolean }
275 | { kind: 'presence'; status: PresenceStatus }
276 | {
277 kind: 'read_marker'
278 room_id: string
279 last_read_message_id: string | null
280 last_read_seq: number
281 }
282 | { kind: 'reminder'; room_id: string; message_id: string }
283 | { kind: 'room_added'; room_id: string }
284 | { kind: 'notification'; notification: WsNotification }
285
286/** The embedded payload of a `notification` user event (`Model::to_event_value`). */
287export interface WsNotification {
288 id: string
289 notification_type: NotificationType
290 title: string
291 body: string | null
292 data: unknown
293 actor_id: string | null
294 read_at: string | null
295 created_at: string
296}
297
298export function isNotificationUserEvent(event: UserEvent): event is {
299 kind: 'notification'
300 notification: WsNotification
301} {
302 return event.kind === 'notification'
303}
304
305/* --------------------------------------------------------------------------
306 * Server -> client frames (ws/frames.rs:29-63)
307 * ------------------------------------------------------------------------ */
308
309export const WS_SERVER_FRAME_TYPES = [
310 'ready',
311 'event',
312 'user_event',
313 'history',
314 'gap',
315 'resync_required',
316 'error',
317 'pong',
318 'bye',
319] as const
320
321/** `resync_required.reason`; the backend uses a plain `String` (frames.rs:52-55). */
322export const WS_RESYNC_REASONS = ['lagged', 'access revoked'] as const
323
324/** `error.message`; plain `String` with four fixed producers (see below). */
325export const WS_ERROR_MESSAGES = [
326 'invalid frame', // ws/mod.rs:256-260 - JSON did not deserialize as ClientFrame
327 'subscription limit reached', // ws/mod.rs:410-419 - max_subscriptions per socket
328 'forbidden', // ws/mod.rs:420-429 - require_room_readable failed
329 'subscribe failed', // ws/mod.rs:430-442 - redis pubsub subscribe failed
330] as const
331
332/** `bye.message`; sent right before a server-initiated close (ws/mod.rs:577-587). */
333export const WS_BYE_MESSAGES = [
334 'internal error', // ws/mod.rs:183-188 - user channel subscribe failed
335 'idle timeout', // ws/mod.rs:310-315 - silent for heartbeat_secs * 3
336] as const
337
338/**
339 * Literal union kept for autocomplete, widened so a backend that adds a new
340 * reason/message cannot break the client's types.
341 */
342type OpenString<T extends readonly string[]> = T[number] | (string & Record<never, never>)
343
344export type WsResyncReason = OpenString<typeof WS_RESYNC_REASONS>
345export type WsErrorMessage = OpenString<typeof WS_ERROR_MESSAGES>
346export type WsByeMessage = OpenString<typeof WS_BYE_MESSAGES>
347
348/**
349 * The 9 `ServerFrame` variants. `event` is typed `| null` on purpose: the server
350 * holds it as `serde_json::Value` and substitutes `Value::Null` when the redis
351 * payload cannot be parsed (frames.rs:36-41, ws/mod.rs:279-283,299-305).
352 */
353export type ServerFrame =
354 /** First frame after the upgrade; `session_id` is a fresh v7 uuid per socket. */
355 | { type: 'ready'; user_id: string; session_id: string }
356 | { type: 'event'; room_id: string; event: RoomEvent | null }
357 | { type: 'user_event'; event: UserEvent | null }
358 | { type: 'history'; room_id: string; messages: WsMessage[] }
359 /** Stream order broke: messages `from..to` were dropped (ws/mod.rs:366-387). */
360 | { type: 'gap'; room_id: string; from: number; to: number }
361 | { type: 'resync_required'; room_id: string; reason: WsResyncReason }
362 | { type: 'error'; message: WsErrorMessage }
363 | { type: 'pong' }
364 | { type: 'bye'; message: WsByeMessage }
365
366/** Coarse classification of `error.message` (the frame carries no room_id). */
367export type WsServerErrorKind = 'invalid_frame' | 'subscription_limit' | 'forbidden' | 'subscribe_failed' | 'unknown'
368
369export function classifyWsError(message: string): WsServerErrorKind {
370 switch (message) {
371 case 'invalid frame':
372 return 'invalid_frame'
373 case 'subscription limit reached':
374 return 'subscription_limit'
375 case 'forbidden':
376 return 'forbidden'
377 case 'subscribe failed':
378 return 'subscribe_failed'
379 default:
380 return 'unknown'
381 }
382}
383
384/** Errors that can only be produced by a `subscribe` frame (best-effort attribution). */
385export function isSubscribeError(kind: WsServerErrorKind): boolean {
386 return kind === 'forbidden' || kind === 'subscription_limit' || kind === 'subscribe_failed'
387}
388
389/* --------------------------------------------------------------------------
390 * Runtime parsing
391 * ------------------------------------------------------------------------ */
392
393/**
394 * Validate a parsed JSON value as a `ServerFrame`.
395 *
396 * Unknown `type` discriminants return `null` so callers can surface drift
397 * instead of casting it into the union. `event` / `user_event` payloads that are
398 * `null` or lack a string `kind` are normalised to `event: null`, which is the
399 * same state the backend produces for an unparseable redis payload.
400 */
401export function asServerFrame(value: unknown): ServerFrame | null {
402 if (typeof value !== 'object' || value === null) return null
403 const candidate = value as Record<string, unknown>
404 if (typeof candidate.type !== 'string') return null
405 if (!WS_SERVER_FRAME_TYPES.includes(candidate.type as (typeof WS_SERVER_FRAME_TYPES)[number])) {
406 return null
407 }
408 if (candidate.type === 'history' && !Array.isArray(candidate.messages)) return null
409 if (candidate.type === 'event' || candidate.type === 'user_event') {
410 const event = candidate.event
411 if (typeof event !== 'object' || event === null || typeof (event as { kind?: unknown }).kind !== 'string') {
412 return { ...candidate, type: candidate.type, event: null } as unknown as ServerFrame
413 }
414 }
415 return value as ServerFrame
416}
417
418/**
419 * Parse one inbound frame. The backend only ever writes text frames
420 * (`Message::Text`, ws/mod.rs:566-575), so non-string payloads are rejected.
421 */
422export function parseServerFrame(data: unknown): ServerFrame | null {
423 if (typeof data !== 'string') return null
424 let parsed: unknown
425 try {
426 parsed = JSON.parse(data)
427 } catch {
428 return null
429 }
430 return asServerFrame(parsed)
431}
Added view/app/src/services/realtime/react.ts view/app/src/services/realtime/react.ts +109 −0 View view/app/src/services/realtime/react.ts
@@ -0,0 +1,109 @@
1/**
2 * React bindings for the realtime hub. They are deliberately thin: each one is
3 * either a store subscription (`useSyncExternalStore`) or an effect that takes a
4 * room intent / registers a listener for the lifetime of the component.
5 *
6 * Every hook is SSR-safe (`getServerSnapshot` + effect-only subscriptions) so a
7 * chat column can be rendered to string in tests without a WebSocket global.
8 */
9import { useCallback, useEffect, useReducer, useRef, useSyncExternalStore } from 'react'
10import type { WsClientIssue } from './client'
11import { getRealtimeHub } from './hub'
12import type { ServerFrame, UserEvent, WsStatus, WsTypingIndicator } from './protocol'
13
14const IDLE: WsStatus = 'idle'
15
16/** Connection lifecycle of the shared socket. */
17export function useWsStatus(): WsStatus {
18 const hub = getRealtimeHub()
19 const subscribe = useCallback((onChange: () => void) => hub.onStatus(onChange), [hub])
20 const getSnapshot = useCallback(() => hub.connectionStatus, [hub])
21 return useSyncExternalStore(subscribe, getSnapshot, () => IDLE)
22}
23
24/** `ready.user_id`, once the socket has handshaked. */
25export function useWsUserId(): string | undefined {
26 const hub = getRealtimeHub()
27 const subscribe = useCallback((onChange: () => void) => hub.onStatus(onChange), [hub])
28 const getSnapshot = useCallback(() => hub.selfId, [hub])
29 return useSyncExternalStore(subscribe, getSnapshot, () => undefined)
30}
31
32/**
33 * Keep a room joined for as long as the caller is mounted. Reference counted in
34 * the hub, so a room opened in both the transcript and the thread panel is
35 * subscribed once.
36 */
37export function useRoomWatch(roomId: string | null | undefined): WsStatus {
38 const hub = getRealtimeHub()
39 useEffect(() => {
40 if (!roomId) return
41 return hub.watchRoom(roomId)
42 }, [hub, roomId])
43 return useWsStatus()
44}
45
46/** Observe every room/user frame; the handler is kept in a ref (stable id). */
47export function useWsFrames(handler: (frame: ServerFrame) => void): void {
48 const ref = useRef(handler)
49 // The latest handler lives in a ref so the subscription is created once; the
50 // write happens after paint, never during it.
51 useEffect(() => {
52 ref.current = handler
53 })
54 const hub = getRealtimeHub()
55 useEffect(() => hub.onFrame((frame) => ref.current(frame)), [hub])
56}
57
58/** Observe protocol conditions (gap, resync, bye, connect failures). */
59export function useWsIssues(handler: (issue: WsClientIssue) => void): void {
60 const ref = useRef(handler)
61 useEffect(() => {
62 ref.current = handler
63 })
64 const hub = getRealtimeHub()
65 useEffect(() => hub.onIssue((issue) => ref.current(issue)), [hub])
66}
67
68/** Observe per-user pushes: mention / presence / read_marker / room_added. */
69export function useWsUserEvents(handler: (event: UserEvent) => void): void {
70 const ref = useRef(handler)
71 useEffect(() => {
72 ref.current = handler
73 })
74 const hub = getRealtimeHub()
75 useEffect(() => hub.onUserEvent((event) => ref.current(event)), [hub])
76}
77
78/**
79 * Live typing indicators for one room/thread. The registry expires each indicator
80 * after its own `ttl_secs`, and it notifies on expiry; the 1s sweep is a safety
81 * net for the case where a broadcast was the only event in a quiet room.
82 */
83export function useTypingIndicators(roomId: string | null | undefined, threadId?: string | null): WsTypingIndicator[] {
84 const hub = getRealtimeHub()
85 // The indicator set is derived from the registry on every render: the listener
86 // only says that something moved, so nothing keeps a second copy of that state.
87 const [, bump] = useReducer((count: number) => count + 1, 0)
88 useEffect(() => {
89 if (!roomId) return
90 const off = hub.onTypingChange((changed) => {
91 if (changed === roomId) bump()
92 })
93 // Indicators expire after their own ttl_secs and the registry notifies on the
94 // sweep; this tick covers a room that went quiet mid-indicator.
95 const timer = setInterval(bump, 1_000)
96 return () => {
97 off()
98 clearInterval(timer)
99 }
100 }, [hub, roomId])
101 // Read straight from the registry: `bump` is what makes a render happen, so a
102 // cached copy here would only be a second source of truth.
103 return roomId ? hub.typingUsers(roomId, threadId ?? null) : []
104}
105
106/** Tell the room you are typing; the client throttles to the server's 3s window. */
107export function sendTypingSignal(roomId: string, threadId?: string | null): void {
108 getRealtimeHub().sendTyping(roomId, threadId)
109}
Added view/app/src/services/realtime/topic.ts view/app/src/services/realtime/topic.ts +62 −0 View view/app/src/services/realtime/topic.ts
@@ -0,0 +1,62 @@
1import { useEffect, useRef } from 'react'
2import { useRealtime } from './context'
3import type { KenEventEnvelope } from './kenEvent'
4
5export interface ResourceEventsSpec {
6 topic: string
7 tags: readonly string[]
8 filter?: Record<string, string>
9 /** 资源 id 缺失时不要订阅（例如详情尚未加载）。 */
10 enabled?: boolean
11 /** 事件到达后执行；默认按 150ms 合并同批事件。 */
12 onEvent: (envelope: KenEventEnvelope) => void
13 debounceMs?: number
14}
15
16/**
17 * 订阅某个资源的若干 tag，事件到达后回调。引用计数与重放由 realtime 层负责，
18 * 组件卸载即退订。回调只在合并窗口内触发一次，避免一次批量写库打爆请求。
19 */
20export function useResourceEvents({
21 topic,
22 tags,
23 filter,
24 enabled = true,
25 onEvent,
26 debounceMs = 150,
27}: ResourceEventsSpec): void {
28 const realtime = useRealtime()
29 const handler = useRef(onEvent)
30 useEffect(() => {
31 handler.current = onEvent
32 })
33 const signature = `${topic}|${tags.join(',')}|${JSON.stringify(filter ?? {})}`
34 const latest = useRef({ tags, filter })
35 latest.current = { tags, filter }
36
37 useEffect(() => {
38 if (!enabled || !realtime.enabled) return
39 const { tags, filter } = latest.current
40 let timer: number | undefined
41 let queued: KenEventEnvelope | null = null
42 const flush = () => {
43 timer = undefined
44 if (!queued) return
45 const envelope = queued
46 queued = null
47 handler.current(envelope)
48 }
49 const releases = tags.map((tag) => realtime.subscribe({ topic, tag, filter, key: `${topic}:${tag}:${signature}` }))
50 const off = realtime.onEnvelope((envelope) => {
51 if (envelope.topic !== topic) return
52 if (!tags.includes(envelope.tag)) return
53 queued = envelope
54 if (timer === undefined) timer = window.setTimeout(flush, debounceMs)
55 })
56 return () => {
57 off()
58 for (const release of releases) release()
59 if (timer !== undefined) window.clearTimeout(timer)
60 }
61 }, [realtime, signature, enabled, debounceMs, topic])
62}
Added view/app/src/services/realtime/typing.ts view/app/src/services/realtime/typing.ts +156 −0 View view/app/src/services/realtime/typing.ts
@@ -0,0 +1,156 @@
1import { type RoomEvent, WS_SERVER_DEFAULTS, type WsTypingIndicator } from './protocol'
2
3/**
4 * TTL bookkeeping for `event{kind:'typing'}` broadcasts.
5 *
6 * The backend is stateless about typing: it re-broadcasts whatever a socket
7 * sends, throttled to one frame per 3s per socket, and stamps the event with
8 * `ttl_secs = max(1, chat.typing_ttl_secs)` (gitdataai/app/im/src/socket.rs,
9 * share/realtime.rs:53-58). There is no "stopped typing" frame, so the *only*
10 * correct way to drop an indicator is to honour `ttl_secs` client-side.
11 */
12
13/** `event{kind:'typing'}` narrowed to the variant the registry consumes. */
14export type WsTypingEvent = Extract<RoomEvent, { kind: 'typing' }>
15
16type WsTypingListener = (room_id: string) => void
17
18export class TypingRegistry {
19 /** room_id -> user_id -> indicator (one slot per user: refreshes overwrite). */
20 private readonly rooms = new Map<string, Map<string, WsTypingIndicator>>()
21 private readonly listeners = new Set<WsTypingListener>()
22 private sweepTimer: ReturnType<typeof setTimeout> | undefined
23
24 /** Record a typing broadcast. Re-broadcasts refresh the expiry, as on the wire. */
25 record(event: WsTypingEvent, now: number = Date.now()): void {
26 // Defence against a payload without a usable ttl_secs: an unbounded
27 // indicator would hang around forever, so fall back to the server default.
28 const raw = event.ttl_secs
29 const ttlMs =
30 typeof raw === 'number' && Number.isFinite(raw) && raw > 0
31 ? Math.max(raw, 1) * 1000
32 : WS_SERVER_DEFAULTS.typingTtlSecs * 1000
33 let byUser = this.rooms.get(event.room_id)
34 if (!byUser) {
35 byUser = new Map()
36 this.rooms.set(event.room_id, byUser)
37 }
38 byUser.set(event.user_id, {
39 room_id: event.room_id,
40 user_id: event.user_id,
41 thread_id: event.thread_id ?? null,
42 expires_at: now + ttlMs,
43 updated_at: now,
44 })
45 this.scheduleSweep(now)
46 this.notify(event.room_id)
47 }
48
49 /**
50 * Live indicators for a room. This is a pure read (no pruning, no listener
51 * callback) so it is safe as a `useSyncExternalStore` snapshot: expiry is
52 * evaluated against `now`, and the sweep timer does the bookkeeping.
53 * `excludeUserId` hides your own echo, because the room channel also reaches
54 * the socket that sent the typing frame.
55 */
56 active(
57 roomId: string,
58 options: { excludeUserId?: string; threadId?: string | null } = {},
59 now: number = Date.now(),
60 ): WsTypingIndicator[] {
61 const byUser = this.rooms.get(roomId)
62 if (!byUser) return []
63 const threadId = options.threadId
64 return [...byUser.values()]
65 .filter((indicator) => indicator.expires_at > now)
66 .filter((indicator) => indicator.user_id !== options.excludeUserId)
67 .filter((indicator) => (threadId === undefined ? true : (indicator.thread_id ?? null) === (threadId ?? null)))
68 .sort((a, b) => a.updated_at - b.updated_at)
69 }
70
71 /** Rooms that currently have at least one live indicator. */
72 activeRoomIds(now: number = Date.now()): string[] {
73 const roomIds: string[] = []
74 for (const [roomId, byUser] of this.rooms) {
75 for (const indicator of byUser.values()) {
76 if (indicator.expires_at > now) {
77 roomIds.push(roomId)
78 break
79 }
80 }
81 }
82 return roomIds
83 }
84
85 /** Drop indicators, e.g. when a room intent is removed or the socket dies. */
86 clear(roomId?: string): void {
87 const changed = roomId === undefined ? [...this.rooms.keys()] : this.rooms.has(roomId) ? [roomId] : []
88 if (roomId === undefined) this.rooms.clear()
89 else this.rooms.delete(roomId)
90 if (this.rooms.size === 0) this.clearSweepTimer()
91 for (const id of changed) this.notify(id)
92 }
93
94 /** Fires with the room id whenever its live indicator set changes. */
95 onTyping(listener: WsTypingListener): () => void {
96 this.listeners.add(listener)
97 return () => this.listeners.delete(listener)
98 }
99
100 dispose(): void {
101 this.clearSweepTimer()
102 this.rooms.clear()
103 this.listeners.clear()
104 }
105
106 private prune(roomId: string, now: number): Map<string, WsTypingIndicator> | undefined {
107 const byUser = this.rooms.get(roomId)
108 if (!byUser) return undefined
109 let removed = false
110 for (const [userId, indicator] of byUser) {
111 if (indicator.expires_at <= now) {
112 byUser.delete(userId)
113 removed = true
114 }
115 }
116 if (byUser.size === 0) {
117 this.rooms.delete(roomId)
118 if (this.rooms.size === 0) this.clearSweepTimer()
119 removed = true
120 }
121 if (removed) this.notify(roomId)
122 return this.rooms.get(roomId)
123 }
124
125 private scheduleSweep(now: number): void {
126 let earliest = Number.POSITIVE_INFINITY
127 for (const byUser of this.rooms.values()) {
128 for (const indicator of byUser.values()) {
129 if (indicator.expires_at < earliest) earliest = indicator.expires_at
130 }
131 }
132 if (!Number.isFinite(earliest)) {
133 this.clearSweepTimer()
134 return
135 }
136 this.clearSweepTimer()
137 this.sweepTimer = setTimeout(
138 () => {
139 this.sweepTimer = undefined
140 for (const roomId of [...this.rooms.keys()]) this.prune(roomId, Date.now())
141 },
142 Math.max(earliest - now, 0),
143 )
144 }
145
146 private clearSweepTimer(): void {
147 if (this.sweepTimer !== undefined) {
148 clearTimeout(this.sweepTimer)
149 this.sweepTimer = undefined
150 }
151 }
152
153 private notify(roomId: string): void {
154 for (const listener of [...this.listeners]) listener(roomId)
155 }
156}