fix(comment-automation): block private replies once the bot quota is reached

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({

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

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,

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<ReturnType<typeof executePrivateReply>> = null
   try {
     // Inside the try, not before it: the integration can be disconnected during

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<boolean> => {
+    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
 }
 

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<boolean> {
+  try {
+    return await isBotMessageQuotaReached(workspaceId)
+  } catch (err) {
+    logger.error(
+      { err, workspaceId },
+      "Failed to check the bot message quota for a private reply",
+    )
+    return false
+  }
+}