From 299ea50fb5bb6874c3fca96277fce187205ac20a Mon Sep 17 00:00:00 2001 From: mono Date: Fri, 16 Oct 2026 13:43:41 +0000 Subject: [PATCH] fix: recover chat replies and organize setup navigation [skip ci] Targeted source-only patch from reviewed local e370485. Adds bounded realtime/catch-up recovery and moves the setup wizard under System. Static checks only; regression tests, full build and runtime verification have not been run. No deployment. Separate hardening-review branch is not merged. Expected tree: 6235478b9ac4b07d0a8a06dce8cc16a603cb7264. --- docs/fixes/chat-realtime-recovery.md | 68 ++++ web/components/support-chat/chat-shell.tsx | 17 +- web/lib/api/im.ts | 6 +- web/lib/navigation.tsx | 11 +- web/lib/realtime-connection.test.mjs | 111 ++++++ web/lib/realtime-connection.ts | 55 ++- web/lib/stores/support-chat-realtime.test.mjs | 321 +++++++++++++++++ web/lib/stores/support-chat.ts | 326 +++++++++++++++--- 8 files changed, 843 insertions(+), 72 deletions(-) create mode 100644 docs/fixes/chat-realtime-recovery.md create mode 100644 web/lib/stores/support-chat-realtime.test.mjs diff --git a/docs/fixes/chat-realtime-recovery.md b/docs/fixes/chat-realtime-recovery.md new file mode 100644 index 0000000..7028e94 --- /dev/null +++ b/docs/fixes/chat-realtime-recovery.md @@ -0,0 +1,68 @@ +# Support chat realtime recovery + +This targeted change is based on canonical Gitea main snapshot +`e57610a27c556e005b9b3b68115477ba1fdd3f1b` (tree +`9cf938c2498ce6b63faaad7cf8b286c1cfe7a5ef`). It does not import the separate +hardening-review branch. + +## Report and source findings + +The reported scenario is opening a new chat from a previous one: sending works, +but replies appear only after a page reload. Source inspection does **not** prove +why the deployed WebSocket fails. The deployed version, reverse proxy, browser +handshake and network have not been observed or tested for this report. + +The source does establish these recovery gaps: + +- A socket in `CONNECTING` has no deadline; subsequent create/select/send paths + see it and return without starting a replacement. +- HTTP fallback checks end after 1.5, 3.5 and 7 seconds, so a later reply is not + retrieved while realtime remains unavailable. +- Reusing an open socket on create/select bypasses the shared status repair path. +- Initial history failure prevents bootstrap/retry from starting realtime. +- A replaced socket's late close can clear the current heartbeat/subscription + bookkeeping. A late close-conversation response can affect a newer selection. + +## Changes + +- Bound connection/authentication attempts to 15 seconds, invalidate abandoned + attempts, and retry using the existing exponential reconnect policy. +- Ignore close events from replaced sockets before touching current state. +- Route new/selected chats through the existing subscription/connection helper. +- Fetch late replies every 5 seconds only while the chat is open, visible, has + an active nonclosed conversation, and realtime is not connected. Never label + HTTP fallback as an online WebSocket. +- Keep latest-message refresh single-flight with a 15-second abort deadline. + Cancel previous-view requests and fence their completion before starting a new + one. Ignore stale identity/view results, stop recovery on connection, hiding, + close or teardown, and recover on focus, visibility or network return. + Background recovery failures remain quiet. The optional API AbortSignal is + backward-compatible with existing history/pagination callers. +- Connect independently of initial HTTP history success and merge history so it + cannot remove a reply already delivered by realtime. +- Restrict a close response to the conversation it actually closed. + +No backend, SDK bundle, credentials, database, deployment configuration or +production data is changed. + +## Validation status + +Completed static checks only: + +- `git diff --check`. +- Isolated strict TypeScript compile of `web/lib/realtime-connection.ts` using an + already installed TypeScript 6.0.3 compiler. +- TypeScript syntax/transpile diagnostics for the four changed production files. +- JavaScript syntax checks for the two focused regression files. + +Added offline mock regressions for delayed replies beyond seven seconds, moving +from a previous chat, resubscription, stuck handshakes, late authentication and +socket close events, hidden/unmounted lifecycle, single-flight fallback, +identity/view changes, abort deadlines, stalled previous-view fetches, history +failures and late close responses. + +**The regression tests have not been executed.** Full project typecheck, lint, +build and browser/runtime validation have not been run; permission for tests, +installation and runtime verification remains pending. No services were started, +real support messages sent, or deployment performed. Do not describe this source +patch as a verified production fix. diff --git a/web/components/support-chat/chat-shell.tsx b/web/components/support-chat/chat-shell.tsx index 4d07c60..5a7b18c 100644 --- a/web/components/support-chat/chat-shell.tsx +++ b/web/components/support-chat/chat-shell.tsx @@ -1,3 +1,4 @@ +// Modified for targeted chat realtime recovery; see docs/fixes/chat-realtime-recovery.md. "use client" import { @@ -139,6 +140,7 @@ function SupportChatPanelContent() { sendAttachment, retry, disconnectSocket, + recoverConnection, markConversationRead, closeConversation, setConversationSatisfaction, @@ -164,6 +166,7 @@ function SupportChatPanelContent() { sendAttachment: state.sendAttachment, retry: state.retry, disconnectSocket: state.disconnectSocket, + recoverConnection: state.recoverConnection, markConversationRead: state.markConversationRead, closeConversation: state.closeConversation, setConversationSatisfaction: state.setConversationSatisfaction, @@ -206,13 +209,9 @@ function SupportChatPanelContent() { useEffect(() => { bootstrap() + }, [isOpen, bootstrap]) - return () => { - if (!isOpen) { - disconnectSocket() - } - } - }, [isOpen, bootstrap, disconnectSocket]) + useEffect(() => () => disconnectSocket(), [disconnectSocket]) useEffect(() => { maybeMarkConversationRead() @@ -220,21 +219,25 @@ function SupportChatPanelContent() { useEffect(() => { const handleVisibilityChange = () => { + recoverConnection() if (document.visibilityState === "visible") { maybeMarkConversationRead() } } const handleFocus = () => { + recoverConnection() maybeMarkConversationRead() } document.addEventListener("visibilitychange", handleVisibilityChange) window.addEventListener("focus", handleFocus) + window.addEventListener("online", recoverConnection) return () => { document.removeEventListener("visibilitychange", handleVisibilityChange) window.removeEventListener("focus", handleFocus) + window.removeEventListener("online", recoverConnection) } - }, [maybeMarkConversationRead]) + }, [maybeMarkConversationRead, recoverConnection]) async function handleSend(content: string) { await handleSendMessage(content) diff --git a/web/lib/api/im.ts b/web/lib/api/im.ts index c2ee877..aeae982 100644 --- a/web/lib/api/im.ts +++ b/web/lib/api/im.ts @@ -1,3 +1,4 @@ +// Modified for targeted chat realtime recovery; see docs/fixes/chat-realtime-recovery.md. import { request } from "@/lib/api/client" import { translateCurrentMessage } from "@/i18n/messages" import { readSupportChatRuntimeConfig } from "@/lib/sdk/runtime-config" @@ -472,11 +473,12 @@ export function fetchImConversationDetail(id: number) { } export function fetchImMessages( - query?: Record + query?: Record, + options?: { signal?: AbortSignal } ) { return request>( `/api/message/list${toQueryString(query)}`, - createRequestOptions() + createRequestOptions({ signal: options?.signal }) ) } diff --git a/web/lib/navigation.tsx b/web/lib/navigation.tsx index 127ac61..a32527d 100644 --- a/web/lib/navigation.tsx +++ b/web/lib/navigation.tsx @@ -1,3 +1,4 @@ +// Modified to move the setup wizard navigation item into the System section. import { BotMessageSquareIcon, BookOpenIcon, @@ -107,11 +108,6 @@ export const dashboardNavSections: DashboardNavSectionConfig[] = [ url: "/dashboard", icon: , }, - { - titleKey: "nav.setupWizard", - url: "/dashboard/setup", - icon: , - }, { titleKey: "nav.conversations", url: "/dashboard/conversations", @@ -268,6 +264,11 @@ export const dashboardNavSections: DashboardNavSectionConfig[] = [ titleKey: "nav.system", icon: , items: [ + { + titleKey: "nav.setupWizard", + url: "/dashboard/setup", + icon: , + }, { titleKey: "nav.users", url: "/dashboard/users", diff --git a/web/lib/realtime-connection.test.mjs b/web/lib/realtime-connection.test.mjs index b11b3dd..604bd94 100644 --- a/web/lib/realtime-connection.test.mjs +++ b/web/lib/realtime-connection.test.mjs @@ -1,3 +1,4 @@ +// Modified for targeted chat realtime recovery; see docs/fixes/chat-realtime-recovery.md. import assert from "node:assert/strict" import test from "node:test" @@ -97,3 +98,113 @@ test("beforeConnect failure triggers onConnectError and disconnect status", asyn assert.ok(statusList.includes("disconnected"), "status should include disconnected") manager.disconnect() }) + +test("a stuck CONNECTING socket times out and retries without reloading", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout", "setInterval"] }) + const sockets = [] + const manager = createRealtimeConnectionManager({ + connectTimeoutMs: 100, + reconnectBaseDelayMs: 10, + createSocket: () => { const socket = new MockWebSocket(); sockets.push(socket); return socket }, + }) + t.after(() => manager.disconnect()) + await manager.connect() + t.mock.timers.tick(100) + assert.equal(sockets[0].readyState, MockWebSocket.CLOSED) + t.mock.timers.tick(10) + assert.equal(sockets.length, 2) + sockets[1].open() + t.mock.timers.tick(100) + assert.equal(manager.getSocket(), sockets[1]) +}) + +test("timed out authentication cannot open a late socket after a newer connection", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout", "setInterval"] }) + let resolveOld, attempts = 0, created = 0 + const old = new Promise((resolve) => { resolveOld = resolve }) + const manager = createRealtimeConnectionManager({ + connectTimeoutMs: 100, + reconnectBaseDelayMs: 10, + beforeConnect: () => ++attempts === 1 ? old : Promise.resolve(), + createSocket: () => { created++; return new MockWebSocket() }, + }) + t.after(() => manager.disconnect()) + const first = manager.connect() + t.mock.timers.tick(100) + t.mock.timers.tick(10) + await Promise.resolve() + resolveOld() + await first + assert.equal(created, 1) +}) + +test("late replaced socket close cannot clear the current heartbeat or subscription", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout", "setInterval"] }) + let closed = 0, pings = 0 + const sockets = [] + const manager = createRealtimeConnectionManager({ + pingIntervalMs: 10, + createSocket: () => { + const socket = new MockWebSocket() + socket.close = () => { socket.readyState = MockWebSocket.CLOSED } + socket.send = () => { pings++ } + sockets.push(socket) + return socket + }, + onClose: () => { closed++ }, + }) + t.after(() => manager.disconnect()) + await manager.connect() + sockets[0].open() + await manager.connect() + sockets[1].open() + sockets[0].emit("close", {}) + t.mock.timers.tick(10) + assert.equal(manager.getSocket(), sockets[1]) + assert.equal(closed, 0) + assert.equal(pings, 1) +}) + +test("an error without a close event detaches a connecting socket and retries", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout", "setInterval"] }) + const sockets = [] + const manager = createRealtimeConnectionManager({ + connectTimeoutMs: 100, + reconnectBaseDelayMs: 10, + createSocket: () => { + const socket = new MockWebSocket() + socket.close = () => { socket.readyState = MockWebSocket.CLOSING } + sockets.push(socket) + return socket + }, + }) + t.after(() => manager.disconnect()) + await manager.connect() + sockets[0].emit("error", {}) + assert.equal(manager.getSocket(), null) + t.mock.timers.tick(10) + assert.equal(sockets.length, 2) + sockets[1].open() + t.mock.timers.tick(100) + assert.equal(manager.getSocket(), sockets[1]) +}) + +test("becoming ineligible during authentication clears the opening deadline", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout", "setInterval"] }) + let resolveAuth, allowed = true, errors = 0 + const auth = new Promise((resolve) => { resolveAuth = resolve }) + const manager = createRealtimeConnectionManager({ + connectTimeoutMs: 100, + canReconnect: () => allowed, + beforeConnect: () => auth, + createSocket: () => assert.fail("ineligible connection must not create a socket"), + onConnectError: () => { errors++ }, + }) + t.after(() => manager.disconnect()) + const connecting = manager.connect() + allowed = false + resolveAuth() + await connecting + t.mock.timers.tick(200) + assert.equal(errors, 0) +}) diff --git a/web/lib/realtime-connection.ts b/web/lib/realtime-connection.ts index 61205fb..79cb015 100644 --- a/web/lib/realtime-connection.ts +++ b/web/lib/realtime-connection.ts @@ -1,3 +1,4 @@ +// Modified for targeted chat realtime recovery; see docs/fixes/chat-realtime-recovery.md. "use client" export type RealtimeConnectionStatus = "connecting" | "connected" | "disconnected" @@ -14,6 +15,7 @@ type RealtimeConnectionManagerOptions = { onError?: (event: Event, socket: WebSocket) => void onConnectError?: (error: unknown) => void buildPingMessage?: () => string + connectTimeoutMs?: number pingIntervalMs?: number reconnectBaseDelayMs?: number reconnectMaxDelayMs?: number @@ -24,6 +26,7 @@ type DisconnectOptions = { updateStatus?: boolean } +const DEFAULT_CONNECT_TIMEOUT_MS = 15000 const DEFAULT_PING_INTERVAL_MS = 20000 const DEFAULT_RECONNECT_BASE_DELAY_MS = 2000 const DEFAULT_RECONNECT_MAX_DELAY_MS = 30000 @@ -32,12 +35,20 @@ export function createRealtimeConnectionManager( options: RealtimeConnectionManagerOptions ) { let socket: WebSocket | null = null + let connectTimer: ReturnType | null = null let reconnectTimer: ReturnType | null = null let pingTimer: ReturnType | null = null let reconnectAttempt = 0 let reconnectEnabled = false let connectEpoch = 0 + const clearConnectTimer = () => { + if (connectTimer !== null) { + globalThis.clearTimeout(connectTimer) + connectTimer = null + } + } + const clearReconnectTimer = () => { if (reconnectTimer !== null) { globalThis.clearTimeout(reconnectTimer) @@ -53,6 +64,7 @@ export function createRealtimeConnectionManager( } const clearTimers = () => { + clearConnectTimer() clearReconnectTimer() clearPingTimer() } @@ -97,6 +109,16 @@ export function createRealtimeConnectionManager( options.onStatusChange?.("connecting") const currentEpoch = ++connectEpoch + connectTimer = globalThis.setTimeout(() => { + if (currentEpoch !== connectEpoch) { + return + } + disconnect({ reconnect: true, updateStatus: false }) + options.onStatusChange?.("disconnected") + // Callers already localize their fallback for non-Error failures. + options.onConnectError?.({ code: "realtime_connect_timeout" }) + scheduleReconnect() + }, options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS) if (options.beforeConnect) { try { @@ -105,6 +127,7 @@ export function createRealtimeConnectionManager( if (currentEpoch !== connectEpoch || !reconnectEnabled) { return } + clearConnectTimer() options.onStatusChange?.("disconnected") options.onConnectError?.(error) scheduleReconnect() @@ -112,11 +135,12 @@ export function createRealtimeConnectionManager( } } - if ( - currentEpoch !== connectEpoch || - !reconnectEnabled || - !(options.canReconnect?.() ?? true) - ) { + if (currentEpoch !== connectEpoch || !reconnectEnabled) { + return + } + if (!(options.canReconnect?.() ?? true)) { + clearConnectTimer() + options.onStatusChange?.("disconnected") return } @@ -124,6 +148,7 @@ export function createRealtimeConnectionManager( try { nextSocket = options.createSocket() } catch (error) { + clearConnectTimer() options.onStatusChange?.("disconnected") options.onConnectError?.(error) scheduleReconnect() @@ -136,6 +161,7 @@ export function createRealtimeConnectionManager( if (socket !== nextSocket) { return } + clearConnectTimer() clearReconnectTimer() clearPingTimer() reconnectAttempt = 0 @@ -157,15 +183,15 @@ export function createRealtimeConnectionManager( }) nextSocket.addEventListener("close", (event) => { - clearPingTimer() - const isCurrentSocket = socket === nextSocket - if (isCurrentSocket) { - setSocket(null) - } - options.onClose?.(event, nextSocket) - if (!isCurrentSocket) { + // A replaced socket may finish closing after the next one has opened. + // It must not clear that connection's heartbeat or subscription state. + if (socket !== nextSocket) { return } + clearConnectTimer() + clearPingTimer() + setSocket(null) + options.onClose?.(event, nextSocket) if (canReconnect()) { scheduleReconnect() } else { @@ -178,12 +204,17 @@ export function createRealtimeConnectionManager( return } options.onError?.(event, nextSocket) + // An error need not be followed promptly by close. Detach it now so + // callers cannot keep reusing a failed OPEN/CONNECTING socket. + disconnect({ reconnect: true, updateStatus: false }) options.onStatusChange?.("disconnected") scheduleReconnect() }) } const disconnect = (disconnectOptions?: DisconnectOptions) => { + // A timed-out or abandoned authentication wait must not open a late socket. + connectEpoch += 1 reconnectEnabled = disconnectOptions?.reconnect ?? false clearTimers() if (!reconnectEnabled) { diff --git a/web/lib/stores/support-chat-realtime.test.mjs b/web/lib/stores/support-chat-realtime.test.mjs new file mode 100644 index 0000000..9b7de03 --- /dev/null +++ b/web/lib/stores/support-chat-realtime.test.mjs @@ -0,0 +1,321 @@ +import assert from "node:assert/strict" +import { readFileSync } from "node:fs" +import test from "node:test" +import vm from "node:vm" +import ts from "typescript" + +const deferred = () => { + let resolve + const promise = new Promise((done) => { resolve = done }) + return { promise, resolve } +} +const conversation = (id) => ({ id, status: 3, customerUnreadCount: 0 }) +const message = (id, conversationId) => ({ id, conversationId, senderType: "agent", + senderId: 1, messageType: "html", content: "Reply", sendStatus: 1, + customerRead: false, agentRead: false }) +const page = (results = []) => ({ results, hasMore: false }) +const flush = async () => { for (let i = 0; i < 20; i++) await Promise.resolve() } + +// Run actual store and merge helpers with fake transport, APIs, visibility and +// clock. These regressions never open a network socket or touch browser data. +function harness(overrides = {}) { + let state, realtime, now = 0, nextTimer = 0, connects = 0, disconnects = 0 + const timers = new Map(), modules = new Map(), sent = [] + const config = { channelId: "fixture", externalId: "visitor-1" } + const document = { visibilityState: "visible" } + const api = { + ensureCustomerSession: async () => ({}), + fetchImWidgetConfig: async () => ({}), + fetchImConversationList: async () => page([conversation(1)]), + createOrMatchImConversation: async () => conversation(1), + createImConversation: async () => conversation(2), + fetchImMessages: async () => page(), + closeImConversation: async () => {}, + applyCustomerSessionRefresh: () => {}, + ...overrides, + } + const mocks = { + zustand: { create: (init) => { + const set = (patch) => { state = { ...state, ...(typeof patch === "function" ? patch(state) : patch) } } + state = init(set, () => state) + return { getState: () => state, setState: set } + } }, + "@/lib/api/im": api, + "@/lib/im-realtime": { createImRealtimeConnection: () => {} }, + "@/lib/im-message": { summarizeIMMessage: (item) => item.content, parseMessageAssetPayload: () => null }, + "@/lib/utils": { generateUUID: () => "fixture" }, + "@/i18n/messages": { translateCurrentMessage: (key) => key }, + "@/lib/sdk/runtime-config": { + readSupportChatRuntimeConfig: () => config, + setSupportChatRuntimeConfig: (next) => Object.assign(config, next), + }, + "@/lib/realtime-connection": { createRealtimeConnectionManager: (options) => { + realtime = options + return { + connect: async () => { connects++; options.onStatusChange("connecting") }, + disconnect: () => { disconnects++; options.onSocketChange(null); options.onStatusChange("disconnected") }, + } + } }, + } + function load(name) { + if (mocks[name]) return mocks[name] + if (modules.has(name)) return modules.get(name) + const url = name.startsWith("@/") + ? new URL(`../../${name.slice(2)}.ts`, import.meta.url) + : new URL(name, import.meta.url) + const source = readFileSync(url, "utf8") + const compiled = ts.transpileModule(source, { compilerOptions: { + target: ts.ScriptTarget.ES2022, module: ts.ModuleKind.CommonJS, + } }) + const module = { exports: {} } + modules.set(name, module.exports) + vm.runInNewContext(compiled.outputText, { module, exports: module.exports, + require: load, console, document, AbortController, WebSocket: { OPEN: 1, CONNECTING: 0 }, + setTimeout: (callback, delay) => { + const id = ++nextTimer + timers.set(id, { callback, at: now + delay }) + return id + }, + clearTimeout: (id) => timers.delete(id), + }) + return module.exports + } + const store = load("./support-chat.ts").useSupportChatStore + store.setState({ initialized: true, isOpen: true, isVisible: true, + conversations: [conversation(1)], conversation: conversation(1), activeConversationId: 1 }) + return { store, api, config, document, timers, sent, + get connects() { return connects }, get disconnects() { return disconnects }, + status: (status) => realtime.onStatusChange(status), + emit: (type, data) => realtime.onMessage({ data: JSON.stringify({ type, data }) }), + open: () => { + const socket = { readyState: 1, send: (value) => sent.push(JSON.parse(value)) } + realtime.onSocketChange(socket) + realtime.onStatusChange("connected") + realtime.onOpen(socket) + return socket + }, + advance: async (ms) => { + const until = now + ms + for (;;) { + const next = [...timers].filter(([, timer]) => timer.at <= until) + .sort((a, b) => a[1].at - b[1].at)[0] + if (!next) break + now = next[1].at + timers.delete(next[0]) + next[1].callback() + await flush() + } + now = until + await flush() + }, + } +} + +test("new chat after previous one receives a reply later than seven seconds while WS is unavailable", async () => { + let reply = false + const h = harness({ fetchImMessages: async ({ conversationId }) => page(reply ? [message(20, conversationId)] : []) }) + await h.store.getState().startNewConversation() + await h.advance(8000) + reply = true + await h.advance(2000) + assert.equal(h.store.getState().messages[0].id, 20) + assert.notEqual(h.store.getState().status, "connected", "HTTP fallback must not claim WS is online") +}) + +test("new chat reuses an open socket, resubscribes and repairs a stale status", async () => { + const h = harness() + await h.store.getState().retry() + const socket = h.open() + await flush() + h.store.setState({ status: "disconnected" }) + await h.store.getState().startNewConversation() + assert.equal(h.store.getState().socket, socket) + assert.equal(h.store.getState().status, "connected") + assert.equal(h.sent.at(-2).type, "unsubscribe") + assert.equal(h.sent.at(-1).topics[0], "conversation:2") +}) + +test("fallback stops on connection, hiding, closing and unmount, and resumes on visibility", async () => { + let reads = 0 + const h = harness({ fetchImMessages: async () => { reads++; return page() } }) + await h.store.getState().retry() + h.document.visibilityState = "hidden" + h.store.getState().recoverConnection() + const hiddenReads = reads + await h.advance(15000) + assert.equal(reads, hiddenReads) + h.document.visibilityState = "visible" + h.store.getState().recoverConnection() + await h.advance(5000) + assert.ok(reads > hiddenReads) + h.store.getState().setIsVisible(false) + const minimizedReads = reads + await h.advance(15000) + assert.equal(reads, minimizedReads) + h.store.getState().setIsVisible(true) + h.open() + await flush() + const connectedReads = reads + await h.advance(15000) + assert.equal(reads, connectedReads) + h.status("connecting") + h.store.getState().disconnectSocket() + const unmountedReads = reads + await h.advance(15000) + assert.equal(reads, unmountedReads) + assert.equal(h.timers.size, 0) +}) + +test("fallback has one in-flight latest refresh and cannot restart after unmount", async () => { + const pending = deferred() + const h = harness() + await h.store.getState().retry() + let reads = 0 + h.api.fetchImMessages = () => { reads++; return pending.promise } + await h.advance(5000) + void h.store.getState().syncLatestMessages() + await h.advance(15000) + assert.equal(reads, 1) + h.store.getState().disconnectSocket() + pending.resolve(page([message(99, 1)])) + await flush() + await h.advance(15000) + assert.equal(reads, 1) + assert.equal(h.store.getState().messages.length, 0) + assert.equal(h.timers.size, 0) +}) + +for (const change of ["conversation", "identity", "round-trip selection"]) { + test(`late fallback cannot paint into a changed ${change}`, async () => { + const pending = deferred() + const h = harness() + await h.store.getState().retry() + h.api.fetchImMessages = () => pending.promise + await h.advance(5000) + h.api.fetchImMessages = async () => page() + if (change === "identity") h.config.externalId = "visitor-2" + else { + await h.store.getState().startNewConversation() + if (change === "round-trip selection") await h.store.getState().selectConversation(1) + } + pending.resolve(page([message(99, 1)])) + await flush() + assert.equal(h.store.getState().messages.length, 0) + h.store.getState().disconnectSocket() + }) +} + +test("bootstrap and retry attempt realtime even when loading history fails", async () => { + const h = harness({ fetchImMessages: async () => { throw new Error("history unavailable") } }) + h.store.setState({ initialized: false, conversation: null, activeConversationId: 0 }) + h.store.getState().bootstrap() + await flush() + assert.ok(h.connects > 0) + const before = h.connects + await h.store.getState().retry() + assert.ok(h.connects > before) + assert.equal(h.store.getState().status, "connecting") + h.store.getState().disconnectSocket() +}) + +test("loading initial history does not erase a newer realtime reply", async () => { + const pending = deferred() + const h = harness() + await h.store.getState().retry() + h.api.fetchImMessages = () => pending.promise + const loading = h.store.getState().refreshMessages() + h.emit("message.created", { conversationId: 1, message: message(20, 1) }) + pending.resolve(page([message(10, 1)])) + await loading + assert.deepEqual(Array.from(h.store.getState().messages, (item) => item.id), [10, 20]) +}) + +test("late close for previous request does not close or disconnect the new chat", async () => { + const pending = deferred() + const h = harness({ closeImConversation: () => pending.promise }) + const closing = h.store.getState().closeConversation() + await h.store.getState().startNewConversation() + const disconnects = h.disconnects + pending.resolve() + await closing + assert.equal(h.store.getState().conversation.id, 2) + assert.equal(h.store.getState().conversation.status, 3) + assert.equal(h.disconnects, disconnects) + assert.equal(h.store.getState().conversations.find((item) => item.id === 1).status, 4) +}) + +test("a successful pending send cannot reconnect or start polling after unmount", async () => { + const pending = deferred() + const h = harness({ sendImMessage: () => pending.promise }) + const sending = h.store.getState().handleSendMessage("hello") + h.store.getState().disconnectSocket() + pending.resolve({ ...message(10, 1), senderType: "customer" }) + await sending + await h.advance(15000) + assert.equal(h.connects, 0) + assert.equal(h.timers.size, 0) +}) + +test("closing the active conversation stops degraded polling", async () => { + let reads = 0 + const h = harness({ fetchImMessages: async () => { reads++; return page() } }) + await h.store.getState().retry() + await h.store.getState().closeConversation() + const before = reads + await h.advance(15000) + assert.equal(reads, before) + assert.equal(h.timers.size, 0) +}) + +test("connection opening during fallback queues one fresh catch-up after its older snapshot", async () => { + const pending = deferred() + const h = harness() + await h.store.getState().retry() + h.api.fetchImMessages = () => pending.promise + await h.advance(5000) + h.open() + h.api.fetchImMessages = async () => page([message(50, 1)]) + pending.resolve(page()) + await flush() + assert.equal(h.store.getState().messages[0].id, 50) + assert.equal(h.store.getState().status, "connected") + assert.equal(h.timers.size, 0) +}) + +test("switching aborts a stalled previous-view fallback without blocking new replies", async () => { + const pending = deferred() + let previousSignal + const h = harness() + await h.store.getState().retry() + h.api.fetchImMessages = (_query, options) => { previousSignal = options?.signal; return pending.promise } + await h.advance(5000) + let replied = false + h.api.fetchImMessages = async ({ conversationId }) => page(replied ? [message(70, conversationId)] : []) + await h.store.getState().startNewConversation() + assert.equal(previousSignal.aborted, true) + replied = true + await h.advance(5000) + assert.equal(h.store.getState().messages[0].conversationId, 2) + pending.resolve(page([message(99, 1)])) + await flush() + assert.deepEqual(Array.from(h.store.getState().messages, (item) => item.id), [70]) +}) + +test("a stalled fallback is aborted at its deadline and can retry", async () => { + const h = harness() + await h.store.getState().retry() + let signal, reads = 0 + h.api.fetchImMessages = (_query, options) => { + reads++ + signal = options.signal + return new Promise((_resolve, reject) => signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true })) + } + await h.advance(5000) + await h.advance(15000) + assert.equal(signal.aborted, true) + h.api.fetchImMessages = async () => { reads++; return page([message(80, 1)]) } + await h.advance(5000) + assert.equal(reads, 2) + assert.equal(h.store.getState().messages[0].id, 80) + assert.equal(h.store.getState().error, "") +}) diff --git a/web/lib/stores/support-chat.ts b/web/lib/stores/support-chat.ts index 4c3071d..64aeced 100644 --- a/web/lib/stores/support-chat.ts +++ b/web/lib/stores/support-chat.ts @@ -1,3 +1,4 @@ +// Modified for targeted chat realtime recovery; see docs/fixes/chat-realtime-recovery.md. "use client" import { create } from "zustand" @@ -60,6 +61,8 @@ type ChatStatus = "connecting" | "connected" | "disconnected" const DEFAULT_PAGE_LIMIT = 50 const CONVERSATION_LIST_PAGE_LIMIT = 50 +const DISCONNECTED_MESSAGE_SYNC_INTERVAL_MS = 5000 +const LATEST_MESSAGE_SYNC_TIMEOUT_MS = 15000 /** * The single writer for the request list. `conversation` is never assigned @@ -196,12 +199,13 @@ export type SupportChatStore = { setIsOpen: (isOpen: boolean) => void setIsVisible: (isVisible: boolean) => void bootstrap: () => void + recoverConnection: () => void disconnectSocket: () => void refreshConversations: () => Promise selectConversation: (conversationId: number) => Promise startNewConversation: (subject?: string) => Promise refreshMessages: () => Promise - syncLatestMessages: () => Promise + syncLatestMessages: (quiet?: boolean) => Promise loadOlderMessages: () => Promise markConversationRead: () => Promise handleSendMessage: (content: string) => Promise @@ -219,6 +223,21 @@ function t(key: string) { return translateCurrentMessage(key) } +// An in-flight fallback must never carry a previous visitor's history into a +// replacement host session. Keep this comparison local; never log its contents. +function messageIdentityScope() { + const config = readSupportChatRuntimeConfig() + return JSON.stringify([ + config.apiBaseUrl || config.baseUrl, + config.channelId, + config.externalId, + config.userToken, + config.externalIdSignature, + config.externalIdSignedAt, + config.requireUserToken, + ]) +} + function buildSelectConversationPatch( state: { conversations: ImConversation[] @@ -244,6 +263,74 @@ function buildSelectConversationPatch( export const useSupportChatStore = create((set, get) => { let subscribedConversationId: number | null = null + let recoveryTimer: ReturnType | null = null + let recoveryInFlight: object | null = null + let latestSync: { + controller: AbortController + timer: ReturnType + quiet: boolean + } | null = null + let latestSyncPending = false + let recoveryEnabled = false + let disposed = false + let viewEpoch = 0 + let socketIdentity = "" + + const cancelLatestSync = () => { + const pending = latestSync + latestSync = null + latestSyncPending = false + if (pending) { + clearTimeout(pending.timer) + pending.controller.abort() + } + } + + const needsMessageRecovery = () => { + const state = get() + return Boolean( + !disposed && + recoveryEnabled && + state.isOpen && + state.isVisible && + (typeof document === "undefined" || document.visibilityState === "visible") && + state.conversation?.id && + state.conversation.status !== CLOSED_IM_CONVERSATION_STATUS && + state.status !== "connected" + ) + } + + const clearMessageRecovery = () => { + recoveryInFlight = null + if (recoveryTimer !== null) { + clearTimeout(recoveryTimer) + recoveryTimer = null + } + } + + const scheduleMessageRecovery = () => { + if (!needsMessageRecovery() || recoveryTimer !== null || recoveryInFlight) { + return + } + recoveryTimer = setTimeout(async () => { + recoveryTimer = null + if (!needsMessageRecovery()) { + return + } + const run = {} + recoveryInFlight = run + try { + // HTTP delivery still works when a proxy/network cannot upgrade WS. + // Keep fetching late replies, not just the first seven seconds. + await get().syncLatestMessages(true) + } finally { + if (recoveryInFlight === run) { + recoveryInFlight = null + scheduleMessageRecovery() + } + } + }, DISCONNECTED_MESSAGE_SYNC_INTERVAL_MS) + } const sendSubscription = ( socket: WebSocket | null, @@ -275,10 +362,19 @@ export const useSupportChatStore = create((set, get) => { conversationId: number, sentMessageId: number ) => { + if (disposed) { + return + } + const epoch = viewEpoch + const identity = messageIdentityScope() for (const delay of [1500, 3500, 7000]) { setTimeout(() => { const current = get() - if (current.conversation?.id === conversationId) { + if ( + epoch === viewEpoch && + identity === messageIdentityScope() && + current.conversation?.id === conversationId + ) { if (current.isOpen && (!current.socket || current.socket.readyState !== WebSocket.OPEN)) { connectSocket() } @@ -299,6 +395,11 @@ export const useSupportChatStore = create((set, get) => { if (get().isOpen || status === "disconnected") { set({ status }) } + if (status === "connected") { + clearMessageRecovery() + } else { + scheduleMessageRecovery() + } }, onSocketChange: (socket) => { set({ socket }) @@ -323,6 +424,9 @@ export const useSupportChatStore = create((set, get) => { subscribedConversationId = null }, onMessage: (messageEvent) => { + if (socketIdentity !== messageIdentityScope()) { + return + } let event: ImRealtimeEnvelope try { event = JSON.parse(messageEvent.data) as ImRealtimeEnvelope @@ -414,6 +518,13 @@ export const useSupportChatStore = create((set, get) => { if (eventConversationId !== activeConversationId) { return } + if ( + event.type === "conversation.closed" || + payload?.status === CLOSED_IM_CONVERSATION_STATUS + ) { + clearMessageRecovery() + cancelLatestSync() + } set((state) => ({ messages: markConversationReadMessages(state.messages, payload), ...(event.type === "conversation.closed" || @@ -426,19 +537,32 @@ export const useSupportChatStore = create((set, get) => { }) const closeSocket = (options?: { reconnect?: boolean }) => { + recoveryEnabled = false + cancelLatestSync() + viewEpoch += 1 subscribedConversationId = null realtime.disconnect({ reconnect: options?.reconnect ?? false, updateStatus: true, }) + clearMessageRecovery() } const connectSocket = () => { + if (disposed) { + return + } const activeConversationId = get().conversation?.id || get().activeConversationId if (!activeConversationId) { return } + const identity = messageIdentityScope() + if (socketIdentity && socketIdentity !== identity) { + closeSocket({ reconnect: false }) + } + socketIdentity = identity + recoveryEnabled = true const socket = get().socket if (socket && socket.readyState === WebSocket.OPEN) { if (activeConversationId !== subscribedConversationId) { @@ -448,9 +572,11 @@ export const useSupportChatStore = create((set, get) => { if (get().status !== "connected") { set({ status: "connected" }) } + clearMessageRecovery() return } if (socket && socket.readyState === WebSocket.CONNECTING) { + scheduleMessageRecovery() return } void realtime.connect() @@ -489,10 +615,21 @@ export const useSupportChatStore = create((set, get) => { setIsVisible: (isVisible: boolean) => { set({ isVisible }) + if (isVisible) { + scheduleMessageRecovery() + } else { + clearMessageRecovery() + if (latestSync?.quiet) cancelLatestSync() + } }, bootstrap: () => { + disposed = false const token = ++bootstrapToken + viewEpoch += 1 + clearMessageRecovery() + cancelLatestSync() + set({ creatingConversation: false }) if (!get().isOpen) { closeSocket({ reconnect: false }) @@ -576,12 +713,10 @@ export const useSupportChatStore = create((set, get) => { })) } - await get().refreshMessages() - if (bootstrapToken !== token || !get().isOpen) { - return - } - + // History and realtime are independent transports. A temporary + // history failure must not prevent the newly created chat connecting. connectSocket() + await get().refreshMessages().catch(() => undefined) } catch (error) { if (bootstrapToken !== token || !get().isOpen) { return @@ -596,7 +731,24 @@ export const useSupportChatStore = create((set, get) => { void activateChat() }, + recoverConnection: () => { + if ( + !get().isOpen || + !get().isVisible || + (typeof document !== "undefined" && document.visibilityState !== "visible") + ) { + clearMessageRecovery() + if (latestSync?.quiet) cancelLatestSync() + return + } + if (recoveryEnabled) { + connectSocket() + } + }, + disconnectSocket: () => { + disposed = true + bootstrapToken += 1 closeSocket({ reconnect: false }) }, @@ -618,12 +770,14 @@ export const useSupportChatStore = create((set, get) => { }, selectConversation: async (conversationId: number) => { - const prevId = get().conversation?.id || get().activeConversationId const patch = buildSelectConversationPatch(get(), conversationId) if (!patch) { return } + viewEpoch += 1 + clearMessageRecovery() + cancelLatestSync() // The patch clears messages and re-derives `conversation` together with // the active id; otherwise refreshMessages would still read the previous // request and show its history under the new header. @@ -632,11 +786,7 @@ export const useSupportChatStore = create((set, get) => { activeAgentActivity: null, }) - const socket = get().socket - if (socket && socket.readyState === WebSocket.OPEN) { - sendSubscription(socket, prevId, conversationId) - subscribedConversationId = conversationId - } else if (get().isOpen) { + if (get().isOpen) { connectSocket() } @@ -653,9 +803,16 @@ export const useSupportChatStore = create((set, get) => { } set({ creatingConversation: true, error: "", activeAgentActivity: null }) + const token = bootstrapToken + const identity = messageIdentityScope() try { - const prevId = get().conversation?.id || get().activeConversationId const created = await createImConversation(subject) + if (token !== bootstrapToken || identity !== messageIdentityScope()) { + return + } + viewEpoch += 1 + clearMessageRecovery() + cancelLatestSync() set((state) => ({ creatingConversation: false, messages: [], @@ -669,18 +826,19 @@ export const useSupportChatStore = create((set, get) => { ), })) - const socket = get().socket - if (socket && socket.readyState === WebSocket.OPEN) { - sendSubscription(socket, prevId, created.id) - subscribedConversationId = created.id - } else if (get().isOpen) { + if (get().isOpen) { connectSocket() } - await get().refreshMessages() + // The conversation is already created. A history outage should not + // make the dialog offer to create it again; realtime/fallback can catch up. + await get().refreshMessages().catch(() => undefined) const latestMsgId = get().messages.at(-1)?.id ?? 0 scheduleMessageReplyCheck(created.id, latestMsgId) } catch (error) { + if (token !== bootstrapToken || identity !== messageIdentityScope()) { + return + } set({ creatingConversation: false, error: @@ -693,6 +851,8 @@ export const useSupportChatStore = create((set, get) => { }, refreshMessages: async () => { + const epoch = viewEpoch + const identity = messageIdentityScope() const conversationId = get().conversation?.id if (!conversationId) { return @@ -703,17 +863,35 @@ export const useSupportChatStore = create((set, get) => { conversationId, limit: DEFAULT_PAGE_LIMIT, }) - if (get().conversation?.id !== conversationId) { + if ( + epoch !== viewEpoch || + identity !== messageIdentityScope() || + get().conversation?.id !== conversationId + ) { return } const results = ensureMessageList(page.results) - set({ - messages: results, - messagesCursor: cursorFromLoadedImMessages(results) || page.cursor || "", - messagesHasMore: Boolean(page.hasMore) || results.length >= DEFAULT_PAGE_LIMIT, + set((state) => { + // The socket may already have delivered a reply newer than the HTTP + // snapshot; history loading must not erase that realtime delivery. + const merged = mergeImMessagesByIdAsc(state.messages, results) + return { + messages: merged, + messagesCursor: cursorFromLoadedImMessages(merged) || page.cursor || "", + messagesHasMore: hasMoreAfterLatestImMessageMerge({ + previousMessages: state.messages, + previousHasMore: state.messagesHasMore, + merged, + apiHasMore: Boolean(page.hasMore) || results.length >= DEFAULT_PAGE_LIMIT, + }), + } }) } catch (error) { - if (get().conversation?.id !== conversationId) { + if ( + epoch !== viewEpoch || + identity !== messageIdentityScope() || + get().conversation?.id !== conversationId + ) { return } set({ @@ -723,19 +901,41 @@ export const useSupportChatStore = create((set, get) => { } }, - syncLatestMessages: async () => { + syncLatestMessages: async (quiet = false) => { + const epoch = viewEpoch + const identity = messageIdentityScope() const conversationId = get().conversation?.id if (!conversationId) { return } + if (latestSync) { + // A connection opening during fallback still needs a fresh catch-up + // after that older HTTP snapshot, without starting parallel requests. + latestSyncPending ||= !quiet + return + } + const controller = new AbortController() + const operation = { + controller, + timer: setTimeout(() => controller.abort(), LATEST_MESSAGE_SYNC_TIMEOUT_MS), + quiet, + } + latestSync = operation try { - const page = await fetchImMessages({ - conversationId, - limit: DEFAULT_PAGE_LIMIT, - }) + const page = await fetchImMessages( + { conversationId, limit: DEFAULT_PAGE_LIMIT }, + { signal: controller.signal } + ) const batch = ensureMessageList(page.results) - if (batch.length === 0) { + if ( + controller.signal.aborted || + latestSync !== operation || + epoch !== viewEpoch || + identity !== messageIdentityScope() || + get().conversation?.id !== conversationId || + batch.length === 0 + ) { return } set((state) => { @@ -752,9 +952,31 @@ export const useSupportChatStore = create((set, get) => { } }) } catch (error) { - set({ - error: error instanceof Error ? error.message : t("supportChat.syncMessagesFailed"), - }) + if ( + controller.signal.aborted || + latestSync !== operation || + epoch !== viewEpoch || + identity !== messageIdentityScope() || + get().conversation?.id !== conversationId + ) { + return + } + if (!quiet) { + set({ + error: error instanceof Error ? error.message : t("supportChat.syncMessagesFailed"), + }) + } + } finally { + clearTimeout(operation.timer) + if (latestSync === operation) { + latestSync = null + if (latestSyncPending) { + latestSyncPending = false + if (!disposed && get().isOpen) { + void get().syncLatestMessages() + } + } + } } }, @@ -950,6 +1172,7 @@ export const useSupportChatStore = create((set, get) => { }, closeConversation: async () => { + const identity = messageIdentityScope() const conversationId = get().conversation?.id if (!conversationId) { return @@ -958,17 +1181,29 @@ export const useSupportChatStore = create((set, get) => { set({ error: "", closingConversation: true }) try { await closeImConversation(conversationId) - closeSocket({ reconnect: false }) + if (identity !== messageIdentityScope()) { + return + } + const isActive = get().conversation?.id === conversationId + if (isActive) { + closeSocket({ reconnect: false }) + } set((state) => ({ closingConversation: false, - status: "disconnected", - activeAgentActivity: null, - ...withActiveConversationPatch(state, { - status: CLOSED_IM_CONVERSATION_STATUS, - customerUnreadCount: 0, - }), + ...(isActive ? { activeAgentActivity: null } : {}), + ...withConversations( + state, + patchConversationList(state.conversations, { + id: conversationId, + status: CLOSED_IM_CONVERSATION_STATUS, + customerUnreadCount: 0, + }) + ), })) } catch (error) { + if (identity !== messageIdentityScope()) { + return + } set({ closingConversation: false, error: error instanceof Error ? error.message : t("supportChat.closeConversationFailed"), @@ -991,14 +1226,13 @@ export const useSupportChatStore = create((set, get) => { } set({ error: "", status: "connecting" }) + if (get().isOpen) { + connectSocket() + } try { await get().refreshMessages() - if (get().isOpen) { - connectSocket() - } } catch (error) { set({ - status: "disconnected", error: error instanceof Error ? error.message : t("supportChat.refreshFailed"), }) }