From fe85af0cd747943fc9c51793837e29e17420347f Mon Sep 17 00:00:00 2001 From: chenhang Date: Fri, 26 Jun 2026 11:11:43 +0800 Subject: [PATCH] fix(chat): queue outgoing messages --- src/app/chat/chat-screen.tsx | 2 +- .../__tests__/chat-machine.helpers.test.ts | 1 + .../chat-machine.transitions.test.ts | 46 +++++++ src/stores/chat/chat-events.ts | 5 + src/stores/chat/chat-machine.actors.ts | 94 ++++++++++++-- src/stores/chat/chat-machine.helpers.ts | 29 ++++- src/stores/chat/chat-machine.ts | 121 ++++++++++++------ src/stores/chat/chat-state.ts | 2 + 8 files changed, 240 insertions(+), 60 deletions(-) diff --git a/src/app/chat/chat-screen.tsx b/src/app/chat/chat-screen.tsx index a85b2ef7..b74cd6ae 100644 --- a/src/app/chat/chat-screen.tsx +++ b/src/app/chat/chat-screen.tsx @@ -202,7 +202,7 @@ export function ChatScreen() { onUnlock={handleMessageLimitUnlock} /> ) : ( - + )} diff --git a/src/stores/chat/__tests__/chat-machine.helpers.test.ts b/src/stores/chat/__tests__/chat-machine.helpers.test.ts index 39803af7..9cb1ae19 100644 --- a/src/stores/chat/__tests__/chat-machine.helpers.test.ts +++ b/src/stores/chat/__tests__/chat-machine.helpers.test.ts @@ -34,6 +34,7 @@ function makeChatState(overrides: Partial = {}): ChatState { return { messages: [], isReplyingAI: true, + pendingReplyCount: 1, wsConnected: false, upgradePromptVisible: false, upgradeReason: null, diff --git a/src/stores/chat/__tests__/chat-machine.transitions.test.ts b/src/stores/chat/__tests__/chat-machine.transitions.test.ts index 0b423782..fdd56f84 100644 --- a/src/stores/chat/__tests__/chat-machine.transitions.test.ts +++ b/src/stores/chat/__tests__/chat-machine.transitions.test.ts @@ -7,6 +7,7 @@ import { type UiMessage, } from "@/data/dto/chat"; import { chatMachine } from "@/stores/chat/chat-machine"; +import type { ChatEvent } from "@/stores/chat/chat-events"; interface LoadHistoryOutput { messages: UiMessage[]; @@ -83,6 +84,23 @@ function createTestChatMachine( reply: null, })), sendMessageWs: fromPromise(async () => {}), + httpMessageQueue: fromCallback(({ receive, sendBack }) => { + receive((event) => { + if (event.type !== "ChatSendMessage") return; + sendBack({ + type: "ChatQueuedHttpDone", + output: { + response: makeChatSendResponse(), + reply: null, + }, + }); + }); + return () => undefined; + }), + wsPreferredMessageQueue: fromCallback(({ receive }) => { + receive(() => undefined); + return () => undefined; + }), unlockPrivateMessage: fromPromise< UnlockPrivateOutput, { messageId: string } @@ -220,4 +238,32 @@ describe("chatMachine transitions", () => { actor.stop(); }); + + it("allows multiple messages to be queued without leaving ready state", async () => { + const actor = createActor(createTestChatMachine()).start(); + + actor.send({ type: "ChatNonVipLogin", token: "token" }); + await waitFor(actor, (snapshot) => + snapshot.matches({ nonVipUserSession: "ready" }), + ); + + actor.send({ type: "ChatSendMessage", content: "hello" }); + actor.send({ type: "ChatSendMessage", content: "still there?" }); + + expect(actor.getSnapshot().matches({ nonVipUserSession: "ready" })).toBe( + true, + ); + expect(actor.getSnapshot().context.messages).toMatchObject([ + { content: "hello", isFromAI: false }, + { content: "still there?", isFromAI: false }, + ]); + + await waitFor( + actor, + (snapshot) => snapshot.context.pendingReplyCount === 0, + ); + expect(actor.getSnapshot().context.isReplyingAI).toBe(false); + + actor.stop(); + }); }); diff --git a/src/stores/chat/chat-events.ts b/src/stores/chat/chat-events.ts index b8e73384..409a2bb1 100644 --- a/src/stores/chat/chat-events.ts +++ b/src/stores/chat/chat-events.ts @@ -27,6 +27,11 @@ export type ChatEvent = | { type: "ChatSendImage"; imageBase64: string } | { type: "ChatUnlockPrivateMessage"; messageId: string } | { type: "ChatLoadMoreHistory" } + | { + type: "ChatQueuedHttpDone"; + output: import("./chat-machine.helpers").HttpSendOutput; + } + | { type: "ChatQueuedSendError"; content: string; errorMessage: string } | { type: "ChatImageReceived"; url: string } | { type: "ChatPaywallStatusReceived"; diff --git a/src/stores/chat/chat-machine.actors.ts b/src/stores/chat/chat-machine.actors.ts index ad60ee60..52dc1b7b 100644 --- a/src/stores/chat/chat-machine.actors.ts +++ b/src/stores/chat/chat-machine.actors.ts @@ -1,10 +1,11 @@ /** - * Chat 状态机：4 个 XState actor + * Chat 状态机：history / send / websocket / queue actors */ import { fromPromise, fromCallback } from "xstate"; import { createChatWebSocket, ChatWebSocket } from "@/core/net/chat-websocket"; +import { MessageQueue } from "@/core/net/message-queue"; import type { ChatSendResponse } from "@/data/dto/chat"; import { Result, Logger } from "@/utils"; @@ -74,19 +75,7 @@ export const sendMessageHttpActor = fromPromise< { response: ChatSendResponse; reply: UiMessage | null }, { content: string } >(async ({ input }) => { - const result = await chatRepo.sendMessage(input.content); - if (Result.isErr(result)) { - log.error("[chat-machine] sendMessageHttpActor failed", { error: result.error }); - throw result.error; - } - const isDailyLimit = - result.data.lockDetail.locked && - result.data.lockDetail.showUpgrade && - result.data.lockDetail.reason === "daily_limit"; - return { - response: result.data, - reply: isDailyLimit ? null : sendResponseToUiMessage(result.data), - }; + return sendMessageViaHttp(input.content); }); /** 翻历史（pagination）—— 不走 local-first，纯 server fetch */ @@ -233,5 +222,82 @@ export const sendMessageWsActor = fromPromise( }, ); +export const httpMessageQueueActor = fromCallback( + ({ sendBack, receive }) => { + return createMessageQueueActor("http", sendBack, receive); + }, +); + +export const wsPreferredMessageQueueActor = fromCallback( + ({ sendBack, receive }) => { + return createMessageQueueActor("wsPreferred", sendBack, receive); + }, +); + +function createMessageQueueActor( + mode: "http" | "wsPreferred", + sendBack: (event: ChatEvent) => void, + receive: (listener: (event: ChatEvent) => void) => void, +): () => void { + const queue = new MessageQueue(); + + queue.setConsumer(async (content) => { + try { + if (mode === "wsPreferred") { + const ws = getActiveChatWebSocket(); + if (ws?.sendMessage(content)) return true; + log.debug("[chat-machine] message queue fallback to HTTP", { + contentLength: content.length, + }); + } + + const output = await sendMessageViaHttp(content); + sendBack({ type: "ChatQueuedHttpDone", output }); + return true; + } catch (error) { + const errorMessage = + error instanceof Error ? error.message : "Message send failed"; + log.error("[chat-machine] message queue send failed", { + error, + mode, + }); + sendBack({ + type: "ChatQueuedSendError", + content, + errorMessage, + }); + return true; + } + }); + + receive((event) => { + if (event.type !== "ChatSendMessage") return; + const content = event.content.trim(); + if (content.length === 0) return; + queue.enqueue(content); + }); + + return () => queue.dispose(); +} + // Re-export `UiMessage` type for the chat-machine.ts public API type UiMessage = import("@/data/dto/chat").UiMessage; + +async function sendMessageViaHttp(content: string): Promise<{ + response: ChatSendResponse; + reply: UiMessage | null; +}> { + const result = await chatRepo.sendMessage(content); + if (Result.isErr(result)) { + log.error("[chat-machine] sendMessageHttpActor failed", { error: result.error }); + throw result.error; + } + const isDailyLimit = + result.data.lockDetail.locked && + result.data.lockDetail.showUpgrade && + result.data.lockDetail.reason === "daily_limit"; + return { + response: result.data, + reply: isDailyLimit ? null : sendResponseToUiMessage(result.data), + }; +} diff --git a/src/stores/chat/chat-machine.helpers.ts b/src/stores/chat/chat-machine.helpers.ts index 4250e532..f5c5ca77 100644 --- a/src/stores/chat/chat-machine.helpers.ts +++ b/src/stores/chat/chat-machine.helpers.ts @@ -136,6 +136,27 @@ export type HttpSendOutput = { reply: UiMessage | null; }; +export function beginPendingReply(context: ChatState): Pick< + ChatState, + "isReplyingAI" | "pendingReplyCount" +> { + return { + pendingReplyCount: context.pendingReplyCount + 1, + isReplyingAI: true, + }; +} + +export function finishPendingReply(context: ChatState): Pick< + ChatState, + "isReplyingAI" | "pendingReplyCount" +> { + const pendingReplyCount = Math.max(0, context.pendingReplyCount - 1); + return { + pendingReplyCount, + isReplyingAI: pendingReplyCount > 0, + }; +} + export function applyHttpSendOutput( context: ChatState, output: HttpSendOutput, @@ -157,7 +178,7 @@ export function applyHttpSendOutput( return { messages, - isReplyingAI: false, + ...finishPendingReply(context), upgradePromptVisible: true, upgradeReason: "daily_limit", upgradeHint: null, @@ -167,14 +188,14 @@ export function applyHttpSendOutput( if (!reply) { return { - isReplyingAI: false, + ...finishPendingReply(context), }; } if (lockDetail.showUpgrade && lockDetail.reason === "private_message") { return { messages: [...context.messages, reply], - isReplyingAI: false, + ...finishPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, @@ -184,7 +205,7 @@ export function applyHttpSendOutput( return { messages: [...context.messages, reply], - isReplyingAI: false, + ...finishPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, diff --git a/src/stores/chat/chat-machine.ts b/src/stores/chat/chat-machine.ts index a9a08a23..0716af43 100644 --- a/src/stores/chat/chat-machine.ts +++ b/src/stores/chat/chat-machine.ts @@ -27,9 +27,10 @@ * - 每个任务都是独立 actor，不互相等待（除非 `always` barrier） * * 消息发送路径（业务事实）： - * - guestSession：走 HTTP，消息数量限制以后端 `lockDetail` 响应为准 - * - nonVipUserSession：走 HTTP，让后端返回 daily_limit lockDetail - * - vipUserSession + WS 已连：走 WS，不受每日免费次数限制 + * - `ChatSendMessage` 只负责即时追加用户消息 + 入队，不进入 sending 子状态 + * - guestSession / nonVipUserSession：队列串行走 HTTP，限制以后端 `lockDetail` 为准 + * - vipUserSession：队列优先走 WS，WS 不可用时 fallback HTTP + * - `pendingReplyCount` 跟踪待回复数量，`isReplyingAI` 由计数派生 * - `ChatWebSocketConnected` 事件 → `wsConnected = true` * - `vipUserSession.exit` action → `wsConnected = false`（WS actor cleanup 时也会跑 exit） * @@ -39,7 +40,7 @@ * */ -import { setup, assign } from "xstate"; +import { setup, assign, sendTo } from "xstate"; import { todayString, Logger } from "@/utils"; @@ -47,6 +48,8 @@ import { ChatState, initialState } from "./chat-state"; import type { ChatEvent } from "./chat-events"; import { applyHttpSendOutput, + beginPendingReply, + finishPendingReply, normalizeLockDetail, } from "./chat-machine.helpers"; import { @@ -55,7 +58,9 @@ import { sendMessageWsActor, loadMoreHistoryActor, chatWebSocketActor, + httpMessageQueueActor, unlockPrivateMessageActor, + wsPreferredMessageQueueActor, } from "./chat-machine.actors"; const log = new Logger("StoresChatChatMachine"); @@ -79,9 +84,13 @@ export const chatMachine = setup({ sendMessageWs: sendMessageWsActor, loadMoreHistory: loadMoreHistoryActor, chatWebSocket: chatWebSocketActor, + httpMessageQueue: httpMessageQueueActor, + wsPreferredMessageQueue: wsPreferredMessageQueueActor, unlockPrivateMessage: unlockPrivateMessageActor, }, actions: { + enqueueMessage: sendTo("messageQueue", ({ event }) => event), + startGuestSession: assign(() => ({ ...initialState, })), @@ -114,7 +123,7 @@ export const chatMachine = setup({ date: today, }, ], - isReplyingAI: true, + ...beginPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, @@ -142,7 +151,7 @@ export const chatMachine = setup({ date: today, }, ], - isReplyingAI: true, + ...beginPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, @@ -168,7 +177,7 @@ export const chatMachine = setup({ imageUrl: event.imageBase64, }, ], - isReplyingAI: true, + ...beginPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, @@ -195,7 +204,7 @@ export const chatMachine = setup({ imageUrl: event.imageBase64, }, ], - isReplyingAI: true, + ...beginPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, @@ -227,7 +236,17 @@ export const chatMachine = setup({ done: event.done, messagesCount: messages.length, }); - return { messages, isReplyingAI: !event.done }; + if (!event.done) { + return { + messages, + pendingReplyCount: Math.max(1, context.pendingReplyCount), + isReplyingAI: true, + }; + } + return { + messages, + ...finishPendingReply(context), + }; }), appendSocketErrorMessage: assign(({ context }) => { @@ -239,7 +258,12 @@ export const chatMachine = setup({ date: todayString(), }, ]; - return { messages, isReplyingAI: false }; + return { messages, ...finishPendingReply(context) }; + }), + + applyQueuedHttpOutput: assign(({ context, event }) => { + if (event.type !== "ChatQueuedHttpDone") return {}; + return applyHttpSendOutput(context, event.output); }), appendAIImage: assign(({ context, event }) => { @@ -310,7 +334,7 @@ export const chatMachine = setup({ applyUnlockPrivateOutput: assign(({ context, event }) => { if (!("output" in event)) return {}; - const output = event.output as { + const output = event.output as unknown as { messageId: string; response: import("@/data/dto/chat").UnlockPrivateResponse; }; @@ -386,6 +410,10 @@ export const chatMachine = setup({ }, guestSession: { + invoke: { + id: "messageQueue", + src: "httpMessageQueue", + }, on: { ChatNonVipLogin: { target: "#chat.nonVipUserSession", @@ -395,6 +423,12 @@ export const chatMachine = setup({ target: "#chat.vipUserSession", actions: "startUserSession", }, + ChatQueuedHttpDone: { + actions: "applyQueuedHttpOutput", + }, + ChatQueuedSendError: { + actions: "appendSocketErrorMessage", + }, }, initial: "initializing", states: { @@ -428,9 +462,8 @@ export const chatMachine = setup({ ready: { on: { ChatSendMessage: { - actions: "appendGuestUserMessage", + actions: ["appendGuestUserMessage", "enqueueMessage"], guard: ({ event }) => event.content.trim().length > 0, - target: "sending", }, ChatSendImage: { actions: "appendGuestUserImage", @@ -482,6 +515,10 @@ export const chatMachine = setup({ }, nonVipUserSession: { + invoke: { + id: "messageQueue", + src: "httpMessageQueue", + }, on: { ChatLogout: { target: "#chat.idle", @@ -496,6 +533,12 @@ export const chatMachine = setup({ target: "#chat.vipUserSession", actions: "startUserSession", }, + ChatQueuedHttpDone: { + actions: "applyQueuedHttpOutput", + }, + ChatQueuedSendError: { + actions: "appendSocketErrorMessage", + }, }, initial: "initializing", states: { @@ -525,8 +568,7 @@ export const chatMachine = setup({ on: { ChatSendMessage: { guard: ({ event }) => event.content.trim().length > 0, - actions: "appendUserMessage", - target: "sendingViaHttp", + actions: ["appendUserMessage", "enqueueMessage"], }, ChatSendImage: { actions: "appendUserImage", @@ -624,16 +666,28 @@ export const chatMachine = setup({ ChatPaywallStatusReceived: { actions: "applyPaywallStatus" }, ChatWebSocketError: { actions: "appendSocketErrorMessage" }, ChatWebSocketConnected: { actions: "setWsConnected" }, + ChatQueuedHttpDone: { + actions: "applyQueuedHttpOutput", + }, + ChatQueuedSendError: { + actions: "appendSocketErrorMessage", + }, }, initial: "initializing", exit: "clearWsConnected", - invoke: { - // 父级：WS 长连接（一进来就连，跨 initializing/ready/sending*/loadingMore） - src: "chatWebSocket", - input: ({ event }) => ({ - token: event.type === "ChatVipLogin" ? event.token : "", - }), - }, + invoke: [ + { + // 父级：WS 长连接（一进来就连，跨 initializing/ready/loadingMore） + src: "chatWebSocket", + input: ({ event }) => ({ + token: event.type === "ChatVipLogin" ? event.token : "", + }), + }, + { + id: "messageQueue", + src: "wsPreferredMessageQueue", + }, + ], states: { initializing: { // barrier：WS 连上 + history 加载完成才进 ready @@ -666,25 +720,10 @@ export const chatMachine = setup({ }, ready: { on: { - // 发送消息：wsConnected → 走 WS（流式回 reply），否则 → 走 HTTP（一次性回 reply） - // 两个 branch 都有 "content 非空" guard（仅作 fallback 兜底：wsConnected=true 也需 - // 内容非空；wsConnected=false 同理） - ChatSendMessage: [ - { - // VIP 用户优先走 WS；WS 尚未可用时由下一个 branch 走 HTTP fallback。 - guard: ({ context, event }) => - context.wsConnected && - event.content.trim().length > 0, - actions: "appendUserMessage", - target: "sendingViaWs", - }, - { - // WS 未连（或 fallback） + 内容非空 → 走 HTTP（sendingViaHttp 一次性回 reply） - guard: ({ event }) => event.content.trim().length > 0, - actions: "appendUserMessage", - target: "sendingViaHttp", - }, - ], + ChatSendMessage: { + guard: ({ event }) => event.content.trim().length > 0, + actions: ["appendUserMessage", "enqueueMessage"], + }, ChatSendImage: { actions: "appendUserImage", }, diff --git a/src/stores/chat/chat-state.ts b/src/stores/chat/chat-state.ts index ad114779..c8969ff5 100644 --- a/src/stores/chat/chat-state.ts +++ b/src/stores/chat/chat-state.ts @@ -3,6 +3,7 @@ import type { UiMessage } from "@/data/dto/chat"; export interface ChatState { messages: UiMessage[]; isReplyingAI: boolean; + pendingReplyCount: number; /** * WebSocket 连接状态（仅 VIP 用户会话期间 true） * - VIP true：走 WS @@ -34,6 +35,7 @@ export interface ChatState { export const initialState: ChatState = { messages: [], isReplyingAI: false, + pendingReplyCount: 0, wsConnected: false, upgradePromptVisible: false, upgradeReason: null,