e33d6e5
import { useEffect, useRef } from 'react'
import { useRealtime } from './context'
import type { KenEventEnvelope } from './kenEvent'
export interface ResourceEventsSpec {
topic: string
tags: readonly string[]
filter?: Record<string, string>
/** 资源 id 缺失时不要订阅（例如详情尚未加载）。 */
enabled?: boolean
/** 事件到达后执行；默认按 150ms 合并同批事件。 */
onEvent: (envelope: KenEventEnvelope) => void
debounceMs?: number
}
/**
* 订阅某个资源的若干 tag，事件到达后回调。引用计数与重放由 realtime 层负责，
* 组件卸载即退订。回调只在合并窗口内触发一次，避免一次批量写库打爆请求。
*/
export function useResourceEvents({
topic,
tags,
filter,
enabled = true,
onEvent,
debounceMs = 150,
}: ResourceEventsSpec): void {
const realtime = useRealtime()
const handler = useRef(onEvent)
useEffect(() => {
handler.current = onEvent
})
const signature = `${topic}|${tags.join(',')}|${JSON.stringify(filter ?? {})}`
const latest = useRef({ tags, filter })
latest.current = { tags, filter }
useEffect(() => {
if (!enabled || !realtime.enabled) return
const { tags, filter } = latest.current
let timer: number | undefined
let queued: KenEventEnvelope | null = null
const flush = () => {
timer = undefined
if (!queued) return
const envelope = queued
queued = null
handler.current(envelope)
}
const releases = tags.map((tag) => realtime.subscribe({ topic, tag, filter, key: `${topic}:${tag}:${signature}` }))
const off = realtime.onEnvelope((envelope) => {
if (envelope.topic !== topic) return
if (!tags.includes(envelope.tag)) return
queued = envelope
if (timer === undefined) timer = window.setTimeout(flush, debounceMs)
})
return () => {
off()
for (const release of releases) release()
if (timer !== undefined) window.clearTimeout(timer)
}
}, [realtime, signature, enabled, debounceMs, topic])
}