From 58236453eda32e67bd58b6813e7413aba7161c96 Mon Sep 17 00:00:00 2001 From: chenhang Date: Fri, 26 Jun 2026 11:42:17 +0800 Subject: [PATCH] fix(chat): track replies by queued batch --- src/core/net/__tests__/message-queue.test.ts | 25 ++--- src/core/net/message-queue.ts | 19 +--- .../chat-machine.transitions.test.ts | 93 +++++++++++++++++++ src/stores/chat/chat-events.ts | 1 + src/stores/chat/chat-machine.actors.ts | 5 +- src/stores/chat/chat-machine.ts | 17 +++- 6 files changed, 126 insertions(+), 34 deletions(-) diff --git a/src/core/net/__tests__/message-queue.test.ts b/src/core/net/__tests__/message-queue.test.ts index b90bfb65..cdc34019 100644 --- a/src/core/net/__tests__/message-queue.test.ts +++ b/src/core/net/__tests__/message-queue.test.ts @@ -12,7 +12,7 @@ describe("MessageQueue", () => { }); it("sends the first message immediately", async () => { - const consumer = vi.fn(() => true); + const consumer = vi.fn(); const queue = new MessageQueue(); queue.setConsumer(consumer); @@ -26,7 +26,7 @@ describe("MessageQueue", () => { }); it("merges pending messages after the interval", async () => { - const consumer = vi.fn(() => true); + const consumer = vi.fn(); const queue = new MessageQueue({ intervalMs: 1000 }); queue.setConsumer(consumer); @@ -48,10 +48,10 @@ describe("MessageQueue", () => { }); it("does not send a pending batch while a previous send is still running", async () => { - let resolveFirstSend: (ok: boolean) => void = () => undefined; + let resolveFirstSend: () => void = () => undefined; const consumer = vi.fn( () => - new Promise((resolve) => { + new Promise((resolve) => { resolveFirstSend = resolve; }), ); @@ -66,7 +66,7 @@ describe("MessageQueue", () => { await vi.advanceTimersByTimeAsync(1000); expect(consumer).toHaveBeenCalledTimes(1); - resolveFirstSend(true); + resolveFirstSend(); await Promise.resolve(); await vi.advanceTimersByTimeAsync(1000); @@ -76,12 +76,12 @@ describe("MessageQueue", () => { queue.dispose(); }); - it("requeues a failed batch and merges it with newer pending messages", async () => { + it("does not retry a failed batch", async () => { const consumer = vi - .fn((content: string) => content.length > 0) - .mockReturnValueOnce(true) - .mockReturnValueOnce(false) - .mockReturnValue(true); + .fn() + .mockResolvedValueOnce(undefined) + .mockRejectedValueOnce(new Error("send failed")) + .mockResolvedValue(undefined); const queue = new MessageQueue({ intervalMs: 1000 }); queue.setConsumer(consumer); @@ -96,13 +96,14 @@ describe("MessageQueue", () => { queue.enqueue("fourth"); await vi.advanceTimersByTimeAsync(1000); - expect(consumer).toHaveBeenLastCalledWith("second\nthird\nfourth"); + expect(consumer).toHaveBeenCalledTimes(3); + expect(consumer).toHaveBeenLastCalledWith("fourth"); queue.dispose(); }); it("supports a custom joiner", async () => { - const consumer = vi.fn(() => true); + const consumer = vi.fn(); const queue = new MessageQueue({ intervalMs: 1000, joiner: " " }); queue.setConsumer(consumer); diff --git a/src/core/net/message-queue.ts b/src/core/net/message-queue.ts index 1c8df807..873940d9 100644 --- a/src/core/net/message-queue.ts +++ b/src/core/net/message-queue.ts @@ -4,12 +4,12 @@ * 业务行为： * - 队列空闲时，第一条消息立即发送 * - 后续消息在冷却窗口内暂存，并按 `intervalMs` 合并成一条发送 - * - 消费者可声明每批消息是否成功消费；不成功则重新入队 + * - 队列只负责发送节奏，不关心每批消息是否成功消费 * - 队列可在任意时刻清空（dispose） */ import { Logger } from "@/utils"; -export type MessageConsumer = (content: string) => boolean | Promise; +export type MessageConsumer = (content: string) => void | Promise; export interface MessageQueueOptions { intervalMs?: number; @@ -38,7 +38,7 @@ export class MessageQueue { this.joiner = options.joiner ?? "\n"; } - /** 注册消费者（消费者必须能处理失败/重试）。 */ + /** 注册消费者。队列只负责调用，不根据结果做重试。 */ setConsumer(consumer: MessageConsumer): void { this.consumer = consumer; } @@ -69,24 +69,13 @@ export class MessageQueue { return; } - let ok = false; try { - ok = await this.consumer(content); + await this.consumer(content); } catch (e) { log.error({ err: e }, "consumer error"); - ok = false; } this.sending = false; - if (!ok) { - // 消费失败，整批放回队首等待下次合并发送。 - this.pending.unshift(content); - log.debug("retry batch", { - pendingCount: this.pending.length, - contentLength: content.length, - }); - } - this.scheduleFlush(); } diff --git a/src/stores/chat/__tests__/chat-machine.transitions.test.ts b/src/stores/chat/__tests__/chat-machine.transitions.test.ts index fdd56f84..0577e815 100644 --- a/src/stores/chat/__tests__/chat-machine.transitions.test.ts +++ b/src/stores/chat/__tests__/chat-machine.transitions.test.ts @@ -87,6 +87,7 @@ function createTestChatMachine( httpMessageQueue: fromCallback(({ receive, sendBack }) => { receive((event) => { if (event.type !== "ChatSendMessage") return; + sendBack({ type: "ChatQueuedSendStarted" }); sendBack({ type: "ChatQueuedHttpDone", output: { @@ -266,4 +267,96 @@ describe("chatMachine transitions", () => { actor.stop(); }); + + it("tracks replying state by queued batch instead of user message count", async () => { + const machine = chatMachine.provide({ + actors: { + loadHistory: fromPromise(async () => ({ + messages: [], + hasMore: false, + newOffset: 0, + localOverwritten: true, + localCount: 0, + networkCount: 0, + })), + loadMoreHistory: fromPromise( + async () => ({ + messages: [], + hasMore: false, + newOffset: 0, + }), + ), + sendMessageHttp: fromPromise< + SendMessageHttpOutput, + { content: string } + >(async () => ({ + response: makeChatSendResponse(), + reply: null, + })), + sendMessageWs: fromPromise(async () => {}), + httpMessageQueue: fromCallback(({ receive, sendBack }) => { + const pending: string[] = []; + receive((event) => { + if (event.type !== "ChatSendMessage") return; + pending.push(event.content); + if (pending.length !== 2) return; + sendBack({ type: "ChatQueuedSendStarted" }); + sendBack({ + type: "ChatQueuedHttpDone", + output: { + response: makeChatSendResponse(), + reply: null, + }, + }); + }); + return () => undefined; + }), + wsPreferredMessageQueue: fromCallback(({ receive }) => { + receive(() => undefined); + return () => undefined; + }), + unlockPrivateMessage: fromPromise< + UnlockPrivateOutput, + { messageId: string } + >(async ({ input }) => ({ + messageId: input.messageId, + response: UnlockPrivateResponse.from({ + unlocked: true, + content: "unlocked", + showUpgrade: false, + paywallTriggered: false, + privateFreeLimit: 0, + privateUsedToday: 0, + reason: "ok", + }), + })), + chatWebSocket: fromCallback(({ sendBack }) => { + sendBack({ + type: "ChatWebSocketConnected", + userId: "user-1", + }); + return () => undefined; + }), + }, + }); + const actor = createActor(machine).start(); + + actor.send({ type: "ChatNonVipLogin", token: "token" }); + await waitFor(actor, (snapshot) => + snapshot.matches({ nonVipUserSession: "ready" }), + ); + + actor.send({ type: "ChatSendMessage", content: "first" }); + actor.send({ type: "ChatSendMessage", content: "second" }); + + await waitFor( + actor, + (snapshot) => snapshot.context.messages.length === 2, + ); + + expect(actor.getSnapshot().context.pendingReplyCount).toBe(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 409a2bb1..da20fab5 100644 --- a/src/stores/chat/chat-events.ts +++ b/src/stores/chat/chat-events.ts @@ -27,6 +27,7 @@ export type ChatEvent = | { type: "ChatSendImage"; imageBase64: string } | { type: "ChatUnlockPrivateMessage"; messageId: string } | { type: "ChatLoadMoreHistory" } + | { type: "ChatQueuedSendStarted" } | { type: "ChatQueuedHttpDone"; output: import("./chat-machine.helpers").HttpSendOutput; diff --git a/src/stores/chat/chat-machine.actors.ts b/src/stores/chat/chat-machine.actors.ts index 52dc1b7b..762f1198 100644 --- a/src/stores/chat/chat-machine.actors.ts +++ b/src/stores/chat/chat-machine.actors.ts @@ -242,10 +242,11 @@ function createMessageQueueActor( const queue = new MessageQueue(); queue.setConsumer(async (content) => { + sendBack({ type: "ChatQueuedSendStarted" }); try { if (mode === "wsPreferred") { const ws = getActiveChatWebSocket(); - if (ws?.sendMessage(content)) return true; + if (ws?.sendMessage(content)) return; log.debug("[chat-machine] message queue fallback to HTTP", { contentLength: content.length, }); @@ -253,7 +254,6 @@ function createMessageQueueActor( const output = await sendMessageViaHttp(content); sendBack({ type: "ChatQueuedHttpDone", output }); - return true; } catch (error) { const errorMessage = error instanceof Error ? error.message : "Message send failed"; @@ -266,7 +266,6 @@ function createMessageQueueActor( content, errorMessage, }); - return true; } }); diff --git a/src/stores/chat/chat-machine.ts b/src/stores/chat/chat-machine.ts index 0716af43..69373c0f 100644 --- a/src/stores/chat/chat-machine.ts +++ b/src/stores/chat/chat-machine.ts @@ -111,7 +111,7 @@ export const chatMachine = setup({ contentLength: event.content.length, contentPreview: event.content.slice(0, 50), quotaMode: "server controlled", - isReplyingAI: true, + isReplyingAI: context.isReplyingAI, }); return { @@ -123,7 +123,6 @@ export const chatMachine = setup({ date: today, }, ], - ...beginPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, @@ -139,7 +138,7 @@ export const chatMachine = setup({ contentLength: event.content.length, contentPreview: event.content.slice(0, 50), wsConnected: context.wsConnected, - isReplyingAI: true, + isReplyingAI: context.isReplyingAI, }); return { @@ -151,7 +150,6 @@ export const chatMachine = setup({ date: today, }, ], - ...beginPendingReply(context), upgradePromptVisible: false, upgradeReason: null, upgradeHint: null, @@ -261,6 +259,8 @@ export const chatMachine = setup({ return { messages, ...finishPendingReply(context) }; }), + markQueuedSendStarted: assign(({ context }) => beginPendingReply(context)), + applyQueuedHttpOutput: assign(({ context, event }) => { if (event.type !== "ChatQueuedHttpDone") return {}; return applyHttpSendOutput(context, event.output); @@ -423,6 +423,9 @@ export const chatMachine = setup({ target: "#chat.vipUserSession", actions: "startUserSession", }, + ChatQueuedSendStarted: { + actions: "markQueuedSendStarted", + }, ChatQueuedHttpDone: { actions: "applyQueuedHttpOutput", }, @@ -533,6 +536,9 @@ export const chatMachine = setup({ target: "#chat.vipUserSession", actions: "startUserSession", }, + ChatQueuedSendStarted: { + actions: "markQueuedSendStarted", + }, ChatQueuedHttpDone: { actions: "applyQueuedHttpOutput", }, @@ -666,6 +672,9 @@ export const chatMachine = setup({ ChatPaywallStatusReceived: { actions: "applyPaywallStatus" }, ChatWebSocketError: { actions: "appendSocketErrorMessage" }, ChatWebSocketConnected: { actions: "setWsConnected" }, + ChatQueuedSendStarted: { + actions: "markQueuedSendStarted", + }, ChatQueuedHttpDone: { actions: "applyQueuedHttpOutput", },