main
/**
* 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<typeof useRoomWatch>
/** 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<HTMLDivElement | null>
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<boolean>
retryPending: (localId: string) => void
dismissPending: (localId: string) => void
editMessage: (messageId: string, content: string) => Promise<boolean>
deleteMessage: (messageId: string) => Promise<boolean>
toggleReaction: (messageId: string, emoji: string, on: boolean) => Promise<void>
togglePin: (messageId: string, pinned: boolean) => Promise<ChatMessage | undefined>
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<HTMLDivElement | null>(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<string | null>(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<ReturnType<typeof setTimeout> | undefined>(undefined)
const inFlight = useRef<Set<string>>(new Set())
const pendingById = useRef<Record<string, PendingMessage>>({})
const slowTimer = useRef<ReturnType<typeof setInterval> | 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<MessagePage>(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<boolean> => {
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<string | null>(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),
}
}