diff --git a/web/src/App.tsx b/web/src/App.tsx index a9745b2b..b672835b 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -11,6 +11,7 @@ import { useSyncingState } from '@/hooks/useSyncingState' import { usePushNotifications } from '@/hooks/usePushNotifications' import { queryKeys } from '@/lib/query-keys' import { AppContextProvider } from '@/lib/app-context' +import { fetchLatestMessages } from '@/lib/message-window-store' import { useAppGoBack } from '@/hooks/useAppGoBack' import { LoginPrompt } from '@/components/LoginPrompt' import { InstallPrompt } from '@/components/InstallPrompt' @@ -152,11 +153,13 @@ export function App() { const invalidations = [ queryClient.invalidateQueries({ queryKey: queryKeys.sessions }), ...(selectedSessionId ? [ - queryClient.invalidateQueries({ queryKey: queryKeys.session(selectedSessionId) }), - queryClient.invalidateQueries({ queryKey: queryKeys.messages(selectedSessionId) }) + queryClient.invalidateQueries({ queryKey: queryKeys.session(selectedSessionId) }) ] : []) ] - Promise.all(invalidations) + const refreshMessages = (selectedSessionId && api) + ? fetchLatestMessages(api, selectedSessionId) + : Promise.resolve() + Promise.all([...invalidations, refreshMessages]) .catch((error) => { console.error('Failed to invalidate queries on SSE connect:', error) }) @@ -166,7 +169,7 @@ export function App() { endSync() } }) - }, [queryClient, selectedSessionId, startSync, endSync]) + }, [api, queryClient, selectedSessionId, startSync, endSync]) const handleSseEvent = useCallback(() => {}, []) diff --git a/web/src/components/AssistantChat/HappyThread.tsx b/web/src/components/AssistantChat/HappyThread.tsx index ceda7685..ec958601 100644 --- a/web/src/components/AssistantChat/HappyThread.tsx +++ b/web/src/components/AssistantChat/HappyThread.tsx @@ -59,14 +59,18 @@ export function HappyThread(props: { disabled: boolean onRefresh: () => void onRetryMessage?: (localId: string) => void + onFlushPending: () => void + onAtBottomChange: (atBottom: boolean) => void isLoadingMessages: boolean messagesWarning: string | null hasMoreMessages: boolean isLoadingMoreMessages: boolean onLoadMore: () => Promise + pendingCount: number rawMessagesCount: number normalizedMessagesCount: number - renderedMessagesCount: number + messagesVersion: number + forceScrollToken: number }) { const viewportRef = useRef(null) const topSentinelRef = useRef(null) @@ -75,22 +79,25 @@ export function HappyThread(props: { const prevLoadingMoreRef = useRef(false) const loadStartedRef = useRef(false) const isLoadingMoreRef = useRef(props.isLoadingMoreMessages) + const atBottomRef = useRef(true) + const onAtBottomChangeRef = useRef(props.onAtBottomChange) + const onFlushPendingRef = useRef(props.onFlushPending) + const forceScrollTokenRef = useRef(props.forceScrollToken) // Smart scroll state: autoScroll enabled when user is near bottom const [autoScrollEnabled, setAutoScrollEnabled] = useState(true) - const [newMessageCount, setNewMessageCount] = useState(0) - const prevRenderedCountRef = useRef(props.renderedMessagesCount) const autoScrollEnabledRef = useRef(autoScrollEnabled) - const newMessageCountRef = useRef(newMessageCount) - const hasBootstrappedRef = useRef(false) // Keep refs in sync with state useEffect(() => { autoScrollEnabledRef.current = autoScrollEnabled }, [autoScrollEnabled]) useEffect(() => { - newMessageCountRef.current = newMessageCount - }, [newMessageCount]) + onAtBottomChangeRef.current = props.onAtBottomChange + }, [props.onAtBottomChange]) + useEffect(() => { + onFlushPendingRef.current = props.onFlushPending + }, [props.onFlushPending]) // Track scroll position to toggle autoScroll (stable listener using refs) useEffect(() => { @@ -105,9 +112,16 @@ export function HappyThread(props: { if (isNearBottom) { if (!autoScrollEnabledRef.current) setAutoScrollEnabled(true) - if (newMessageCountRef.current > 0) setNewMessageCount(0) - } else { - if (autoScrollEnabledRef.current) setAutoScrollEnabled(false) + } else if (autoScrollEnabledRef.current) { + setAutoScrollEnabled(false) + } + + if (isNearBottom !== atBottomRef.current) { + atBottomRef.current = isNearBottom + onAtBottomChangeRef.current(isNearBottom) + if (isNearBottom) { + onFlushPendingRef.current() + } } } @@ -115,43 +129,6 @@ export function HappyThread(props: { return () => viewport.removeEventListener('scroll', handleScroll) }, []) // Stable: no dependencies, reads from refs - // Track new messages when autoScroll is disabled - const wasLoadingMoreRef = useRef(props.isLoadingMoreMessages) - useEffect(() => { - const prevCount = prevRenderedCountRef.current - const currentCount = props.renderedMessagesCount - const wasLoadingMore = wasLoadingMoreRef.current - wasLoadingMoreRef.current = props.isLoadingMoreMessages - - if (props.isLoadingMessages) { - prevRenderedCountRef.current = currentCount - return - } - - if (!hasBootstrappedRef.current) { - hasBootstrappedRef.current = true - prevRenderedCountRef.current = currentCount - return - } - - prevRenderedCountRef.current = currentCount - - // Skip during loading states - if (props.isLoadingMoreMessages) { - return - } - - // Skip if load-more just finished (older messages, not new ones) - if (wasLoadingMore) { - return - } - - const newCount = currentCount - prevCount - if (newCount > 0 && !autoScrollEnabled) { - setNewMessageCount((prev) => prev + newCount) - } - }, [props.renderedMessagesCount, props.isLoadingMoreMessages, props.isLoadingMessages, autoScrollEnabled]) - // Scroll to bottom handler for the indicator button const scrollToBottom = useCallback(() => { const viewport = viewportRef.current @@ -159,17 +136,29 @@ export function HappyThread(props: { viewport.scrollTo({ top: viewport.scrollHeight, behavior: 'smooth' }) } setAutoScrollEnabled(true) - setNewMessageCount(0) + if (!atBottomRef.current) { + atBottomRef.current = true + onAtBottomChangeRef.current(true) + } + onFlushPendingRef.current() }, []) // Reset state when session changes useEffect(() => { setAutoScrollEnabled(true) - setNewMessageCount(0) - prevRenderedCountRef.current = 0 - hasBootstrappedRef.current = false + atBottomRef.current = true + onAtBottomChangeRef.current(true) + forceScrollTokenRef.current = props.forceScrollToken }, [props.sessionId]) + useEffect(() => { + if (forceScrollTokenRef.current === props.forceScrollToken) { + return + } + forceScrollTokenRef.current = props.forceScrollToken + scrollToBottom() + }, [props.forceScrollToken, scrollToBottom]) + const handleLoadMore = useCallback(() => { if (props.isLoadingMessages || !props.hasMoreMessages || props.isLoadingMoreMessages || loadLockRef.current) { return @@ -242,7 +231,7 @@ export function HappyThread(props: { viewport.scrollTop = pending.scrollTop + delta pendingScrollRef.current = null loadLockRef.current = false - }, [props.rawMessagesCount]) + }, [props.messagesVersion]) useEffect(() => { isLoadingMoreRef.current = props.isLoadingMoreMessages @@ -320,7 +309,7 @@ export function HappyThread(props: { - + ) diff --git a/web/src/components/SessionChat.tsx b/web/src/components/SessionChat.tsx index f9d2eb9c..56f477bd 100644 --- a/web/src/components/SessionChat.tsx +++ b/web/src/components/SessionChat.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useRef } from 'react' +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import { useNavigate } from '@tanstack/react-router' import { AssistantRuntimeProvider } from '@assistant-ui/react' import type { ApiClient } from '@/api/client' @@ -24,10 +24,14 @@ export function SessionChat(props: { isLoadingMessages: boolean isLoadingMoreMessages: boolean isSending: boolean + pendingCount: number + messagesVersion: number onBack: () => void onRefresh: () => void onLoadMore: () => Promise onSend: (text: string) => void + onFlushPending: () => void + onAtBottomChange: (atBottom: boolean) => void onRetryMessage?: (localId: string) => void autocompleteSuggestions?: (query: string) => Promise }) { @@ -36,6 +40,7 @@ export function SessionChat(props: { const controlsDisabled = !props.session.active const normalizedCacheRef = useRef>(new Map()) const blocksByIdRef = useRef>(new Map()) + const [forceScrollToken, setForceScrollToken] = useState(0) const agentFlavor = props.session.metadata?.flavor ?? null const { abortSession, switchSession, setPermissionMode, setModelMode } = useSessionActions( props.api, @@ -134,11 +139,16 @@ export function SessionChat(props: { }) }, [navigate, props.session.id]) + const handleSend = useCallback((text: string) => { + props.onSend(text) + setForceScrollToken((token) => token + 1) + }, [props.onSend]) + const runtime = useHappyRuntime({ session: props.session, blocks: reconciled.blocks, isSending: props.isSending, - onSendMessage: props.onSend, + onSendMessage: handleSend, onAbort: handleAbort }) @@ -170,14 +180,18 @@ export function SessionChat(props: { disabled={controlsDisabled} onRefresh={props.onRefresh} onRetryMessage={props.onRetryMessage} + onFlushPending={props.onFlushPending} + onAtBottomChange={props.onAtBottomChange} isLoadingMessages={props.isLoadingMessages} messagesWarning={props.messagesWarning} hasMoreMessages={props.hasMoreMessages} isLoadingMoreMessages={props.isLoadingMoreMessages} onLoadMore={props.onLoadMore} + pendingCount={props.pendingCount} rawMessagesCount={props.messages.length} normalizedMessagesCount={normalizedMessages.length} - renderedMessagesCount={reconciled.blocks.length} + messagesVersion={props.messagesVersion} + forceScrollToken={forceScrollToken} /> | undefined, - localId: string, - status: DecryptedMessage['status'], -): InfiniteData | undefined { - if (!data) return data - - const pages = data.pages.map((page) => ({ - ...page, - messages: page.messages.map((message) => - message.localId === localId - ? { ...message, status } - : message - ), - })) - - return { - ...data, - pages, - } -} - function findMessageByLocalId( - data: InfiniteData | undefined, + sessionId: string, localId: string, ): DecryptedMessage | null { - if (!data) return null - for (const page of data.pages) { - const match = page.messages.find((message) => message.localId === localId) - if (match) return match + const state = getMessageWindowState(sessionId) + for (const message of state.messages) { + if (message.localId === localId) return message + } + for (const message of state.pending) { + if (message.localId === localId) return message } return null } @@ -51,7 +35,6 @@ export function useSendMessage(api: ApiClient | null, sessionId: string | null): retryMessage: (localId: string) => void isSending: boolean } { - const queryClient = useQueryClient() const { haptic } = usePlatform() const mutation = useMutation({ @@ -72,23 +55,14 @@ export function useSendMessage(api: ApiClient | null, sessionId: string | null): originalText: input.text, } - queryClient.setQueryData>( - queryKeys.messages(input.sessionId), - (data) => upsertMessagesInCache(data, [optimisticMessage]), - ) + appendOptimisticMessage(input.sessionId, optimisticMessage) }, onSuccess: (_, input) => { - queryClient.setQueryData>( - queryKeys.messages(input.sessionId), - (data) => updateMessageStatus(data, input.localId, 'sent'), - ) + updateMessageStatus(input.sessionId, input.localId, 'sent') haptic.notification('success') }, onError: (_, input) => { - queryClient.setQueryData>( - queryKeys.messages(input.sessionId), - (data) => updateMessageStatus(data, input.localId, 'failed'), - ) + updateMessageStatus(input.sessionId, input.localId, 'failed') haptic.notification('error') }, }) @@ -109,14 +83,10 @@ export function useSendMessage(api: ApiClient | null, sessionId: string | null): if (!api || !sessionId) return if (mutation.isPending) return - const data = queryClient.getQueryData>(queryKeys.messages(sessionId)) - const message = findMessageByLocalId(data, localId) + const message = findMessageByLocalId(sessionId, localId) if (!message?.originalText) return - queryClient.setQueryData>( - queryKeys.messages(sessionId), - (current) => updateMessageStatus(current, localId, 'sending'), - ) + updateMessageStatus(sessionId, localId, 'sending') mutation.mutate({ sessionId, diff --git a/web/src/hooks/mutations/useSessionActions.ts b/web/src/hooks/mutations/useSessionActions.ts index 62b832a6..91bacae4 100644 --- a/web/src/hooks/mutations/useSessionActions.ts +++ b/web/src/hooks/mutations/useSessionActions.ts @@ -3,6 +3,7 @@ import { isPermissionModeAllowedForFlavor } from '@hapi/protocol' import type { ApiClient } from '@/api/client' import type { ModelMode, PermissionMode } from '@/types/api' import { queryKeys } from '@/lib/query-keys' +import { clearMessageWindow } from '@/lib/message-window-store' export function useSessionActions( api: ApiClient | null, @@ -100,7 +101,7 @@ export function useSessionActions( onSuccess: async () => { if (!sessionId) return queryClient.removeQueries({ queryKey: queryKeys.session(sessionId) }) - queryClient.removeQueries({ queryKey: queryKeys.messages(sessionId) }) + clearMessageWindow(sessionId) await queryClient.invalidateQueries({ queryKey: queryKeys.sessions }) }, }) diff --git a/web/src/hooks/queries/useMessages.ts b/web/src/hooks/queries/useMessages.ts index 84067c82..237e16f8 100644 --- a/web/src/hooks/queries/useMessages.ts +++ b/web/src/hooks/queries/useMessages.ts @@ -1,9 +1,31 @@ -import { useInfiniteQuery } from '@tanstack/react-query' -import { useMemo } from 'react' +import { useCallback, useEffect, useSyncExternalStore } from 'react' import type { ApiClient } from '@/api/client' -import type { DecryptedMessage, MessagesResponse } from '@/types/api' -import { mergeMessages } from '@/lib/messages' -import { queryKeys } from '@/lib/query-keys' +import type { DecryptedMessage } from '@/types/api' +import { + clearMessageWindow, + fetchLatestMessages, + fetchOlderMessages, + flushPendingMessages, + getMessageWindowState, + setAtBottom as setMessageWindowAtBottom, + subscribeMessageWindow, + type MessageWindowState, +} from '@/lib/message-window-store' + +const EMPTY_STATE: MessageWindowState = { + sessionId: 'unknown', + messages: [], + pending: [], + pendingCount: 0, + hasMore: false, + oldestSeq: null, + newestSeq: null, + isLoading: false, + isLoadingMore: false, + warning: null, + atBottom: true, + messagesVersion: 0, +} export function useMessages(api: ApiClient | null, sessionId: string | null): { messages: DecryptedMessage[] @@ -11,51 +33,80 @@ export function useMessages(api: ApiClient | null, sessionId: string | null): { isLoading: boolean isLoadingMore: boolean hasMore: boolean + pendingCount: number + messagesVersion: number loadMore: () => Promise refetch: () => Promise + flushPending: () => Promise + setAtBottom: (atBottom: boolean) => void } { - const resolvedSessionId = sessionId ?? 'unknown' - const query = useInfiniteQuery({ - queryKey: queryKeys.messages(resolvedSessionId), - queryFn: async ({ pageParam }) => { - if (!api || !sessionId) { - throw new Error('Messages unavailable') + const state = useSyncExternalStore( + useCallback((listener) => { + if (!sessionId) { + return () => {} } - const beforeSeq = typeof pageParam === 'number' ? pageParam : null - return await api.getMessages(sessionId, { limit: 50, beforeSeq }) - }, - initialPageParam: null, - getNextPageParam: (lastPage) => - lastPage.page.hasMore ? lastPage.page.nextBeforeSeq : undefined, - enabled: Boolean(api && sessionId), - }) + return subscribeMessageWindow(sessionId, listener) + }, [sessionId]), + useCallback(() => { + if (!sessionId) { + return EMPTY_STATE + } + return getMessageWindowState(sessionId) + }, [sessionId]), + () => EMPTY_STATE + ) - const messages = useMemo(() => { - const pages = query.data?.pages ?? [] - let merged: DecryptedMessage[] = [] - for (const page of pages) { - merged = mergeMessages(merged, page.messages) + useEffect(() => { + if (!api || !sessionId) { + return } - return merged - }, [query.data?.pages]) + void fetchLatestMessages(api, sessionId) + }, [api, sessionId]) - const warning = useMemo(() => { - if (!query.error) return null - return query.error instanceof Error ? query.error.message : 'Failed to load messages' - }, [query.error]) + useEffect(() => { + if (!sessionId) { + return + } + return () => { + clearMessageWindow(sessionId) + } + }, [sessionId]) - const loadMore = async () => { - if (!query.hasNextPage || query.isFetchingNextPage) return - await query.fetchNextPage() - } + const loadMore = useCallback(async () => { + if (!api || !sessionId) return + if (!state.hasMore || state.isLoadingMore) return + await fetchOlderMessages(api, sessionId) + }, [api, sessionId, state.hasMore, state.isLoadingMore]) + + const refetch = useCallback(async () => { + if (!api || !sessionId) return + await fetchLatestMessages(api, sessionId) + }, [api, sessionId]) + + const flushPending = useCallback(async () => { + if (!sessionId) return + const needsRefresh = flushPendingMessages(sessionId) + if (needsRefresh && api) { + await fetchLatestMessages(api, sessionId) + } + }, [api, sessionId]) + + const setAtBottom = useCallback((atBottom: boolean) => { + if (!sessionId) return + setMessageWindowAtBottom(sessionId, atBottom) + }, [sessionId]) return { - messages, - warning, - isLoading: query.isLoading, - isLoadingMore: query.isFetchingNextPage, - hasMore: Boolean(query.hasNextPage), + messages: state.messages, + warning: state.warning, + isLoading: state.isLoading, + isLoadingMore: state.isLoadingMore, + hasMore: state.hasMore, + pendingCount: state.pendingCount, + messagesVersion: state.messagesVersion, loadMore, - refetch: query.refetch, + refetch, + flushPending, + setAtBottom, } } diff --git a/web/src/hooks/useSSE.ts b/web/src/hooks/useSSE.ts index 32408ed8..bdd675e9 100644 --- a/web/src/hooks/useSSE.ts +++ b/web/src/hooks/useSSE.ts @@ -1,8 +1,8 @@ import { useEffect, useMemo, useRef } from 'react' -import { useQueryClient, type InfiniteData } from '@tanstack/react-query' -import type { MessagesResponse, SyncEvent } from '@/types/api' +import { useQueryClient } from '@tanstack/react-query' +import type { SyncEvent } from '@/types/api' import { queryKeys } from '@/lib/query-keys' -import { upsertMessagesInCache } from '@/lib/messages' +import { clearMessageWindow, ingestIncomingMessages } from '@/lib/message-window-store' function isObject(value: unknown): value is Record { return Boolean(value) && typeof value === 'object' @@ -86,15 +86,7 @@ export function useSSE(options: { const handleSyncEvent = (event: SyncEvent) => { if (event.type === 'message-received') { - queryClient.setQueryData>( - queryKeys.messages(event.sessionId), - (data) => upsertMessagesInCache(data, [event.message]) - ) - // Mark stale so the initial query still fetches history when it mounts. - void queryClient.invalidateQueries({ - queryKey: queryKeys.messages(event.sessionId), - refetchType: 'none' - }) + ingestIncomingMessages(event.sessionId, [event.message]) } if (event.type === 'session-added' || event.type === 'session-updated' || event.type === 'session-removed') { @@ -102,7 +94,7 @@ export function useSSE(options: { if ('sessionId' in event) { if (event.type === 'session-removed') { void queryClient.removeQueries({ queryKey: queryKeys.session(event.sessionId) }) - void queryClient.removeQueries({ queryKey: queryKeys.messages(event.sessionId) }) + clearMessageWindow(event.sessionId) } else { void queryClient.invalidateQueries({ queryKey: queryKeys.session(event.sessionId) }) } diff --git a/web/src/lib/message-window-store.ts b/web/src/lib/message-window-store.ts new file mode 100644 index 00000000..59be8434 --- /dev/null +++ b/web/src/lib/message-window-store.ts @@ -0,0 +1,351 @@ +import type { ApiClient } from '@/api/client' +import type { DecryptedMessage, MessageStatus } from '@/types/api' +import { mergeMessages } from '@/lib/messages' + +export type MessageWindowState = { + sessionId: string + messages: DecryptedMessage[] + pending: DecryptedMessage[] + pendingCount: number + hasMore: boolean + oldestSeq: number | null + newestSeq: number | null + isLoading: boolean + isLoadingMore: boolean + warning: string | null + atBottom: boolean + messagesVersion: number +} + +export const VISIBLE_WINDOW_SIZE = 400 +export const PENDING_WINDOW_SIZE = 200 +const PAGE_SIZE = 50 +const PENDING_OVERFLOW_WARNING = 'New messages arrived while you were away. Scroll to bottom to refresh.' + +type InternalState = MessageWindowState & { + pendingOverflowCount: number +} + +const states = new Map() +const listeners = new Map void>>() + +function createState(sessionId: string): InternalState { + return { + sessionId, + messages: [], + pending: [], + pendingCount: 0, + hasMore: false, + oldestSeq: null, + newestSeq: null, + isLoading: false, + isLoadingMore: false, + warning: null, + atBottom: true, + messagesVersion: 0, + pendingOverflowCount: 0, + } +} + +function getState(sessionId: string): InternalState { + const existing = states.get(sessionId) + if (existing) { + return existing + } + const created = createState(sessionId) + states.set(sessionId, created) + return created +} + +function notify(sessionId: string): void { + const subs = listeners.get(sessionId) + if (!subs) return + for (const listener of subs) { + listener() + } +} + +function setState(sessionId: string, next: InternalState): void { + states.set(sessionId, next) + notify(sessionId) +} + +function updateState(sessionId: string, updater: (prev: InternalState) => InternalState): void { + const prev = getState(sessionId) + const next = updater(prev) + if (next !== prev) { + setState(sessionId, next) + } +} + +function deriveSeqBounds(messages: DecryptedMessage[]): { oldestSeq: number | null; newestSeq: number | null } { + let oldest: number | null = null + let newest: number | null = null + for (const message of messages) { + if (typeof message.seq !== 'number') { + continue + } + if (oldest === null || message.seq < oldest) { + oldest = message.seq + } + if (newest === null || message.seq > newest) { + newest = message.seq + } + } + return { oldestSeq: oldest, newestSeq: newest } +} + +function buildState( + prev: InternalState, + updates: { + messages?: DecryptedMessage[] + pending?: DecryptedMessage[] + pendingOverflowCount?: number + hasMore?: boolean + isLoading?: boolean + isLoadingMore?: boolean + warning?: string | null + atBottom?: boolean + } +): InternalState { + const messages = updates.messages ?? prev.messages + const pending = updates.pending ?? prev.pending + const pendingOverflowCount = updates.pendingOverflowCount ?? prev.pendingOverflowCount + const pendingCount = pending.length + pendingOverflowCount + const { oldestSeq, newestSeq } = deriveSeqBounds(messages) + const messagesVersion = messages === prev.messages ? prev.messagesVersion : prev.messagesVersion + 1 + + return { + ...prev, + messages, + pending, + pendingOverflowCount, + pendingCount, + oldestSeq, + newestSeq, + hasMore: updates.hasMore !== undefined ? updates.hasMore : prev.hasMore, + isLoading: updates.isLoading !== undefined ? updates.isLoading : prev.isLoading, + isLoadingMore: updates.isLoadingMore !== undefined ? updates.isLoadingMore : prev.isLoadingMore, + warning: updates.warning !== undefined ? updates.warning : prev.warning, + atBottom: updates.atBottom !== undefined ? updates.atBottom : prev.atBottom, + messagesVersion, + } +} + +function trimVisible(messages: DecryptedMessage[], mode: 'append' | 'prepend'): DecryptedMessage[] { + if (messages.length <= VISIBLE_WINDOW_SIZE) { + return messages + } + if (mode === 'prepend') { + return messages.slice(0, VISIBLE_WINDOW_SIZE) + } + return messages.slice(messages.length - VISIBLE_WINDOW_SIZE) +} + +function trimPending(messages: DecryptedMessage[]): { pending: DecryptedMessage[]; dropped: number } { + if (messages.length <= PENDING_WINDOW_SIZE) { + return { pending: messages, dropped: 0 } + } + const dropped = messages.length - PENDING_WINDOW_SIZE + return { pending: messages.slice(messages.length - PENDING_WINDOW_SIZE), dropped } +} + +function filterPendingAgainstVisible(pending: DecryptedMessage[], visible: DecryptedMessage[]): DecryptedMessage[] { + if (pending.length === 0 || visible.length === 0) { + return pending + } + const visibleIds = new Set(visible.map((message) => message.id)) + return pending.filter((message) => !visibleIds.has(message.id)) +} + +function isOptimisticMessage(message: DecryptedMessage): boolean { + return Boolean(message.localId && message.id === message.localId) +} + +function mergeIntoPending( + prev: InternalState, + incoming: DecryptedMessage[] +): { pending: DecryptedMessage[]; pendingOverflowCount: number; warning: string | null } { + if (incoming.length === 0) { + return { pending: prev.pending, pendingOverflowCount: prev.pendingOverflowCount, warning: prev.warning } + } + const mergedPending = mergeMessages(prev.pending, incoming) + const filtered = filterPendingAgainstVisible(mergedPending, prev.messages) + const { pending, dropped } = trimPending(filtered) + const pendingOverflowCount = prev.pendingOverflowCount + dropped + const warning = dropped > 0 && !prev.warning ? PENDING_OVERFLOW_WARNING : prev.warning + return { pending, pendingOverflowCount, warning } +} + +export function getMessageWindowState(sessionId: string): MessageWindowState { + return getState(sessionId) +} + +export function subscribeMessageWindow(sessionId: string, listener: () => void): () => void { + const subs = listeners.get(sessionId) ?? new Set() + subs.add(listener) + listeners.set(sessionId, subs) + return () => { + const current = listeners.get(sessionId) + if (!current) return + current.delete(listener) + if (current.size === 0) { + listeners.delete(sessionId) + states.delete(sessionId) + } + } +} + +export function clearMessageWindow(sessionId: string): void { + if (!states.has(sessionId)) { + return + } + setState(sessionId, createState(sessionId)) +} + +export async function fetchLatestMessages(api: ApiClient, sessionId: string): Promise { + const initial = getState(sessionId) + if (initial.isLoading) { + return + } + updateState(sessionId, (prev) => buildState(prev, { isLoading: true, warning: null })) + + try { + const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: null }) + updateState(sessionId, (prev) => { + if (prev.atBottom) { + const merged = mergeMessages(prev.messages, [...prev.pending, ...response.messages]) + const trimmed = trimVisible(merged, 'append') + return buildState(prev, { + messages: trimmed, + pending: [], + pendingOverflowCount: 0, + hasMore: response.page.hasMore, + isLoading: false, + warning: null, + }) + } + const pendingResult = mergeIntoPending(prev, response.messages) + return buildState(prev, { + pending: pendingResult.pending, + pendingOverflowCount: pendingResult.pendingOverflowCount, + isLoading: false, + warning: pendingResult.warning, + }) + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Failed to load messages' + updateState(sessionId, (prev) => buildState(prev, { isLoading: false, warning: message })) + } +} + +export async function fetchOlderMessages(api: ApiClient, sessionId: string): Promise { + const initial = getState(sessionId) + if (initial.isLoadingMore || !initial.hasMore) { + return + } + if (initial.oldestSeq === null) { + return + } + updateState(sessionId, (prev) => buildState(prev, { isLoadingMore: true })) + + try { + const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: initial.oldestSeq }) + updateState(sessionId, (prev) => { + const merged = mergeMessages(response.messages, prev.messages) + const trimmed = trimVisible(merged, 'prepend') + return buildState(prev, { + messages: trimmed, + hasMore: response.page.hasMore, + isLoadingMore: false, + }) + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Failed to load messages' + updateState(sessionId, (prev) => buildState(prev, { isLoadingMore: false, warning: message })) + } +} + +export function ingestIncomingMessages(sessionId: string, incoming: DecryptedMessage[]): void { + if (incoming.length === 0) { + return + } + updateState(sessionId, (prev) => { + if (prev.atBottom) { + const merged = mergeMessages(prev.messages, incoming) + const trimmed = trimVisible(merged, 'append') + const pending = filterPendingAgainstVisible(prev.pending, trimmed) + return buildState(prev, { messages: trimmed, pending }) + } + const pendingResult = mergeIntoPending(prev, incoming) + return buildState(prev, { + pending: pendingResult.pending, + pendingOverflowCount: pendingResult.pendingOverflowCount, + warning: pendingResult.warning, + }) + }) +} + +export function flushPendingMessages(sessionId: string): boolean { + const current = getState(sessionId) + if (current.pending.length === 0 && current.pendingOverflowCount === 0) { + return false + } + const needsRefresh = current.pendingOverflowCount > 0 + updateState(sessionId, (prev) => { + const merged = mergeMessages(prev.messages, prev.pending) + const trimmed = trimVisible(merged, 'append') + return buildState(prev, { + messages: trimmed, + pending: [], + pendingOverflowCount: 0, + warning: needsRefresh ? (prev.warning ?? PENDING_OVERFLOW_WARNING) : prev.warning, + }) + }) + return needsRefresh +} + +export function setAtBottom(sessionId: string, atBottom: boolean): void { + updateState(sessionId, (prev) => { + if (prev.atBottom === atBottom) { + return prev + } + return buildState(prev, { atBottom }) + }) +} + +export function appendOptimisticMessage(sessionId: string, message: DecryptedMessage): void { + updateState(sessionId, (prev) => { + const merged = mergeMessages(prev.messages, [message]) + const trimmed = trimVisible(merged, 'append') + const pending = filterPendingAgainstVisible(prev.pending, trimmed) + return buildState(prev, { messages: trimmed, pending, atBottom: true }) + }) +} + +export function updateMessageStatus(sessionId: string, localId: string, status: MessageStatus): void { + if (!localId) { + return + } + updateState(sessionId, (prev) => { + let changed = false + const updateList = (list: DecryptedMessage[]) => { + return list.map((message) => { + if (message.localId !== localId || !isOptimisticMessage(message)) { + return message + } + if (message.status === status) { + return message + } + changed = true + return { ...message, status } + }) + } + const messages = updateList(prev.messages) + const pending = updateList(prev.pending) + if (!changed) { + return prev + } + return buildState(prev, { messages, pending }) + }) +} diff --git a/web/src/lib/messages.ts b/web/src/lib/messages.ts index 0f960e77..d832624b 100644 --- a/web/src/lib/messages.ts +++ b/web/src/lib/messages.ts @@ -16,6 +16,10 @@ function isUserMessage(msg: DecryptedMessage): boolean { return false } +function isOptimisticMessage(msg: DecryptedMessage): boolean { + return Boolean(msg.localId && msg.id === msg.localId) +} + function compareMessages(a: DecryptedMessage, b: DecryptedMessage): number { const aSeq = typeof a.seq === 'number' ? a.seq : null const bSeq = typeof b.seq === 'number' ? b.seq : null @@ -46,33 +50,32 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM let merged = Array.from(byId.values()) - const incomingLocalIds = new Set() + const incomingStoredLocalIds = new Set() for (const msg of incoming) { - if (msg.localId) { - incomingLocalIds.add(msg.localId) + if (msg.localId && !isOptimisticMessage(msg)) { + incomingStoredLocalIds.add(msg.localId) } } - // If we received a stored message with a localId, drop any optimistic bubble with the same localId. - if (incomingLocalIds.size > 0) { + // If we received stored messages with a localId, drop any optimistic bubbles with the same localId. + if (incomingStoredLocalIds.size > 0) { merged = merged.filter((msg) => { - if (!msg.localId || !incomingLocalIds.has(msg.localId)) { + if (!msg.localId || !incomingStoredLocalIds.has(msg.localId)) { return true } - return !msg.status + return !isOptimisticMessage(msg) }) } // Fallback: if an optimistic message was marked as sent but we didn't get a localId echo, // drop it when a server user message appears close in time. - const optimisticMessages = merged.filter((m) => m.localId && m.status) - const nonOptimisticMessages = merged.filter((m) => !m.localId || !m.status) + const optimisticMessages = merged.filter((m) => isOptimisticMessage(m)) + const nonOptimisticMessages = merged.filter((m) => !isOptimisticMessage(m)) const result: DecryptedMessage[] = [...nonOptimisticMessages] for (const optimistic of optimisticMessages) { if (optimistic.status === 'sent') { const hasServerUserMessage = nonOptimisticMessages.some((m) => - !m.status && isUserMessage(m) && Math.abs(m.createdAt - optimistic.createdAt) < 10_000 ) diff --git a/web/src/router.tsx b/web/src/router.tsx index 38ac17f4..eed9d29e 100644 --- a/web/src/router.tsx +++ b/web/src/router.tsx @@ -133,6 +133,10 @@ function SessionPage() { hasMore: messagesHasMore, loadMore: loadMoreMessages, refetch: refetchMessages, + pendingCount, + messagesVersion, + flushPending, + setAtBottom, } = useMessages(api, sessionId) const { sendMessage, @@ -169,10 +173,14 @@ function SessionPage() { isLoadingMessages={messagesLoading} isLoadingMoreMessages={messagesLoadingMore} isSending={isSending} + pendingCount={pendingCount} + messagesVersion={messagesVersion} onBack={goBack} onRefresh={refreshSelectedSession} onLoadMore={loadMoreMessages} onSend={sendMessage} + onFlushPending={flushPending} + onAtBottomChange={setAtBottom} onRetryMessage={retryMessage} autocompleteSuggestions={getSlashSuggestions} />