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

fix(chat): track replies by queued batch

Browse Source
This commit is contained in:
chenhang
2026-06-26 11:42:17 +08:00
parent ceaa3ebfcf
commit 58236453ed
6 changed files with 126 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/net/__tests__/message-queue.test.ts
+13 -12
View File
@@ -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<boolean>((resolve) => {
new Promise<void>((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);
src/core/net/message-queue.ts
+4 -15
View File
@@ -4,12 +4,12 @@
* 业务行为：
* - 队列空闲时，第一条消息立即发送
* - 后续消息在冷却窗口内暂存，并按 `intervalMs` 合并成一条发送
* - 消费者可声明每批消息是否成功消费；不成功则重新入队
* - 队列只负责发送节奏，不关心每批消息是否成功消费
* - 队列可在任意时刻清空（dispose）
*/
import { Logger } from "@/utils";
export type MessageConsumer = (content: string) => boolean | Promise<boolean>;
export type MessageConsumer = (content: string) => void | Promise<void>;
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();
}
src/stores/chat/__tests__/chat-machine.transitions.test.ts
+93
View File
@@ -87,6 +87,7 @@ function createTestChatMachine(
httpMessageQueue: fromCallback<ChatEvent>(({ 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<LoadHistoryOutput>(async () => ({
messages: [],
hasMore: false,
newOffset: 0,
localOverwritten: true,
localCount: 0,
networkCount: 0,
})),
loadMoreHistory: fromPromise<LoadMoreHistoryOutput, { offset: number }>(
async () => ({
messages: [],
hasMore: false,
newOffset: 0,
}),
),
sendMessageHttp: fromPromise<
SendMessageHttpOutput,
{ content: string }
>(async () => ({
response: makeChatSendResponse(),
reply: null,
})),
sendMessageWs: fromPromise<void, { content: string }>(async () => {}),
httpMessageQueue: fromCallback<ChatEvent>(({ 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<ChatEvent>(({ 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();
});
});
src/stores/chat/chat-events.ts
+1
View File
@@ -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;
src/stores/chat/chat-machine.actors.ts
+2 -3
View File
@@ -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;
}
});
src/stores/chat/chat-machine.ts
+13 -4
View File
@@ -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",
},