From 6f18fd53dff32bb52a4dbdbbe9fec0ba2cc5b5ee Mon Sep 17 00:00:00 2001 From: Real Codesiman Date: Wed, 16 Sep 2026 13:30:42 +0700 Subject: [PATCH 1/4] perf(worker): make chat queue concurrency and rate limit configurable --- apps/worker/src/chat/worker.ts | 6 ++++++ packages/worker-config/src/keys.ts | 11 +++++++++++ 2 files changed, 17 insertions(+) diff --git a/apps/worker/src/chat/worker.ts b/apps/worker/src/chat/worker.ts index 31008417eb..cb44df7e19 100644 --- a/apps/worker/src/chat/worker.ts +++ b/apps/worker/src/chat/worker.ts @@ -8,6 +8,7 @@ import { getRedisConnection, queueNames, } from "@chatbotx.io/worker-config" +import { keys } from "@chatbotx.io/worker-config/keys" import { type Job, Worker } from "bullmq" import { ensureBootstrapped } from "../lib/bootstrap" import { isBlockedWorkspace } from "../lib/is-blocked-workspace" @@ -139,6 +140,11 @@ async function startChatWorker() { { connection: getRedisConnection(), ...defaultWorkerOptions, + concurrency: keys().CHAT_WORKER_CONCURRENCY, + limiter: { + max: keys().CHAT_WORKER_RATE_LIMIT_MAX, + duration: keys().CHAT_WORKER_RATE_LIMIT_DURATION_MS, + }, }, ) diff --git a/packages/worker-config/src/keys.ts b/packages/worker-config/src/keys.ts index 3500b99a59..990327a3aa 100644 --- a/packages/worker-config/src/keys.ts +++ b/packages/worker-config/src/keys.ts @@ -6,6 +6,17 @@ export const keys = () => server: { REDIS_URL: z.url(), NEXT_PHASE: z.string().default(""), + CHAT_WORKER_CONCURRENCY: z.coerce.number().int().positive().default(20), + CHAT_WORKER_RATE_LIMIT_MAX: z.coerce + .number() + .int() + .positive() + .default(80), + CHAT_WORKER_RATE_LIMIT_DURATION_MS: z.coerce + .number() + .int() + .positive() + .default(1000), }, experimental__runtimeEnv: {}, skipValidation: process.env.SKIP_ENV_CHECK === "true", From 3f566399e946d64643cbe414a9cdf6c7b58653a6 Mon Sep 17 00:00:00 2001 From: Real Codesiman Date: Wed, 16 Sep 2026 13:31:30 +0700 Subject: [PATCH 2/4] fix(integrations): dedupe incomingMessage jobs with deterministic jobId --- .../__tests__/webhook-job-id.test.ts | 45 +++++++++++++++++ .../instagram/src/handlers/webhook.ts | 26 +++++++--- .../__tests__/webhook-job-id.test.ts | 47 +++++++++++++++++ .../messenger/src/handlers/webhook.ts | 50 +++++++++++++------ .../telegram/__tests__/webhook-job-id.test.ts | 31 ++++++++++++ integrations/telegram/src/handlers/webhook.ts | 41 +++++++++------ .../tiktok/__tests__/webhook-job-id.test.ts | 42 ++++++++++++++++ integrations/tiktok/src/handlers/webhook.ts | 16 ++++-- integrations/tiktok/src/schema.ts | 2 + .../whatsapp/__tests__/webhook-job-id.test.ts | 41 +++++++++++++++ integrations/whatsapp/src/handlers/webhook.ts | 28 +++++++---- .../zalo/__tests__/webhook-job-id.test.ts | 29 +++++++++++ integrations/zalo/src/handlers/webhook.ts | 23 ++++++--- packages/utils/__tests__/id.test.ts | 10 ++++ packages/utils/src/id.ts | 3 ++ 15 files changed, 377 insertions(+), 57 deletions(-) create mode 100644 integrations/instagram/__tests__/webhook-job-id.test.ts create mode 100644 integrations/messenger/__tests__/webhook-job-id.test.ts create mode 100644 integrations/telegram/__tests__/webhook-job-id.test.ts create mode 100644 integrations/tiktok/__tests__/webhook-job-id.test.ts create mode 100644 integrations/whatsapp/__tests__/webhook-job-id.test.ts create mode 100644 integrations/zalo/__tests__/webhook-job-id.test.ts create mode 100644 packages/utils/__tests__/id.test.ts diff --git a/integrations/instagram/__tests__/webhook-job-id.test.ts b/integrations/instagram/__tests__/webhook-job-id.test.ts new file mode 100644 index 0000000000..34ccbe756c --- /dev/null +++ b/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" }, + ) + }) +}) diff --git a/integrations/instagram/src/handlers/webhook.ts b/integrations/instagram/src/handlers/webhook.ts index 40e5f3de05..1254c95357 100644 --- a/integrations/instagram/src/handlers/webhook.ts +++ b/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) { diff --git a/integrations/messenger/__tests__/webhook-job-id.test.ts b/integrations/messenger/__tests__/webhook-job-id.test.ts new file mode 100644 index 0000000000..1c993a28cd --- /dev/null +++ b/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" }, + ) + }) +}) diff --git a/integrations/messenger/src/handlers/webhook.ts b/integrations/messenger/src/handlers/webhook.ts index a79d9f9713..56264bc5ac 100644 --- a/integrations/messenger/src/handlers/webhook.ts +++ b/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) { diff --git a/integrations/telegram/__tests__/webhook-job-id.test.ts b/integrations/telegram/__tests__/webhook-job-id.test.ts new file mode 100644 index 0000000000..ec6d3478e4 --- /dev/null +++ b/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" }, + ) + }) +}) diff --git a/integrations/telegram/src/handlers/webhook.ts b/integrations/telegram/src/handlers/webhook.ts index 8cc8d41863..6740efde60 100644 --- a/integrations/telegram/src/handlers/webhook.ts +++ b/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,14 +24,20 @@ 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" } @@ -38,14 +45,20 @@ 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" } diff --git a/integrations/tiktok/__tests__/webhook-job-id.test.ts b/integrations/tiktok/__tests__/webhook-job-id.test.ts new file mode 100644 index 0000000000..dd0fc28d52 --- /dev/null +++ b/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", + }, + ) + }) +}) diff --git a/integrations/tiktok/src/handlers/webhook.ts b/integrations/tiktok/src/handlers/webhook.ts index 8e5097b631..810096c665 100644 --- a/integrations/tiktok/src/handlers/webhook.ts +++ b/integrations/tiktok/src/handlers/webhook.ts @@ -1,4 +1,5 @@ import type { HandleRequestProps } from "@chatbotx.io/sdk" +import { toBullMqSafeIdSegment } from "@chatbotx.io/utils" import { TiktokWebhookException } from "../exception" import { logger } from "../lib/logger" import { hmacSha256Hex, timingSafeStringEqual } from "../lib/webhook" @@ -101,6 +102,17 @@ export const webhookHandler = async ( return "ok" } + const sourceMessageId = event.data.message_id ?? event.data.unique_identifier + const echoDelayOptions = + event.data.event === "im_send_msg" ? { delay: 2000 } : undefined + const queueOptions = + sourceMessageId === undefined + ? echoDelayOptions + : { + ...echoDelayOptions, + jobId: `incoming-tiktok-${toBullMqSafeIdSegment(integrationIdentifier)}-${event.data.event}-${toBullMqSafeIdSegment(sourceMessageId)}`, + } + await queue?.add( "incomingMessage", { @@ -111,9 +123,7 @@ export const webhookHandler = async ( payload: event.data, }, }, - // Add delay for echo events to avoid race condition where echo arrives - // before the send message API response completes - event.data.event === "im_send_msg" ? { delay: 2000 } : undefined, + queueOptions, ) return "ok" diff --git a/integrations/tiktok/src/schema.ts b/integrations/tiktok/src/schema.ts index 41ce9aab82..4acdcd64eb 100644 --- a/integrations/tiktok/src/schema.ts +++ b/integrations/tiktok/src/schema.ts @@ -23,6 +23,8 @@ export const tiktokWebhookEventSchema = z.object({ create_time: z.number(), user_openid: z.string(), content: z.string(), + message_id: z.string().optional(), + unique_identifier: z.string().optional(), }) export type TiktokWebhookEvent = z.infer diff --git a/integrations/whatsapp/__tests__/webhook-job-id.test.ts b/integrations/whatsapp/__tests__/webhook-job-id.test.ts new file mode 100644 index 0000000000..4f89b490e4 --- /dev/null +++ b/integrations/whatsapp/__tests__/webhook-job-id.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, test, vi } from "vitest" + +vi.mock("whatsapp-api-js/middleware/next", () => ({ + WhatsAppAPI: class { + on: { message?: (args: unknown) => void } = {} + + async handle_post(): Promise { + await Promise.resolve() + this.on.message?.({ + phoneID: "phone-1", + message: { id: "wamid:1/2" }, + }) + return 200 + } + }, +})) + +const { webhookHandler } = await import("../src/handlers/webhook") + +describe("WhatsApp webhook incomingMessage job IDs", () => { + test("uses a BullMQ-safe deterministic job ID for message events", async () => { + const add = vi.fn() + + await expect( + webhookHandler({ + config: { verifyToken: "verify-token" }, + req: new Request("https://example.test/webhook", { + method: "POST", + body: JSON.stringify({ entry: [] }), + }), + queue: { add }, + } as never), + ).resolves.toBe("ok") + + expect(add).toHaveBeenCalledWith( + "incomingMessage", + expect.objectContaining({ type: "incomingMessage" }), + { jobId: "incoming-whatsapp-wamid_1_2" }, + ) + }) +}) diff --git a/integrations/whatsapp/src/handlers/webhook.ts b/integrations/whatsapp/src/handlers/webhook.ts index b29a1f8044..5d0b18d8a1 100644 --- a/integrations/whatsapp/src/handlers/webhook.ts +++ b/integrations/whatsapp/src/handlers/webhook.ts @@ -3,6 +3,7 @@ import { type ReceivedMessageProps, SdkException, } from "@chatbotx.io/sdk" +import { toBullMqSafeIdSegment } from "@chatbotx.io/utils" import type { OnMessageArgs, OnStatusArgs } from "whatsapp-api-js/emitters" import { WhatsAppAPI as Middleware } from "whatsapp-api-js/middleware/next" import type { GetParams } from "whatsapp-api-js/types" @@ -164,9 +165,6 @@ const automaticEventFieldExtractors: Record< }, } -const toBullMqSafeIdSegment = (value: string): string => - value.replace(/[^a-zA-Z0-9._-]/g, "_") - export const extractAutomaticEventPayloads = ( rawBody: unknown, ): AutomaticEventPayload[] => { @@ -362,14 +360,22 @@ const dispatchWebhookResult = async ( | null, ): Promise => { if (result?.type === "message" && result.data.message) { - await queue?.add("incomingMessage", { - type: "incomingMessage", - data: { - integrationType: "whatsapp", - integrationIdentifier: result.data.phoneID, - payload: result.data, - } as ReceivedMessageProps, - }) + await queue?.add( + "incomingMessage", + { + type: "incomingMessage", + data: { + integrationType: "whatsapp", + integrationIdentifier: result.data.phoneID, + payload: result.data, + } as ReceivedMessageProps, + }, + result.data.message.id === undefined + ? undefined + : { + jobId: `incoming-whatsapp-${toBullMqSafeIdSegment(result.data.message.id)}`, + }, + ) } if (result?.type === "status") { diff --git a/integrations/zalo/__tests__/webhook-job-id.test.ts b/integrations/zalo/__tests__/webhook-job-id.test.ts new file mode 100644 index 0000000000..8f65a631f7 --- /dev/null +++ b/integrations/zalo/__tests__/webhook-job-id.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, test, vi } from "vitest" +import { webhookHandler } from "../src/handlers/webhook" + +describe("zalo webhook incomingMessage job IDs", () => { + test("uses a BullMQ-safe deterministic job ID for message events", async () => { + const add = vi.fn() + + await webhookHandler({ + config: { clientId: "app-1" }, + req: new Request("https://example.test/webhook", { + method: "POST", + body: JSON.stringify({ + app_id: "app-1", + event_name: "user_send_text", + sender: { id: "user-1" }, + recipient: { id: "oa-1" }, + message: { msg_id: "message:1/2", text: "hi" }, + }), + }), + queue: { add }, + } as never) + + expect(add).toHaveBeenCalledWith( + "incomingMessage", + expect.objectContaining({ type: "incomingMessage" }), + { jobId: "incoming-zalo-message_1_2" }, + ) + }) +}) diff --git a/integrations/zalo/src/handlers/webhook.ts b/integrations/zalo/src/handlers/webhook.ts index 4f894f73d1..bffab26970 100644 --- a/integrations/zalo/src/handlers/webhook.ts +++ b/integrations/zalo/src/handlers/webhook.ts @@ -3,6 +3,7 @@ import { type HandleRequestProps, SdkException, } from "@chatbotx.io/sdk" +import { toBullMqSafeIdSegment } from "@chatbotx.io/utils" import { sha256Hex, timingSafeStringEqual } from "../lib/webhook" import type { ZaloConfig } from "../schema/definition" import { @@ -110,14 +111,22 @@ const handleWebhookEvent = async ( ? webhookData.recipient.id : webhookData.sender.id - await queue.add("incomingMessage", { - type: "incomingMessage", - data: { - integrationType: "zalo", - integrationIdentifier, - payload: webhookData, + await queue.add( + "incomingMessage", + { + type: "incomingMessage", + data: { + integrationType: "zalo", + integrationIdentifier, + payload: webhookData, + }, }, - }) + webhookData.message?.msg_id === undefined + ? undefined + : { + jobId: `incoming-zalo-${toBullMqSafeIdSegment(webhookData.message.msg_id)}`, + }, + ) } } catch (error) { const errorMessage = diff --git a/packages/utils/__tests__/id.test.ts b/packages/utils/__tests__/id.test.ts new file mode 100644 index 0000000000..5fb9ee794e --- /dev/null +++ b/packages/utils/__tests__/id.test.ts @@ -0,0 +1,10 @@ +import { describe, expect, test } from "vitest" +import { toBullMqSafeIdSegment } from "../src/id" + +describe("toBullMqSafeIdSegment", () => { + test("replaces BullMQ-unsafe characters while preserving safe segments", () => { + expect(toBullMqSafeIdSegment("wamid:abc/123?x=1._-")).toBe( + "wamid_abc_123_x_1._-", + ) + }) +}) diff --git a/packages/utils/src/id.ts b/packages/utils/src/id.ts index edeac2ff3b..54a4a241e3 100644 --- a/packages/utils/src/id.ts +++ b/packages/utils/src/id.ts @@ -37,3 +37,6 @@ export const getIdFromParams = < const NUMERIC_ID_REGEX = /^\d+$/ export const isNumericId = (value: string): boolean => NUMERIC_ID_REGEX.test(value) + +export const toBullMqSafeIdSegment = (value: string): string => + value.replace(/[^a-zA-Z0-9._-]/g, "_") From 6620d6fc771438201f924af82f1fc82fe374ace5 Mon Sep 17 00:00:00 2001 From: Real Codesiman Date: Wed, 16 Sep 2026 13:34:09 +0700 Subject: [PATCH 3/4] perf(worker-config): stronger retry policy for chat and integration queues --- .../__tests__/enqueue-integration-job.test.ts | 4 ++ .../__tests__/resilient-queue-options.test.ts | 43 +++++++++++++++++++ packages/worker-config/src/lib/connection.ts | 8 ++++ .../worker-config/src/queues/chat/index.ts | 4 +- .../src/queues/integration/index.ts | 4 +- 5 files changed, 59 insertions(+), 4 deletions(-) create mode 100644 packages/worker-config/__tests__/resilient-queue-options.test.ts diff --git a/packages/worker-config/__tests__/enqueue-integration-job.test.ts b/packages/worker-config/__tests__/enqueue-integration-job.test.ts index d75647b588..f993f74481 100644 --- a/packages/worker-config/__tests__/enqueue-integration-job.test.ts +++ b/packages/worker-config/__tests__/enqueue-integration-job.test.ts @@ -20,6 +20,10 @@ vi.mock("../src/lib/connection", () => ({ attempts: 2, backoff: { type: "exponential", delay: 5000 }, }, + resilientJobOptions: { + attempts: 5, + backoff: { type: "exponential", delay: 10_000 }, + }, fakeQueue: { add: vi.fn() }, getRedisConnection: () => ({}), isNoRedisEnv: () => false, diff --git a/packages/worker-config/__tests__/resilient-queue-options.test.ts b/packages/worker-config/__tests__/resilient-queue-options.test.ts new file mode 100644 index 0000000000..3624443e6b --- /dev/null +++ b/packages/worker-config/__tests__/resilient-queue-options.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, test, vi } from "vitest" + +const mocks = vi.hoisted(() => ({ + queueOptions: new Map(), +})) + +vi.mock("bullmq", () => ({ + Queue: class { + constructor(name: string, options: unknown) { + mocks.queueOptions.set(name, options) + } + }, +})) + +vi.mock("../src/lib/connection", () => ({ + fakeQueue: { add: vi.fn() }, + getRedisConnection: () => ({}), + isNoRedisEnv: () => false, + resilientJobOptions: { + attempts: 5, + backoff: { type: "exponential", delay: 10_000 }, + }, +})) + +const { queueNames } = await import("../src/lib/types") +await import("../src/queues/chat") +await import("../src/queues/integration") + +describe("resilient queue defaults", () => { + test("gives chat and integration queues five exponential retry attempts", () => { + for (const queueName of [ + queueNames.enum.chat, + queueNames.enum.integration, + ]) { + expect(mocks.queueOptions.get(queueName)).toMatchObject({ + defaultJobOptions: { + attempts: 5, + backoff: { type: "exponential", delay: 10_000 }, + }, + }) + } + }) +}) diff --git a/packages/worker-config/src/lib/connection.ts b/packages/worker-config/src/lib/connection.ts index 2ff08d6f21..a48bbcb8de 100644 --- a/packages/worker-config/src/lib/connection.ts +++ b/packages/worker-config/src/lib/connection.ts @@ -56,6 +56,14 @@ export const defaultJobOptions = { }, } +export const resilientJobOptions = { + attempts: 5, + backoff: { + type: "exponential", + delay: 10_000, + }, +} + export const defaultWorkerOptions = { concurrency: 5, removeOnComplete: { count: 1000 }, diff --git a/packages/worker-config/src/queues/chat/index.ts b/packages/worker-config/src/queues/chat/index.ts index ec1115c17e..807760d84e 100644 --- a/packages/worker-config/src/queues/chat/index.ts +++ b/packages/worker-config/src/queues/chat/index.ts @@ -25,10 +25,10 @@ import type { import type { CommentAnchor, MessageButtonTemplate } from "@chatbotx.io/sdk" import { Queue } from "bullmq" import { - defaultJobOptions, fakeQueue, getRedisConnection, isNoRedisEnv, + resilientJobOptions, } from "../../lib/connection" import { queueNames } from "../../lib/types" import type { BotResponseTrackingContext } from "../types" @@ -235,5 +235,5 @@ export const chatQueue = isNoRedisEnv() ? fakeQueue : new Queue(queueNames.enum.chat, { connection: getRedisConnection(), - defaultJobOptions, + defaultJobOptions: resilientJobOptions, }) diff --git a/packages/worker-config/src/queues/integration/index.ts b/packages/worker-config/src/queues/integration/index.ts index be1b0a19d1..92fb07595e 100644 --- a/packages/worker-config/src/queues/integration/index.ts +++ b/packages/worker-config/src/queues/integration/index.ts @@ -10,10 +10,10 @@ import type { import type { CommentAnchor, OutgoingMessage } from "@chatbotx.io/sdk" import { type JobsOptions, Queue } from "bullmq" import { - defaultJobOptions, fakeQueue, getRedisConnection, isNoRedisEnv, + resilientJobOptions, } from "../../lib/connection" import { queueNames } from "../../lib/types" import type { BotResponseTrackingContext } from "../types" @@ -689,7 +689,7 @@ export const integrationQueue = isNoRedisEnv() ? fakeQueue : new Queue(queueNames.enum.integration, { connection: getRedisConnection(), - defaultJobOptions, + defaultJobOptions: resilientJobOptions, }) // Ads-conversion jobs need a stronger retry policy than the integration From a4645cb6ee952a799d180666aae4575b1f62752a Mon Sep 17 00:00:00 2001 From: Real Codesiman Date: Wed, 16 Sep 2026 14:35:51 +0700 Subject: [PATCH 4/4] refactor(worker): move chat concurrency env vars next to sibling queues --- apps/worker/src/chat/worker.ts | 8 ++++---- apps/worker/src/env.ts | 15 +++++++++++++++ packages/worker-config/src/keys.ts | 11 ----------- 3 files changed, 19 insertions(+), 15 deletions(-) diff --git a/apps/worker/src/chat/worker.ts b/apps/worker/src/chat/worker.ts index cb44df7e19..9528cdd68d 100644 --- a/apps/worker/src/chat/worker.ts +++ b/apps/worker/src/chat/worker.ts @@ -8,8 +8,8 @@ import { getRedisConnection, queueNames, } from "@chatbotx.io/worker-config" -import { keys } from "@chatbotx.io/worker-config/keys" 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" @@ -140,10 +140,10 @@ async function startChatWorker() { { connection: getRedisConnection(), ...defaultWorkerOptions, - concurrency: keys().CHAT_WORKER_CONCURRENCY, + concurrency: env.CHAT_WORKER_CONCURRENCY, limiter: { - max: keys().CHAT_WORKER_RATE_LIMIT_MAX, - duration: keys().CHAT_WORKER_RATE_LIMIT_DURATION_MS, + max: env.CHAT_WORKER_RATE_LIMIT_MAX, + duration: env.CHAT_WORKER_RATE_LIMIT_DURATION_MS, }, }, ) diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts index e6b1e0f8ee..f82187f816 100644 --- a/apps/worker/src/env.ts +++ b/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() diff --git a/packages/worker-config/src/keys.ts b/packages/worker-config/src/keys.ts index 990327a3aa..3500b99a59 100644 --- a/packages/worker-config/src/keys.ts +++ b/packages/worker-config/src/keys.ts @@ -6,17 +6,6 @@ export const keys = () => server: { REDIS_URL: z.url(), NEXT_PHASE: z.string().default(""), - CHAT_WORKER_CONCURRENCY: z.coerce.number().int().positive().default(20), - CHAT_WORKER_RATE_LIMIT_MAX: z.coerce - .number() - .int() - .positive() - .default(80), - CHAT_WORKER_RATE_LIMIT_DURATION_MS: z.coerce - .number() - .int() - .positive() - .default(1000), }, experimental__runtimeEnv: {}, skipValidation: process.env.SKIP_ENV_CHECK === "true",