/** * The chat room's data engine: one hook that owns the transcript of a scope (the * room's main stream, or one thread inside it) and keeps it correct across HTTP * pagination, WebSocket live traffic, reconnects and the local user's own * optimistic posts. * * Division of labour, and why: * - **HTTP** seeds the window and pages it (`before_seq` / `after_seq`), because * the socket only ever walks *forward* from a cursor. * - **The socket** (`@/shared/ws`) is the live source: `subscribe` answers with a * `history` snapshot that is merged in, then `event` frames stream updates. The * client resolves `gap` and `resync_required` itself; the cases it cannot * (lost cursor, truncated walk) fall back to an HTTP re-seed here. * - **The reducer** (`./transcript`) is the only writer of list state, so dedupe, * ordering and the viewer's own reaction flags stay in exactly one place. * * Sends use the backend's `nonce` idempotency key: retrying after a timeout * returns the original message instead of posting a duplicate. */ import { type Attachment, type ChatMessage, clientNonce, deleteMessage as deleteMessageApi, describeFailure, editMessage as editMessageApi, failureReason, findRetryAfterSeconds, getMessage, getReadState, listMessages, MESSAGE_PAGE_SIZE, type MessagePage, markRead as markReadApi, type PendingMessage, sendMessage, setMessagePin, setReaction, } from '@app/domains/chat/model' import { t } from '@app/i18n' import { chatSeedKey } from '@app/lib' import { useRoomWatch, useTypingIndicators, useWsFrames, useWsIssues, useWsUserEvents } from '@app/services/realtime' import { readCache, writeCache } from '@app/services/storage' import { useCallback, useEffect, useMemo, useReducer, useRef, useState } from 'react' import { belongsToScope, createTranscriptState, findEchoedPending, type TranscriptState, transcriptReducer, } from './transcript' export interface UseRoomStreamOptions { roomId: string /** Set to read one thread instead of the room's main stream. */ threadId?: string | null /** Local user id: identifies your own posts and ignores your typing echo. */ viewerId?: number /** Stops all traffic (no room selected, voice channel, outage gate open …). */ enabled?: boolean /** The unread cursor moved: the shell refreshes its sidebar badges. */ onUnreadChanged?: () => void /** Room's `rate_limit_per_user`, mirrored so the countdown starts *before* the * server has to refuse a second post. */ slowModeSeconds?: number /** Side-channel room events (typing excluded — that is the registry's job). */ onRoomEvent?: (kind: string) => void } export interface SendInput { content: string attachments?: Attachment[] replyToId?: string | null /** Thread the post goes into; the main stream when omitted. */ threadId?: string | null /** Also surface a thread reply in the main stream (`thread_broadcast`). */ threadBroadcast?: boolean } export interface RoomStream { state: TranscriptState /** Connection lifecycle of the shared socket. */ status: ReturnType /** Ids currently typing in this scope (your own echo already filtered out). */ typingUserIds: number[] /** Seq the viewer last read; the「新消息」divider sits right after it. */ lastReadSeq: number /** True while the viewport is pinned to the newest row. */ atBottom: boolean /** Rows that arrived while the viewport was scrolled away from the tail. */ missedCount: number /** Seconds still blocked by the room's slow mode (0 when free to send). */ slowModeRemaining: number /** Transient action failure (edit / delete / react / pin), shown above the composer. */ notice: string | null dismissNotice: () => void listRef: React.RefObject reportAtBottom: (atBottom: boolean) => void scrollToBottom: (options?: { smooth?: boolean }) => void loadOlder: () => void jumpTo: (messageId: string) => void /** Back to the newest page: replaces the window instead of leaving a hole. */ returnToLatest: () => void send: (input: SendInput) => Promise retryPending: (localId: string) => void dismissPending: (localId: string) => void editMessage: (messageId: string, content: string) => Promise deleteMessage: (messageId: string) => Promise toggleReaction: (messageId: string, emoji: string, on: boolean) => Promise togglePin: (messageId: string, pinned: boolean) => Promise refresh: () => void } /** Coalesces the read-marker write: scrolling must not become a request storm. */ const MARK_READ_DEBOUNCE_MS = 800 export function useRoomStream({ roomId, threadId = null, viewerId, enabled = true, onUnreadChanged, slowModeSeconds = 0, onRoomEvent, }: UseRoomStreamOptions): RoomStream { const [state, dispatch] = useReducer(transcriptReducer, { roomId, threadId }, (scope) => createTranscriptState(scope.roomId, scope.threadId), ) const status = useRoomWatch(enabled ? roomId : null) const scopeKey = `${roomId}\u0000${threadId ?? ''}` const listRef = useRef(null) // Scope-tagged state: a room/thread switch needs no reset effect, because a // value stamped with the previous scope simply does not read as anything. const [bottom, setBottom] = useState<{ scope: string; value: boolean; baseline: number }>({ scope: scopeKey, value: true, baseline: 0, }) const [read, setRead] = useState<{ scope: string; seq: number }>({ scope: scopeKey, seq: 0 }) const [slowModeRemaining, setSlowModeRemaining] = useState(0) const [notice, setNotice] = useState(null) const [reloadKey, setReloadKey] = useState(0) const typing = useTypingIndicators(enabled ? roomId : null, threadId) const atBottom = bottom.scope === scopeKey ? bottom.value : true /** Tip captured when the reader left the tail; everything above it is seen. */ const missedBaseline = bottom.scope === scopeKey ? bottom.baseline : 0 const lastReadSeq = read.scope === scopeKey ? read.seq : 0 const readCursor = useRef(0) const atBottomRef = useRef(true) /** Newest row the transcript holds, mirrored for the debounced read marker. */ const tipRef = useRef(0) const markReadTimer = useRef | undefined>(undefined) const inFlight = useRef>(new Set()) const pendingById = useRef>({}) const slowTimer = useRef | undefined>(undefined) /* ------------------------------------------------------------ seeding */ useEffect(() => { if (!enabled || !roomId) return dispatch({ type: 'reset', roomId, threadId }) readCursor.current = 0 tipRef.current = 0 atBottomRef.current = true let active = true // A *successful* network seed outranks a slower disk read; a failed one // must not block it — offline, the stored page is the whole transcript. let seedSucceeded = false dispatch({ type: 'seedStart' }) const seedKey = chatSeedKey(roomId, threadId ?? null) // Local-first prefill: paint the stored tail page while the fetch runs. void readCache(seedKey).then((hit) => { if (!active || seedSucceeded || !hit) return dispatch({ type: 'seedCached', items: hit.data.items, hasMore: hit.data.has_more, oldestSeq: hit.data.oldest_seq ?? null, newestSeq: hit.data.newest_seq ?? null, }) }) listMessages(roomId, { threadId: threadId ?? undefined, limit: MESSAGE_PAGE_SIZE }).then( (page) => { if (!active) return seedSucceeded = true // The tail page is what the next visit prefills from. writeCache(seedKey, page) dispatch({ type: 'seedDone', items: page.items, hasMore: page.has_more, oldestSeq: page.oldest_seq ?? null, newestSeq: page.newest_seq ?? null, }) }, (error: unknown) => { if (active) dispatch({ type: 'seedError', message: describeFailure(error) }) }, ) return () => { active = false } }, [enabled, roomId, threadId, reloadKey]) /** * The unread cursor is room-level state (`thread_id` is not part of it), so a * thread panel never touches it. The response also reports the room tip, which * is what stops the first render from marking a room as read before the reader * has actually seen anything. */ useEffect(() => { if (!enabled || !roomId || threadId) return let active = true getReadState(roomId).then( (current) => { if (!active) return readCursor.current = current.last_read_seq setRead({ scope: scopeKey, seq: current.last_read_seq }) }, () => { // No read state yet means nothing was ever read: the divider will sit at // the very first row, which is the honest rendering of that. }, ) return () => { active = false } }, [enabled, roomId, threadId, reloadKey, scopeKey]) /* -------------------------------------------------------- live traffic */ useWsFrames((frame) => { if (!enabled) return if (!('room_id' in frame) || frame.room_id !== roomId) return if (frame.type === 'history') { const items = frame.messages.filter((message) => belongsToScope(message, threadId)) if (items.length > 0) dispatch({ type: 'liveHistory', items }) return } if (frame.type === 'event') { const event = frame.event if (!event) return switch (event.kind) { case 'message_published': if (belongsToScope(event.message, threadId)) { dispatch({ type: 'published', message: event.message }) } return case 'message_edited': if (belongsToScope(event.message, threadId)) { dispatch({ type: 'edited', message: event.message }) } return case 'message_deleted': dispatch({ type: 'deleted', messageIds: [event.message_id] }) return case 'messages_deleted': dispatch({ type: 'deleted', messageIds: event.message_ids }) return case 'message_pinned': dispatch({ type: 'pinned', messageId: event.message_id, pinned: event.pinned }) return case 'reaction_updated': dispatch({ type: 'reactions', messageId: event.message_id, reactions: event.reactions }) return default: // room_updated / member_* / thread_updated / typing: panels consume // these; the transcript only needs to know the room moved. onRoomEvent?.(event.kind) dispatch({ type: 'liveTick' }) } return } if (frame.type === 'resync_required' && frame.reason === 'access revoked') { dispatch({ type: 'accessRevoked' }) } }) /** Your own read marker, mirrored from another tab. */ useWsUserEvents((event) => { if (event.kind !== 'read_marker' || event.room_id !== roomId || threadId) return const seq = Number.isFinite(event.last_read_seq) ? event.last_read_seq : 0 if (seq <= readCursor.current) return readCursor.current = seq setRead({ scope: scopeKey, seq }) }) /** * Conditions the socket cannot resolve on its own. A gap or a lagged resync it * already backfills; these two mean the local window is no longer trustworthy, * so re-seed it over HTTP instead of showing a transcript with a hole in it. */ useWsIssues((issue) => { if (!enabled) return if ('room_id' in issue && issue.room_id !== undefined && issue.room_id !== roomId) return if (issue.kind === 'server_error' && issue.error === 'forbidden') { dispatch({ type: 'accessRevoked' }) setNotice(t('chat.room.accessDenied')) return } if ( issue.kind === 'cursor_lost' || issue.kind === 'history_truncated' || (issue.kind === 'resync_required' && !issue.dropped) ) { setReloadKey((key) => key + 1) } }) /* --------------------------------------------------------- read marker */ const flushRead = useCallback(() => { if (!enabled || threadId) return const target = atBottomRef.current ? tipRef.current : readCursor.current if (target <= readCursor.current) return markReadApi(roomId, { seq: target }).then( (current) => { if (current.last_read_seq > readCursor.current) { readCursor.current = current.last_read_seq setRead({ scope: scopeKey, seq: current.last_read_seq }) onUnreadChanged?.() } }, () => { // A lost read marker is cosmetic and self-heals on the next scroll. }, ) }, [enabled, onUnreadChanged, roomId, scopeKey, threadId]) const scheduleRead = useCallback(() => { if (markReadTimer.current) clearTimeout(markReadTimer.current) markReadTimer.current = setTimeout(flushRead, MARK_READ_DEBOUNCE_MS) }, [flushRead]) useEffect( () => () => { if (markReadTimer.current) clearTimeout(markReadTimer.current) }, [], ) // The tail is what the reader has seen. Mirroring it into a ref keeps the // debounced marker honest without a second render, and the marker itself is // only ever scheduled while the viewport is pinned there. useEffect(() => { if (state.newestSeq !== null && state.newestSeq > tipRef.current) { tipRef.current = state.newestSeq } if (atBottom && tipRef.current > readCursor.current) scheduleRead() }, [atBottom, scheduleRead, state.newestSeq]) // Leaving the room (navigating to another channel) must not lose what was read // while sitting at the tail, so the pending marker is flushed on unmount. const flushRef = useRef(flushRead) useEffect(() => { flushRef.current = flushRead }) useEffect( () => () => { if (atBottomRef.current) flushRef.current() }, [], ) /* ------------------------------------------------------------ slow mode */ const startSlowMode = useCallback((seconds: number) => { const until = Date.now() + seconds * 1_000 if (slowTimer.current) clearInterval(slowTimer.current) slowTimer.current = setInterval(() => { const remaining = Math.max(0, Math.ceil((until - Date.now()) / 1_000)) setSlowModeRemaining(remaining) if (remaining === 0 && slowTimer.current) { clearInterval(slowTimer.current) slowTimer.current = undefined } }, 1_000) setSlowModeRemaining(seconds) }, []) useEffect( () => () => { if (slowTimer.current) clearInterval(slowTimer.current) }, [], ) /* ------------------------------------------------------------ sending */ const deliver = useCallback( async (item: PendingMessage): Promise => { if (inFlight.current.has(item.localId)) return false inFlight.current.add(item.localId) try { const message = await sendMessage(item.room_id, { content: item.content, nonce: item.nonce, reply_to_id: item.reply_to_id ?? null, thread_id: item.thread_id ?? null, thread_broadcast: item.thread_broadcast ?? false, attachments: item.attachments.map((attachment) => attachment.id), }) delete pendingById.current[item.localId] dispatch({ type: 'pendingSent', localId: item.localId, message }) // The server answers with the room's own slow-mode failure; starting the // countdown here keeps a fast typist from collecting 429s one by one. if (slowModeSeconds > 0) startSlowMode(slowModeSeconds) return true } catch (error) { const retry = findRetryAfterSeconds(error) if (retry) startSlowMode(retry) dispatch({ type: 'pendingFail', localId: item.localId, error: failureReason(error) }) return false } finally { inFlight.current.delete(item.localId) } }, [slowModeSeconds, startSlowMode], ) const send = useCallback( ({ content, attachments = [], replyToId, threadId: target, threadBroadcast }: SendInput) => { const body = content.replace(/\s+$/, '') if (!enabled || !roomId || (!body && attachments.length === 0)) return Promise.resolve(false) const nonce = clientNonce() const item: PendingMessage = { localId: `local:${nonce}`, nonce, room_id: roomId, author_id: viewerId ?? 0, content: body, created_at: new Date().toISOString(), reply_to_id: replyToId ?? null, thread_id: target ?? threadId ?? null, thread_broadcast: threadBroadcast ?? false, attachments, state: 'sending', sending: true, } pendingById.current[item.localId] = item dispatch({ type: 'pendingAdd', message: item }) // The author must see their own post: pin the viewport back to the tail. atBottomRef.current = true setBottom({ scope: scopeKey, value: true, baseline: 0 }) return deliver(item) }, [deliver, enabled, roomId, scopeKey, threadId, viewerId], ) const retryPending = useCallback( (localId: string) => { const item = pendingById.current[localId] if (!item) return dispatch({ type: 'pendingRetry', localId }) void deliver(item) }, [deliver], ) const dismissPending = useCallback((localId: string) => { delete pendingById.current[localId] dispatch({ type: 'pendingDrop', localId }) }, []) /** Reconcile a failed row whose message actually reached the server. */ useEffect(() => { if (state.pending.length === 0) return for (const row of state.rows) { const echo = findEchoedPending(state.pending, row, viewerId) if (echo) { delete pendingById.current[echo.localId] dispatch({ type: 'pendingDrop', localId: echo.localId }) } } }, [state.pending, state.rows, viewerId]) /* ---------------------------------------------------- message actions */ const reportFailure = useCallback((error: unknown) => { setNotice(describeFailure(error)) if (typeof window !== 'undefined') { window.setTimeout(() => setNotice(null), 6_000) } }, []) const editMessage = useCallback( async (messageId: string, content: string) => { try { dispatch({ type: 'edited', message: await editMessageApi(roomId, messageId, content) }) return true } catch (error) { reportFailure(error) return false } }, [reportFailure, roomId], ) const deleteMessage = useCallback( async (messageId: string) => { try { await deleteMessageApi(roomId, messageId) dispatch({ type: 'deleted', messageIds: [messageId] }) return true } catch (error) { reportFailure(error) return false } }, [reportFailure, roomId], ) const toggleReaction = useCallback( async (messageId: string, emoji: string, on: boolean) => { const local = state.rows.find((row) => row.id === messageId) if (!local) return const reactions = local.reactions ?? [] const without = reactions.filter((reaction) => reaction.emoji !== emoji) // Optimistic, keyed off the *previous* aggregate so a concurrent live // update can be out by one but never loses the viewer's own flag. const current = reactions.find((reaction) => reaction.emoji === emoji) const next = on ? [...without, { emoji, count: (current?.count ?? 0) + 1, me: true }] : without dispatch({ type: 'reactions', messageId, reactions: next }) try { await setReaction(roomId, messageId, emoji, on) } catch (error) { dispatch({ type: 'reactions', messageId, reactions }) reportFailure(error) } }, [reportFailure, roomId, state.rows], ) const togglePin = useCallback( async (messageId: string, pinned: boolean) => { try { const message = await setMessagePin(roomId, messageId, pinned) dispatch({ type: 'pinned', messageId, pinned }) return message } catch (error) { reportFailure(error) return undefined } }, [reportFailure, roomId], ) /* --------------------------------------------------------- pagination */ const loadOlder = useCallback(() => { if (!enabled || state.loadingOlder || !state.hasOlder) return const anchor = state.oldestSeq ?? state.rows[0]?.seq if (anchor === undefined || anchor <= 1) return dispatch({ type: 'olderStart' }) listMessages(roomId, { beforeSeq: anchor, limit: MESSAGE_PAGE_SIZE, threadId: threadId ?? undefined, }).then( (page) => dispatch({ type: 'olderDone', items: page.items, hasMore: page.has_more, oldestSeq: page.oldest_seq ?? null, }), () => dispatch({ type: 'olderDone', items: [], hasMore: false, oldestSeq: null }), ) }, [enabled, roomId, state.hasOlder, state.loadingOlder, state.oldestSeq, state.rows, threadId]) /** * Return to the newest page after a backwards jump. The window is *replaced*, * not extended: rows between where the reader was and the tail were never * fetched, and merging would show a transcript with an invisible hole in it. */ const returnToLatest = useCallback(() => { if (!enabled || !roomId) return dispatch({ type: 'newerStart' }) listMessages(roomId, { limit: MESSAGE_PAGE_SIZE, threadId: threadId ?? undefined }).then( (page) => { // Back at the tail: this is exactly the window the next visit prefills // from, so the stored seed follows the reader. writeCache(chatSeedKey(roomId, threadId ?? null), page) dispatch({ type: 'tailReplaced', items: page.items, hasMore: page.has_more, oldestSeq: page.oldest_seq ?? null, newestSeq: page.newest_seq ?? null, }) tipRef.current = Math.max(tipRef.current, page.newest_seq ?? 0) atBottomRef.current = true setBottom({ scope: scopeKey, value: true, baseline: tipRef.current }) }, () => dispatch({ type: 'seedError', message: t('chat.transcript.seedFailed') }), ) }, [enabled, roomId, scopeKey, threadId]) /** * Open the window around a row that is not loaded — a pin, a search hit, a * reply target. `get_message` yields its seq first, because a page can only be * anchored on a cursor, and one call cannot straddle a row. */ const jumpTo = useCallback( (messageId: string) => { if (!enabled || !roomId) return dispatch({ type: 'highlight', messageId }) if (state.rows.some((row) => row.id === messageId)) return void (async () => { try { const target = await getMessage(roomId, messageId) const page = await listMessages(roomId, { beforeSeq: target.seq + 1, limit: MESSAGE_PAGE_SIZE, threadId: threadId ?? undefined, }) dispatch({ type: 'seedDone', items: page.items, hasMore: page.has_more, oldestSeq: page.oldest_seq ?? null, newestSeq: page.newest_seq ?? null, }) // "Missed" restarts at the row the reader asked for, not at the tail. tipRef.current = Math.max(tipRef.current, page.newest_seq ?? target.seq) atBottomRef.current = false setBottom({ scope: scopeKey, value: false, baseline: target.seq }) dispatch({ type: 'highlight', messageId }) } catch (error) { reportFailure(error) } })() }, [enabled, reportFailure, roomId, scopeKey, state.rows, threadId], ) /* ------------------------------------------------------- reply lookup */ const missingRefs = useMemo( () => state.rows .map((row) => row.reply_to_id) .filter((id): id is string => Boolean(id)) .filter((id) => !state.refs[id] && !state.rows.some((row) => row.id === id)), [state.refs, state.rows], ) useEffect(() => { if (missingRefs.length === 0 || !enabled) return let active = true for (const id of missingRefs) { dispatch({ type: 'refStart', messageId: id }) getMessage(roomId, id).then( (message) => active && dispatch({ type: 'refDone', messageId: id, message }), () => active && dispatch({ type: 'refDone', messageId: id, message: null }), ) } return () => { active = false } }, [enabled, missingRefs, roomId]) /* -------------------------------------------------------- scroll glue */ const reportAtBottom = useCallback( (value: boolean) => { if (atBottomRef.current === value) return atBottomRef.current = value // The baseline is captured at the moment the reader leaves the tail, which // is what makes the「N 条新消息」pill count only what they have not seen. // The baseline is captured at the moment the reader leaves the tail, which // is what makes the「N 条新消息」pill count only what they have not seen. setBottom({ scope: scopeKey, value, baseline: value ? 0 : tipRef.current }) }, [scopeKey], ) const scrollToBottom = useCallback( (options?: { smooth?: boolean }) => { const node = listRef.current if (node) { node.scrollTo({ top: node.scrollHeight, behavior: options?.smooth ? 'smooth' : 'auto', }) } reportAtBottom(true) }, [reportAtBottom], ) // Pinned means "follow the tail": only a *change of the last row* scrolls, so // prepending an older page never moves the viewport out from under the reader. const lastTail = useRef(null) useEffect(() => { const tail = state.rows[state.rows.length - 1]?.id ?? '' const pendingTail = state.pending[state.pending.length - 1]?.localId ?? '' const signature = `${tail}|${pendingTail}` const changed = lastTail.current !== signature lastTail.current = signature if (!changed || !atBottomRef.current) return const frame = requestAnimationFrame(() => { const node = listRef.current if (node) node.scrollTop = node.scrollHeight }) return () => cancelAnimationFrame(frame) }, [state.rows, state.pending]) const missedCount = useMemo(() => { if (atBottom) return 0 let count = 0 for (const row of state.rows) if (row.seq > missedBaseline) count += 1 return count }, [atBottom, missedBaseline, state.rows]) return { state, status, typingUserIds: typing.map((indicator) => Number(indicator.user_id)), lastReadSeq, atBottom, missedCount, slowModeRemaining, notice, dismissNotice: () => setNotice(null), listRef, reportAtBottom, scrollToBottom, loadOlder, jumpTo, returnToLatest, send, retryPending, dismissPending, editMessage, deleteMessage, toggleReaction, togglePin, refresh: () => setReloadKey((key) => key + 1), } }