banlv/cozsweet-frontend-nextjs
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(chat): queue outgoing messages

Browse Source
This commit is contained in:
chenhang
2026-06-26 11:11:43 +08:00
parent 2858ceccdc
commit fe85af0cd7
8 changed files with 240 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/chat/chat-screen.tsx
+1 -1
View File
@@ -202,7 +202,7 @@ export function ChatScreen() {
onUnlock={handleMessageLimitUnlock}
/>
) : (
<ChatInputBar disabled={state.isReplyingAI} />
<ChatInputBar disabled={!state.historyLoaded} />
)}
</div>
src/stores/chat/__tests__/chat-machine.helpers.test.ts
+1
View File
@@ -34,6 +34,7 @@ function makeChatState(overrides: Partial<ChatState> = {}): ChatState {
return {
messages: [],
isReplyingAI: true,
pendingReplyCount: 1,
wsConnected: false,
upgradePromptVisible: false,
upgradeReason: null,
src/stores/chat/__tests__/chat-machine.transitions.test.ts
+46
View File
@@ -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<void, { content: string }>(async () => {}),
httpMessageQueue: fromCallback<ChatEvent>(({ receive, sendBack }) => {
receive((event) => {
if (event.type !== "ChatSendMessage") return;
sendBack({
type: "ChatQueuedHttpDone",
output: {
response: makeChatSendResponse(),
reply: null,
},
});
});
return () => undefined;
}),
wsPreferredMessageQueue: fromCallback<ChatEvent>(({ 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();
});
});
src/stores/chat/chat-events.ts
+5
View File
@@ -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";
src/stores/chat/chat-machine.actors.ts
+80 -14
View File
@@ -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<void, { content: string }>(
},
);
export const httpMessageQueueActor = fromCallback<ChatEvent>(
({ sendBack, receive }) => {
return createMessageQueueActor("http", sendBack, receive);
},
);
export const wsPreferredMessageQueueActor = fromCallback<ChatEvent>(
({ 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),
};
}
src/stores/chat/chat-machine.helpers.ts
+25 -4
View File
@@ -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,
src/stores/chat/chat-machine.ts
+80 -41
View File
@@ -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",
},
src/stores/chat/chat-state.ts
+2
View File
@@ -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,