main
/**
* KenEvent transport: the Socket.IO wire contract of the realtime gateway
* (`KenEvent/src/socket`, Engine.IO v4, path `/socket.io`).
*
* 与旧 im 网关（`GET /ws/v1`，裸 WebSocket + `gitdata.jwt.*` 子协议）的区别：
* - **连接**：Socket.IO，默认路径 `/socket.io`；浏览器握手带不了自定义头，
* 所以 token 由会话 cookie（`gitdata_session`）或 CONNECT 包的 `auth.token`
* 提供，KenEvent 侧见 `socket/response.rs` 与 `socket/session/state.rs`。
* - **订阅**：客户端发 `subscribe {topic, tag, filter, id}`，服务端按
* RocketMQ 的 topic/tag 精确匹配（`events/subscriptions.rs:validate`），
* 没有通配，所以一个房间要按关心的 tag 各订阅一条。
* - **推送**：服务端只发两种事件名——`event` 与 `resync`，载荷是
* `{topic, tag, user_id, recipient_user_id, properties, body}` 信封
* （`events/action/common.rs:payload`）。**详情一律走 REST 回拉**。
* - **个人事件**：`gitdata-notify`/`notification-create` 与 chat 的
* mention/read_state/presence 等由服务端强制绑定 `filter.user_id` 自己。
*/
import { io, type Socket } from 'socket.io-client'
export const KEN_EVENT_PATH = '/socket.io'
export const CHAT_TOPIC = 'gitdata-chat'
export const NOTIFY_TOPIC = 'gitdata-notify'
/** 房间级事件：一条订阅一个 tag，`filter` 固定为 `{room_id}`。 */
export const ROOM_TAGS = [
'message-create',
'message-update',
'message-delete',
'messages-delete',
'pin-create',
'pin-delete',
'reaction-update',
'room-update',
'member-create',
'member-delete',
'thread-update',
'voice-presence',
'voice-hand',
'voice-layout',
'voice-moderate',
] as const
/** 用户级事件：不带 room 过滤，服务端会补上 `filter.user_id = 自己`。 */
export const PERSONAL_CHAT_TAGS = [
'mention-create',
'reminder-create',
'membership-create',
'read_state-update',
'presence-update',
] as const
export const PERSONAL_NOTIFY_TAGS = ['notification-create'] as const
/** 服务端推送的信封（`events/action/common.rs:80`）。 */
export interface KenEventEnvelope {
topic: string
tag: string
user_id?: number | null
recipient_user_id?: number
properties?: Record<string, unknown> | null
body?: unknown
/** 断流补齐时 KenEvent 会把事件名换成 `resync` 并置这个标记。 */
resync_required?: boolean
}
export interface SubscriptionAck {
ok?: boolean
data?: unknown
error?: string
}
export interface KenEventSocketOptions {
/** 默认同源（`window.location.origin`）；vite dev 与网关都由 `/socket.io` 反代。 */
url?: string
onOpen: (info: { sid: string }) => void
onClose: (reason: string, wasOpen: boolean) => void
onAuthRequired: (message: string) => void
onConnectError: (message: string) => void
onEnvelope: (envelope: KenEventEnvelope) => void
onResync: (envelope: KenEventEnvelope) => void
onServerError: (message: string) => void
onSubscribeError: (subscription: string, message: string) => void
}
const DEVICE_KEY = 'gitdata.ws.device_id'
/** 设备标识：KenEvent 用它做多端会话隔离（`session/state.rs`）。 */
export function deviceId(): string {
try {
const stored = window.localStorage.getItem(DEVICE_KEY)
if (stored) return stored
const created = crypto.randomUUID()
window.localStorage.setItem(DEVICE_KEY, created)
return created
} catch {
// 隐私模式 / SSR：退化成会话级 id，重连后仍能复用（同一模块生命周期内）。
return crypto.randomUUID()
}
}
function asRecord(value: unknown): Record<string, unknown> | null {
return typeof value === 'object' && value !== null ? (value as Record<string, unknown>) : null
}
export function asEnvelope(value: unknown): KenEventEnvelope | null {
const record = asRecord(value)
if (!record) return null
const topic = record.topic
const tag = record.tag
if (typeof topic !== 'string' || typeof tag !== 'string') return null
return record as unknown as KenEventEnvelope
}
/** `properties` 里的字符串字段（RocketMQ 用户属性全是字符串）。 */
export function prop(envelope: KenEventEnvelope, key: string): string | null {
const value = envelope.properties?.[key]
if (typeof value === 'string' && value.length > 0) return value
const fromBody = asRecord(envelope.body)?.[key]
return typeof fromBody === 'string' && fromBody.length > 0 ? fromBody : null
}
/**
* 订阅管理器：把「打开了哪些房间」这份意图翻译成 KenEvent 的逐 tag 订阅。
*
* 重连后服务端的订阅是空的（per-socket 状态），所以每次 `connect` 都要重放；
* 房间的 refcount 由上层 hub 维护，这里只负责 idempotent 增删。
*/
export class KenEventSocket {
private socket: Socket | null = null
/** roomId -> tag -> 订阅 id（KenEvent 要求 id 是 UUID，由客户端生成）。 */
private readonly rooms = new Map<string, Map<string, string>>()
/** `${topic}|${tag}` -> 个人订阅 id。 */
private readonly personalIds = new Map<string, string>()
/** 通用资源订阅：key -> 订阅意图（issue / pull / org / kenshell 等）。 */
private readonly topics = new Map<
string,
{ topic: string; tag: string; filter: Record<string, string>; refs: number; id: string }
>()
private personal = false
private disposed = false
private readonly options: KenEventSocketOptions
constructor(options: KenEventSocketOptions) {
this.options = options
}
get connected(): boolean {
return this.socket?.connected ?? false
}
/** 幂等：连接中/已连接时什么也不做。 */
start(): void {
if (this.disposed || this.socket) return
const socket = io(this.options.url, {
path: KEN_EVENT_PATH,
// 先 polling 再升级 websocket：网关（KenGateWay）不代理 Upgrade，会回 501，
// 从 websocket 起步会直接连不上；polling 起步在两种部署下都能连上，
// 拿到连接后 socket.io 会自己尝试升级到 WebSocket（nginx 环境即可升级成功）。
transports: ['polling', 'websocket'],
// 兜底：某个传输被拒时继续尝试下一个，而不是直接失败。
tryAllTransports: true,
withCredentials: true,
auth: (cb) => cb({ device_id: deviceId() }),
reconnection: true,
reconnectionDelay: 500,
reconnectionDelayMax: 30_000,
timeout: 20_000,
})
this.socket = socket
socket.on('connect', () => {
this.options.onOpen({ sid: socket.id ?? '' })
this.replay()
})
socket.on('disconnect', (reason) => {
this.options.onClose(reason, true)
})
socket.on('connect_error', (error) => {
const message = error instanceof Error ? error.message : String(error)
// KenEvent 在鉴权失败时发 `connect_error('authentication failed')`
// （`socket/session/state.rs:connect`）。
if (/auth|unauthor|missing access token/i.test(message)) {
this.options.onAuthRequired(message)
return
}
this.options.onConnectError(message)
})
socket.on('event', (payload: unknown) => {
const envelope = asEnvelope(payload)
if (!envelope) {
this.options.onServerError('unrecognized event envelope')
return
}
if (envelope.resync_required) {
this.options.onResync(envelope)
return
}
this.options.onEnvelope(envelope)
})
socket.on('resync', (payload: unknown) => {
const envelope = asEnvelope(payload)
if (envelope) this.options.onResync(envelope)
})
socket.on('error', (payload: unknown) => {
const message = (asRecord(payload)?.message as string | undefined) ?? 'realtime gateway error'
this.options.onServerError(message)
})
}
stop(): void {
const socket = this.socket
this.socket = null
if (!socket) return
socket.removeAllListeners()
socket.disconnect()
}
dispose(): void {
this.disposed = true
this.rooms.clear()
this.personal = false
this.stop()
}
/** 加入一个房间：按 `ROOM_TAGS` 逐条订阅。 */
subscribeRoom(roomId: string): void {
if (this.rooms.has(roomId)) return
this.rooms.set(roomId, new Map())
this.start()
if (this.connected) this.emitRoom(roomId)
}
unsubscribeRoom(roomId: string): void {
const ids = this.rooms.get(roomId)
if (!ids) return
this.rooms.delete(roomId)
if (!this.connected) return
for (const [tag, id] of ids) this.emitUnsubscribe(CHAT_TOPIC, tag, id)
}
/**
* 订阅任意 topic/tag（带资源过滤）。同一 `key` 只发一次订阅，引用计数归零时退订；
* 返回的函数可安全重复调用。
*/
subscribeTopic(spec: { key: string; topic: string; tag: string; filter?: Record<string, string> }): () => void {
const existing = this.topics.get(spec.key)
if (existing) {
existing.refs += 1
return () => this.releaseTopic(spec.key)
}
const entry = {
topic: spec.topic,
tag: spec.tag,
filter: spec.filter ?? {},
refs: 1,
id: crypto.randomUUID(),
}
this.topics.set(spec.key, entry)
this.start()
if (this.connected) this.emit(entry.topic, entry.tag, { id: entry.id, filter: entry.filter })
return () => this.releaseTopic(spec.key)
}
private releaseTopic(key: string): void {
const entry = this.topics.get(key)
if (!entry) return
entry.refs -= 1
if (entry.refs > 0) return
this.topics.delete(key)
if (this.connected) this.emitUnsubscribe(entry.topic, entry.tag, entry.id)
}
/** 个人通道（通知、提及、已读游标）：一次连接订阅一次。 */
subscribePersonal(): void {
if (this.personal) return
this.personal = true
this.start()
if (this.connected) this.emitPersonal()
}
/** 重放全部订阅意图；每次 `connect` 都会调用。 */
private replay(): void {
this.personalIds.clear()
for (const ids of this.rooms.values()) ids.clear()
if (this.personal) this.emitPersonal()
for (const roomId of this.rooms.keys()) this.emitRoom(roomId)
for (const entry of this.topics.values()) {
entry.id = crypto.randomUUID()
this.emit(entry.topic, entry.tag, { id: entry.id, filter: entry.filter })
}
}
private emitRoom(roomId: string): void {
const ids = this.rooms.get(roomId)
if (!ids) return
for (const tag of ROOM_TAGS) {
if (ids.has(tag)) continue
const id = crypto.randomUUID()
ids.set(tag, id)
this.emit(CHAT_TOPIC, tag, { id, filter: { room_id: roomId } })
}
}
private emitPersonal(): void {
for (const tag of PERSONAL_CHAT_TAGS) this.emitPersonalOne(CHAT_TOPIC, tag)
for (const tag of PERSONAL_NOTIFY_TAGS) this.emitPersonalOne(NOTIFY_TOPIC, tag)
}
private emitPersonalOne(topic: string, tag: string): void {
const key = `${topic}|${tag}`
if (this.personalIds.has(key)) return
const id = crypto.randomUUID()
this.personalIds.set(key, id)
this.emit(topic, tag, { id })
}
private emit(topic: string, tag: string, extra: { id: string; filter?: unknown }): void {
const socket = this.socket
if (!socket) return
const payload = { topic, tag, ...extra }
// ack 里的 `{ok:false,error}` 是唯一能归因到具体订阅的失败信号。
socket.timeout(10_000).emit('subscribe', payload, (error: Error | null, ack?: SubscriptionAck) => {
if (error) {
this.options.onSubscribeError(`${topic}/${tag}`, 'subscribe ack timeout')
return
}
if (ack && ack.ok === false) {
this.options.onSubscribeError(`${topic}/${tag}`, ack.error ?? 'subscribe rejected')
}
})
}
private emitUnsubscribe(topic: string, tag: string, id: string): void {
this.socket?.emit('unsubscribe', { topic, tag, id })
}
}