forked from bai/curriculum-project-hub
feat: add message burst debouncing (MessageBatcher)
- Create MessageBatcher: debounce 600ms, adaptive delay for long chunks, max 8 messages/4000 chars before immediate flush, per (chatId, sender) key - Integrate into trigger: text messages enqueued, slash commands and file/image bypass batching, locked projects respond immediately - Add unit tests (8) for batcher lifecycle and edge cases
This commit is contained in:
@@ -0,0 +1,125 @@
|
||||
export interface MessageBatcherOptions {
|
||||
/** Default: 600. */
|
||||
readonly debounceMs?: number;
|
||||
/** Default: 8. */
|
||||
readonly maxMessages?: number;
|
||||
/** Default: 4000. */
|
||||
readonly maxChars?: number;
|
||||
/** Default: 3500. */
|
||||
readonly splitThreshold?: number;
|
||||
/** Default: 1200. */
|
||||
readonly extendedDebounceMs?: number;
|
||||
}
|
||||
|
||||
export type MessageBatcherCallback = (mergedText: string, key?: string) => Promise<void>;
|
||||
|
||||
interface PendingBatch {
|
||||
readonly texts: string[];
|
||||
charCount: number;
|
||||
timer: ReturnType<typeof setTimeout> | null;
|
||||
}
|
||||
|
||||
const DEFAULT_OPTIONS = {
|
||||
debounceMs: 600,
|
||||
maxMessages: 8,
|
||||
maxChars: 4000,
|
||||
splitThreshold: 3500,
|
||||
extendedDebounceMs: 1200,
|
||||
} as const;
|
||||
|
||||
export class MessageBatcher {
|
||||
private readonly debounceMs: number;
|
||||
private readonly maxMessages: number;
|
||||
private readonly maxChars: number;
|
||||
private readonly splitThreshold: number;
|
||||
private readonly extendedDebounceMs: number;
|
||||
private readonly pending = new Map<string, PendingBatch>();
|
||||
private readonly chains = new Map<string, Promise<void>>();
|
||||
|
||||
constructor(
|
||||
private readonly onFlush: MessageBatcherCallback,
|
||||
options: MessageBatcherOptions = {},
|
||||
) {
|
||||
this.debounceMs = options.debounceMs ?? DEFAULT_OPTIONS.debounceMs;
|
||||
this.maxMessages = options.maxMessages ?? DEFAULT_OPTIONS.maxMessages;
|
||||
this.maxChars = options.maxChars ?? DEFAULT_OPTIONS.maxChars;
|
||||
this.splitThreshold = options.splitThreshold ?? DEFAULT_OPTIONS.splitThreshold;
|
||||
this.extendedDebounceMs = options.extendedDebounceMs ?? DEFAULT_OPTIONS.extendedDebounceMs;
|
||||
}
|
||||
|
||||
async enqueue(chatId: string, senderOpenId: string, text: string): Promise<void> {
|
||||
const key = messageBatchKey(chatId, senderOpenId);
|
||||
await this.runSerial(key, async () => {
|
||||
const existing = this.pending.get(key);
|
||||
const batch =
|
||||
existing ??
|
||||
{
|
||||
texts: [],
|
||||
charCount: 0,
|
||||
timer: null,
|
||||
};
|
||||
|
||||
if (existing === undefined) {
|
||||
this.pending.set(key, batch);
|
||||
}
|
||||
|
||||
batch.texts.push(text);
|
||||
batch.charCount += text.length + (batch.texts.length === 1 ? 0 : 1);
|
||||
|
||||
if (batch.texts.length >= this.maxMessages || batch.charCount >= this.maxChars) {
|
||||
await this.flushPendingBatch(key, batch);
|
||||
return;
|
||||
}
|
||||
|
||||
this.scheduleFlush(key, batch, text.length >= this.splitThreshold ? this.extendedDebounceMs : this.debounceMs);
|
||||
});
|
||||
}
|
||||
|
||||
async flushNow(key: string): Promise<void> {
|
||||
await this.runSerial(key, async () => {
|
||||
const batch = this.pending.get(key);
|
||||
if (batch === undefined) return;
|
||||
await this.flushPendingBatch(key, batch);
|
||||
});
|
||||
}
|
||||
|
||||
async flushAll(): Promise<void> {
|
||||
const keys = [...this.pending.keys()];
|
||||
await Promise.all(keys.map((key) => this.flushNow(key)));
|
||||
}
|
||||
|
||||
private scheduleFlush(key: string, batch: PendingBatch, delayMs: number): void {
|
||||
if (batch.timer !== null) {
|
||||
clearTimeout(batch.timer);
|
||||
}
|
||||
batch.timer = setTimeout(() => {
|
||||
void this.flushNow(key).catch(() => undefined);
|
||||
}, delayMs);
|
||||
(batch.timer as { unref?: () => void }).unref?.();
|
||||
}
|
||||
|
||||
private async flushPendingBatch(key: string, batch: PendingBatch): Promise<void> {
|
||||
if (batch.timer !== null) {
|
||||
clearTimeout(batch.timer);
|
||||
batch.timer = null;
|
||||
}
|
||||
if (this.pending.get(key) !== batch) return;
|
||||
this.pending.delete(key);
|
||||
await this.onFlush(batch.texts.join("\n"), key);
|
||||
}
|
||||
|
||||
private runSerial(key: string, operation: () => Promise<void>): Promise<void> {
|
||||
const previous = this.chains.get(key) ?? Promise.resolve();
|
||||
const next = previous.catch(() => undefined).then(operation);
|
||||
this.chains.set(key, next);
|
||||
return next.finally(() => {
|
||||
if (this.chains.get(key) === next) {
|
||||
this.chains.delete(key);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function messageBatchKey(chatId: string, senderOpenId: string): string {
|
||||
return `${chatId}:${senderOpenId}`;
|
||||
}
|
||||
+193
-134
@@ -20,6 +20,7 @@ import { writeAudit } from "../audit.js";
|
||||
import { PatchableTextStream } from "./textStream.js";
|
||||
import { createFileDeliveryMcpServer } from "./fileDeliveryTool.js";
|
||||
import { readFeishuContext } from "./read.js";
|
||||
import { MessageBatcher, messageBatchKey, type MessageBatcherOptions } from "./messageBatcher.js";
|
||||
|
||||
interface TriggerDeps {
|
||||
readonly prisma: PrismaClient;
|
||||
@@ -27,6 +28,21 @@ interface TriggerDeps {
|
||||
readonly logger: FastifyBaseLogger;
|
||||
readonly runAgent?: (req: RunRequest) => Promise<RunResult>;
|
||||
readonly authorizer?: PermissionAuthorizer | undefined;
|
||||
readonly messageBatcherOptions?: MessageBatcherOptions | undefined;
|
||||
}
|
||||
|
||||
interface TriggerActor {
|
||||
readonly feishuOpenId: string;
|
||||
readonly chatId: string;
|
||||
}
|
||||
|
||||
interface TriggerRunContext {
|
||||
readonly msg: MessageReceiveEvent["message"];
|
||||
readonly rt: FeishuRuntime;
|
||||
readonly chatId: string;
|
||||
readonly projectId: string;
|
||||
readonly senderOpenId: string;
|
||||
readonly actor: TriggerActor;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -36,145 +52,29 @@ interface TriggerDeps {
|
||||
*/
|
||||
export function makeTriggerHandler(deps: TriggerDeps) {
|
||||
const authorizer = deps.authorizer ?? createPermissionAuthorizer(deps.prisma);
|
||||
return async (event: MessageReceiveEvent, rt: FeishuRuntime): Promise<void> => {
|
||||
const runAgent = deps.runAgent ?? defaultRunAgent;
|
||||
const msg = event.message;
|
||||
const sender = event.sender;
|
||||
void sender; // sender→user mapping is OPEN (principal sub-typology)
|
||||
|
||||
// Dedup: the lark ws client redelivers at-least-once. A duplicate event
|
||||
// would spawn a second AgentRun — the lock would refuse it, but a FAILED
|
||||
// run row + a busy reply would leak. Record the event_id up front; if it
|
||||
// already exists, this is a redelivery → stop.
|
||||
const eventId = feishuEventId(event);
|
||||
if (eventId !== undefined && eventId !== "") {
|
||||
const existing = await deps.prisma.feishuEventReceipt.findUnique({
|
||||
where: { eventId },
|
||||
select: { id: true },
|
||||
});
|
||||
if (existing !== null) {
|
||||
deps.logger.debug({ eventId }, "feishu trigger: duplicate event, skipping");
|
||||
return;
|
||||
}
|
||||
await deps.prisma.feishuEventReceipt.create({
|
||||
data: {
|
||||
eventId,
|
||||
eventType: feishuEventType(event),
|
||||
messageId: msg.message_id,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// Only react to text, file, or image messages in group chats.
|
||||
if (msg.message_type !== "text" && msg.message_type !== "file" && msg.message_type !== "image") return;
|
||||
const chatId = msg.chat_id;
|
||||
if (chatId === "") return;
|
||||
|
||||
// ADR-0001: resolve chat → project. Unknown chat ⇒ not a project group.
|
||||
const binding = await deps.prisma.projectGroupBinding.findUnique({
|
||||
where: { chatId },
|
||||
select: { projectId: true },
|
||||
});
|
||||
if (binding === null) {
|
||||
deps.logger.debug({ chatId }, "feishu trigger: chat not bound to any project");
|
||||
const runAgent = deps.runAgent ?? defaultRunAgent;
|
||||
const batchContexts = new Map<string, TriggerRunContext>();
|
||||
const messageBatcher = new MessageBatcher(async (mergedText, key) => {
|
||||
if (key === undefined) {
|
||||
deps.logger.warn("feishu trigger: message batch flushed without a key");
|
||||
return;
|
||||
}
|
||||
const projectId = binding.projectId;
|
||||
// ADR-0019: triggerAgent is authorized through actor → principal-set
|
||||
// resolution, then grant/settings composition.
|
||||
// A missing open_id is treated as an untrusted event: deny explicitly
|
||||
// rather than skipping the check (fail closed).
|
||||
const senderOpenId = event.sender.sender_id.open_id ?? "";
|
||||
if (senderOpenId === "") {
|
||||
deps.logger.warn({ projectId, chatId }, "feishu trigger: sender has no open_id, denying");
|
||||
await writeAudit(deps.prisma, { projectId, action: "trigger.denied", metadata: { reason: "missing open_id" } });
|
||||
await sendText(rt, chatId, "无法识别发送者,拒绝触发。");
|
||||
const context = batchContexts.get(key);
|
||||
if (context === undefined) {
|
||||
deps.logger.warn({ key }, "feishu trigger: message batch flushed without trigger context");
|
||||
return;
|
||||
}
|
||||
const actor = { feishuOpenId: senderOpenId, chatId };
|
||||
const triggerDecision = await authorizer.can({
|
||||
actor,
|
||||
action: "agent.trigger",
|
||||
resource: { type: "PROJECT", id: projectId },
|
||||
});
|
||||
if (!triggerDecision.allowed) {
|
||||
deps.logger.info({ projectId, senderOpenId, reason: triggerDecision.reason }, "feishu trigger: permission denied");
|
||||
await writeAudit(deps.prisma, {
|
||||
projectId,
|
||||
...(triggerDecision.actorUserId !== undefined ? { actorUserId: triggerDecision.actorUserId } : {}),
|
||||
action: "trigger.denied",
|
||||
metadata: { decision: auditDecision(triggerDecision) },
|
||||
});
|
||||
await sendText(rt, chatId, "无权限触发。");
|
||||
return;
|
||||
batchContexts.delete(key);
|
||||
try {
|
||||
await startAgentRun(context, mergedText);
|
||||
} catch (e) {
|
||||
deps.logger.error({ key, err: e instanceof Error ? e.message : String(e) }, "feishu trigger: batched run setup failed");
|
||||
throw e;
|
||||
}
|
||||
}, deps.messageBatcherOptions);
|
||||
|
||||
let cleanPrompt: string | null = null;
|
||||
let downloadedFiles: string[] = [];
|
||||
if (msg.message_type === "text") {
|
||||
cleanPrompt = extractPrompt(msg);
|
||||
} else if (msg.message_type === "file" || msg.message_type === "image") {
|
||||
// Download the file/image to the workspace, build a prompt around it.
|
||||
const project = await deps.prisma.project.findUnique({
|
||||
where: { id: projectId },
|
||||
select: { workspaceDir: true },
|
||||
});
|
||||
if (project === null) return;
|
||||
try {
|
||||
const content = FileMessageContentSchema.parse(JSON.parse(msg.content));
|
||||
const key = content.file_key ?? content.image_key ?? "";
|
||||
if (key !== "") {
|
||||
const fileName = `${msg.message_type}-${Date.now()}.${msg.message_type === "image" ? "png" : "bin"}`;
|
||||
const savePath = `${project.workspaceDir}/.cph/inbox/${fileName}`;
|
||||
await downloadMessageFile(rt, msg.message_id, key, savePath);
|
||||
downloadedFiles.push(savePath);
|
||||
cleanPrompt = `用户发来一个文件: ${savePath}`;
|
||||
}
|
||||
} catch { return; }
|
||||
}
|
||||
if (cleanPrompt === null) return;
|
||||
// Slash commands: session management, not agent runs. These bypass the
|
||||
// lock — they don't create a run, so they can't conflict with one.
|
||||
if (cleanPrompt.startsWith("/")) {
|
||||
const cmd = cleanPrompt.split(/\s+/)[0];
|
||||
switch (cmd) {
|
||||
case "/new": {
|
||||
await deps.prisma.agentSession.updateMany({
|
||||
where: { projectId, archivedAt: null },
|
||||
data: { archivedAt: new Date() },
|
||||
});
|
||||
await sendText(rt, chatId, "已开新会话,下次 @bot 将从头开始。");
|
||||
return;
|
||||
}
|
||||
case "/resume": {
|
||||
const latest = await deps.prisma.agentSession.findFirst({
|
||||
where: { projectId, archivedAt: { not: null } },
|
||||
orderBy: { archivedAt: "desc" },
|
||||
select: { id: true },
|
||||
});
|
||||
if (latest === null) {
|
||||
await sendText(rt, chatId, "没有可恢复的会话。");
|
||||
return;
|
||||
}
|
||||
await deps.prisma.agentSession.update({
|
||||
where: { id: latest.id },
|
||||
data: { archivedAt: null },
|
||||
});
|
||||
await sendText(rt, chatId, "已恢复上一个会话。");
|
||||
return;
|
||||
}
|
||||
case "/reset": {
|
||||
await deps.prisma.agentSession.updateMany({
|
||||
where: { projectId, archivedAt: null },
|
||||
data: { archivedAt: new Date() },
|
||||
});
|
||||
await sendText(rt, chatId, "已重置,下次 @bot 将从头开始。");
|
||||
return;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
async function startAgentRun(context: TriggerRunContext, cleanPrompt: string): Promise<void> {
|
||||
const { msg, rt, chatId, projectId, senderOpenId, actor } = context;
|
||||
|
||||
// ADR-0002: if locked by another run, reply busy.
|
||||
const existing = await currentLockRunId(deps.prisma, projectId);
|
||||
@@ -302,7 +202,6 @@ export function makeTriggerHandler(deps: TriggerDeps) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
const projectCtx: ProjectContext = {
|
||||
projectId,
|
||||
boundChatId: chatId, // ADR-0003: Feishu reads stay in this chat
|
||||
@@ -408,6 +307,166 @@ export function makeTriggerHandler(deps: TriggerDeps) {
|
||||
deps.logger.warn({ runId: run.id, err: e instanceof Error ? e.message : String(e) }, "trigger: could not release lock");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
return async (event: MessageReceiveEvent, rt: FeishuRuntime): Promise<void> => {
|
||||
const msg = event.message;
|
||||
const sender = event.sender;
|
||||
void sender; // sender→user mapping is OPEN (principal sub-typology)
|
||||
|
||||
// Dedup: the lark ws client redelivers at-least-once. A duplicate event
|
||||
// would spawn a second AgentRun — the lock would refuse it, but a FAILED
|
||||
// run row + a busy reply would leak. Record the event_id up front; if it
|
||||
// already exists, this is a redelivery → stop.
|
||||
const eventId = feishuEventId(event);
|
||||
if (eventId !== undefined && eventId !== "") {
|
||||
const existing = await deps.prisma.feishuEventReceipt.findUnique({
|
||||
where: { eventId },
|
||||
select: { id: true },
|
||||
});
|
||||
if (existing !== null) {
|
||||
deps.logger.debug({ eventId }, "feishu trigger: duplicate event, skipping");
|
||||
return;
|
||||
}
|
||||
await deps.prisma.feishuEventReceipt.create({
|
||||
data: {
|
||||
eventId,
|
||||
eventType: feishuEventType(event),
|
||||
messageId: msg.message_id,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// Only react to text, file, or image messages in group chats.
|
||||
if (msg.message_type !== "text" && msg.message_type !== "file" && msg.message_type !== "image") return;
|
||||
const chatId = msg.chat_id;
|
||||
if (chatId === "") return;
|
||||
|
||||
// ADR-0001: resolve chat → project. Unknown chat ⇒ not a project group.
|
||||
const binding = await deps.prisma.projectGroupBinding.findUnique({
|
||||
where: { chatId },
|
||||
select: { projectId: true },
|
||||
});
|
||||
if (binding === null) {
|
||||
deps.logger.debug({ chatId }, "feishu trigger: chat not bound to any project");
|
||||
return;
|
||||
}
|
||||
const projectId = binding.projectId;
|
||||
// ADR-0019: triggerAgent is authorized through actor → principal-set
|
||||
// resolution, then grant/settings composition.
|
||||
// A missing open_id is treated as an untrusted event: deny explicitly
|
||||
// rather than skipping the check (fail closed).
|
||||
const senderOpenId = event.sender.sender_id.open_id ?? "";
|
||||
if (senderOpenId === "") {
|
||||
deps.logger.warn({ projectId, chatId }, "feishu trigger: sender has no open_id, denying");
|
||||
await writeAudit(deps.prisma, { projectId, action: "trigger.denied", metadata: { reason: "missing open_id" } });
|
||||
await sendText(rt, chatId, "无法识别发送者,拒绝触发。");
|
||||
return;
|
||||
}
|
||||
const actor = { feishuOpenId: senderOpenId, chatId };
|
||||
const triggerDecision = await authorizer.can({
|
||||
actor,
|
||||
action: "agent.trigger",
|
||||
resource: { type: "PROJECT", id: projectId },
|
||||
});
|
||||
if (!triggerDecision.allowed) {
|
||||
deps.logger.info({ projectId, senderOpenId, reason: triggerDecision.reason }, "feishu trigger: permission denied");
|
||||
await writeAudit(deps.prisma, {
|
||||
projectId,
|
||||
...(triggerDecision.actorUserId !== undefined ? { actorUserId: triggerDecision.actorUserId } : {}),
|
||||
action: "trigger.denied",
|
||||
metadata: { decision: auditDecision(triggerDecision) },
|
||||
});
|
||||
await sendText(rt, chatId, "无权限触发。");
|
||||
return;
|
||||
}
|
||||
|
||||
let cleanPrompt: string | null = null;
|
||||
if (msg.message_type === "text") {
|
||||
cleanPrompt = extractPrompt(msg);
|
||||
} else if (msg.message_type === "file" || msg.message_type === "image") {
|
||||
// Download the file/image to the workspace, build a prompt around it.
|
||||
const project = await deps.prisma.project.findUnique({
|
||||
where: { id: projectId },
|
||||
select: { workspaceDir: true },
|
||||
});
|
||||
if (project === null) return;
|
||||
try {
|
||||
const content = FileMessageContentSchema.parse(JSON.parse(msg.content));
|
||||
const key = content.file_key ?? content.image_key ?? "";
|
||||
if (key !== "") {
|
||||
const fileName = `${msg.message_type}-${Date.now()}.${msg.message_type === "image" ? "png" : "bin"}`;
|
||||
const savePath = `${project.workspaceDir}/.cph/inbox/${fileName}`;
|
||||
await downloadMessageFile(rt, msg.message_id, key, savePath);
|
||||
cleanPrompt = `用户发来一个文件: ${savePath}`;
|
||||
}
|
||||
} catch { return; }
|
||||
}
|
||||
if (cleanPrompt === null) return;
|
||||
const runContext: TriggerRunContext = { msg, rt, chatId, projectId, senderOpenId, actor };
|
||||
|
||||
// Slash commands: session management, not agent runs. These bypass the
|
||||
// batcher. Session commands bypass the lock because they don't create runs;
|
||||
// role/unknown slash prompts still run immediately through the normal lock.
|
||||
if (cleanPrompt.startsWith("/")) {
|
||||
const cmd = cleanPrompt.split(/\s+/)[0];
|
||||
switch (cmd) {
|
||||
case "/new": {
|
||||
await deps.prisma.agentSession.updateMany({
|
||||
where: { projectId, archivedAt: null },
|
||||
data: { archivedAt: new Date() },
|
||||
});
|
||||
await sendText(rt, chatId, "已开新会话,下次 @bot 将从头开始。");
|
||||
return;
|
||||
}
|
||||
case "/resume": {
|
||||
const latest = await deps.prisma.agentSession.findFirst({
|
||||
where: { projectId, archivedAt: { not: null } },
|
||||
orderBy: { archivedAt: "desc" },
|
||||
select: { id: true },
|
||||
});
|
||||
if (latest === null) {
|
||||
await sendText(rt, chatId, "没有可恢复的会话。");
|
||||
return;
|
||||
}
|
||||
await deps.prisma.agentSession.update({
|
||||
where: { id: latest.id },
|
||||
data: { archivedAt: null },
|
||||
});
|
||||
await sendText(rt, chatId, "已恢复上一个会话。");
|
||||
return;
|
||||
}
|
||||
case "/reset": {
|
||||
await deps.prisma.agentSession.updateMany({
|
||||
where: { projectId, archivedAt: null },
|
||||
data: { archivedAt: new Date() },
|
||||
});
|
||||
await sendText(rt, chatId, "已重置,下次 @bot 将从头开始。");
|
||||
return;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
await startAgentRun(runContext, cleanPrompt);
|
||||
return;
|
||||
}
|
||||
|
||||
if (msg.message_type === "text") {
|
||||
const existing = await currentLockRunId(deps.prisma, projectId);
|
||||
if (existing !== null) {
|
||||
await reactToMessage(rt, msg.message_id, "OnIt");
|
||||
await sendText(rt, chatId, "项目正在处理中,请稍候。");
|
||||
return;
|
||||
}
|
||||
const key = messageBatchKey(chatId, senderOpenId);
|
||||
if (!batchContexts.has(key)) {
|
||||
batchContexts.set(key, runContext);
|
||||
}
|
||||
await messageBatcher.enqueue(chatId, senderOpenId, cleanPrompt);
|
||||
return;
|
||||
}
|
||||
|
||||
await startAgentRun(runContext, cleanPrompt);
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user