perf(worker): harden chat queue dispatch (concurrency, dedup, retries)

apps/worker/src/chat/worker.ts

@@ -9,6 +9,7 @@ import {
   queueNames,
 } from "@chatbotx.io/worker-config"
 import { type Job, Worker } from "bullmq"
+import { env } from "../env"
 import { ensureBootstrapped } from "../lib/bootstrap"
 import { isBlockedWorkspace } from "../lib/is-blocked-workspace"
 import { isBotMessageQuotaReached } from "../lib/is-bot-message-quota-reached"
@@ -139,6 +140,11 @@ async function startChatWorker() {
     {
       connection: getRedisConnection(),
       ...defaultWorkerOptions,
+      concurrency: env.CHAT_WORKER_CONCURRENCY,
+      limiter: {
+        max: env.CHAT_WORKER_RATE_LIMIT_MAX,
+        duration: env.CHAT_WORKER_RATE_LIMIT_DURATION_MS,
+      },
     },
   )
 

apps/worker/src/env.ts

@@ -9,6 +9,21 @@ export const env = createEnv({
   server: {
     NEXT_PUBLIC_EDITION: editionRule,
     QUOTA_SYNC_INTERVAL_SECONDS: z.coerce.number().int().min(10).default(60),
+    CHAT_WORKER_CONCURRENCY: z.coerce
+      .number()
+      .int()
+      .min(1)
+      .max(200)
+      .default(20),
+    // BullMQ's own rate limiter is queue-wide, not per-channel/page — a
+    // coarse throughput cap on the whole chat queue, not a substitute for a
+    // per-inbox token bucket in front of each provider API call.
+    CHAT_WORKER_RATE_LIMIT_MAX: z.coerce.number().int().min(1).default(80),
+    CHAT_WORKER_RATE_LIMIT_DURATION_MS: z.coerce
+      .number()
+      .int()
+      .min(1)
+      .default(1000),
     WEBHOOK_WORKER_CONCURRENCY: z.coerce
       .number()
       .int()

integrations/instagram/__tests__/webhook-job-id.test.ts

@@ -0,0 +1,45 @@
+import { describe, expect, test, vi } from "vitest"
+import { webhookHandler } from "../src/handlers/webhook"
+import { hmacSha256Hex } from "../src/lib/webhook"
+
+const CLIENT_SECRET = "webhook-secret"
+
+describe("instagram webhook incomingMessage job IDs", () => {
+  test("uses a BullMQ-safe deterministic job ID for message events", async () => {
+    const body = JSON.stringify({
+      object: "instagram",
+      entry: [
+        {
+          id: "instagram-1",
+          time: 1_700_000_000,
+          messaging: [
+            {
+              sender: { id: "contact-1" },
+              recipient: { id: "instagram-1" },
+              timestamp: 1_700_000_000,
+              message: { mid: "mid:1/2", text: "hi" },
+            },
+          ],
+        },
+      ],
+    })
+    const signature = await hmacSha256Hex(CLIENT_SECRET, body)
+    const add = vi.fn()
+
+    await webhookHandler({
+      config: { clientSecret: CLIENT_SECRET },
+      req: new Request("https://example.test/webhook", {
+        method: "POST",
+        body,
+        headers: { "x-hub-signature-256": `sha256=${signature}` },
+      }),
+      queue: { add },
+    } as never)
+
+    expect(add).toHaveBeenCalledWith(
+      "incomingMessage",
+      expect.objectContaining({ type: "incomingMessage" }),
+      { jobId: "incoming-instagram-mid_1_2" },
+    )
+  })
+})

integrations/instagram/src/handlers/webhook.ts

@@ -1,4 +1,5 @@
 import type { ContextQueue, HandleRequestProps } from "@chatbotx.io/sdk"
+import { toBullMqSafeIdSegment } from "@chatbotx.io/utils"
 import z from "zod"
 import { InstagramWebhookException } from "../exception"
 import { logger } from "../lib/logger"
@@ -211,14 +212,25 @@ const handleWebhookEvent = async (
           ? messagingEvent.sender.id
           : messagingEvent.recipient.id
 
-        await queue?.add("incomingMessage", {
-          type: "incomingMessage",
-          data: {
-            integrationType: "instagram",
-            integrationIdentifier,
-            payload: singleEventPayload,
+        const sourceMessageId =
+          messagingEvent.message?.mid ?? messagingEvent.postback?.mid
+
+        await queue?.add(
+          "incomingMessage",
+          {
+            type: "incomingMessage",
+            data: {
+              integrationType: "instagram",
+              integrationIdentifier,
+              payload: singleEventPayload,
+            },
           },
-        })
+          sourceMessageId === undefined
+            ? undefined
+            : {
+                jobId: `incoming-instagram-${toBullMqSafeIdSegment(sourceMessageId)}`,
+              },
+        )
       }
     }
   } catch (error) {

integrations/messenger/__tests__/webhook-job-id.test.ts

@@ -0,0 +1,47 @@
+import { createHmac } from "node:crypto"
+import { describe, expect, test, vi } from "vitest"
+import { webhookHandler } from "../src/handlers/webhook"
+
+const CLIENT_SECRET = "webhook-secret"
+
+describe("messenger webhook incomingMessage job IDs", () => {
+  test("uses a BullMQ-safe deterministic job ID for message events", async () => {
+    const body = JSON.stringify({
+      object: "page",
+      entry: [
+        {
+          id: "page-1",
+          time: 1_700_000_000,
+          messaging: [
+            {
+              sender: { id: "contact-1" },
+              recipient: { id: "page-1" },
+              timestamp: 1_700_000_000,
+              message: { mid: "mid:1/2", text: "hi" },
+            },
+          ],
+        },
+      ],
+    })
+    const signature = createHmac("sha256", CLIENT_SECRET)
+      .update(body)
+      .digest("hex")
+    const add = vi.fn()
+
+    await webhookHandler({
+      config: { clientSecret: CLIENT_SECRET },
+      req: new Request("https://example.test/webhook", {
+        method: "POST",
+        body,
+        headers: { "x-hub-signature-256": `sha256=${signature}` },
+      }),
+      queue: { add },
+    } as never)
+
+    expect(add).toHaveBeenCalledWith(
+      "incomingMessage",
+      expect.objectContaining({ type: "incomingMessage" }),
+      { jobId: "incoming-messenger-mid_1_2" },
+    )
+  })
+})

integrations/messenger/src/handlers/webhook.ts

@@ -1,4 +1,5 @@
 import type { ContextQueue, HandleRequestProps } from "@chatbotx.io/sdk"
+import { toBullMqSafeIdSegment } from "@chatbotx.io/utils"
 import z from "zod"
 import { MessengerWebhookException } from "../exception"
 import { logger } from "../lib/logger"
@@ -263,16 +264,27 @@ const handleWebhookEvent = async (
           ? messagingEvent.sender.id
           : messagingEvent.recipient.id
 
+        const sourceMessageId =
+          messagingEvent.postback?.mid ?? messagingEvent.message?.mid
+
         if (messagingEvent.postback) {
-          await queue?.add("incomingMessage", {
-            type: "incomingMessage",
-            data: {
-              integrationType: "messenger",
-              integrationIdentifier,
-              payload: singleEventPayload,
-              action: messagingEvent.postback.payload,
+          await queue?.add(
+            "incomingMessage",
+            {
+              type: "incomingMessage",
+              data: {
+                integrationType: "messenger",
+                integrationIdentifier,
+                payload: singleEventPayload,
+                action: messagingEvent.postback.payload,
+              },
             },
-          })
+            sourceMessageId === undefined
+              ? undefined
+              : {
+                  jobId: `incoming-messenger-${toBullMqSafeIdSegment(sourceMessageId)}`,
+                },
+          )
           continue
         }
 
@@ -284,14 +296,22 @@ const handleWebhookEvent = async (
           continue
         }
 
-        await queue?.add("incomingMessage", {
-          type: "incomingMessage",
-          data: {
-            integrationType: "messenger",
-            integrationIdentifier,
-            payload: singleEventPayload,
+        await queue?.add(
+          "incomingMessage",
+          {
+            type: "incomingMessage",
+            data: {
+              integrationType: "messenger",
+              integrationIdentifier,
+              payload: singleEventPayload,
+            },
           },
-        })
+          sourceMessageId === undefined
+            ? undefined
+            : {
+                jobId: `incoming-messenger-${toBullMqSafeIdSegment(sourceMessageId)}`,
+              },
+        )
       }
     }
   } catch (error) {

integrations/telegram/__tests__/webhook-job-id.test.ts

@@ -0,0 +1,31 @@
+import { describe, expect, test, vi } from "vitest"
+import { webhookHandler } from "../src/handlers/webhook"
+
+describe("telegram webhook incomingMessage job IDs", () => {
+  test("uses the update ID and safe integration identifier", async () => {
+    const add = vi.fn()
+
+    await webhookHandler({
+      config: { botId: "bot:1/2" },
+      req: new Request("https://example.test/webhook", {
+        method: "POST",
+        body: JSON.stringify({
+          update_id: 42,
+          message: {
+            message_id: 1,
+            chat: { id: 123, type: "private" },
+            date: 1_700_000_000,
+            text: "hi",
+          },
+        }),
+      }),
+      queue: { add },
+    } as never)
+
+    expect(add).toHaveBeenCalledWith(
+      "incomingMessage",
+      expect.objectContaining({ type: "incomingMessage" }),
+      { jobId: "incoming-telegram-bot_1_2-42" },
+    )
+  })
+})

integrations/telegram/src/handlers/webhook.ts

@@ -1,4 +1,5 @@
 import type { HandleRequestProps } from "@chatbotx.io/sdk"
+import { toBullMqSafeIdSegment } from "@chatbotx.io/utils"
 import { TelegramWebhookException } from "../exception"
 import type { TelegramConfig } from "../schema"
 import { telegramUpdateSchema } from "../schema"
@@ -23,29 +24,41 @@ export const webhookHandler = async (
       return "ok"
     }
 
-    await queue?.add("incomingMessage", {
-      type: "incomingMessage",
-      data: {
-        integrationType: "telegram",
-        integrationIdentifier,
-        payload: update,
+    await queue?.add(
+      "incomingMessage",
+      {
+        type: "incomingMessage",
+        data: {
+          integrationType: "telegram",
+          integrationIdentifier,
+          payload: update,
+        },
+      },
+      {
+        jobId: `incoming-telegram-${toBullMqSafeIdSegment(integrationIdentifier)}-${update.update_id}`,
       },
-    })
+    )
     return "ok"
   }
 
   if (!update.message) {
     return "ok"
   }
 
-  await queue?.add("incomingMessage", {
-    type: "incomingMessage",
-    data: {
-      integrationType: "telegram",
-      integrationIdentifier,
-      payload: update,
+  await queue?.add(
+    "incomingMessage",
+    {
+      type: "incomingMessage",
+      data: {
+        integrationType: "telegram",
+        integrationIdentifier,
+        payload: update,
+      },
+    },
+    {
+      jobId: `incoming-telegram-${toBullMqSafeIdSegment(integrationIdentifier)}-${update.update_id}`,
     },
-  })
+  )
 
   return "ok"
 }

integrations/tiktok/__tests__/webhook-job-id.test.ts

@@ -0,0 +1,42 @@
+import { createHmac } from "node:crypto"
+import { describe, expect, test, vi } from "vitest"
+import { webhookHandler } from "../src/handlers/webhook"
+
+const CLIENT_SECRET = "webhook-secret"
+
+describe("TikTok webhook incomingMessage job IDs", () => {
+  test("preserves echo delay while adding a BullMQ-safe deterministic job ID", async () => {
+    const timestamp = Math.floor(Date.now() / 1000)
+    const body = JSON.stringify({
+      client_key: "client-1",
+      event: "im_send_msg",
+      create_time: timestamp,
+      user_openid: "user-1",
+      content: "{}",
+      message_id: "message:1/2",
+    })
+    const signature = createHmac("sha256", CLIENT_SECRET)
+      .update(`${timestamp}.${body}`)
+      .digest("hex")
+    const add = vi.fn()
+
+    await webhookHandler({
+      config: { clientSecret: CLIENT_SECRET, openId: "business:1/2" },
+      req: new Request("https://example.test/webhook", {
+        method: "POST",
+        body,
+        headers: { "TikTok-Signature": `t=${timestamp},s=${signature}` },
+      }),
+      queue: { add },
+    } as never)
+
+    expect(add).toHaveBeenCalledWith(
+      "incomingMessage",
+      expect.objectContaining({ type: "incomingMessage" }),
+      {
+        delay: 2000,
+        jobId: "incoming-tiktok-business_1_2-im_send_msg-message_1_2",
+      },
+    )
+  })
+})