diff --git a/apps/worker/__tests__/chat-worker-quota-gate.test.ts b/apps/worker/__tests__/chat-worker-quota-gate.test.ts index 2f70b01c88..ef18439193 100644 --- a/apps/worker/__tests__/chat-worker-quota-gate.test.ts +++ b/apps/worker/__tests__/chat-worker-quota-gate.test.ts @@ -7,6 +7,7 @@ const mocks = vi.hoisted(() => ({ isBlockedWorkspace: vi.fn(), isBotMessageQuotaReached: vi.fn(), processJob: undefined as undefined | ((job: unknown) => Promise), + recordQuotaSkippedSend: vi.fn(), resolveWorkspaceId: vi.fn(), sendFlowStep: vi.fn(), sendMessageToChannel: vi.fn(), @@ -103,6 +104,9 @@ vi.mock("../src/chat/handlers/send-message", () => ({ sendMessageToChannel: mocks.sendMessageToChannel, sendTypingToChannel: mocks.sendTypingToChannel, })) +vi.mock("../src/chat/handlers/record-quota-skipped-send", () => ({ + recordQuotaSkippedSend: mocks.recordQuotaSkippedSend, +})) vi.mock("../src/chat/handlers/send-messenger-template", () => ({ sendMessengerTemplateMessage: vi.fn(), })) @@ -222,6 +226,27 @@ describe("chat worker bot message quota gate", () => { expect(mocks.sendMessageToChannel).not.toHaveBeenCalled() }) + test("records the skipped send so it is not left looking delivered", async () => { + mocks.isBotMessageQuotaReached.mockResolvedValue(true) + const data = { + type: "sendChannelMessage", + data: { message: { senderType: "bot" } }, + } + + await mocks.processJob?.({ id: "job-1", data }) + + expect(mocks.recordQuotaSkippedSend).toHaveBeenCalledWith(data) + }) + + test("records nothing when the send goes ahead below the quota", async () => { + await mocks.processJob?.({ + id: "job-1", + data: { type: "sendFlowMessage", data: {} }, + }) + + expect(mocks.recordQuotaSkippedSend).not.toHaveBeenCalled() + }) + test("invokes a user channel-message handler even when the quota is reached", async () => { mocks.isBotMessageQuotaReached.mockResolvedValue(true) diff --git a/apps/worker/__tests__/record-quota-skipped-send.test.ts b/apps/worker/__tests__/record-quota-skipped-send.test.ts new file mode 100644 index 0000000000..67fbe8e02d --- /dev/null +++ b/apps/worker/__tests__/record-quota-skipped-send.test.ts @@ -0,0 +1,164 @@ +import { beforeEach, describe, expect, test, vi } from "vitest" + +const mocks = vi.hoisted(() => ({ + recordMessageSendError: vi.fn(), + settleCommentAutomationFailure: vi.fn(), + loggerError: vi.fn(), +})) + +vi.mock("../src/chat/handlers/send-message", () => ({ + recordMessageSendError: mocks.recordMessageSendError, +})) +vi.mock("../src/lib/comment-automation-anchor", () => ({ + settleCommentAutomationFailure: mocks.settleCommentAutomationFailure, +})) +vi.mock("../src/lib/logger", () => ({ + logger: { error: mocks.loggerError, info: vi.fn(), warn: vi.fn() }, +})) + +const { BOT_MESSAGE_QUOTA_REACHED_ERROR, recordQuotaSkippedSend } = + await import("../src/chat/handlers/record-quota-skipped-send") + +type JobData = Parameters[0] + +const createdAt = "2026-11-14T00:00:00.000Z" +const commentReplyAttributes = { + replyToCommentId: "comment-1", + commentAutomation: { automationId: "automation-1", replyChannel: "public" }, +} + +function channelMessageJob( + message: Record, + extra: Record = {}, +): JobData { + return { + type: "sendChannelMessage", + data: { + conversation: { id: "conversation-1", workspaceId: "workspace-1" }, + contactInbox: { id: "contact-inbox-1" }, + message: { + id: "message-1", + clientId: "client-1", + createdAt, + senderType: "bot", + ...message, + }, + ...extra, + }, + } as unknown as JobData +} + +function flowMessageJob(commentAnchor?: Record): JobData { + return { + type: "sendFlowMessage", + data: { commentAnchor }, + } as unknown as JobData +} + +beforeEach(() => { + vi.clearAllMocks() +}) + +describe("recordQuotaSkippedSend", () => { + test("marks an existing channel message failed with the quota reason", async () => { + await recordQuotaSkippedSend( + channelMessageJob({ contentAttributes: commentReplyAttributes }), + ) + + expect(mocks.recordMessageSendError).toHaveBeenCalledWith( + "message-1", + "client-1", + "workspace-1", + new Date(createdAt), + BOT_MESSAGE_QUOTA_REACHED_ERROR, + { conversationId: "conversation-1", silent: false }, + ) + expect(mocks.settleCommentAutomationFailure).toHaveBeenCalledWith({ + contentAttributes: commentReplyAttributes, + errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR, + }) + }) + + test("keeps a bulk broadcast message out of realtime", async () => { + await recordQuotaSkippedSend( + channelMessageJob( + {}, + { isBulkBroadcast: true, metadata: { type: "broadcast" } }, + ), + ) + + expect(mocks.recordMessageSendError).toHaveBeenCalledWith( + expect.anything(), + expect.anything(), + expect.anything(), + expect.anything(), + BOT_MESSAGE_QUOTA_REACHED_ERROR, + { conversationId: "conversation-1", silent: true }, + ) + }) + + test("settles a comment-automation flow reply failed", async () => { + await recordQuotaSkippedSend( + flowMessageJob({ + commentId: "comment-1", + replyChannel: "private", + automationId: "automation-1", + }), + ) + + expect(mocks.recordMessageSendError).not.toHaveBeenCalled() + expect(mocks.settleCommentAutomationFailure).toHaveBeenCalledWith({ + contentAttributes: { + replyToCommentId: "comment-1", + commentAutomation: { + automationId: "automation-1", + replyChannel: "private", + }, + }, + errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR, + }) + }) + + test.each([ + ["no anchor", undefined], + [ + "an anchor without an automation", + { commentId: "comment-1", replyChannel: "private" }, + ], + [ + "a spent anchor (first DM already delivered)", + { + commentId: "comment-1", + replyChannel: "private", + automationId: "automation-1", + spent: true, + }, + ], + ])("leaves a flow message with %s untouched", async (_name, anchor) => { + await recordQuotaSkippedSend(flowMessageJob(anchor)) + + expect(mocks.settleCommentAutomationFailure).not.toHaveBeenCalled() + }) + + test("does nothing for a send whose message is created by the handler", async () => { + await recordQuotaSkippedSend({ + type: "sendChatMessage", + data: {}, + } as unknown as JobData) + + expect(mocks.recordMessageSendError).not.toHaveBeenCalled() + expect(mocks.settleCommentAutomationFailure).not.toHaveBeenCalled() + }) + + test("swallows its own failures so the job is still dropped", async () => { + mocks.recordMessageSendError.mockRejectedValueOnce(new Error("db down")) + + await expect( + recordQuotaSkippedSend(channelMessageJob({})), + ).resolves.toBeUndefined() + expect(mocks.loggerError).toHaveBeenCalledWith( + expect.objectContaining({ err: expect.any(Error) }), + expect.any(String), + ) + }) +}) diff --git a/apps/worker/src/chat/handlers/record-quota-skipped-send.ts b/apps/worker/src/chat/handlers/record-quota-skipped-send.ts new file mode 100644 index 0000000000..3aed5e2fb5 --- /dev/null +++ b/apps/worker/src/chat/handlers/record-quota-skipped-send.ts @@ -0,0 +1,92 @@ +import { isBulkOutboundMetadata } from "@chatbotx.io/flow-config" +import { ChatJobAction, type ChatJobData } from "@chatbotx.io/worker-config" +import { settleCommentAutomationFailure } from "../../lib/comment-automation-anchor" +import { logger } from "../../lib/logger" +import { recordMessageSendError } from "./send-message" + +/** The `sendError` / Error Logs detail for a send the bot quota dropped. */ +export const BOT_MESSAGE_QUOTA_REACHED_ERROR = "Bot message limit reached" + +/** + * Records a bot send that the chat worker's quota gate dropped, so it does not + * keep looking delivered. + * + * - `sendChannelMessage`: the outgoing `Message` row already exists (a public + * comment reply is written to the inbox before it is enqueued), so it gets a + * `sendError` and a realtime `messageFailed`, and its comment-automation event + * flips from the optimistic `sent` to `failed`. + * - `sendFlowMessage` carrying a comment-automation anchor: no message exists + * yet, so only the automation event is settled `failed`. A `spent` anchor + * means the comment's first DM already went out, so that event stays as is. + * - Every other bot send creates its message inside the handler, so there is + * nothing to mark. + * + * Never throws: the gate must still drop the job without a retry. + */ +export async function recordQuotaSkippedSend( + jobData: ChatJobData, +): Promise { + try { + if (jobData.type === ChatJobAction.sendChannelMessage) { + await recordSkippedChannelMessage(jobData.data) + return + } + if (jobData.type === ChatJobAction.sendFlowMessage) { + await recordSkippedFlowMessage(jobData.data) + } + } catch (err) { + logger.error( + { err, jobType: jobData.type }, + "Failed to record a send skipped by the bot message quota", + ) + } +} + +async function recordSkippedChannelMessage( + data: Extract< + ChatJobData, + { type: typeof ChatJobAction.sendChannelMessage } + >["data"], +): Promise { + const { conversation, message, metadata, isBulkBroadcast } = data + const isBulkOutbound = + isBulkOutboundMetadata(metadata, isBulkBroadcast) || + isBulkOutboundMetadata(message.contentAttributes?.metadata, isBulkBroadcast) + + await recordMessageSendError( + message.id, + message.clientId, + conversation.workspaceId, + message.createdAt ? new Date(message.createdAt) : undefined, + BOT_MESSAGE_QUOTA_REACHED_ERROR, + { conversationId: conversation.id, silent: isBulkOutbound }, + ) + await settleCommentAutomationFailure({ + contentAttributes: message.contentAttributes, + errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR, + }) +} + +async function recordSkippedFlowMessage( + data: Extract< + ChatJobData, + { type: typeof ChatJobAction.sendFlowMessage } + >["data"], +): Promise { + const anchor = data.commentAnchor + if (!anchor?.automationId || anchor.spent) { + return + } + + // The shape `readCommentAutomationAnchor` reads, as `sendFlowStep` stamps it. + await settleCommentAutomationFailure({ + contentAttributes: { + replyToCommentId: anchor.commentId, + commentAutomation: { + automationId: anchor.automationId, + replyChannel: anchor.replyChannel, + }, + }, + errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR, + }) +} diff --git a/apps/worker/src/chat/worker.ts b/apps/worker/src/chat/worker.ts index a30908472e..fd9a788e0a 100644 --- a/apps/worker/src/chat/worker.ts +++ b/apps/worker/src/chat/worker.ts @@ -15,6 +15,7 @@ import { resolveWorkspaceId } from "../lib/resolve-workspace-id" import { runJobWithAuditContext } from "../lib/run-job-with-audit-context" import { runWorker } from "../lib/shutdown" import { checkOutboundAutomatedResponse } from "./handlers/outbound-automated-response" +import { recordQuotaSkippedSend } from "./handlers/record-quota-skipped-send" import { sendChatMessage, sendFlowStep } from "./handlers/send-flow-step" import { changeMessageStateOnChannel, @@ -76,6 +77,7 @@ function startChatWorker() { { jobId: job.id, workspaceId }, "Skipping bot send — quota reached", ) + await recordQuotaSkippedSend(job.data) return }