From 968a0ba950eec1ea29b69d73a601af24eb017943 Mon Sep 17 00:00:00 2001 From: Deathgiver Date: Sat, 14 Nov 2026 12:12:15 +0700 Subject: [PATCH] fix(comment-automation): block private replies once the bot quota is reached A `text` or `AIAgent` private reply goes straight out through the comment-anchored Send API, never through the chat worker, so the bot message quota gate never saw it: an over-quota workspace kept sending these DMs while every other bot send was blocked. The private reply gate now checks the owner's bot message quota (read at most once per comment, fail-open on a lookup error) and, when it is reached, records a `failed` event with "Private reply not sent: Bot message limit reached" instead of sending. The deferred TikTok private reply re-checks it at send time. The reply-once slot is released as for any other blocked private reply. Co-Authored-By: Claude Opus 5.5 --- .../__tests__/comment-automation.test.ts | 59 ++++++++++++ .../comment-private-reply-quota.test.ts | 95 +++++++++++++++++++ .../__tests__/deferred-private-reply.test.ts | 27 ++++++ .../deferred-private-reply.ts | 26 +++++ .../handlers/comment-automation/index.ts | 24 +++++ .../comment-automation/private-reply-quota.ts | 32 +++++++ 6 files changed, 263 insertions(+) create mode 100644 apps/worker/__tests__/comment-private-reply-quota.test.ts create mode 100644 apps/worker/src/integration/handlers/comment-automation/private-reply-quota.ts diff --git a/apps/worker/__tests__/comment-automation.test.ts b/apps/worker/__tests__/comment-automation.test.ts index b44895590c..e35faf2f55 100644 --- a/apps/worker/__tests__/comment-automation.test.ts +++ b/apps/worker/__tests__/comment-automation.test.ts @@ -28,6 +28,7 @@ const { mockIntegrationQueueAdd, mockChatQueueAdd, mockSendPrivateReply, + mockIsBotMessageQuotaReached, mockSendInstagramPrivateReply, mockSendInstagramFacebookPrivateReply, mockSendTiktokPrivateReply, @@ -80,6 +81,7 @@ const { mockClaimContentAttributes: vi.fn(), mockNeedsAttachmentInfo: vi.fn(), mockResolveAttachmentInfo: vi.fn(), + mockIsBotMessageQuotaReached: vi.fn(), })) const mockLogProviderError = vi.fn().mockResolvedValue(undefined) @@ -212,6 +214,10 @@ vi.mock("../src/chat/handlers/send-message", () => ({ recordDeliveredWhenNoReceipts: mockRecordDeliveredWhenNoReceipts, })) +vi.mock("../src/lib/is-bot-message-quota-reached", () => ({ + isBotMessageQuotaReached: mockIsBotMessageQuotaReached, +})) + vi.mock("../src/lib/logger", () => ({ logger: { error: vi.fn(), @@ -437,6 +443,7 @@ beforeEach(() => { // `clearAllMocks` wipes call history but keeps implementations, so a test // that makes a sender reject would leak that into every later test. mockSendPrivateReply.mockResolvedValue(undefined) + mockIsBotMessageQuotaReached.mockResolvedValue(false) mockSendInstagramPrivateReply.mockResolvedValue(undefined) mockSendInstagramFacebookPrivateReply.mockResolvedValue(undefined) mockChatQueueAdd.mockResolvedValue(undefined) @@ -3298,6 +3305,58 @@ describe("processCommentAutomation blocked private reply", () => { expect(mockLogProviderError).not.toHaveBeenCalled() }) + // A `text` private reply goes straight to the Send API, past the chat + // worker's quota gate, so it is checked before sending. + test("records the bot message quota as a failed event and sends nothing", async () => { + mockIsBotMessageQuotaReached.mockResolvedValue(true) + mockFindActiveAutomations.mockResolvedValue([ + buildAutomation({ privateReply: { type: "text", value: "over quota" } }), + ]) + + await processCommentAutomation(buildJobData() as any) + + expect(mockSendPrivateReply).not.toHaveBeenCalled() + expect(mockRecordEvent).toHaveBeenCalledWith( + expect.objectContaining({ + replyChannel: "private", + replyType: "text", + status: "failed", + errorDetail: "Private reply not sent: Bot message limit reached", + }), + ) + expect(mockLogProviderError).not.toHaveBeenCalled() + }) + + test("reads the quota once per comment, and not at all without a private reply", async () => { + mockFindActiveAutomations.mockResolvedValue([ + buildAutomation({ + id: "automation-1", + publicReply: { type: "text", value: "public only" }, + privateReply: { type: "none", value: null }, + }), + ]) + + await processCommentAutomation(buildJobData() as any) + + expect(mockIsBotMessageQuotaReached).not.toHaveBeenCalled() + + mockIsBotMessageQuotaReached.mockResolvedValue(true) + mockFindActiveAutomations.mockResolvedValue([ + buildAutomation({ + id: "automation-1", + privateReply: { type: "text", value: "first" }, + }), + buildAutomation({ + id: "automation-2", + privateReply: { type: "text", value: "second" }, + }), + ]) + + await processCommentAutomation(buildJobData() as any) + + expect(mockIsBotMessageQuotaReached).toHaveBeenCalledOnce() + }) + test("the blocked automation's public reply still goes out", async () => { mockFindActiveAutomations.mockResolvedValue([ buildAutomation({ diff --git a/apps/worker/__tests__/comment-private-reply-quota.test.ts b/apps/worker/__tests__/comment-private-reply-quota.test.ts new file mode 100644 index 0000000000..fe0cc2af20 --- /dev/null +++ b/apps/worker/__tests__/comment-private-reply-quota.test.ts @@ -0,0 +1,95 @@ +import { beforeEach, describe, expect, test, vi } from "vitest" + +const mocks = vi.hoisted(() => ({ + isBotMessageQuotaReached: vi.fn(), + loggerError: vi.fn(), +})) + +vi.mock("../src/lib/is-bot-message-quota-reached", () => ({ + isBotMessageQuotaReached: mocks.isBotMessageQuotaReached, +})) +vi.mock("../src/lib/logger", () => ({ + logger: { + error: mocks.loggerError, + warn: vi.fn(), + info: vi.fn(), + debug: vi.fn(), + }, +})) + +const { isPrivateReplyQuotaReached, PRIVATE_REPLY_QUOTA_REACHED_ERROR } = + await import( + "../src/integration/handlers/comment-automation/private-reply-quota" + ) +const { resolvePrivateReplyBlockedReason } = await import( + "../src/integration/handlers/comment-automation" +) + +beforeEach(() => { + vi.clearAllMocks() +}) + +describe("isPrivateReplyQuotaReached", () => { + test.each([ + true, + false, + ])("returns the owner's quota verdict (%s)", async (reached) => { + mocks.isBotMessageQuotaReached.mockResolvedValue(reached) + + await expect(isPrivateReplyQuotaReached("workspace-1")).resolves.toBe( + reached, + ) + expect(mocks.isBotMessageQuotaReached).toHaveBeenCalledWith("workspace-1") + }) + + test("fails open when the quota lookup throws", async () => { + mocks.isBotMessageQuotaReached.mockRejectedValue(new Error("redis down")) + + await expect(isPrivateReplyQuotaReached("workspace-1")).resolves.toBe(false) + expect(mocks.loggerError).toHaveBeenCalledWith( + expect.objectContaining({ err: expect.any(Error) }), + expect.any(String), + ) + }) +}) + +describe("resolvePrivateReplyBlockedReason bot message quota", () => { + const base = { + privateReply: { type: "text" as const, value: "hi" }, + privateReplyClaimed: false, + channelType: "instagram" as const, + createdTime: Math.floor(Date.now() / 1000), + delay: 0, + } + + test("blocks a private reply once the quota is reached", () => { + expect( + resolvePrivateReplyBlockedReason({ + ...base, + botMessageQuotaReached: true, + }), + ).toEqual({ + logReason: "bot message quota reached", + errorDetail: PRIVATE_REPLY_QUOTA_REACHED_ERROR, + }) + }) + + test("lets a private reply through below the quota", () => { + expect( + resolvePrivateReplyBlockedReason({ + ...base, + botMessageQuotaReached: false, + }), + ).toBeNull() + }) + + test("reports the channel's own rule first when both apply", () => { + expect( + resolvePrivateReplyBlockedReason({ + ...base, + privateReplyClaimed: true, + botMessageQuotaReached: true, + })?.logReason, + ).toBe("private reply already claimed for this comment") + }) +}) diff --git a/apps/worker/__tests__/deferred-private-reply.test.ts b/apps/worker/__tests__/deferred-private-reply.test.ts index 34c4bed5aa..5b6ed81cae 100644 --- a/apps/worker/__tests__/deferred-private-reply.test.ts +++ b/apps/worker/__tests__/deferred-private-reply.test.ts @@ -12,6 +12,7 @@ const { mockContactVariableGetAll, mockContactVariableReplaceAll, mockLoggerInfo, + mockIsBotMessageQuotaReached, } = vi.hoisted(() => ({ mockFindContactInboxBy: vi.fn(), mockIncrementRepliesCount: vi.fn(), @@ -24,6 +25,7 @@ const { mockContactVariableGetAll: vi.fn(), mockContactVariableReplaceAll: vi.fn(), mockLoggerInfo: vi.fn(), + mockIsBotMessageQuotaReached: vi.fn(), })) vi.mock("@chatbotx.io/business", () => ({ @@ -90,6 +92,10 @@ vi.mock("../src/chat/handlers/send-message", () => ({ emitBotMessageSentEvents: vi.fn(), })) +vi.mock("../src/lib/is-bot-message-quota-reached", () => ({ + isBotMessageQuotaReached: mockIsBotMessageQuotaReached, +})) + vi.mock("../src/lib/logger", () => ({ logger: { error: vi.fn(), @@ -172,6 +178,7 @@ describe("runDeferredCommentPrivateReply", () => { ({ text }: { text: string }) => Promise.resolve(text), ) mockIntegrationQueueAdd.mockResolvedValue(undefined) + mockIsBotMessageQuotaReached.mockResolvedValue(false) }) test("re-enqueues itself while the comment is still unflagged", async () => { @@ -247,6 +254,26 @@ describe("runDeferredCommentPrivateReply", () => { ) }) + // The DM goes straight to TikTok's Send API, past the chat worker's quota + // gate, and the quota can run out during the deferral wait. + test("records a blocked event when the bot message quota is reached", async () => { + mockFindBySourceId.mockResolvedValue(flaggedComment) + mockIsBotMessageQuotaReached.mockResolvedValue(true) + + await runDeferredCommentPrivateReply(buildData()) + + expect(mockIsBotMessageQuotaReached).toHaveBeenCalledWith("workspace-1") + expect(mockSendTiktokPrivateReply).not.toHaveBeenCalled() + expect(mockIncrementRepliesCount).not.toHaveBeenCalled() + expect(mockRecordEvent).toHaveBeenCalledWith( + expect.objectContaining({ + replyChannel: "private", + status: "failed", + errorDetail: "Private reply not sent: Bot message limit reached", + }), + ) + }) + // The integration can be disconnected during the 45s–13min wait. With // `attempts: 1` a throw out of this handler dies leaving no `failed` row and // no Error Log entry — while the dedup row has already claimed the comment, diff --git a/apps/worker/src/integration/handlers/comment-automation/deferred-private-reply.ts b/apps/worker/src/integration/handlers/comment-automation/deferred-private-reply.ts index 52d7b3fd9e..812587271a 100644 --- a/apps/worker/src/integration/handlers/comment-automation/deferred-private-reply.ts +++ b/apps/worker/src/integration/handlers/comment-automation/deferred-private-reply.ts @@ -18,6 +18,10 @@ import { type PrivateReplyAuth, privateReplyWindowLabel, } from "./private-reply" +import { + isPrivateReplyQuotaReached, + PRIVATE_REPLY_QUOTA_REACHED_ERROR, +} from "./private-reply-quota" import { recordAndDispatchReply, recordBlockedPrivateReply, @@ -195,6 +199,28 @@ export async function runDeferredCommentPrivateReply( return } + // Re-checked at send time: the quota can run out during the deferral wait, + // and this DM goes straight to the Send API, past the chat worker's gate. + if (await isPrivateReplyQuotaReached(workspaceId)) { + logger.info( + { automationId, commentId, workspaceId }, + "Deferred private reply skipped: bot message quota reached", + ) + await recordBlockedPrivateReply({ + workspaceId, + automationId, + contactInbox, + commentId, + postId, + message, + occurredAt, + replyChannel: "private", + replyType: privateReply.type, + errorDetail: PRIVATE_REPLY_QUOTA_REACHED_ERROR, + }) + return + } + let outcome: Awaited> = null try { // Inside the try, not before it: the integration can be disconnected during diff --git a/apps/worker/src/integration/handlers/comment-automation/index.ts b/apps/worker/src/integration/handlers/comment-automation/index.ts index bba0b664aa..ab28d27061 100644 --- a/apps/worker/src/integration/handlers/comment-automation/index.ts +++ b/apps/worker/src/integration/handlers/comment-automation/index.ts @@ -57,6 +57,10 @@ import { privateReplyWindowLabel, supportsPrivateReply, } from "./private-reply" +import { + isPrivateReplyQuotaReached, + PRIVATE_REPLY_QUOTA_REACHED_ERROR, +} from "./private-reply-quota" import { executePublicReply } from "./public-reply" import { logAutomationSkipped, @@ -281,6 +285,14 @@ export async function processCommentAutomation( }) : false + // Read at most once per comment, and only once an automation actually has a + // private reply to send — see `isPrivateReplyQuotaReached`. + let privateReplyQuotaReached: boolean | undefined + const resolvePrivateReplyQuotaReached = async (): Promise => { + privateReplyQuotaReached ??= await isPrivateReplyQuotaReached(workspaceId) + return privateReplyQuotaReached + } + // Every automation that declines this comment, flushed in ONE insert after // the loop. `findActiveAutomations` scopes by workspace + channel + account, // not by post, so a single comment is shown to every active automation of @@ -702,6 +714,9 @@ export async function processCommentAutomation( const privateReplyBlockedReason = resolvePrivateReplyBlockedReason({ privateReply: automation.privateReply, privateReplyClaimed, + botMessageQuotaReached: + willSendReply(automation.privateReply) && + (await resolvePrivateReplyQuotaReached()), channelType, createdTime, delay, @@ -944,6 +959,8 @@ export async function processCommentAutomation( export function resolvePrivateReplyBlockedReason(props: { privateReply: CommentReply privateReplyClaimed: boolean + /** See `isPrivateReplyQuotaReached`: an inline DM bypasses the chat gate. */ + botMessageQuotaReached: boolean channelType: CommentAutomationChannelType createdTime: number delay: number @@ -976,6 +993,13 @@ export function resolvePrivateReplyBlockedReason(props: { } } + if (props.botMessageQuotaReached) { + return { + logReason: "bot message quota reached", + errorDetail: PRIVATE_REPLY_QUOTA_REACHED_ERROR, + } + } + return null } diff --git a/apps/worker/src/integration/handlers/comment-automation/private-reply-quota.ts b/apps/worker/src/integration/handlers/comment-automation/private-reply-quota.ts new file mode 100644 index 0000000000..ba61318e20 --- /dev/null +++ b/apps/worker/src/integration/handlers/comment-automation/private-reply-quota.ts @@ -0,0 +1,32 @@ +import { isBotMessageQuotaReached } from "../../../lib/is-bot-message-quota-reached" +import { logger } from "../../../lib/logger" + +/** Error Logs detail for a private reply the bot message quota blocked. */ +export const PRIVATE_REPLY_QUOTA_REACHED_ERROR = + "Private reply not sent: Bot message limit reached" + +/** + * Whether the workspace owner's bot message quota blocks a private reply. + * + * A `text` or `AIAgent` private reply goes straight out through the + * comment-anchored Send API, never through the chat worker, so the chat + * worker's quota gate never sees it. It is checked here instead, before the + * DM is sent or queued. A `flow` reply is checked too, so a blocked one is + * recorded the same way rather than spending the comment's single DM. + * + * Fail-open on a lookup error, like the chat worker's gate on a job with no + * owner: a quota read that breaks must not silence every private reply. + */ +export async function isPrivateReplyQuotaReached( + workspaceId: string, +): Promise { + try { + return await isBotMessageQuotaReached(workspaceId) + } catch (err) { + logger.error( + { err, workspaceId }, + "Failed to check the bot message quota for a private reply", + ) + return false + } +}