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 + } +}