main
import type { ChatMessage } from '@app/domains/chat/model'
import { useToast } from '@app/providers/toast-provider'
import { useRealtime } from '@app/services/realtime/context'
import { asEnvelope, CHAT_TOPIC, type KenEventEnvelope } from '@app/services/realtime/kenEvent'
import { createContext, type ReactNode, useCallback, useContext, useEffect, useMemo, useRef, useState } from 'react'
import { useNavigate } from 'react-router-dom'
export interface RoomMeta {
name: string
href: string
}
interface RoomSnapshot {
messages: ChatMessage[]
updatedAt: number
}
interface MessagePipelineApi {
snapshot(roomId: string): RoomSnapshot | undefined
watch(roomId: string): () => void
activate(roomId: string | null): void
setRoomMeta(meta: Record<string, RoomMeta>): void
}
const MessagePipelineContext = createContext<MessagePipelineApi | null>(null)
const MAX_PER_ROOM = 200
const MESSAGE_TAGS = ['message-create', 'message-update', 'message-delete'] as const
function messageOf(envelope: KenEventEnvelope): ChatMessage | null {
const body = envelope.body as { message?: ChatMessage } | null
if (!body || typeof body !== 'object') return null
const message = body.message
if (!message || typeof message.id !== 'string') return null
return message
}
function merge(list: ChatMessage[], incoming: ChatMessage): ChatMessage[] {
const index = list.findIndex((item) => item.id === incoming.id)
const next = index >= 0 ? [...list.slice(0, index), incoming, ...list.slice(index + 1)] : [...list, incoming]
next.sort((a, b) => (a.seq ?? 0) - (b.seq ?? 0))
return next.length > MAX_PER_ROOM ? next.slice(next.length - MAX_PER_ROOM) : next
}
export function MessagePipelineProvider({ children }: { children: ReactNode }) {
const realtime = useRealtime()
const toast = useToast()
const navigate = useNavigate()
const rooms = useRef(new Map<string, RoomSnapshot>())
const watchRefs = useRef(new Map<string, () => void>())
const meta = useRef(new Map<string, RoomMeta>())
const activeRoom = useRef<string | null>(null)
const selfId = useRef<number | undefined>(undefined)
const [, bump] = useState(0)
selfId.current = realtime.selfId
const activate = useCallback((roomId: string | null) => {
activeRoom.current = roomId
}, [])
const setRoomMeta = useCallback((next: Record<string, RoomMeta>) => {
meta.current = new Map(Object.entries(next))
}, [])
const watch = useCallback(
(roomId: string) => {
const existing = watchRefs.current.get(roomId)
if (existing) return existing
const releases = MESSAGE_TAGS.map((tag) =>
realtime.subscribe({ topic: CHAT_TOPIC, tag, filter: { room_id: roomId }, key: `chat:${tag}:${roomId}` }),
)
const release = () => {
for (const off of releases) off()
watchRefs.current.delete(roomId)
}
watchRefs.current.set(roomId, release)
return release
},
[realtime],
)
useEffect(() => {
return realtime.onEnvelope((envelope) => {
if (envelope.topic !== CHAT_TOPIC) return
if (!MESSAGE_TAGS.includes(envelope.tag as (typeof MESSAGE_TAGS)[number])) return
const roomId = (envelope.properties as { room_id?: string } | undefined)?.room_id
if (typeof roomId !== 'string' || !roomId) return
const message = messageOf(envelope)
if (!message) return
const current = rooms.current.get(roomId) ?? { messages: [], updatedAt: 0 }
if (envelope.tag === 'message-delete') {
rooms.current.set(roomId, {
messages: current.messages.filter((item) => item.id !== message.id),
updatedAt: Date.now(),
})
} else {
rooms.current.set(roomId, { messages: merge(current.messages, message), updatedAt: Date.now() })
}
bump((value) => value + 1)
if (envelope.tag !== 'message-create') return
if (activeRoom.current === roomId) return
if (selfId.current !== undefined && message.author_id === selfId.current) return
const info = meta.current.get(roomId)
const author = message.author?.display_name || message.author?.username || ''
toast.push({
title: info ? `${info.name}${author ? ` · ${author}` : ''}` : '新消息',
description: message.content?.slice(0, 120),
action: info
? { label: '查看', onClick: () => navigate(info.href) }
: { label: '查看', onClick: () => navigate(`/app/@me/chat/${roomId}`) },
})
})
}, [realtime, toast, navigate])
const api = useMemo<MessagePipelineApi>(
() => ({ snapshot: (roomId) => rooms.current.get(roomId), watch, activate, setRoomMeta }),
[watch, activate, setRoomMeta],
)
return <MessagePipelineContext.Provider value={api}>{children}</MessagePipelineContext.Provider>
}
export function useMessagePipeline(): MessagePipelineApi {
const api = useContext(MessagePipelineContext)
if (!api) throw new Error('useMessagePipeline 必须在 MessagePipelineProvider 内使用')
return api
}
/** 订阅一个房间的消息流（引用计数），返回该房间的实时消息窗口。 */
export function useRoomMessages(roomId: string | null | undefined): ChatMessage[] {
const api = useMessagePipeline()
const [, bump] = useState(0)
useEffect(() => {
if (!roomId) return
const release = api.watch(roomId)
return () => {
release()
bump((value) => value + 1)
}
}, [api, roomId])
return roomId ? (api.snapshot(roomId)?.messages ?? []) : []
}
/** 标记“当前正在看这个房间”，该房间的新消息不再弹 toast。 */
export function useActiveRoom(roomId: string | null | undefined): void {
const api = useMessagePipeline()
useEffect(() => {
api.activate(roomId ?? null)
return () => api.activate(null)
}, [api, roomId])
}
/** 注册房间名与跳转地址，供“不在当前房间”的 toast 使用。 */
export function useRoomMeta(meta: Record<string, RoomMeta>): void {
const api = useMessagePipeline()
const serialized = JSON.stringify(meta)
useEffect(() => {
api.setRoomMeta(JSON.parse(serialized) as Record<string, RoomMeta>)
}, [api, serialized])
}
/** 订阅房间但不挂载消息视图（侧栏未读、后台提醒）。 */
export function useWatchRoom(roomId: string | null | undefined): void {
const api = useMessagePipeline()
useEffect(() => {
if (!roomId) return
return api.watch(roomId)
}, [api, roomId])
}
export { asEnvelope }