main
import { createContext, type ReactNode, useCallback, useContext, useEffect, useMemo, useRef, useState } from 'react'
import { type KenEventEnvelope, KenEventSocket, prop } from './kenEvent'
export type RealtimeStatus = 'disabled' | 'connecting' | 'online' | 'offline' | 'auth_required' | 'failed'
export interface TopicSubscription {
topic: string
tag: string
filter?: Record<string, string>
key?: string
}
export interface RealtimeApi {
readonly enabled: boolean
readonly status: RealtimeStatus
readonly selfId?: number
subscribe(spec: TopicSubscription): () => void
onEnvelope(handler: (envelope: KenEventEnvelope) => void): () => void
retry(): void
lastError?: string
}
const DISABLED: RealtimeApi = {
enabled: false,
status: 'disabled',
subscribe: () => () => {},
onEnvelope: () => () => {},
retry: () => {},
}
const RealtimeContext = createContext<RealtimeApi | null>(null)
function endpoint(): string | null {
const configured = import.meta.env.VITE_EVENT_URL as string | undefined
if (configured?.trim()) return configured.trim()
return null
}
export function RealtimeProvider({ children }: { children: ReactNode }) {
const url = endpoint()
const socketRef = useRef<KenEventSocket | null>(null)
const listeners = useRef(new Set<(envelope: KenEventEnvelope) => void>())
const pending = useRef(new Map<string, () => void>())
const [status, setStatus] = useState<RealtimeStatus>(url ? 'connecting' : 'disabled')
const [selfId, setSelfId] = useState<number | undefined>(undefined)
const [lastError, setLastError] = useState<string | undefined>(undefined)
useEffect(() => {
if (!url) return
const socket = new KenEventSocket({
url,
onOpen: () => {
setStatus('online')
setLastError(undefined)
},
onClose: (_reason, wasOpen) => {
setStatus(wasOpen ? 'offline' : 'failed')
},
onAuthRequired: (message) => {
setStatus('auth_required')
setLastError(message)
},
onConnectError: (message) => {
setStatus('failed')
setLastError(message)
},
onEnvelope: (envelope) => {
const recipient = envelope.recipient_user_id
if (typeof recipient === 'number') setSelfId((current) => current ?? recipient)
for (const listener of [...listeners.current]) listener(envelope)
},
onResync: (envelope) => {
for (const listener of [...listeners.current]) listener({ ...envelope, resync_required: true })
},
onServerError: (message) => setLastError(message),
onSubscribeError: (subscription, message) => setLastError(`${subscription}: ${message}`),
})
socketRef.current = socket
socket.subscribePersonal()
return () => {
socket.dispose()
socketRef.current = null
pending.current.clear()
}
}, [url])
const subscribe = useCallback((spec: TopicSubscription) => {
const socket = socketRef.current
if (!socket) return () => {}
const key = spec.key ?? `${spec.topic}|${spec.tag}|${JSON.stringify(spec.filter ?? {})}`
const existing = pending.current.get(key)
if (existing) return existing
const off = socket.subscribeTopic({ key, topic: spec.topic, tag: spec.tag, filter: spec.filter })
const release = () => {
pending.current.delete(key)
off()
}
pending.current.set(key, release)
return release
}, [])
const onEnvelope = useCallback((handler: (envelope: KenEventEnvelope) => void) => {
listeners.current.add(handler)
return () => {
listeners.current.delete(handler)
}
}, [])
const retry = useCallback(() => {
socketRef.current?.stop()
socketRef.current?.start()
}, [])
const value = useMemo<RealtimeApi>(
() => ({ enabled: Boolean(url), status, selfId, subscribe, onEnvelope, retry, lastError }),
[url, status, selfId, subscribe, onEnvelope, retry, lastError],
)
return <RealtimeContext.Provider value={value}>{children}</RealtimeContext.Provider>
}
export function useRealtime(): RealtimeApi {
return useContext(RealtimeContext) ?? DISABLED
}
export function useRealtimeEnvelope(handler: (envelope: KenEventEnvelope) => void): void {
const realtime = useRealtime()
const ref = useRef(handler)
useEffect(() => {
ref.current = handler
})
useEffect(() => realtime.onEnvelope((envelope) => ref.current(envelope)), [realtime])
}
/** 读取信封里的资源属性（RocketMQ 用户属性都是字符串）。 */
export const envelopeProperty = prop