From 825ccd24fefa36f0df20213c2688bd3faed9c97a Mon Sep 17 00:00:00 2001 From: Hung Phan Viet Date: Sun, 30 Aug 2026 15:29:22 +0700 Subject: [PATCH 1/2] refactor(worker): move coexist jobs to dedicated heavy worker --- .agents/skills/worker-development/SKILL.md | 14 + .../__tests__/coexist-whatsapp-api.test.ts | 8 + .../integration-sendgrid-api.test.ts | 8 + .../integration-webhook-freeze.test.ts | 31 +- .../__tests__/whatsapp-webhook-route.test.ts | 94 ++++ .../app/developer/queues/[[...path]]/route.ts | 2 + .../integrations/[...integration]/webhook.ts | 5 +- .../whatsapp/webhook/[integrationId]/route.ts | 3 +- .../__tests__/bulk-historical-import.test.ts | 2 +- .../__tests__/bulk-import-messages.test.ts | 2 +- .../coexist-attachment-download.test.ts | 2 +- .../__tests__/coexist-historical-id.test.ts | 2 +- .../coexist-instagram-adapter.test.ts | 2 +- ...coexist-instagram-facebook-adapter.test.ts | 2 +- .../__tests__/coexist-instagram-sync.test.ts | 37 +- .../__tests__/coexist-messenger-sync.test.ts | 16 +- .../coexist-messenger-template.test.ts | 2 +- .../__tests__/coexist-usage-throttle.test.ts | 2 +- .../__tests__/coexist-whatsapp-buffer.test.ts | 6 +- .../coexist-whatsapp-flush-bsuid.test.ts | 2 +- .../__tests__/coexist-whatsapp-flush.test.ts | 12 +- .../__tests__/docker-entrypoint.test.ts | 7 + .../__tests__/heavy-worker-boot.test.ts | 204 ++++++++ .../__tests__/integration-worker-boot.test.ts | 23 +- .../integration-worker-legacy-drain.test.ts | 406 +++++++++++++++ .../__tests__/scan-coexist-runs.test.ts | 4 +- apps/worker/package.json | 1 + apps/worker/src/env.ts | 16 +- .../handlers/coexist/attachment-download.ts | 7 +- .../coexist/bulk-historical-import.ts | 0 .../handlers/coexist/instagram-adapter.ts | 0 .../coexist/instagram-facebook-adapter.ts | 0 .../handlers/coexist/instagram-normalize.ts | 0 .../handlers/coexist/instagram-sync.ts | 28 +- .../handlers/coexist/messenger-helpers.ts | 0 .../handlers/coexist/messenger-sync.ts | 18 +- .../handlers/coexist/pull-adapter.ts | 0 .../handlers/coexist/usage-throttle.ts | 0 .../handlers/coexist/whatsapp-buffer.ts | 14 +- .../handlers/coexist/whatsapp-flush.ts | 20 +- apps/worker/src/heavy/worker.ts | 123 +++++ apps/worker/src/integration/utils/message.ts | 3 +- apps/worker/src/integration/worker.ts | 81 +-- .../schedule/handlers/scan-coexist-runs.ts | 18 +- apps/worker/tsdown.config.ts | 1 + .../2026-08-30-heavy-worker-coexist-split.md | 490 ++++++++++++++++++ .../__tests__/coexist-webhook.test.ts | 49 +- integrations/whatsapp/src/handlers/webhook.ts | 33 +- .../__tests__/coexist.service.test.ts | 2 +- packages/business/src/coexist/service.ts | 14 +- packages/sdk/src/lib/shared/index.ts | 8 + .../__tests__/heavy-job-data-schema.test.ts | 109 ++++ .../__tests__/no-redis-env.test.ts | 23 + packages/worker-config/src/index.ts | 1 + packages/worker-config/src/lib/types.ts | 1 + .../worker-config/src/queues/heavy/index.ts | 166 ++++++ .../src/queues/integration/index.ts | 74 --- scripts/check-coexist-drain.mts | 90 ++++ 58 files changed, 2052 insertions(+), 236 deletions(-) create mode 100644 apps/builder/__tests__/whatsapp-webhook-route.test.ts create mode 100644 apps/worker/__tests__/heavy-worker-boot.test.ts create mode 100644 apps/worker/__tests__/integration-worker-legacy-drain.test.ts rename apps/worker/src/{integration => heavy}/handlers/coexist/attachment-download.ts (97%) rename apps/worker/src/{integration => heavy}/handlers/coexist/bulk-historical-import.ts (100%) rename apps/worker/src/{integration => heavy}/handlers/coexist/instagram-adapter.ts (100%) rename apps/worker/src/{integration => heavy}/handlers/coexist/instagram-facebook-adapter.ts (100%) rename apps/worker/src/{integration => heavy}/handlers/coexist/instagram-normalize.ts (100%) rename apps/worker/src/{integration => heavy}/handlers/coexist/instagram-sync.ts (95%) rename apps/worker/src/{integration => heavy}/handlers/coexist/messenger-helpers.ts (100%) rename apps/worker/src/{integration => heavy}/handlers/coexist/messenger-sync.ts (98%) rename apps/worker/src/{integration => heavy}/handlers/coexist/pull-adapter.ts (100%) rename apps/worker/src/{integration => heavy}/handlers/coexist/usage-throttle.ts (100%) rename apps/worker/src/{integration => heavy}/handlers/coexist/whatsapp-buffer.ts (91%) rename apps/worker/src/{integration => heavy}/handlers/coexist/whatsapp-flush.ts (98%) create mode 100644 apps/worker/src/heavy/worker.ts create mode 100644 docs/plans/2026-08-30-heavy-worker-coexist-split.md create mode 100644 packages/worker-config/__tests__/heavy-job-data-schema.test.ts create mode 100644 packages/worker-config/src/queues/heavy/index.ts create mode 100644 scripts/check-coexist-drain.mts diff --git a/.agents/skills/worker-development/SKILL.md b/.agents/skills/worker-development/SKILL.md index d3ac2f9947..ffa89aa4c0 100644 --- a/.agents/skills/worker-development/SKILL.md +++ b/.agents/skills/worker-development/SKILL.md @@ -19,6 +19,7 @@ Workers run as separate Node processes in `apps/worker/`. They consume jobs from | Worker | Queue/Topic | Entry | |--------|------------|-------| | integration | `integration` | `src/integration/worker.ts` | +| heavy | `heavy` | `src/heavy/worker.ts` | | chat | `chat` | `src/chat/worker.ts` | | ai-agent | `aiAgent` | `src/ai-agent/worker.ts` | | default | `default` | `src/default/worker.ts` | @@ -28,6 +29,19 @@ Workers run as separate Node processes in `apps/worker/`. They consume jobs from | sequence-scheduler | Kafka | `src/sequence-scheduler/worker*.ts` | | notification | `notification` | `src/notification/worker.ts` | +The `heavy` queue/worker is a **workload-class** queue (not a domain queue): +long-lock (10 min), throughput-oriented, latency-tolerant jobs that would +otherwise starve latency-sensitive integration jobs of concurrency slots. +Coexist historical sync (Messenger/Instagram pulls, WhatsApp staging +flushes, attachment downloads) is its first tenant — handlers live under +`src/heavy/handlers/coexist/`, domain-grouped one level down. A future heavy +workload (e.g. a contact-import backfill) should join this queue with its +own `handlers//` folder rather than spawning a new worker. See +`docs/plans/2026-08-30-heavy-worker-coexist-split.md` for the full rationale, +including why the integration worker's `lockDuration`/`stalledInterval` stay +at 10 minutes after the split (bounded by `CHAT_JOB_WAIT_TIMEOUT_MS`, not by +coexist chunk sizing anymore). + ## Creating a New Queue ### 1. Define Queue Name diff --git a/apps/builder/__tests__/coexist-whatsapp-api.test.ts b/apps/builder/__tests__/coexist-whatsapp-api.test.ts index f4446596be..2f752c261a 100644 --- a/apps/builder/__tests__/coexist-whatsapp-api.test.ts +++ b/apps/builder/__tests__/coexist-whatsapp-api.test.ts @@ -76,6 +76,14 @@ const mockQueueAdd = vi.fn< vi.mock("@chatbotx.io/worker-config", () => ({ IntegrationJobAction: {}, + // coexist/service.ts sources its job-action strings from HeavyJobAction — + // the coexist actions moved to the `heavy` queue. See + // docs/plans/2026-08-30-heavy-worker-coexist-split.md. + HeavyJobAction: { + coexistMessengerSync: "coexistMessengerSync", + coexistInstagramSync: "coexistInstagramSync", + coexistWhatsappFlush: "coexistWhatsappFlush", + }, integrationQueue: { add: mockQueueAdd, }, diff --git a/apps/builder/__tests__/integration-sendgrid-api.test.ts b/apps/builder/__tests__/integration-sendgrid-api.test.ts index 60090f53eb..72d611d922 100644 --- a/apps/builder/__tests__/integration-sendgrid-api.test.ts +++ b/apps/builder/__tests__/integration-sendgrid-api.test.ts @@ -67,6 +67,14 @@ vi.mock("@chatbotx.io/worker-config", () => ({ coexistMessengerSync: "coexistMessengerSync", coexistInstagramSync: "coexistInstagramSync", }, + // coexist/service.ts now sources its job-action strings from HeavyJobAction + // (the coexist actions moved to the `heavy` queue) — see + // docs/plans/2026-08-30-heavy-worker-coexist-split.md. + HeavyJobAction: { + coexistMessengerSync: "coexistMessengerSync", + coexistInstagramSync: "coexistInstagramSync", + coexistWhatsappFlush: "coexistWhatsappFlush", + }, PURGE_WORKSPACES_INTERVAL_MINUTES: 30, })) diff --git a/apps/builder/__tests__/integration-webhook-freeze.test.ts b/apps/builder/__tests__/integration-webhook-freeze.test.ts index cb6b6e797f..f1894e91e7 100644 --- a/apps/builder/__tests__/integration-webhook-freeze.test.ts +++ b/apps/builder/__tests__/integration-webhook-freeze.test.ts @@ -39,8 +39,19 @@ vi.mock("@chatbotx.io/database/schema", () => ({ inboxModel: {}, })) +// Distinct object identity so tests can assert the exact `heavyQueue` passed +// through to `integration.handleRequest`. +const mockHeavyQueue = { name: "heavy" } +const mockIntegrationQueue = { name: "integration" } + vi.mock("@chatbotx.io/worker-config", () => ({ - integrationQueue: {}, + integrationQueue: mockIntegrationQueue, + // The webhook handler now also threads `heavyQueue` through to + // `integration.handleRequest` (coexist actions moved to the `heavy` queue — + // see docs/plans/2026-08-30-heavy-worker-coexist-split.md). It must be + // exported here or referencing it inside the try block throws (surfacing + // as the integration's handleRequest mock never being called). + heavyQueue: mockHeavyQueue, })) const isCloud = vi.fn(() => false) @@ -116,6 +127,15 @@ describe("telegram webhook freeze", () => { expect(telegramHandleRequest).toHaveBeenCalledOnce() }) + test("passes the heavyQueue through to handleRequest so coexist buffers are not stranded", async () => { + await handleWebhook("telegram", request()) + + expect(telegramHandleRequest).toHaveBeenCalledOnce() + const call = telegramHandleRequest.mock.calls[0][0] + expect(call.heavyQueue).toBe(mockHeavyQueue) + expect(call.queue).toBe(mockIntegrationQueue) + }) + test("skips the update when the workspace is scheduled for deletion", async () => { workspaceFind.mockResolvedValue({ ...liveWorkspace, @@ -175,6 +195,15 @@ describe("tiktok webhook freeze", () => { expect(tiktokHandleRequest).toHaveBeenCalledOnce() }) + test("passes the heavyQueue through to handleRequest so coexist buffers are not stranded", async () => { + await handleWebhook("tiktok", request()) + + expect(tiktokHandleRequest).toHaveBeenCalledOnce() + const call = tiktokHandleRequest.mock.calls[0][0] + expect(call.heavyQueue).toBe(mockHeavyQueue) + expect(call.queue).toBe(mockIntegrationQueue) + }) + test("skips the event when the workspace is scheduled for deletion", async () => { workspaceFind.mockResolvedValue({ ...liveWorkspace, diff --git a/apps/builder/__tests__/whatsapp-webhook-route.test.ts b/apps/builder/__tests__/whatsapp-webhook-route.test.ts new file mode 100644 index 0000000000..acacd25452 --- /dev/null +++ b/apps/builder/__tests__/whatsapp-webhook-route.test.ts @@ -0,0 +1,94 @@ +// @vitest-environment node +import { beforeEach, describe, expect, test, vi } from "vitest" + +const findIntegrationWhatsappById = vi.fn() +const markWhatsappWebhookVerified = vi.fn() +const whatsappHandleRequest = vi.fn() + +// Distinct mock objects so we can assert `handleRequest` receives the exact +// `heavyQueue` identity — the webhook handler now threads `heavyQueue` through +// to `integration.handleRequest` (coexist actions moved to the `heavy` queue — +// see docs/plans/2026-08-30-heavy-worker-coexist-split.md). It must be +// exported here or referencing it inside the try block throws. +const heavyQueue = { name: "heavy" } +const integrationQueue = { name: "integration" } + +vi.mock("@chatbotx.io/worker-config", () => ({ + heavyQueue, + integrationQueue, +})) + +vi.mock("@/features/integration-whatsapp/queries", () => ({ + findIntegrationWhatsappById, + markWhatsappWebhookVerified, +})) + +vi.mock("@/integration", () => ({ + integrations: { + whatsapp: { name: "whatsapp", handleRequest: whatsappHandleRequest }, + }, +})) + +vi.mock("@/lib/log", () => ({ + logger: { debug: vi.fn(), error: vi.fn(), info: vi.fn(), warn: vi.fn() }, +})) + +vi.mock("@/lib/webhook-log", () => ({ + logWebhookRequestBody: vi.fn(async () => undefined), +})) + +const { POST } = await import( + "../src/app/integrations/whatsapp/webhook/[integrationId]/route" +) + +const asPostRequest = (body: string) => + new Request("http://localhost/integrations/whatsapp/webhook/int-1", { + method: "POST", + body, + headers: { "x-hub-signature-256": "sha256=deadbeef" }, + }) as never + +const verifiedIntegration = { + row: { id: "int-1" }, + auth: { + authType: "oauth2", + clientId: "id", + clientSecret: "secret", + redirectUrl: "https://x", + verifyToken: "verify-token", + tokens: { accessToken: "token" }, + metadata: { + wabaId: "waba-1", + businessId: "biz-1", + phoneNumber: {}, + webhookUrl: "https://x", + isManual: true, + webhookVerifiedAt: "2026-01-01T00:00:00Z", + }, + }, +} + +beforeEach(() => { + vi.clearAllMocks() + findIntegrationWhatsappById.mockResolvedValue(verifiedIntegration.row) + whatsappHandleRequest.mockResolvedValue("ok") +}) + +describe("whatsapp dedicated webhook route", () => { + test("passes heavyQueue through to integration.handleRequest", async () => { + // loadManualIntegration re-reads auth off the found row. + findIntegrationWhatsappById.mockResolvedValue({ + ...verifiedIntegration.row, + auth: verifiedIntegration.auth, + }) + + await POST(asPostRequest(JSON.stringify({ entry: [] })), { + params: Promise.resolve({ integrationId: "int-1" }), + } as never) + + expect(whatsappHandleRequest).toHaveBeenCalledOnce() + const call = whatsappHandleRequest.mock.calls[0][0] + expect(call.heavyQueue).toBe(heavyQueue) + expect(call.queue).toBe(integrationQueue) + }) +}) diff --git a/apps/builder/src/app/developer/queues/[[...path]]/route.ts b/apps/builder/src/app/developer/queues/[[...path]]/route.ts index d19027905a..a93cb41cc3 100644 --- a/apps/builder/src/app/developer/queues/[[...path]]/route.ts +++ b/apps/builder/src/app/developer/queues/[[...path]]/route.ts @@ -7,6 +7,7 @@ import { chatQueue, defaultQueue, getSequenceSchedulerQueue, + heavyQueue, integrationQueue, quotaQueue, scheduleQueue, @@ -52,6 +53,7 @@ async function buildApp() { webhookQueue, defaultQueue, integrationQueue, + heavyQueue, quotaQueue, scheduleQueue, ...(sequenceSchedulerQueue ? [sequenceSchedulerQueue] : []), diff --git a/apps/builder/src/app/integrations/[...integration]/webhook.ts b/apps/builder/src/app/integrations/[...integration]/webhook.ts index ef438a227f..0507d5d01e 100644 --- a/apps/builder/src/app/integrations/[...integration]/webhook.ts +++ b/apps/builder/src/app/integrations/[...integration]/webhook.ts @@ -13,7 +13,7 @@ import type { TiktokAuthValue, TiktokConfig, } from "@chatbotx.io/integration-tiktok" -import { integrationQueue } from "@chatbotx.io/worker-config" +import { heavyQueue, integrationQueue } from "@chatbotx.io/worker-config" import type { NextRequest } from "next/server" import { isCloud } from "@/env" import { findIntegrationTelegramByBotId } from "@/features/integration-telegram/queries" @@ -167,6 +167,7 @@ export const handleWebhook = async ( } as any, req, queue: integrationQueue, + heavyQueue, }) return new Response(result as BodyInit) @@ -238,6 +239,7 @@ const handleTelegramWebhook = async (req: NextRequest) => { } as any, req, queue: integrationQueue, + heavyQueue, }) return new Response(result as BodyInit) @@ -348,6 +350,7 @@ const handleTiktokWebhook = async (req: NextRequest) => { config: tiktokConfig, req: reqWithBody, queue: integrationQueue, + heavyQueue, }) return new Response(result as BodyInit) diff --git a/apps/builder/src/app/integrations/whatsapp/webhook/[integrationId]/route.ts b/apps/builder/src/app/integrations/whatsapp/webhook/[integrationId]/route.ts index 7217fc1807..d4d006cb5c 100644 --- a/apps/builder/src/app/integrations/whatsapp/webhook/[integrationId]/route.ts +++ b/apps/builder/src/app/integrations/whatsapp/webhook/[integrationId]/route.ts @@ -1,5 +1,5 @@ import type { WhatsappAuthValue } from "@chatbotx.io/integration-whatsapp" -import { integrationQueue } from "@chatbotx.io/worker-config" +import { heavyQueue, integrationQueue } from "@chatbotx.io/worker-config" import type { NextRequest } from "next/server" import { findIntegrationWhatsappById, @@ -109,6 +109,7 @@ const handlePost = async (req: NextRequest, integrationId: string) => { } as any, req, queue: integrationQueue, + heavyQueue, }) return new Response(handlerResult as BodyInit) diff --git a/apps/worker/__tests__/bulk-historical-import.test.ts b/apps/worker/__tests__/bulk-historical-import.test.ts index a1fce00c5e..66bf05e93d 100644 --- a/apps/worker/__tests__/bulk-historical-import.test.ts +++ b/apps/worker/__tests__/bulk-historical-import.test.ts @@ -150,7 +150,7 @@ vi.mock("@chatbotx.io/utils", async (importOriginal) => { // Import after mocks // --------------------------------------------------------------------------- -import { bulkImportHistorical } from "../src/integration/handlers/coexist/bulk-historical-import" +import { bulkImportHistorical } from "../src/heavy/handlers/coexist/bulk-historical-import" // --------------------------------------------------------------------------- // Helpers — chain builders mirroring Drizzle's fluent API diff --git a/apps/worker/__tests__/bulk-import-messages.test.ts b/apps/worker/__tests__/bulk-import-messages.test.ts index 07421f9f6e..4d1dc3d2da 100644 --- a/apps/worker/__tests__/bulk-import-messages.test.ts +++ b/apps/worker/__tests__/bulk-import-messages.test.ts @@ -168,7 +168,7 @@ vi.mock("../src/lib/logger", () => ({ // --------------------------------------------------------------------------- const { bulkImportMessages, applyCoexistActivityUpdates } = await import( - "../src/integration/handlers/coexist/bulk-historical-import" + "../src/heavy/handlers/coexist/bulk-historical-import" ) // --------------------------------------------------------------------------- diff --git a/apps/worker/__tests__/coexist-attachment-download.test.ts b/apps/worker/__tests__/coexist-attachment-download.test.ts index 3f17cccda4..f836e4bb87 100644 --- a/apps/worker/__tests__/coexist-attachment-download.test.ts +++ b/apps/worker/__tests__/coexist-attachment-download.test.ts @@ -125,7 +125,7 @@ vi.mock("../src/lib/logger", () => ({ import { coexistAttachmentDownload, MAX_ATTACHMENT_BYTES, -} from "../src/integration/handlers/coexist/attachment-download" +} from "../src/heavy/handlers/coexist/attachment-download" // --------------------------------------------------------------------------- // Fixtures diff --git a/apps/worker/__tests__/coexist-historical-id.test.ts b/apps/worker/__tests__/coexist-historical-id.test.ts index 6b951eb58c..58e2f2ca8d 100644 --- a/apps/worker/__tests__/coexist-historical-id.test.ts +++ b/apps/worker/__tests__/coexist-historical-id.test.ts @@ -5,7 +5,7 @@ import { describe, expect, it } from "vitest" import { createHistoricalIdFactory, decodeHistoricalId, -} from "../src/integration/handlers/coexist/bulk-historical-import" +} from "../src/heavy/handlers/coexist/bulk-historical-import" // Must match COEXIST_EPOCH_MS in bulk-historical-import.ts (the uuniq epoch, // so historical IDs decode back to real wall-clock createdAt). diff --git a/apps/worker/__tests__/coexist-instagram-adapter.test.ts b/apps/worker/__tests__/coexist-instagram-adapter.test.ts index 5e7a87b991..b14c89d18b 100644 --- a/apps/worker/__tests__/coexist-instagram-adapter.test.ts +++ b/apps/worker/__tests__/coexist-instagram-adapter.test.ts @@ -14,7 +14,7 @@ vi.mock("@chatbotx.io/business", () => ({ import { type InstagramCoexistContext, instagramCoexistAdapter, -} from "../src/integration/handlers/coexist/instagram-adapter" +} from "../src/heavy/handlers/coexist/instagram-adapter" const context = { integration: { diff --git a/apps/worker/__tests__/coexist-instagram-facebook-adapter.test.ts b/apps/worker/__tests__/coexist-instagram-facebook-adapter.test.ts index 5bcfaf9ea4..5f5bfcb551 100644 --- a/apps/worker/__tests__/coexist-instagram-facebook-adapter.test.ts +++ b/apps/worker/__tests__/coexist-instagram-facebook-adapter.test.ts @@ -27,7 +27,7 @@ vi.mock("@chatbotx.io/integration-instagram-facebook/apis/sync", () => ({ import { type InstagramFacebookCoexistContext, instagramFacebookCoexistAdapter, -} from "../src/integration/handlers/coexist/instagram-facebook-adapter" +} from "../src/heavy/handlers/coexist/instagram-facebook-adapter" const context = { integration: { diff --git a/apps/worker/__tests__/coexist-instagram-sync.test.ts b/apps/worker/__tests__/coexist-instagram-sync.test.ts index 27ce07be6f..d811783728 100644 --- a/apps/worker/__tests__/coexist-instagram-sync.test.ts +++ b/apps/worker/__tests__/coexist-instagram-sync.test.ts @@ -55,11 +55,11 @@ vi.mock("@chatbotx.io/business", () => ({ })) vi.mock("@chatbotx.io/worker-config", () => ({ - IntegrationJobAction: { + HeavyJobAction: { coexistAttachmentDownload: "coexistAttachmentDownload", coexistInstagramSync: "coexistInstagramSync", }, - integrationQueue: { + heavyQueue: { add: mockQueueAdd, addBulk: mockQueueAddBulk, }, @@ -73,14 +73,14 @@ vi.mock("../src/lib/logger", () => ({ }, })) -vi.mock("../src/integration/handlers/coexist/bulk-historical-import", () => ({ +vi.mock("../src/heavy/handlers/coexist/bulk-historical-import", () => ({ applyCoexistActivityUpdates: mockApplyCoexistActivityUpdates, bulkImportContacts: mockBulkImportContacts, bulkImportMessages: mockBulkImportMessages, createHistoricalIdFactory: vi.fn(() => () => "historical-id"), })) -vi.mock("../src/integration/handlers/coexist/instagram-adapter", () => ({ +vi.mock("../src/heavy/handlers/coexist/instagram-adapter", () => ({ instagramCoexistAdapter: { channel: "instagram", discoverContactEnrichment: vi.fn(() => ({})), @@ -96,24 +96,21 @@ vi.mock("../src/integration/handlers/coexist/instagram-adapter", () => ({ // Provider routing imports the Facebook adapter too; stub it so the native // (`type: "instagram"`) path stays isolated in this suite. -vi.mock( - "../src/integration/handlers/coexist/instagram-facebook-adapter", - () => ({ - instagramFacebookCoexistAdapter: { - channel: "instagram", - discoverContactEnrichment: vi.fn(() => ({})), - fetchConversationMessages: vi.fn(), - getConversationUpdatedAt: vi.fn(), - listConversations: vi.fn(), - loadContext: mockFbLoadContext, - resolveContact: vi.fn(), - toHistoricalMessage: vi.fn(), - }, - }), -) +vi.mock("../src/heavy/handlers/coexist/instagram-facebook-adapter", () => ({ + instagramFacebookCoexistAdapter: { + channel: "instagram", + discoverContactEnrichment: vi.fn(() => ({})), + fetchConversationMessages: vi.fn(), + getConversationUpdatedAt: vi.fn(), + listConversations: vi.fn(), + loadContext: mockFbLoadContext, + resolveContact: vi.fn(), + toHistoricalMessage: vi.fn(), + }, +})) const { coexistInstagramSync } = await import( - "../src/integration/handlers/coexist/instagram-sync" + "../src/heavy/handlers/coexist/instagram-sync" ) const syncData = { diff --git a/apps/worker/__tests__/coexist-messenger-sync.test.ts b/apps/worker/__tests__/coexist-messenger-sync.test.ts index 9c71bc84e1..7f046b04e7 100644 --- a/apps/worker/__tests__/coexist-messenger-sync.test.ts +++ b/apps/worker/__tests__/coexist-messenger-sync.test.ts @@ -71,17 +71,23 @@ vi.mock("@chatbotx.io/database/client", () => ({ })) vi.mock("@chatbotx.io/worker-config", () => ({ - IntegrationJobAction: { + HeavyJobAction: { coexistWhatsappBuffer: "coexistWhatsappBuffer", coexistWhatsappFlush: "coexistWhatsappFlush", coexistMessengerSync: "coexistMessengerSync", - updateContactAvatar: "updateContactAvatar", coexistAttachmentDownload: "coexistAttachmentDownload", }, - integrationQueue: { + heavyQueue: { add: mockQueueAdd, addBulk: vi.fn().mockResolvedValue(undefined), }, + IntegrationJobAction: { + updateContactAvatar: "updateContactAvatar", + }, + integrationQueue: { + add: vi.fn().mockResolvedValue(undefined), + addBulk: vi.fn().mockResolvedValue(undefined), + }, })) vi.mock("@chatbotx.io/database/schema", () => ({ @@ -123,7 +129,7 @@ vi.mock("@chatbotx.io/integration-messenger/apis/usage", () => ({ concurrencyForUsage: mockConcurrencyForUsage, })) -vi.mock("../src/integration/handlers/coexist/bulk-historical-import", () => ({ +vi.mock("../src/heavy/handlers/coexist/bulk-historical-import", () => ({ bulkImportHistorical: mockBulkImportHistorical, bulkImportMessages: mockBulkImportMessages, bulkImportContacts: mockBulkImportContacts, @@ -155,7 +161,7 @@ vi.mock("@chatbotx.io/business", () => ({ // Import handler after mocks // --------------------------------------------------------------------------- -import { coexistMessengerSync } from "../src/integration/handlers/coexist/messenger-sync" +import { coexistMessengerSync } from "../src/heavy/handlers/coexist/messenger-sync" // --------------------------------------------------------------------------- // Fixtures diff --git a/apps/worker/__tests__/coexist-messenger-template.test.ts b/apps/worker/__tests__/coexist-messenger-template.test.ts index 41f635e977..5cbc69e28c 100644 --- a/apps/worker/__tests__/coexist-messenger-template.test.ts +++ b/apps/worker/__tests__/coexist-messenger-template.test.ts @@ -21,7 +21,7 @@ vi.mock("../src/lib/logger", () => ({ logger: { error: vi.fn(), warn: vi.fn(), info: vi.fn(), debug: vi.fn() }, })) -import { fetchConvMessages } from "../src/integration/handlers/coexist/messenger-helpers" +import { fetchConvMessages } from "../src/heavy/handlers/coexist/messenger-helpers" type GraphAttachment = Record diff --git a/apps/worker/__tests__/coexist-usage-throttle.test.ts b/apps/worker/__tests__/coexist-usage-throttle.test.ts index ac1061a402..b1d49e444a 100644 --- a/apps/worker/__tests__/coexist-usage-throttle.test.ts +++ b/apps/worker/__tests__/coexist-usage-throttle.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest" -import { resolveUsageThrottle } from "../src/integration/handlers/coexist/usage-throttle" +import { resolveUsageThrottle } from "../src/heavy/handlers/coexist/usage-throttle" describe("resolveUsageThrottle", () => { it("keeps default concurrency when usage is absent or low", () => { diff --git a/apps/worker/__tests__/coexist-whatsapp-buffer.test.ts b/apps/worker/__tests__/coexist-whatsapp-buffer.test.ts index b5fcc8b8a1..2904178cd9 100644 --- a/apps/worker/__tests__/coexist-whatsapp-buffer.test.ts +++ b/apps/worker/__tests__/coexist-whatsapp-buffer.test.ts @@ -32,12 +32,12 @@ vi.mock("@chatbotx.io/database/client", () => ({ })) vi.mock("@chatbotx.io/worker-config", () => ({ - IntegrationJobAction: { + HeavyJobAction: { coexistWhatsappBuffer: "coexistWhatsappBuffer", coexistWhatsappFlush: "coexistWhatsappFlush", coexistMessengerSync: "coexistMessengerSync", }, - integrationQueue: { add: mockQueueAdd }, + heavyQueue: { add: mockQueueAdd }, })) vi.mock("@chatbotx.io/database/schema", () => ({ @@ -63,7 +63,7 @@ vi.mock("@chatbotx.io/utils", async (importOriginal) => { // Import handler after mocks are registered // --------------------------------------------------------------------------- -import { coexistWhatsappBuffer } from "../src/integration/handlers/coexist/whatsapp-buffer" +import { coexistWhatsappBuffer } from "../src/heavy/handlers/coexist/whatsapp-buffer" // --------------------------------------------------------------------------- // Test helpers diff --git a/apps/worker/__tests__/coexist-whatsapp-flush-bsuid.test.ts b/apps/worker/__tests__/coexist-whatsapp-flush-bsuid.test.ts index 5eb2da2a00..9f0122f2ef 100644 --- a/apps/worker/__tests__/coexist-whatsapp-flush-bsuid.test.ts +++ b/apps/worker/__tests__/coexist-whatsapp-flush-bsuid.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "vitest" -import { extractFromValue } from "../src/integration/handlers/coexist/whatsapp-flush" +import { extractFromValue } from "../src/heavy/handlers/coexist/whatsapp-flush" // --------------------------------------------------------------------------- // WhatsApp Coexistence — BSUID/username extraction (D7, P7) diff --git a/apps/worker/__tests__/coexist-whatsapp-flush.test.ts b/apps/worker/__tests__/coexist-whatsapp-flush.test.ts index 594275a70f..0b986c0d8d 100644 --- a/apps/worker/__tests__/coexist-whatsapp-flush.test.ts +++ b/apps/worker/__tests__/coexist-whatsapp-flush.test.ts @@ -93,12 +93,16 @@ vi.mock("@chatbotx.io/database/client", () => ({ })) vi.mock("@chatbotx.io/worker-config", () => ({ - IntegrationJobAction: { + HeavyJobAction: { coexistWhatsappBuffer: "coexistWhatsappBuffer", coexistWhatsappFlush: "coexistWhatsappFlush", coexistMessengerSync: "coexistMessengerSync", + coexistAttachmentDownload: "coexistAttachmentDownload", + }, + heavyQueue: { + add: mockQueueAdd, + addBulk: vi.fn().mockResolvedValue(undefined), }, - integrationQueue: { add: mockQueueAdd }, })) vi.mock("@chatbotx.io/database/schema", () => ({ @@ -131,7 +135,7 @@ vi.mock("@chatbotx.io/database/schema", () => ({ }, })) -vi.mock("../src/integration/handlers/coexist/bulk-historical-import", () => ({ +vi.mock("../src/heavy/handlers/coexist/bulk-historical-import", () => ({ bulkImportHistorical: mockBulkImport, })) @@ -139,7 +143,7 @@ vi.mock("../src/integration/handlers/coexist/bulk-historical-import", () => ({ // Import handler after mocks // --------------------------------------------------------------------------- -import { coexistWhatsappFlush } from "../src/integration/handlers/coexist/whatsapp-flush" +import { coexistWhatsappFlush } from "../src/heavy/handlers/coexist/whatsapp-flush" // --------------------------------------------------------------------------- // Fixtures diff --git a/apps/worker/__tests__/docker-entrypoint.test.ts b/apps/worker/__tests__/docker-entrypoint.test.ts index eb09f24f1c..4179a2f6a5 100644 --- a/apps/worker/__tests__/docker-entrypoint.test.ts +++ b/apps/worker/__tests__/docker-entrypoint.test.ts @@ -16,6 +16,7 @@ const STANDARD_WORKERS = [ "chat", "events", "integration", + "heavy", "ai-agent", "default", "trigger", @@ -71,6 +72,12 @@ describe("docker-entrypoint worker discovery", () => { expect(stdout.trim()).toBe(join(distDir, "chat", "worker.mjs")) }) + test("resolves the heavy worker (coexist split) to its dist bundle", () => { + const { stdout, status } = run("worker", "heavy") + expect(status).toBe(0) + expect(stdout.trim()).toBe(join(distDir, "heavy", "worker.mjs")) + }) + test("aliases sequence variants to their historical CLI names", () => { expect(run("worker", "sequence-producer").stdout.trim()).toBe( join(distDir, "sequence-scheduler", "worker-producer.mjs"), diff --git a/apps/worker/__tests__/heavy-worker-boot.test.ts b/apps/worker/__tests__/heavy-worker-boot.test.ts new file mode 100644 index 0000000000..b7ab09b768 --- /dev/null +++ b/apps/worker/__tests__/heavy-worker-boot.test.ts @@ -0,0 +1,204 @@ +import { getAuditActor } from "@chatbotx.io/business/audit" +import type { HeavyJobData } from "@chatbotx.io/worker-config" +import { describe, expect, test, vi } from "vitest" + +// This test boots the real `src/heavy/worker.ts` module (it starts itself on +// import) to assert: the heavy worker process boots exactly one BullMQ +// `Worker`, on the `heavy` queue, with the long-lock/coarse-concurrency +// options the coexist workload needs, and every `HeavyJobData` variant routes +// to its handler via the exhaustive switch. Every import worker.ts pulls in +// is mocked below so this stays a fast, isolated unit test. + +type CapturedWorker = { + queueName: unknown + processor: (job: { data: unknown }) => Promise + options: Record +} + +const workerState = vi.hoisted(() => ({ + capturedWorkers: [] as CapturedWorker[], + coexistAttachmentDownload: vi.fn(async () => undefined), + coexistInstagramSync: vi.fn(async () => undefined), + coexistMessengerSync: vi.fn(async () => undefined), + coexistWhatsappBuffer: vi.fn(async () => undefined), + coexistWhatsappFlush: vi.fn(async () => undefined), + ensureBootstrapped: vi.fn(async () => undefined), + isBlockedWorkspace: vi.fn(async () => false), + resolveWorkspaceId: vi.fn(async () => undefined), + workerClose: vi.fn(async () => undefined), + workerOn: vi.fn(), +})) + +vi.mock("bullmq", () => { + class WorkerMock { + close = workerState.workerClose + on = workerState.workerOn + + constructor( + queueName: unknown, + processor: CapturedWorker["processor"], + options: Record, + ) { + workerState.capturedWorkers.push({ queueName, processor, options }) + } + } + + return { Worker: WorkerMock } +}) + +vi.mock("@chatbotx.io/worker-config", () => ({ + HeavyJobAction: { + coexistWhatsappBuffer: "coexistWhatsappBuffer", + coexistWhatsappFlush: "coexistWhatsappFlush", + coexistMessengerSync: "coexistMessengerSync", + coexistInstagramSync: "coexistInstagramSync", + coexistAttachmentDownload: "coexistAttachmentDownload", + }, + defaultWorkerOptions: { + concurrency: 5, + removeOnComplete: { count: 1000 }, + removeOnFail: { count: 5000 }, + }, + getRedisConnection: () => ({}), + queueNames: { + enum: { + heavy: "heavy", + }, + }, +})) + +vi.mock("../src/env", () => ({ + env: { HEAVY_WORKER_CONCURRENCY: 5 }, +})) + +vi.mock("../src/lib/bootstrap", () => ({ + ensureBootstrapped: workerState.ensureBootstrapped, +})) + +vi.mock("../src/lib/is-blocked-workspace", () => ({ + isBlockedWorkspace: workerState.isBlockedWorkspace, +})) + +vi.mock("../src/lib/logger", () => ({ + logger: { error: vi.fn(), info: vi.fn(), warn: vi.fn() }, +})) + +vi.mock("../src/lib/resolve-workspace-id", () => ({ + resolveWorkspaceId: workerState.resolveWorkspaceId, +})) + +vi.mock("../src/heavy/handlers/coexist/attachment-download", () => ({ + coexistAttachmentDownload: workerState.coexistAttachmentDownload, +})) +vi.mock("../src/heavy/handlers/coexist/instagram-sync", () => ({ + coexistInstagramSync: workerState.coexistInstagramSync, +})) +vi.mock("../src/heavy/handlers/coexist/messenger-sync", () => ({ + coexistMessengerSync: workerState.coexistMessengerSync, +})) +vi.mock("../src/heavy/handlers/coexist/whatsapp-buffer", () => ({ + coexistWhatsappBuffer: workerState.coexistWhatsappBuffer, +})) +vi.mock("../src/heavy/handlers/coexist/whatsapp-flush", () => ({ + coexistWhatsappFlush: workerState.coexistWhatsappFlush, +})) + +// Importing the worker module boots it exactly once (ESM module cache) — the +// single `new Worker(...)` call happens as a side effect of this import, so +// it must happen once, before any assertions, rather than per-test. +await import("../src/heavy/worker") +await vi.waitFor(() => { + expect(workerState.capturedWorkers).toHaveLength(1) +}) + +describe("heavy worker process boot", () => { + test("boots exactly one Worker, on the heavy queue", () => { + expect(workerState.capturedWorkers).toHaveLength(1) + expect(workerState.capturedWorkers[0]?.queueName).toBe("heavy") + }) + + test("uses the env-tunable concurrency and the long coexist lock", () => { + const [heavyWorker] = workerState.capturedWorkers + + expect(heavyWorker?.options.concurrency).toBe(5) + expect(heavyWorker?.options.lockDuration).toBe(10 * 60 * 1000) + expect(heavyWorker?.options.stalledInterval).toBe(10 * 60 * 1000) + expect(heavyWorker?.options.maxStalledCount).toBe(1) + }) +}) + +describe("heavy worker dispatch (exhaustive switch over HeavyJobData)", () => { + const jobs: HeavyJobData[] = [ + { + type: "coexistWhatsappBuffer", + data: { phoneNumberId: "phone-1", payload: {} }, + }, + { + type: "coexistWhatsappFlush", + data: { phoneNumberId: "phone-1" }, + }, + { + type: "coexistMessengerSync", + data: { runId: "run-1", integrationId: "int-1", workspaceId: "ws-1" }, + }, + { + type: "coexistInstagramSync", + data: { runId: "run-1", integrationId: "int-1", workspaceId: "ws-1" }, + }, + { + type: "coexistAttachmentDownload", + data: { + attachmentId: "att-1", + workspaceId: "ws-1", + channel: "messenger", + integrationId: "int-1", + }, + }, + ] + + const handlerByType: Record = + { + coexistWhatsappBuffer: "coexistWhatsappBuffer", + coexistWhatsappFlush: "coexistWhatsappFlush", + coexistMessengerSync: "coexistMessengerSync", + coexistInstagramSync: "coexistInstagramSync", + coexistAttachmentDownload: "coexistAttachmentDownload", + } + + test.each(jobs)("dispatches $type to its own handler", async (jobData) => { + for (const fn of Object.values(handlerByType)) { + ;(workerState[fn] as ReturnType).mockClear() + } + const [heavyWorker] = workerState.capturedWorkers + + await heavyWorker?.processor({ data: jobData }) + + const handlerKey = handlerByType[jobData.type] + expect(workerState[handlerKey]).toHaveBeenCalledWith(jobData.data) + }) + + test("populates the audit actor with the heavy job source before dispatching", async () => { + let capturedActor: ReturnType + workerState.coexistWhatsappBuffer.mockImplementationOnce(() => { + capturedActor = getAuditActor() + return Promise.resolve(undefined) + }) + const [heavyWorker] = workerState.capturedWorkers + + await heavyWorker?.processor({ data: jobs[0] }) + + expect(capturedActor).toEqual( + expect.objectContaining({ source: "heavy:coexistWhatsappBuffer" }), + ) + }) + + test("short-circuits before dispatch when the workspace is blocked", async () => { + workerState.isBlockedWorkspace.mockResolvedValueOnce(true) + workerState.coexistWhatsappBuffer.mockClear() + const [heavyWorker] = workerState.capturedWorkers + + await heavyWorker?.processor({ data: jobs[0] }) + + expect(workerState.coexistWhatsappBuffer).not.toHaveBeenCalled() + }) +}) diff --git a/apps/worker/__tests__/integration-worker-boot.test.ts b/apps/worker/__tests__/integration-worker-boot.test.ts index cd96e21a95..90bf551b06 100644 --- a/apps/worker/__tests__/integration-worker-boot.test.ts +++ b/apps/worker/__tests__/integration-worker-boot.test.ts @@ -48,6 +48,14 @@ vi.mock("@chatbotx.io/worker-config", () => ({ removeOnFail: { count: 5000 }, }, getRedisConnection: () => ({}), + closeIntegrationQueueEvents: vi.fn(async () => undefined), + // Forward-only-shim schema — never matches an `IntegrationJobData` fixture + // built by this suite (none of them carry one of the 5 coexist `type` + // strings), so `safeParse` always fails here and every job falls through + // to the normal switch. The dedicated legacy-drain suite exercises match + // cases. + heavyJobDataSchema: { safeParse: () => ({ success: false }) }, + heavyQueue: { add: vi.fn() }, IntegrationJobAction: { evaluateTemplateSent: "evaluateTemplateSent", evaluateConversionTrigger: "evaluateConversionTrigger", @@ -111,21 +119,6 @@ vi.mock("../src/integration/handlers/automated-response", () => ({ vi.mock("../src/integration/handlers/challenge", () => ({ runChallenge: vi.fn(), })) -vi.mock("../src/integration/handlers/coexist/attachment-download", () => ({ - coexistAttachmentDownload: vi.fn(), -})) -vi.mock("../src/integration/handlers/coexist/instagram-sync", () => ({ - coexistInstagramSync: vi.fn(), -})) -vi.mock("../src/integration/handlers/coexist/messenger-sync", () => ({ - coexistMessengerSync: vi.fn(), -})) -vi.mock("../src/integration/handlers/coexist/whatsapp-buffer", () => ({ - coexistWhatsappBuffer: vi.fn(), -})) -vi.mock("../src/integration/handlers/coexist/whatsapp-flush", () => ({ - coexistWhatsappFlush: vi.fn(), -})) vi.mock("../src/integration/handlers/comment-automation", () => ({ processCommentAutomation: vi.fn(), })) diff --git a/apps/worker/__tests__/integration-worker-legacy-drain.test.ts b/apps/worker/__tests__/integration-worker-legacy-drain.test.ts new file mode 100644 index 0000000000..a6da61f846 --- /dev/null +++ b/apps/worker/__tests__/integration-worker-legacy-drain.test.ts @@ -0,0 +1,406 @@ +import { beforeEach, describe, expect, test, vi } from "vitest" + +// This suite boots the real `src/integration/worker.ts` module and exercises +// the FORWARD-ONLY SHIM: `bull:integration` survives a deploy, so legacy +// coexist jobs (typed `IntegrationJobData` before the split, structurally +// identical to `HeavyJobData` after it) can still show up on this worker. +// Production deploys are Docker Swarm stop-first (no old/new code overlap), +// so every recognized legacy action — including flush — is forwarded into +// the `heavy` queue's single jobId namespace instead of any of them running +// in place; the worker no longer imports the coexist handlers at all. +// `heavyJobDataSchema.safeParse` delegates to the REAL schema (via +// `importOriginal` in the `@chatbotx.io/worker-config` mock below) so a +// recognized-but-malformed legacy payload is proven to fall through safely, +// not just an unrecognized `type` string. Full schema coverage still lives in +// `packages/worker-config/__tests__/heavy-job-data-schema.test.ts`; this suite +// stays focused on worker.ts's own dispatch/forwarding logic. + +type CapturedWorker = { + queueName: unknown + processor: (job: { + data: unknown + name?: string + id?: string + opts?: Record + }) => Promise + options: Record +} + +const workerState = vi.hoisted(() => ({ + capturedWorkers: [] as CapturedWorker[], + heavyQueueAdd: vi.fn(async () => undefined), + ensureBootstrapped: vi.fn(async () => undefined), + isBlockedWorkspace: vi.fn(async () => false), + loggerWarn: vi.fn(), + resolveWorkspaceId: vi.fn(async () => "ws-1"), + // Implementation is wired to the REAL heavyJobDataSchema.safeParse inside + // the "@chatbotx.io/worker-config" mock factory below (via importOriginal) + // so this stays a genuine schema check, not a fake type-only lookup, while + // remaining a spy for the blocked-workspace call-count assertion. + safeParse: vi.fn(), + workerClose: vi.fn(async () => undefined), + workerOn: vi.fn(), +})) + +vi.mock("bullmq", () => { + class WorkerMock { + close = workerState.workerClose + on = workerState.workerOn + + constructor( + queueName: unknown, + processor: CapturedWorker["processor"], + options: Record, + ) { + workerState.capturedWorkers.push({ queueName, processor, options }) + } + } + + return { + Worker: WorkerMock, + // Instantiated at module scope by worker-config's queue setup when the + // real @chatbotx.io/worker-config module is loaded below (via + // importOriginal) to source the real heavyJobDataSchema; not exercised by + // this unit test beyond needing to construct successfully. isNoRedisEnv() + // is true under vitest, so heavyQueue itself resolves to fakeQueue and + // never calls this constructor — it only guards other queues in the + // barrel module. + Queue: class Queue { + add() { + return Promise.resolve() + } + }, + } +}) + +vi.mock("@chatbotx.io/worker-config", async (importOriginal) => { + const actual = + await importOriginal() + + // Delegate to the REAL schema so a recognized-but-malformed legacy payload + // (e.g. coexistWhatsappFlush missing phoneNumberId, or coexistMessengerSync + // missing workspaceId/integrationId) is rejected the same way it would be + // in production, instead of a fake check that only inspected `type`. + workerState.safeParse.mockImplementation((data: unknown) => + actual.heavyJobDataSchema.safeParse(data), + ) + + return { + defaultWorkerOptions: { + concurrency: 5, + removeOnComplete: { count: 1000 }, + removeOnFail: { count: 5000 }, + }, + getRedisConnection: () => ({}), + closeIntegrationQueueEvents: vi.fn(async () => undefined), + heavyJobDataSchema: { safeParse: workerState.safeParse }, + heavyQueue: { add: workerState.heavyQueueAdd }, + IntegrationJobAction: { + evaluateTemplateSent: "evaluateTemplateSent", + evaluateConversionTrigger: "evaluateConversionTrigger", + sendConversionEvent: "sendConversionEvent", + syncRetargetAudience: "syncRetargetAudience", + }, + integrationQueue: { add: vi.fn() }, + queueNames: { + enum: { + integration: "integration", + }, + }, + } +}) + +vi.mock("@chatbotx.io/automated-response", () => ({ + automatedResponseService: { enqueue: vi.fn() }, +})) + +vi.mock("@chatbotx.io/business", () => ({ + conversationService: { ensureActive: vi.fn() }, +})) + +vi.mock("@chatbotx.io/event-bus", () => ({ + emit: vi.fn(), +})) + +vi.mock("@chatbotx.io/sdk", async (importOriginal) => ({ + ...(await importOriginal()), + getStoryReply: vi.fn(), +})) + +vi.mock("../src/env", () => ({ + env: { INTEGRATION_WORKER_CONCURRENCY: 10 }, +})) + +vi.mock("../src/lib/bootstrap", () => ({ + ensureBootstrapped: workerState.ensureBootstrapped, +})) + +vi.mock("../src/lib/is-blocked-workspace", () => ({ + isBlockedWorkspace: workerState.isBlockedWorkspace, +})) + +vi.mock("../src/lib/logger", () => ({ + logger: { error: vi.fn(), info: vi.fn(), warn: workerState.loggerWarn }, +})) + +vi.mock("../src/lib/resolve-workspace-id", () => ({ + resolveWorkspaceId: workerState.resolveWorkspaceId, +})) + +vi.mock("../src/integration/handlers/ads-automatic-event", () => ({ + handleAdsAutomaticEvent: vi.fn(), +})) +vi.mock("../src/integration/handlers/ads-conversion/registry", () => ({ + dispatchAdsConversionJob: vi.fn(), +})) +vi.mock("../src/integration/handlers/automated-response", () => ({ + processAutomatedResponse: vi.fn(), +})) +vi.mock("../src/integration/handlers/challenge", () => ({ + runChallenge: vi.fn(), +})) +vi.mock("../src/integration/handlers/comment-automation", () => ({ + processCommentAutomation: vi.fn(), +})) +vi.mock("../src/integration/handlers/comment-automation/ai-reply", () => ({ + processCommentAIReply: vi.fn(), +})) +vi.mock("../src/integration/handlers/contact/update-avatar", () => ({ + updateContactAvatar: vi.fn(), +})) +vi.mock("../src/integration/handlers/conversation", () => ({ + agentMarkAsRead: vi.fn(), + contactMarkAsRead: vi.fn(), +})) +vi.mock("../src/integration/handlers/flow", () => ({ + runFlowNode: vi.fn(), + runFlowPostback: vi.fn(), + runFlowQuickReply: vi.fn(), +})) +vi.mock("../src/integration/handlers/follow-up", () => ({ + runFollowUpResume: vi.fn(), +})) +vi.mock("../src/integration/handlers/inbox_labels", () => ({ + handleChannelLabelWebhook: vi.fn(), +})) +vi.mock("../src/integration/handlers/lead-ads", () => ({ + processLeadgen: vi.fn(), +})) +vi.mock("../src/integration/handlers/message-status", () => ({ + handleMessageStatus: vi.fn(), +})) +vi.mock("../src/integration/handlers/received-message", () => ({ + deleteIncomingComment: vi.fn(), + receiveComment: vi.fn(), + receiveMessage: vi.fn(async () => ({ message: null })), + updateIncomingComment: vi.fn(), +})) +vi.mock("../src/integration/handlers/ref", () => ({ + runRef: vi.fn(), +})) +vi.mock("../src/integration/handlers/sequence-flow", () => ({ + handleSendSequenceFlow: vi.fn(), +})) +vi.mock("../src/integration/handlers/story-reply-automation", () => ({ + processStoryReplyAutomation: vi.fn(), +})) +vi.mock("../src/integration/handlers/template-flow-response", () => ({ + captureTemplateFlowResponse: vi.fn(), +})) +vi.mock("../src/integration/handlers/wait-resume", () => ({ + runWaitResume: vi.fn(), +})) +vi.mock("../src/integration/job-context", () => ({ + runIntegrationJobWithWebhookContext: vi.fn( + async (_job: unknown, callback: () => Promise) => callback(), + ), +})) +vi.mock("../src/integration/routing", () => ({ + resolveIncomingTextRouting: vi.fn(), +})) +vi.mock("../src/integration/utils/message", () => ({ + closeChatQueueEvents: vi.fn(async () => undefined), +})) + +await import("../src/integration/worker") +await vi.waitFor(() => { + expect(workerState.capturedWorkers).toHaveLength(1) +}) + +const getProcessor = () => { + const [integrationWorker] = workerState.capturedWorkers + if (!integrationWorker) { + throw new Error("integration worker was not captured") + } + return integrationWorker.processor +} + +beforeEach(() => { + workerState.heavyQueueAdd.mockClear() + workerState.isBlockedWorkspace.mockClear() + workerState.isBlockedWorkspace.mockResolvedValue(false) + workerState.resolveWorkspaceId.mockClear() + workerState.resolveWorkspaceId.mockResolvedValue("ws-1") + workerState.safeParse.mockClear() + workerState.loggerWarn.mockClear() +}) + +describe("integration worker forward-only shim — every legacy coexist action forwards", () => { + const cases: Array<{ type: string; payload: Record }> = [ + { + type: "coexistWhatsappBuffer", + payload: { phoneNumberId: "phone-1", payload: { entry: [] } }, + }, + { + type: "coexistWhatsappFlush", + payload: { runId: "run-3", phoneNumberId: "phone-3" }, + }, + { + type: "coexistMessengerSync", + payload: { + runId: "run-1", + integrationId: "int-1", + workspaceId: "ws-1", + }, + }, + { + type: "coexistInstagramSync", + payload: { + runId: "run-2", + integrationId: "int-2", + workspaceId: "ws-1", + }, + }, + { + type: "coexistAttachmentDownload", + payload: { + attachmentId: "att-1", + workspaceId: "ws-1", + channel: "messenger" as const, + integrationId: "int-1", + }, + }, + ] + + test.each( + cases, + )("forwards a legacy $type job to heavyQueue preserving jobId/opts (minus delay/repeat)", async ({ + type, + payload, + }) => { + const job = { + data: { type, data: payload }, + name: type, + id: `raw-job-id-${type}`, + opts: { + jobId: `custom-jobid-${type}`, + delay: 60_000, + attempts: 2, + removeOnComplete: true, + }, + } + + await getProcessor()(job) + + expect(workerState.heavyQueueAdd).toHaveBeenCalledTimes(1) + const [name, data, opts] = workerState.heavyQueueAdd.mock.calls[0] as [ + string, + unknown, + Record, + ] + expect(name).toBe(type) + expect(data).toEqual(job.data) + expect(opts.jobId).toBe(`custom-jobid-${type}`) + expect(opts.attempts).toBe(2) + expect(opts.removeOnComplete).toBe(true) + expect(opts).not.toHaveProperty("delay") + expect(opts).not.toHaveProperty("repeat") + }) + + test("falls back to job.id as the forwarded jobId when opts carries none", async () => { + const job = { + data: { + type: "coexistWhatsappFlush", + data: { phoneNumberId: "phone-4" }, + }, + name: "coexistWhatsappFlush", + id: "raw-job-id-fallback", + opts: {}, + } + + await getProcessor()(job) + + const [, , opts] = workerState.heavyQueueAdd.mock.calls[0] as [ + string, + unknown, + Record, + ] + expect(opts.jobId).toBe("raw-job-id-fallback") + }) +}) + +describe("integration worker forward-only shim — malformed/unknown payload", () => { + test("an unrecognized type fails safeParse and falls through to the normal switch (unhandled-type warn), no throw", async () => { + await expect( + getProcessor()({ + data: { type: "totallyUnknownAction", data: {} }, + }), + ).resolves.toBeUndefined() + + expect(workerState.heavyQueueAdd).not.toHaveBeenCalled() + expect(workerState.loggerWarn).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ type: "totallyUnknownAction" }), + }), + "Unhandled integration job type", + ) + }) + + test("a recognized coexistWhatsappFlush payload missing phoneNumberId fails the real schema and falls through, no throw", async () => { + await expect( + getProcessor()({ + data: { type: "coexistWhatsappFlush", data: {} }, + }), + ).resolves.toBeUndefined() + + expect(workerState.heavyQueueAdd).not.toHaveBeenCalled() + expect(workerState.loggerWarn).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ type: "coexistWhatsappFlush" }), + }), + "Unhandled integration job type", + ) + }) + + test("a recognized coexistMessengerSync payload missing workspaceId/integrationId fails the real schema and falls through, no throw", async () => { + await expect( + getProcessor()({ + data: { type: "coexistMessengerSync", data: { runId: "r" } }, + }), + ).resolves.toBeUndefined() + + expect(workerState.heavyQueueAdd).not.toHaveBeenCalled() + expect(workerState.loggerWarn).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ type: "coexistMessengerSync" }), + }), + "Unhandled integration job type", + ) + }) +}) + +describe("integration worker forward-only shim — blocked-workspace guard ordering", () => { + test("a blocked workspace short-circuits BEFORE the forward-only shim parse runs", async () => { + workerState.isBlockedWorkspace.mockResolvedValueOnce(true) + + await getProcessor()({ + data: { + type: "coexistWhatsappBuffer", + data: { phoneNumberId: "phone-blocked", payload: {} }, + }, + }) + + expect(workerState.safeParse).not.toHaveBeenCalled() + expect(workerState.heavyQueueAdd).not.toHaveBeenCalled() + }) +}) diff --git a/apps/worker/__tests__/scan-coexist-runs.test.ts b/apps/worker/__tests__/scan-coexist-runs.test.ts index 3793766a61..9beda88cc4 100644 --- a/apps/worker/__tests__/scan-coexist-runs.test.ts +++ b/apps/worker/__tests__/scan-coexist-runs.test.ts @@ -29,12 +29,12 @@ vi.mock("@chatbotx.io/business", () => ({ })) vi.mock("@chatbotx.io/worker-config", () => ({ - IntegrationJobAction: { + HeavyJobAction: { coexistWhatsappFlush: "coexistWhatsappFlush", coexistMessengerSync: "coexistMessengerSync", coexistInstagramSync: "coexistInstagramSync", }, - integrationQueue: { add: mockQueueAdd }, + heavyQueue: { add: mockQueueAdd }, })) vi.mock("@chatbotx.io/logger", () => ({ diff --git a/apps/worker/package.json b/apps/worker/package.json index 35b8257588..8d3b531783 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -15,6 +15,7 @@ "worker:chat": "dotenv -e ../../.env -- tsx --watch src/chat/worker.ts", "worker:default": "dotenv -e ../../.env -- tsx --watch src/default/worker.ts", "worker:events": "dotenv -e ../../.env -- tsx --watch src/events/worker.ts", + "worker:heavy": "dotenv -e ../../.env -- tsx --watch src/heavy/worker.ts", "worker:integration": "dotenv -e ../../.env -- tsx --watch src/integration/worker.ts", "worker:notification": "dotenv -e ../../.env -- tsx --watch src/notification/worker.ts", "worker:sequence-consumer": "dotenv -e ../../.env -- tsx --watch src/sequence-scheduler/worker-consumer.ts", diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts index 73dd1aedfd..6261bfef24 100644 --- a/apps/worker/src/env.ts +++ b/apps/worker/src/env.ts @@ -24,13 +24,27 @@ export const env = createEnv({ // Bounds each chat-job wait (awaitChatJob). Capped below the integration // worker lockDuration (10 min) so a wait can never outlive the job lock — // otherwise BullMQ would treat the job as stalled and reprocess it (double - // send). Validated so a bad value can't become NaN (= wait forever). + // send). Validated so a bad value can't become NaN (= wait forever). Since + // the coexist split (heavy worker), this max (9 min) is the ONLY reason the + // integration worker's lockDuration/stalledInterval stay at 10 min — see + // `apps/worker/src/integration/worker.ts`. CHAT_JOB_WAIT_TIMEOUT_MS: z.coerce .number() .int() .min(1000) .max(9 * 60 * 1000) .default(120_000), + // Coexist historical sync (Messenger/Instagram pulls, WhatsApp staging + // flushes, attachment-download fan-outs) runs on its own `heavy` worker so + // it no longer competes with latency-sensitive integration jobs. Handlers + // also self-throttle via the BUC adaptive throttle (usage-throttle.ts), so + // this concurrency is a coarse upper bound, not the primary rate control. + HEAVY_WORKER_CONCURRENCY: z.coerce + .number() + .int() + .min(1) + .max(200) + .default(5), NOTIFICATION_WORKER_CONCURRENCY: z.coerce .number() .int() diff --git a/apps/worker/src/integration/handlers/coexist/attachment-download.ts b/apps/worker/src/heavy/handlers/coexist/attachment-download.ts similarity index 97% rename from apps/worker/src/integration/handlers/coexist/attachment-download.ts rename to apps/worker/src/heavy/handlers/coexist/attachment-download.ts index b2c033bb3e..3b9a0d997e 100644 --- a/apps/worker/src/integration/handlers/coexist/attachment-download.ts +++ b/apps/worker/src/heavy/handlers/coexist/attachment-download.ts @@ -9,7 +9,7 @@ import { } from "@chatbotx.io/integration-whatsapp" import { SdkException } from "@chatbotx.io/sdk" import { createId } from "@chatbotx.io/utils" -import type { IntegrationJobCoexistAttachmentDownload } from "@chatbotx.io/worker-config" +import type { HeavyJobCoexistAttachmentDownload } from "@chatbotx.io/worker-config" import imageSize from "image-size" import { logger } from "../../../lib/logger" @@ -205,8 +205,7 @@ const downloadWhatsappMedia = async ( } } -type AttachmentChannel = - IntegrationJobCoexistAttachmentDownload["data"]["channel"] +type AttachmentChannel = HeavyJobCoexistAttachmentDownload["data"]["channel"] type BearerTokenAuth = { tokens: { accessToken: string } } type AttachmentDownloadContext = { auth: BearerTokenAuth } @@ -261,7 +260,7 @@ const mediaDownloaders = { * — BullMQ `jobId: att-` dedup plus this prefix guard cover idempotency. */ export const coexistAttachmentDownload = async ( - data: IntegrationJobCoexistAttachmentDownload["data"], + data: HeavyJobCoexistAttachmentDownload["data"], ): Promise => { const { attachmentId, workspaceId, channel, integrationId } = data diff --git a/apps/worker/src/integration/handlers/coexist/bulk-historical-import.ts b/apps/worker/src/heavy/handlers/coexist/bulk-historical-import.ts similarity index 100% rename from apps/worker/src/integration/handlers/coexist/bulk-historical-import.ts rename to apps/worker/src/heavy/handlers/coexist/bulk-historical-import.ts diff --git a/apps/worker/src/integration/handlers/coexist/instagram-adapter.ts b/apps/worker/src/heavy/handlers/coexist/instagram-adapter.ts similarity index 100% rename from apps/worker/src/integration/handlers/coexist/instagram-adapter.ts rename to apps/worker/src/heavy/handlers/coexist/instagram-adapter.ts diff --git a/apps/worker/src/integration/handlers/coexist/instagram-facebook-adapter.ts b/apps/worker/src/heavy/handlers/coexist/instagram-facebook-adapter.ts similarity index 100% rename from apps/worker/src/integration/handlers/coexist/instagram-facebook-adapter.ts rename to apps/worker/src/heavy/handlers/coexist/instagram-facebook-adapter.ts diff --git a/apps/worker/src/integration/handlers/coexist/instagram-normalize.ts b/apps/worker/src/heavy/handlers/coexist/instagram-normalize.ts similarity index 100% rename from apps/worker/src/integration/handlers/coexist/instagram-normalize.ts rename to apps/worker/src/heavy/handlers/coexist/instagram-normalize.ts diff --git a/apps/worker/src/integration/handlers/coexist/instagram-sync.ts b/apps/worker/src/heavy/handlers/coexist/instagram-sync.ts similarity index 95% rename from apps/worker/src/integration/handlers/coexist/instagram-sync.ts rename to apps/worker/src/heavy/handlers/coexist/instagram-sync.ts index be83fedecf..40fa72003d 100644 --- a/apps/worker/src/integration/handlers/coexist/instagram-sync.ts +++ b/apps/worker/src/heavy/handlers/coexist/instagram-sync.ts @@ -4,9 +4,9 @@ import type { IntegrationInstagramModel, } from "@chatbotx.io/database/types" import { - IntegrationJobAction, - type IntegrationJobCoexistInstagramSync, - integrationQueue, + HeavyJobAction, + type HeavyJobCoexistInstagramSync, + heavyQueue, } from "@chatbotx.io/worker-config" import pLimit from "p-limit" import { logger } from "../../../lib/logger" @@ -46,7 +46,7 @@ const runInstagramCoexistPull = async < Msg, >( adapter: PullCoexistAdapter, - data: IntegrationJobCoexistInstagramSync["data"], + data: HeavyJobCoexistInstagramSync["data"], ): Promise => { const { runId, integrationId, workspaceId } = data const jobStart = Date.now() @@ -355,11 +355,11 @@ const runInstagramCoexistPull = async < failedTotal += pageFailed if (attachmentIds.length > 0) { - await integrationQueue.addBulk( + await heavyQueue.addBulk( attachmentIds.map((attachmentId) => ({ - name: IntegrationJobAction.coexistAttachmentDownload, + name: HeavyJobAction.coexistAttachmentDownload, data: { - type: IntegrationJobAction.coexistAttachmentDownload, + type: HeavyJobAction.coexistAttachmentDownload, data: { attachmentId, workspaceId, @@ -404,10 +404,10 @@ const runInstagramCoexistPull = async < } if (continueLater) { - await integrationQueue.add( - IntegrationJobAction.coexistInstagramSync, + await heavyQueue.add( + HeavyJobAction.coexistInstagramSync, { - type: IntegrationJobAction.coexistInstagramSync, + type: HeavyJobAction.coexistInstagramSync, data: { runId, integrationId, workspaceId }, }, { @@ -451,17 +451,17 @@ const runInstagramCoexistPull = async < // run rows, and job action — only the pull source differs. Adding a provider is // one entry here plus its adapter; there is no branching to touch. const instagramCoexistProvidersByType = { - instagram: (data: IntegrationJobCoexistInstagramSync["data"]) => + instagram: (data: HeavyJobCoexistInstagramSync["data"]) => runInstagramCoexistPull(instagramCoexistAdapter, data), - facebook: (data: IntegrationJobCoexistInstagramSync["data"]) => + facebook: (data: HeavyJobCoexistInstagramSync["data"]) => runInstagramCoexistPull(instagramFacebookCoexistAdapter, data), } satisfies Record< InstagramIntegrationType, - (data: IntegrationJobCoexistInstagramSync["data"]) => Promise + (data: HeavyJobCoexistInstagramSync["data"]) => Promise > export const coexistInstagramSync = async ( - data: IntegrationJobCoexistInstagramSync["data"], + data: HeavyJobCoexistInstagramSync["data"], ): Promise => { const { runId, integrationId, workspaceId } = data diff --git a/apps/worker/src/integration/handlers/coexist/messenger-helpers.ts b/apps/worker/src/heavy/handlers/coexist/messenger-helpers.ts similarity index 100% rename from apps/worker/src/integration/handlers/coexist/messenger-helpers.ts rename to apps/worker/src/heavy/handlers/coexist/messenger-helpers.ts diff --git a/apps/worker/src/integration/handlers/coexist/messenger-sync.ts b/apps/worker/src/heavy/handlers/coexist/messenger-sync.ts similarity index 98% rename from apps/worker/src/integration/handlers/coexist/messenger-sync.ts rename to apps/worker/src/heavy/handlers/coexist/messenger-sync.ts index c09d1914f0..3339e37784 100644 --- a/apps/worker/src/integration/handlers/coexist/messenger-sync.ts +++ b/apps/worker/src/heavy/handlers/coexist/messenger-sync.ts @@ -25,8 +25,10 @@ import { } from "@chatbotx.io/integration-messenger/apis/usage" import type { IncomingContact } from "@chatbotx.io/sdk" import { + HeavyJobAction, + type HeavyJobCoexistMessengerSync, + heavyQueue, IntegrationJobAction, - type IntegrationJobCoexistMessengerSync, integrationQueue, } from "@chatbotx.io/worker-config" import pLimit from "p-limit" @@ -585,11 +587,11 @@ async function runMessagesPhase(ctx: SyncContext): Promise { // re-enqueues the same jobIds harmlessly. if (pageAttachmentIds.length > 0) { try { - await integrationQueue.addBulk( + await heavyQueue.addBulk( pageAttachmentIds.map((attachmentId) => ({ - name: IntegrationJobAction.coexistAttachmentDownload, + name: HeavyJobAction.coexistAttachmentDownload, data: { - type: IntegrationJobAction.coexistAttachmentDownload, + type: HeavyJobAction.coexistAttachmentDownload, data: { attachmentId, workspaceId, @@ -649,7 +651,7 @@ async function runMessagesPhase(ctx: SyncContext): Promise { * Idempotent via `Message_(contactInboxId, sourceId)_key`. */ export const coexistMessengerSync = async ( - data: IntegrationJobCoexistMessengerSync["data"], + data: HeavyJobCoexistMessengerSync["data"], ): Promise => { const { runId, integrationId, workspaceId } = data const jobStart = Date.now() @@ -901,10 +903,10 @@ export const coexistMessengerSync = async ( if (continueLater) { try { - await integrationQueue.add( - IntegrationJobAction.coexistMessengerSync, + await heavyQueue.add( + HeavyJobAction.coexistMessengerSync, { - type: IntegrationJobAction.coexistMessengerSync, + type: HeavyJobAction.coexistMessengerSync, data: { runId, integrationId, workspaceId }, }, { diff --git a/apps/worker/src/integration/handlers/coexist/pull-adapter.ts b/apps/worker/src/heavy/handlers/coexist/pull-adapter.ts similarity index 100% rename from apps/worker/src/integration/handlers/coexist/pull-adapter.ts rename to apps/worker/src/heavy/handlers/coexist/pull-adapter.ts diff --git a/apps/worker/src/integration/handlers/coexist/usage-throttle.ts b/apps/worker/src/heavy/handlers/coexist/usage-throttle.ts similarity index 100% rename from apps/worker/src/integration/handlers/coexist/usage-throttle.ts rename to apps/worker/src/heavy/handlers/coexist/usage-throttle.ts diff --git a/apps/worker/src/integration/handlers/coexist/whatsapp-buffer.ts b/apps/worker/src/heavy/handlers/coexist/whatsapp-buffer.ts similarity index 91% rename from apps/worker/src/integration/handlers/coexist/whatsapp-buffer.ts rename to apps/worker/src/heavy/handlers/coexist/whatsapp-buffer.ts index d561f27573..75c86e339b 100644 --- a/apps/worker/src/integration/handlers/coexist/whatsapp-buffer.ts +++ b/apps/worker/src/heavy/handlers/coexist/whatsapp-buffer.ts @@ -3,9 +3,9 @@ import { db } from "@chatbotx.io/database/client" import { whatsappCoexistStagingModel } from "@chatbotx.io/database/schema" import { createId } from "@chatbotx.io/utils" import { - IntegrationJobAction, - type IntegrationJobCoexistWhatsappBuffer, - integrationQueue, + HeavyJobAction, + type HeavyJobCoexistWhatsappBuffer, + heavyQueue, } from "@chatbotx.io/worker-config" import { logger } from "../../../lib/logger" @@ -27,7 +27,7 @@ const FLUSH_DELAY_MS = 60_000 * Run rows are created by the popup-enable action (builder api/coexist.ts). */ export const coexistWhatsappBuffer = async ( - data: IntegrationJobCoexistWhatsappBuffer["data"], + data: HeavyJobCoexistWhatsappBuffer["data"], ): Promise => { const { phoneNumberId, payload } = data @@ -70,10 +70,10 @@ export const coexistWhatsappBuffer = async ( // drains, it re-checks for unprocessed rows and self-enqueues one follow-up // (also coalesced) — see coexistWhatsappFlush. That keeps the queue free of // a per-webhook follow-up storm during a multi-hour history backfill. - await integrationQueue.add( - IntegrationJobAction.coexistWhatsappFlush, + await heavyQueue.add( + HeavyJobAction.coexistWhatsappFlush, { - type: IntegrationJobAction.coexistWhatsappFlush, + type: HeavyJobAction.coexistWhatsappFlush, data: { phoneNumberId }, }, { diff --git a/apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts b/apps/worker/src/heavy/handlers/coexist/whatsapp-flush.ts similarity index 98% rename from apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts rename to apps/worker/src/heavy/handlers/coexist/whatsapp-flush.ts index a59102cde0..faa1f339a4 100644 --- a/apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts +++ b/apps/worker/src/heavy/handlers/coexist/whatsapp-flush.ts @@ -29,9 +29,9 @@ import { } from "@chatbotx.io/sdk" import { createId } from "@chatbotx.io/utils" import { - IntegrationJobAction, - type IntegrationJobCoexistWhatsappFlush, - integrationQueue, + HeavyJobAction, + type HeavyJobCoexistWhatsappFlush, + heavyQueue, } from "@chatbotx.io/worker-config" import { z } from "zod" import { logger } from "../../../lib/logger" @@ -827,7 +827,7 @@ const CHUNK_BUDGET_MS = 4 * 60 * 1000 * window Meta uses to push it. */ export const coexistWhatsappFlush = async ( - data: IntegrationJobCoexistWhatsappFlush["data"], + data: HeavyJobCoexistWhatsappFlush["data"], ): Promise => { const { phoneNumberId } = data const jobStart = Date.now() @@ -1114,11 +1114,11 @@ export const coexistWhatsappFlush = async ( ] if (attachmentIdsToDownload.length > 0) { try { - await integrationQueue.addBulk( + await heavyQueue.addBulk( attachmentIdsToDownload.map((attachmentId) => ({ - name: IntegrationJobAction.coexistAttachmentDownload, + name: HeavyJobAction.coexistAttachmentDownload, data: { - type: IntegrationJobAction.coexistAttachmentDownload, + type: HeavyJobAction.coexistAttachmentDownload, data: { attachmentId, workspaceId: integration.workspaceId, @@ -1233,10 +1233,10 @@ export const coexistWhatsappFlush = async ( // and a continuation is enqueued to keep draining. if (continueLater) { try { - await integrationQueue.add( - IntegrationJobAction.coexistWhatsappFlush, + await heavyQueue.add( + HeavyJobAction.coexistWhatsappFlush, { - type: IntegrationJobAction.coexistWhatsappFlush, + type: HeavyJobAction.coexistWhatsappFlush, data: { runId, phoneNumberId }, }, { diff --git a/apps/worker/src/heavy/worker.ts b/apps/worker/src/heavy/worker.ts new file mode 100644 index 0000000000..4905387e81 --- /dev/null +++ b/apps/worker/src/heavy/worker.ts @@ -0,0 +1,123 @@ +import { + defaultWorkerOptions, + getRedisConnection, + HeavyJobAction, + type HeavyJobData, + 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 { logger } from "../lib/logger" +import { resolveWorkspaceId } from "../lib/resolve-workspace-id" +import { runJobWithAuditContext } from "../lib/run-job-with-audit-context" +import { coexistAttachmentDownload } from "./handlers/coexist/attachment-download" +import { coexistInstagramSync } from "./handlers/coexist/instagram-sync" +import { coexistMessengerSync } from "./handlers/coexist/messenger-sync" +import { coexistWhatsappBuffer } from "./handlers/coexist/whatsapp-buffer" +import { coexistWhatsappFlush } from "./handlers/coexist/whatsapp-flush" + +async function startHeavyWorker() { + try { + await ensureBootstrapped() + } catch (err) { + logger.error({ err }, "Failed to bootstrap heavy worker") + process.exit(1) + } + + const worker = new Worker( + queueNames.enum.heavy, + async (job: Job) => { + const workspaceId = await resolveWorkspaceId(job.data.data) + if (await isBlockedWorkspace(workspaceId)) { + return + } + + // No `runIntegrationJobWithWebhookContext` wrapper here — coexist + // actions are never channel-originated (see + // `apps/worker/src/integration/channel-origin.ts`), so the webhook + // execution-context wrapper the integration worker uses has nothing to + // do for this queue. + return await runJobWithAuditContext( + { workspaceId, source: `heavy:${job.data.type}` }, + async () => { + switch (job.data.type) { + case HeavyJobAction.coexistWhatsappBuffer: { + await coexistWhatsappBuffer(job.data.data) + return + } + case HeavyJobAction.coexistWhatsappFlush: { + await coexistWhatsappFlush(job.data.data) + return + } + case HeavyJobAction.coexistMessengerSync: { + await coexistMessengerSync(job.data.data) + return + } + case HeavyJobAction.coexistInstagramSync: { + await coexistInstagramSync(job.data.data) + return + } + case HeavyJobAction.coexistAttachmentDownload: { + await coexistAttachmentDownload(job.data.data) + return + } + default: { + // Exhaustiveness guard — adding a new HeavyJobData variant + // without handling it here becomes a compile error. + const _exhaustive: never = job.data + logger.warn({ data: _exhaustive }, "Unhandled heavy job type") + return + } + } + }, + ) + }, + { + connection: getRedisConnection(), + ...defaultWorkerOptions, + // Env-tunable — coexist handlers also self-throttle via the BUC + // adaptive throttle (usage-throttle.ts), so this is a coarse cap. + concurrency: env.HEAVY_WORKER_CONCURRENCY, + // Coexist historical sync chunks are bounded to ~4 min via + // self-continuation (see coexist-messenger-sync / coexist-whatsapp-flush). + // Lock sized as: 4 min active + 4 min Graph 5xx retry tail + 2 min bulk + // INSERT tail. + lockDuration: 10 * 60 * 1000, + stalledInterval: 10 * 60 * 1000, + maxStalledCount: 1, + }, + ) + + worker.on("failed", (job, err) => { + if (job) { + logger.error({ err }, `Job ${job.id} has failed`) + } + }) + + let isShuttingDown = false + async function shutdown() { + if (isShuttingDown) { + return + } + isShuttingDown = true + try { + // The moved coexist handlers hold no QueueEvents open (they don't await + // chat or integration job completion), so — unlike the integration + // worker — there is nothing else to close here. + await worker.close() + process.exit(0) + } catch (err) { + logger.error(err, "[HeavyWorker] Error during shutdown") + process.exit(1) + } + } + process.once("SIGINT", shutdown) + process.once("SIGTERM", shutdown) +} + +startHeavyWorker().catch((err) => { + logger.error({ err }, "Failed to start heavy worker") + process.exit(1) +}) diff --git a/apps/worker/src/integration/utils/message.ts b/apps/worker/src/integration/utils/message.ts index 5d7f5df6fa..aacb69bcd9 100644 --- a/apps/worker/src/integration/utils/message.ts +++ b/apps/worker/src/integration/utils/message.ts @@ -16,7 +16,8 @@ import { logger } from "../../lib/logger" // job.waitUntilFinished() pending forever, and each pending wait keeps a // QueueEvents listener plus its captured closures alive — the slow leak that // OOM-crashed the integration worker. Validated + capped below the integration -// worker lockDuration in env.ts. +// worker lockDuration in env.ts — that lock (10 min) is sized to stay above +// this max wait, not (post coexist-split) for any coexist-chunking reason. const CHAT_JOB_WAIT_TIMEOUT_MS = env.CHAT_JOB_WAIT_TIMEOUT_MS let chatQueueEvents: QueueEvents | null = null diff --git a/apps/worker/src/integration/worker.ts b/apps/worker/src/integration/worker.ts index 924cfc70c8..47d39ff40f 100644 --- a/apps/worker/src/integration/worker.ts +++ b/apps/worker/src/integration/worker.ts @@ -6,6 +6,8 @@ import { closeIntegrationQueueEvents, defaultWorkerOptions, getRedisConnection, + heavyJobDataSchema, + heavyQueue, IntegrationJobAction, type IntegrationJobData, integrationQueue, @@ -22,11 +24,6 @@ import { handleAdsAutomaticEvent } from "./handlers/ads-automatic-event" import { dispatchAdsConversionJob } from "./handlers/ads-conversion/registry" import { processAutomatedResponse } from "./handlers/automated-response" import { runChallenge } from "./handlers/challenge" -import { coexistAttachmentDownload } from "./handlers/coexist/attachment-download" -import { coexistInstagramSync } from "./handlers/coexist/instagram-sync" -import { coexistMessengerSync } from "./handlers/coexist/messenger-sync" -import { coexistWhatsappBuffer } from "./handlers/coexist/whatsapp-buffer" -import { coexistWhatsappFlush } from "./handlers/coexist/whatsapp-flush" import { processCommentAutomation } from "./handlers/comment-automation" import { processCommentAIReply } from "./handlers/comment-automation/ai-reply" import { updateContactAvatar } from "./handlers/contact/update-avatar" @@ -78,6 +75,43 @@ async function startIntegrationWorker() { runJobWithAuditContext( { workspaceId, source: `integration:${job.data.type}` }, async () => { + // ── FORWARD-ONLY SHIM (remove after bull:integration holds no + // coexist jobs; see scripts/check-coexist-drain.mts) ──────── + // + // The 5 coexist actions moved to the `heavy` queue/worker, but + // `bull:integration` survives a deploy: legacy coexist jobs + // (waiting, delayed ~60s flushes, retrying) are still there after + // cutover. `job.data` is typed `IntegrationJobData`, which no + // longer includes these actions, so a legacy payload can only be + // recognized by parsing — not casting — the raw value against the + // runtime `heavyJobDataSchema`. + // + // Production deploys are Docker Swarm stop-first: the old + // integration worker image is fully stopped before the new one + // (which owns none of the coexist handlers) starts, so old and + // new code never run concurrently. That makes it safe to forward + // EVERY recognized legacy action — including flush — into the + // `heavy` queue's single jobId namespace instead of executing any + // of them in place, which lets this worker drop all coexist + // handler knowledge. The forward preserves the original jobId + // (dedup-safe) and opts, minus `delay`/`repeat` (already consumed + // by having reached execution here; re-adding them would + // re-schedule/re-repeat the job on the heavy queue). + // + // Caveat: attempts state restarts fresh on the forwarded job — + // acceptable, since the default `attempts: 2` retry plus the + // scan-runs DB-side attempt tracking (`CoexistSyncRun.attempts`) + // remains the real retry authority for sync runs. + const legacy = heavyJobDataSchema.safeParse(job.data) + if (legacy.success) { + const { delay: _delay, repeat: _repeat, ...opts } = job.opts + await heavyQueue.add(job.name, legacy.data, { + ...opts, + jobId: job.opts.jobId ?? job.id, + }) + return + } + switch (job.data.type) { case IntegrationJobAction.incomingMessage: { const { @@ -255,30 +289,10 @@ async function startIntegrationWorker() { await handleMessageStatus(job.data.data) return } - case IntegrationJobAction.coexistWhatsappBuffer: { - await coexistWhatsappBuffer(job.data.data) - return - } case IntegrationJobAction.channelLabelChange: { await handleChannelLabelWebhook(job.data.data) return } - case IntegrationJobAction.coexistWhatsappFlush: { - await coexistWhatsappFlush(job.data.data) - return - } - case IntegrationJobAction.coexistMessengerSync: { - await coexistMessengerSync(job.data.data) - return - } - case IntegrationJobAction.coexistInstagramSync: { - await coexistInstagramSync(job.data.data) - return - } - case IntegrationJobAction.coexistAttachmentDownload: { - await coexistAttachmentDownload(job.data.data) - return - } case IntegrationJobAction.adsAutomaticEvent: { await handleAdsAutomaticEvent(job.data.data) return @@ -343,11 +357,20 @@ async function startIntegrationWorker() { // Override the shared default (5). I/O-bound webhook handling tolerates // more parallelism; env-tunable via INTEGRATION_WORKER_CONCURRENCY. concurrency: env.INTEGRATION_WORKER_CONCURRENCY, - // Coexist historical sync chunks are bounded to ~4 min via self-continuation - // (see coexist-messenger-sync / coexist-whatsapp-flush). Lock sized as: - // 4 min active + 4 min Graph 5xx retry tail + 2 min bulk INSERT tail. + // Coexist historical sync moved to its own `heavy` worker (see + // apps/worker/src/heavy/worker.ts), which now owns the sizing rationale + // this comment used to describe. The 10-minute lock stays here for a + // different reason: it must stay above CHAT_JOB_WAIT_TIMEOUT_MS (max + // 9 min, env.ts) — a shorter lock could let BullMQ treat an + // in-progress job as stalled and double-process it while it's still + // legitimately waiting on a chat job. A future follow-up could lower + // the max wait to tighten this lock (see plan §7, out of scope here). lockDuration: 10 * 60 * 1000, - stalledInterval: 10 * 60 * 1000, + // No stalledInterval override: with coexist gone there are no + // long-running jobs here, so the BullMQ default (30s) applies — a + // crashed worker's jobs are re-picked in seconds instead of minutes. + // Live workers auto-renew their locks, so the frequent check cannot + // false-positive an in-progress job. maxStalledCount: 1, }, ) diff --git a/apps/worker/src/schedule/handlers/scan-coexist-runs.ts b/apps/worker/src/schedule/handlers/scan-coexist-runs.ts index 46d4878d38..94cf33d50a 100644 --- a/apps/worker/src/schedule/handlers/scan-coexist-runs.ts +++ b/apps/worker/src/schedule/handlers/scan-coexist-runs.ts @@ -9,9 +9,9 @@ import type { } from "@chatbotx.io/database/repositories" import { getChildLogger } from "@chatbotx.io/logger" import { - IntegrationJobAction, - type IntegrationJobData, - integrationQueue, + HeavyJobAction, + type HeavyJobData, + heavyQueue, } from "@chatbotx.io/worker-config" const log = getChildLogger("scan-coexist-runs") @@ -22,8 +22,8 @@ const MAX_ATTEMPTS = 5 type CoexistRunEnqueuer = (run: PickedCoexistRun) => Promise const pullSyncActions = { - messenger: IntegrationJobAction.coexistMessengerSync, - instagram: IntegrationJobAction.coexistInstagramSync, + messenger: HeavyJobAction.coexistMessengerSync, + instagram: HeavyJobAction.coexistInstagramSync, } satisfies Partial< Record< CoexistChannel, @@ -34,7 +34,7 @@ const pullSyncActions = { const createPullSyncPayload = ( run: PickedCoexistRun, action: (typeof pullSyncActions)[keyof typeof pullSyncActions], -): IntegrationJobData => ({ +): HeavyJobData => ({ type: action, data: { runId: run.id, @@ -70,7 +70,7 @@ const coexistRunEnqueuers = { } await enqueueRun(run, { - type: IntegrationJobAction.coexistWhatsappFlush, + type: HeavyJobAction.coexistWhatsappFlush, data: { runId: run.id, phoneNumberId: integration.phoneNumberId }, }) }, @@ -78,9 +78,9 @@ const coexistRunEnqueuers = { async function enqueueRun( run: PickedCoexistRun, - payload: IntegrationJobData, + payload: HeavyJobData, ): Promise { - await integrationQueue.add(payload.type, payload, { + await heavyQueue.add(payload.type, payload, { jobId: `coexist-run-${run.id}-${run.attempts}`, attempts: 1, removeOnComplete: true, diff --git a/apps/worker/tsdown.config.ts b/apps/worker/tsdown.config.ts index 2b31ee0915..822e0ab35e 100644 --- a/apps/worker/tsdown.config.ts +++ b/apps/worker/tsdown.config.ts @@ -3,6 +3,7 @@ import { defineConfig } from "tsdown" export default defineConfig({ format: ["esm"], entry: [ + "src/heavy/worker.ts", "src/chat/worker.ts", "src/integration/worker.ts", "src/ai-agent/worker.ts", diff --git a/docs/plans/2026-08-30-heavy-worker-coexist-split.md b/docs/plans/2026-08-30-heavy-worker-coexist-split.md new file mode 100644 index 0000000000..8067cddee0 --- /dev/null +++ b/docs/plans/2026-08-30-heavy-worker-coexist-split.md @@ -0,0 +1,490 @@ +# Plan: Dedicated `heavy` worker + queue (coexist split out of `integration`) + +**Status:** REVIEWED — Codex rounds 1–4 incorporated (`[R#]` round 1, +`[R2-#]` round 2 move-completeness/flow audit, `[R4-#]` round 4 review of the +hybrid legacy-drain deploy strategy) +**Date:** 2026-08-30 +**Complexity:** MEDIUM-HIGH (~25 files across `packages/worker-config`, `packages/business`, `packages/sdk`, `apps/worker`, `apps/builder`, `integrations/whatsapp`) + +## Amendment (post-review) + +1. **Queue renamed `bulk` → `heavy`.** `bulk` collides with BullMQ's own + `addBulk` vocabulary (this repo's coexist handlers already batch-enqueue + via `Queue.addBulk`), which made `bulkQueue`/`BulkJobAction` read as if + they were part of that batching API instead of naming a workload-class + queue. Every `bulk`-prefixed identifier, path, env var, and script + described below (Section 2 onward) is superseded by its `heavy` + equivalent (`HeavyJobAction`, `HeavyJobData`, `heavyQueue`, + `heavyJobDataSchema`, `src/heavy/`, `queues/heavy/`, + `HEAVY_WORKER_CONCURRENCY`, `worker:heavy`). Section 2's naming rationale + is kept below as the historical record of the original decision, but its + conclusion (picking `bulk` over `heavy`) no longer holds. +2. **Phase 3 changed from execute-in-place drain to a forward-only shim.** + The hybrid design below (execute 4 of 5 legacy actions in place, forward + only `coexistWhatsappFlush`) assumed old and new worker code could be + running at the same time during a rolling deploy. Production deploys are + actually **Docker Swarm stop-first**: the old integration worker image is + fully stopped before the new one starts, so old and new code never run + concurrently. That makes it safe — and simpler — to forward **every** + recognized legacy action (including flush) into the `heavy` queue's + single jobId namespace, instead of executing four of them in place. The + integration worker now imports none of the coexist handlers at all. See + `apps/worker/src/integration/worker.ts` and + `apps/worker/__tests__/integration-worker-legacy-drain.test.ts` for the + implemented shape; `scripts/check-coexist-drain.mts` remains the removal + gate, unchanged in purpose. + +## 1. Motivation + +The integration worker's BullMQ options are sized entirely around Coexist +historical sync (`apps/worker/src/integration/worker.ts:346-351`): + +```ts +// Coexist historical sync chunks are bounded to ~4 min via self-continuation +// (see coexist-messenger-sync / coexist-whatsapp-flush). Lock sized as: +// 4 min active + 4 min Graph 5xx retry tail + 2 min bulk INSERT tail. +lockDuration: 10 * 60 * 1000, +stalledInterval: 10 * 60 * 1000, +``` + +Problems this causes: + +1. **Head-of-line blocking.** Coexist syncs (Messenger/Instagram history pulls, + WhatsApp staging flushes, attachment-download fan-outs) are long-running, + Graph-API-throttled bulk jobs. They occupy integration-worker concurrency + slots (default 10) that latency-sensitive jobs (`incomingMessage`, + `sendFlow`, postbacks) need for real-time chat. +2. **Config coupling.** Every knob on the integration worker (lock, stalled + interval, `maxStalledCount: 1`) is tuned for coexist, and every + latency-sensitive job inherits the 10-minute stall-detection latency: a + genuinely stuck `incomingMessage` job isn't retried for up to ~10 minutes. +3. **Blast radius.** A coexist sync storm (large page backfill) can starve all + webhook-driven message processing for a workspace-wide outage-like effect. + +Industry precedent for the fix: queue-per-workload-class separation +(Sidekiq `critical/default/low/bulk`, GitLab resource-class queues, Celery +dedicated heavy queues routed to their own worker pools). This repo already +follows a **domain-named queue** convention (`integration`, `chat`, `trigger`, +`webhook`, `notification`, `quota`, `sequenceScheduler`). + +## 2. Naming decision + +> **Superseded — see Amendment above.** This section is the historical +> record of the original decision (`bulk`); the shipped name is `heavy` +> because `bulk` collides with BullMQ's `addBulk` vocabulary. + +**Queue/worker name (as originally decided): `bulk` — a workload-class queue, not a domain queue.** + +Rationale (revised per review): naming the queue after its first tenant +(`coexist`) paints us into a corner — the next heavy workload (contact +import backfill, mass export, media re-processing) would need yet another +queue+worker+deployment unit, or would squat awkwardly in a queue named +after an unrelated feature. Large systems name this tier by **workload +class**, orthogonal to domain: + +- **Sidekiq** (canonical): `critical` / `default` / `low` / `bulk` — `bulk` + is the established name for high-throughput, latency-tolerant work. +- **GitLab**: sidekiq queues routed by `urgency`/resource class; heavy + backfills run on `low_urgency`/bulk shards. +- **Celery**: dedicated "heavy"/"bulk" queues routed to their own worker + pools with long time limits. + +At the time, `bulk` was judged to beat `heavy` (describes cost, not +contract; rare as an actual queue name), `sync`/`backfill` (too narrow — +future heavy jobs may be neither), and `low` (priority naming without a +priority system here). Post-review, `heavy` won anyway: `bulk` turned out to +collide with BullMQ's own `addBulk` vocabulary in this same codebase (see +Amendment above). + +The queue contract is explicit: **long-lock (10 min), throughput-oriented, +latency-tolerant jobs.** Coexist is its first tenant; future heavy actions +join this queue (with per-action `JobsOptions` if needed) instead of +spawning new workers. Domain grouping is preserved one level down — handlers +live in `handlers/coexist/`, a future import backfill would add +`handlers/import/`. + +Concrete names (mirror existing patterns exactly; `heavy` is the shipped +name — see Amendment): + +| Thing | Name | +|---|---| +| Queue name (`queueNames` enum) | `heavy` | +| Queue config module | `packages/worker-config/src/queues/heavy/index.ts` | +| Action const / types | `HeavyJobAction`, `HeavyJobData`, `heavyQueue` | +| Worker entry | `apps/worker/src/heavy/worker.ts` | +| Handlers dir | `apps/worker/src/heavy/handlers/coexist/` | +| Dev script | `worker:heavy` (auto-joins `pnpm dev` via `concurrently pnpm:worker:*`) | +| Concurrency env | `HEAVY_WORKER_CONCURRENCY` (default 5) | + +**Job action strings stay identical** (`coexistWhatsappBuffer`, +`coexistWhatsappFlush`, `coexistMessengerSync`, `coexistInstagramSync`, +`coexistAttachmentDownload`) — keeps jobIds/dedup semantics unchanged and makes +the legacy-to-`heavy` forwarding (Phase 3) trivial. + +## 3. Jobs that move (all 5 coexist actions) + +| Action | Producer(s) today | +|---|---| +| `coexistWhatsappBuffer` | `integrations/whatsapp/src/handlers/webhook.ts` via injected `props.queue` | +| `coexistWhatsappFlush` | `whatsapp-buffer.ts` (delayed follow-up), `scan-coexist-runs.ts` (schedule cron), self-continuation in `whatsapp-flush.ts` | +| `coexistMessengerSync` | `scan-coexist-runs.ts`, self-continuation in `messenger-sync.ts` | +| `coexistInstagramSync` | `scan-coexist-runs.ts`, self-continuation in `instagram-sync.ts` | +| `coexistAttachmentDownload` | `addBulk` fan-outs in `messenger-sync.ts`, `instagram-sync.ts`, `whatsapp-flush.ts` | + +**Stays on `integration`:** the `updateContactAvatar` `addBulk` fan-out from +`messenger-sync.ts:351` (it is an integration-domain job). + +**Stays on `schedule`:** `scanCoexistRuns` and `purgeCoexistStaging` cron +handlers (they are schedulers/janitors, not heavy work; only their enqueue +target changes). + +## 4. Implementation phases + +### Phase 1 — `packages/worker-config`: new heavy queue module + +1. `lib/types.ts`: add `"heavy"` to the `queueNames` z.enum. +2. New `queues/heavy/index.ts`: + - `HeavyJobAction` const (5 actions, same strings). + - Move the 5 job-data types verbatim from `queues/integration/index.ts` + (`IntegrationJobCoexistWhatsappBuffer` → `HeavyJobCoexistWhatsappBuffer`, etc.). + - `HeavyJobData` union; `heavyQueue = isNoRedisEnv() ? fakeQueue : new Queue(queueNames.enum.heavy, { connection, defaultJobOptions })`. + - **[R4-3]** a runtime `heavyJobDataSchema` (zod discriminated union on + `type`) exported alongside the TS types — consumed by the Phase 3 + forward-only shim parse. + - Carry over any per-action `JobsOptions` the moved jobs relied on (audit + `jobOptionsByAction` — currently no coexist entries, so default retry + `attempts: 2` is preserved; the scan-runs DB-side attempt tracking is the + real retry authority). +3. `queues/integration/index.ts`: remove the 5 action keys, 5 types, and the 5 + union members. TypeScript now turns every stale + `integrationQueue.add()` into a compile error — this is the safety + net that finds all typed producer sites. +4. Export the new module from the package index (`export * from "./queues/heavy"`). + +### Phase 2 — move handlers + new worker entry (`apps/worker`) + +1. `git mv apps/worker/src/integration/handlers/coexist apps/worker/src/heavy/handlers/coexist` + (12 files incl. `bulk-historical-import.ts`, adapters, `usage-throttle.ts`). + **[R2] Import depth:** handlers currently import `../../../lib/*` (3 + levels deep under `src/integration/handlers/coexist/`); the new location + `src/heavy/handlers/coexist/` is also 3 levels deep, so `../../../lib/*` + stays valid — still verify every relative import compiles (notably + `messenger-helpers`/`pull-adapter` internal imports and any + `../../utils/*` references, which now resolve to `src/heavy/utils/*` and + must be repointed or the util moved/shared). +2. Inside moved handlers, retarget enqueues — **every coexist follow-up**: + - `whatsapp-buffer.ts:73` delayed `coexistWhatsappFlush` → `heavyQueue` + (**[R4-2]** explicitly — this is what makes the legacy queue's coexist + subset drain monotonically); + - flush/sync self-continuations (`whatsapp-flush.ts:1236`, + `messenger-sync.ts:904`, `instagram-sync.ts:407`) → `heavyQueue`; + - `coexistAttachmentDownload` `addBulk` fan-outs → `heavyQueue`; + - `updateContactAvatar` `addBulk` → stays `integrationQueue`. +3. New `apps/worker/src/heavy/worker.ts`, mirroring the integration worker: + - `ensureBootstrapped()`, `resolveWorkspaceId` + `isBlockedWorkspace` guard + (**invariant 15** — blocked owner = silent no-op `return`, no retry), + - `runJobWithAuditContext({ workspaceId, source: `heavy:${type}` }, …)`, + - switch over `HeavyJobData` with the `never` exhaustiveness guard, + - worker options: `concurrency: env.HEAVY_WORKER_CONCURRENCY`, + `lockDuration: 10 * 60 * 1000`, `stalledInterval: 10 * 60 * 1000`, + `maxStalledCount: 1` — move the chunk-sizing comment here, + - SIGINT/SIGTERM shutdown closing the worker (audit which QueueEvents the + moved handlers hold open — e.g. if any use `awaitChatJob`/chat + QueueEvents, close those like the integration worker does; current grep + says coexist handlers do **not** use chat waits). + - **`runIntegrationJobWithWebhookContext`: skip it.** Confirmed safe by + review — coexist actions are not channel-originated + (`apps/worker/src/integration/channel-origin.ts:6`), so the heavy + worker does not need the webhook-context wrapper. +4. `apps/worker/src/env.ts`: add `HEAVY_WORKER_CONCURRENCY` (int, 1–200, + default 5) with a comment noting coexist handlers also self-throttle via + BUC (`usage-throttle.ts`), so worker concurrency is a coarse cap. + +### Phase 3 — integration worker: forward-only shim (superseded original: legacy drain execute-in-place), then delete + +**Strategy as shipped (see Amendment above): forward-only shim, not a +parallel execute-in-place drain.** The Redis queue `bull:integration` +survives a deploy, so legacy coexist jobs (waiting, delayed ~60s flushes, +retrying) are still there after cutover. Because production deploys are +Docker Swarm stop-first (old and new worker code never run concurrently), +the integration worker simply **forwards every recognized legacy coexist +job into `heavyQueue`** under its original jobId — it owns none of the +coexist handlers at all, unlike the original execute-in-place design below +(kept for historical context): + +1. Remove the 5 typed cases and old handler imports from + `apps/worker/src/integration/worker.ts`. +2. Add one compact, clearly-marked forward-only shim block **before** the + typed switch (not five case declarations — so the follow-up removal is + one deletion). **[R4-3] Typing contract:** Phase 1 additionally exports a + runtime `heavyJobDataSchema` (zod discriminated union on `type`, reusing + the per-action payload schemas) so the shim block can *parse* — not + cast — the untyped legacy payload. As shipped, every parsed action is + forwarded uniformly (no inner switch needed): + + ```ts + // ── FORWARD-ONLY SHIM (remove after bull:integration holds no coexist + // jobs; see scripts/check-coexist-drain.mts) ──────────────────────── + const legacy = heavyJobDataSchema.safeParse(job.data) + if (legacy.success) { + const { delay: _delay, repeat: _repeat, ...opts } = job.opts + await heavyQueue.add(job.name, legacy.data, { + ...opts, + jobId: job.opts.jobId ?? job.id, + }) + return + } + ``` + + **Original design (superseded, kept for context):** the hybrid plan + below assumed old and new worker code could run concurrently during a + rolling deploy, so it executed 4 of the 5 actions in place using the + *moved* handlers and forwarded only the coalescing-sensitive + `coexistWhatsappFlush`: + + ```ts + // ── LEGACY DRAIN (remove after bull:integration holds no coexist jobs; + // see scripts/check-coexist-drain.mts) ───────────────────────────── + const legacy = heavyJobDataSchema.safeParse(job.data) + if (legacy.success) { + switch (legacy.data.type) { + // [R4-1][R5-1] Flush is the ONLY coalescing-sensitive action: two + // queues = two jobId namespaces, and whatsapp-flush has no + // staging-row claim (only processedAt), so running a legacy flush + // here while a fresh same-jobId flush runs on heavy is NOT safe. + // Re-enqueue into heavy under the original jobId — this restores + // SAME-SCHEME dedup (`coexist-flush-` vs its fresh twin, + // `coexist-run--` vs its fresh twin). Cross-scheme + // overlap (buffer-flush vs scan-run-flush for one phone) is + // PRE-EXISTING behavior in today's single integration queue: + // same-run overlap is serialized by the atomic 10-min run-row lease + // (whatsapp-flush.ts:874), but two DISTINCT active runs for one + // phone are not globally phone-serialized (the unique index only + // covers status='init', coexist-sync-run.ts:126). This migration + // leaves that behavior exactly as it is — it neither fixes nor + // worsens it. Never execute a legacy flush here. + case HeavyJobAction.coexistWhatsappFlush: { + const { delay: _d, repeat: _r, ...opts } = job.opts + await heavyQueue.add(job.name, legacy.data, { + ...opts, + jobId: job.opts.jobId ?? job.id, + }) + return + } + // The rest are safe to execute in place with their original opts: + // buffer is append-only staging; sync runs are claimRun-leased; + // attachment download is an idempotent per-message mirror (already + // required to tolerate stalled-job reprocessing). + case HeavyJobAction.coexistWhatsappBuffer: + return await coexistWhatsappBuffer(legacy.data.data) + case HeavyJobAction.coexistMessengerSync: + return await coexistMessengerSync(legacy.data.data) + case HeavyJobAction.coexistInstagramSync: + return await coexistInstagramSync(legacy.data.data) + case HeavyJobAction.coexistAttachmentDownload: + return await coexistAttachmentDownload(legacy.data.data) + default: { + const _exhaustive: never = legacy.data + return + } + } + } + ``` + + Why the original hybrid was thought to beat pure forwarding *and* pure + execute-in-place (superseded — see Amendment): + - Execute-in-place jobs keep their original `job.opts` + (backoff/retention/jobId) with nothing to copy, and the legacy queue's + coexist subset only shrinks (all follow-ups target `heavyQueue`); + delayed buffer flushes fire within ~1–2 min of deploy. + - The single unsafe-overlap action (flush) is funneled into one jobId + namespace via dedup-preserving forward — no concurrent flush possible. + - The integration worker still has the 10-min lock during the window + (unchanged in this PR), so long legacy chunks are safe. + + This concern is moot under the shipped forward-only design: Docker Swarm + stop-first means there is never a window where old code executes any + coexist chunk concurrently with new code, so uniformly forwarding + everything (including flush) is simplest and equally safe. +4. **[R4] Drain criterion & removal PR (~1 day later):** ship + `scripts/check-coexist-drain.mts` in this PR (reads + `integrationQueue.getJobs(["delayed","waiting","active","failed"])`, + filters the 5 type strings). The follow-up PR — gated on the script + reporting zero — deletes the forward-only shim block and the script + itself (the integration worker already imports no coexist handlers, so + there are no handler imports left to remove). +3. **Integration worker options after the split:** keep + `lockDuration`/`stalledInterval` at 10 min **but rewrite the comment** — the + surviving constraint is `CHAT_JOB_WAIT_TIMEOUT_MS` (max 9 min; env schema + says the wait must stay below the lock or BullMQ double-processes). Update + the stale references in `apps/worker/src/env.ts` (CHAT_JOB_WAIT_TIMEOUT_MS + comment) and `apps/worker/src/integration/utils/message.ts:19`. Optionally + note a future follow-up: lowering the max wait would allow a tighter lock. + +### Phase 4 — producer sites outside `apps/worker` + +1. `apps/worker/src/schedule/handlers/scan-coexist-runs.ts`: enqueue into + `heavyQueue` (jobId scheme `coexist-run-${id}-${attempts}` unchanged — + jobIds are per-queue, no dedup collision). +2. `packages/business/src/coexist/service.ts`: `coexistJobStrategies` action + types reference `IntegrationJobAction.coexist*` — retarget to + `HeavyJobAction`. +3. **WhatsApp webhook buffer enqueue** (`integrations/whatsapp/src/handlers/webhook.ts` + uses the injected `props.queue`, typed `ContextQueue` from `packages/sdk`): + - **Chosen approach:** add an optional `heavyQueue?: …` prop to + `HandleRequestProps` in `packages/sdk/src/lib/shared/index.ts` (typed with + a minimal `add` signature like `ContextQueue`) and switch + `enqueueCoexistPayloads` to it. + - **[R3] Inject it from BOTH builder webhook routes — mandatory, not + conditional.** The generic catch-all + (`apps/builder/src/app/integrations/[...integration]/webhook.ts:158-169`) + is a real WhatsApp ingress (it resolves `integrations[integrationType]` + and passes only `integrationQueue` today), alongside + `apps/builder/src/app/integrations/whatsapp/webhook/[integrationId]/route.ts:111`. + Missing either one silently keeps coexist payloads on `integration`. + - **[R5] Optional-prop footgun guard:** in `enqueueCoexistPayloads`, when + coexist payloads are present but `heavyQueue` is absent, log an + **error** (never silently `?.`-drop). Add tests covering both builder + routes and a direct SDK `handleRequest` invocation asserting the buffer + job lands on the coexist queue. + - Rejected: parsing coexist payloads in the builder route (leaks + channel-specific webhook parsing out of the integration package). + - ⚠️ `queue?.add("coexistWhatsappBuffer", …)` is **loosely typed** — the + compile-error safety net does NOT cover this site. Grep + `coexistWhatsappBuffer|coexistWhatsappFlush|coexistMessengerSync|coexistInstagramSync|coexistAttachmentDownload` + repo-wide at the end of the phase to catch any other stringly-typed + producer. +4. **[R2-2] Bull Board registration:** add `heavyQueue` to the queue array in + `apps/builder/src/app/developer/queues/[[...path]]/route.ts` (the + super-admin dashboard) — otherwise the moved jobs vanish from the ops UI. + (Codex verified this is the only queue-name-keyed admin/metric/alert + surface.) +5. Sweep remaining references: `apps/builder/src/features/messages/queries/index.ts` + (comment only), `packages/business/src/contact-inbox/service.ts` + (bulk-import reference), test files. **Do NOT touch** the DB enum + `coexistMessengerSyncPhase` (`packages/database/src/schema/coexist-sync-run.ts:47`) + — persistence naming stays unchanged. + +### Phase 5 — build & deploy wiring + +1. `apps/worker/tsdown.config.ts`: add `"src/heavy/worker.ts"` to `entry`. +2. `apps/worker/package.json`: add + `"worker:heavy": "dotenv -e ../../.env -- tsx --watch src/heavy/worker.ts"`. +3. Docker: **no change needed** — `docker-entrypoint.sh` auto-discovers + `dist/heavy/worker.mjs` → roster name `heavy`; `CMD ["worker","all"]` + starts it. Operators can also run it isolated (`worker heavy`) on a + beefier node later. +4. No new package ⇒ no `CI=true pnpm install` needed. + +### Phase 6 — tests & docs + +1. **[R6] Full test inventory** — update every coexist-touching suite, not + just the obvious six: + - `apps/worker/__tests__/coexist-*.test.ts` (all of them — attachment + download, instagram sync, messenger sync, whatsapp buffer, whatsapp + flush, whatsapp-flush-bsuid, plus any other suite importing the moved + handlers): new import paths; mocked queue modules `integrationQueue` → + `heavyQueue` where handlers now enqueue. + - **`integrations/whatsapp/__tests__/coexist-webhook.test.ts`** — the + producer regression suite for the actual buffer enqueue + (lines ~330-351): pass and assert the new `heavyQueue` prop. + - `apps/worker/__tests__/scan-coexist-runs.test.ts`: assert enqueue on + `heavyQueue` with unchanged jobId scheme. + - **[R2-3] `apps/builder/__tests__/integration-sendgrid-api.test.ts`** — + mocks `IntegrationJobAction.coexist*` (line ~59) because + `@chatbotx.io/business` transitively exports the coexist service; the + mock must gain `HeavyJobAction` or the suite fails at module init. + - **[R2-4] `apps/worker/__tests__/docker-entrypoint.test.ts`** — add + `heavy` to the roster fixture. +2. `apps/worker/__tests__/integration-worker-boot.test.ts`: keep passing after + case removal; add a sibling `heavy-worker-boot.test.ts` (mirror its + structure) plus forward-only-shim tests: every legacy coexist job type is + forwarded to `heavyQueue` preserving jobId and opts (minus + `delay`/`repeat`), with no handler executed in place — the integration + worker imports none of the coexist handlers; a malformed legacy payload + fails the schema parse and falls through to the normal switch. +3. **[R7] Build-phase regression:** extend + `packages/worker-config/__tests__/no-redis-env.test.ts` (same + pattern as existing queues) so `heavyQueue` resolves to `fakeQueue` + under `NEXT_PHASE=phase-production-build` / no-Redis env — the builder + routes import it at build time. +4. **[R2-5] Stale-doc policy:** historic plan docs referencing the old + handler paths (`docs/plans/coexist-skip-ai-context-plan.md:33`, + `docs/plans/2026-08-17-whatsapp-bsuid-username.md:68`) are declared + historical records — exempt, not updated. +5. Docs: worker table + queue list in + `.agents/skills/worker-development/SKILL.md`; mention the split where + integration-worker lock sizing is documented. + +### Phase 7 — verification (definition of done) + +- `pnpm lint` +- `pnpm --filter worker check-types`, `--filter @chatbotx.io/worker-config`, + `--filter @chatbotx.io/business`, `--filter @chatbotx.io/sdk`, + `--filter builder check-types` +- `pnpm --filter worker test` (at minimum: 6 coexist suites, + `scan-coexist-runs`, both worker-boot suites) and + `pnpm --filter @chatbotx.io/business test` (coexist service suite) +- Manual dev smoke: `pnpm --filter worker dev` boots the heavy worker; + trigger a WhatsApp coexist webhook → job lands in Redis key + `bull:heavy:*`, flush completes; `scanCoexistRuns` cron enqueues into the + new queue. + +## 5. Risks + +| Risk | Severity | Mitigation | +|---|---|---| +| Legacy coexist jobs (incl. delayed flushes) in `integration` queue at deploy get dropped | HIGH | Phase 3 forward-only shim re-enqueues them into `heavyQueue` under their original jobId; drain-check script gates the removal PR; `scan-coexist-runs`/`pickDueRuns` re-picks sync runs as backstop | +| Cross-queue concurrent flush for one phoneNumberId during drain window (verified UNSAFE under the original execute-in-place design — no staging-row claim) | HIGH (superseded) | Moot under the shipped forward-only shim: Docker Swarm stop-first means old and new code never run concurrently, and every legacy job (including flush) is forwarded, never executed in place | +| Stringly-typed producer missed (only `props.queue?.add("coexistWhatsappBuffer")` today) → jobs enqueued into a queue with a consumer that warn-drops them | HIGH | Repo-wide string grep in Phase 4; typed sites are compile-enforced by the union removal | +| New worker forgets blocked-owner guard / audit context (invariant 15) | MED | Explicit Phase 2 checklist item + boot test | +| Builder route passes only `queue`, coexist payloads silently dropped via `?.` | MED | Error log in `enqueueCoexistPayloads` when payloads present but no `heavyQueue`; route + SDK-level tests | +| `moduleNameMapper`-style test mocks pointing at old paths silently green | LOW | Run the exact suites in Phase 7, not just typecheck | +| +1 Redis connection per deployment unit | LOW | Negligible; same pattern as every existing queue | + +## 6. Code-quality requirements (binding for implementation) + +1. **Dispatch follows the repo's worker idiom: exhaustive `switch` with the + `never` guard** (as `apps/worker/src/integration/worker.ts:81` does). + Rationale [R2-6]: the five payloads are a discriminated union; a + homogeneous `Record` map loses the type correlation + between action and payload at the call site (forcing casts), while the + exhaustive switch keeps compile-time completeness without casts. With only + 5 actions the switch stays small; no `if/else` chains. (The shipped + forward-only shim in the integration worker itself needs no inner switch + at all — see Amendment — but the `heavy` worker's own dispatch still + follows this idiom.) +2. **Reuse existing handlers verbatim.** The 5 coexist handler functions move + as-is; no rewrites, no new wrapper functions where an existing one + (`runJobWithAuditContext`, `resolveWorkspaceId`, `isBlockedWorkspace`) + already exists. +3. **No channel hard-coding in shared files.** The SDK `HandleRequestProps` + gains a generic `heavyQueue` capability prop (typed like `ContextQueue`), + not a WhatsApp-specific field; only the WhatsApp integration consumes it. + Existing channel-keyed logic stays table-driven + (`coexistJobStrategies`/`coexistRunEnqueuers` maps — extend, don't fork). +4. **No `any`.** The forward-only shim narrows via the typed + parser/type-guard defined in Phase 3 (returning `HeavyJobData | null`), + not `as any`; `as never` only in the established exhaustiveness-guard + idiom. +5. **Business layer intact.** All DB access stays behind + `coexistService`/repositories; the move introduces zero raw SQL and zero + direct `db` imports in `apps/`. +6. **No duplicate code.** Worker boilerplate (bootstrap, guards, shutdown) + mirrors the integration worker's existing helpers; if any block would be + copy-pasted a third time, extract it to `apps/worker/src/lib/` instead. +7. **High-load posture unchanged.** Same BUC adaptive throttle, same + `addBulk` batching, same jobId dedup; concurrency env-tunable. +8. **Every touched behavior has a test** (Phase 6 inventory) — including the + forward-only shim, the missing-`heavyQueue` error log, and build-phase + `fakeQueue`. + +## 7. Explicitly out of scope + +- Retuning `INTEGRATION_WORKER_CONCURRENCY` / lowering the integration lock + below 10 min (blocked on `CHAT_JOB_WAIT_TIMEOUT_MS` max; note as follow-up). +- Any behavior change inside the coexist sync algorithms (BUC throttle, + chunking, staging tables). +- Moving `scanCoexistRuns`/`purgeCoexistStaging` off the schedule worker. +- Generic "heavy" queue abstraction for future domains beyond what already + exists. diff --git a/integrations/whatsapp/__tests__/coexist-webhook.test.ts b/integrations/whatsapp/__tests__/coexist-webhook.test.ts index de3df958aa..a34867814a 100644 --- a/integrations/whatsapp/__tests__/coexist-webhook.test.ts +++ b/integrations/whatsapp/__tests__/coexist-webhook.test.ts @@ -329,11 +329,14 @@ describe("webhookHandler — H2 setTimeout race", () => { const queueAdd = vi.fn().mockResolvedValue(undefined) const queue = { add: queueAdd } as never + const heavyQueueAdd = vi.fn().mockResolvedValue(undefined) + const heavyQueue = { add: heavyQueueAdd } as never const handlerPromise = webhookHandler({ config: baseConfig, req: makePostRequest(coexistBody), queue, + heavyQueue, }) // Advance past the internal 300 ms guard — handle_post is still pending. @@ -347,10 +350,16 @@ describe("webhookHandler — H2 setTimeout race", () => { // Let remaining microtasks and promises settle. await handlerPromise - expect(queueAdd).toHaveBeenCalledWith( + // The coexist buffer enqueue lands on the `heavy` queue, not the + // channel-domain `integration` queue. + expect(heavyQueueAdd).toHaveBeenCalledWith( "coexistWhatsappBuffer", expect.objectContaining({ type: "coexistWhatsappBuffer" }), ) + expect(queueAdd).not.toHaveBeenCalledWith( + "coexistWhatsappBuffer", + expect.anything(), + ) }) it("(b) no unhandled rejection when handle_post rejects after the timeout", async () => { @@ -366,11 +375,14 @@ describe("webhookHandler — H2 setTimeout race", () => { const queueAdd = vi.fn().mockResolvedValue(undefined) const queue = { add: queueAdd } as never + const heavyQueueAdd = vi.fn().mockResolvedValue(undefined) + const heavyQueue = { add: heavyQueueAdd } as never const handlerPromise = webhookHandler({ config: baseConfig, req: makePostRequest(coexistBody), queue, + heavyQueue, }) // Advance past the 300 ms guard; handle_post is still pending. @@ -384,9 +396,44 @@ describe("webhookHandler — H2 setTimeout race", () => { await expect(handlerPromise).rejects.toThrow() // Coexist payloads must NOT be enqueued (HMAC not verified). + expect(heavyQueueAdd).not.toHaveBeenCalled() + expect(queueAdd).not.toHaveBeenCalledWith( + "coexistWhatsappBuffer", + expect.anything(), + ) + }) + + it("(c) logs an error (no throw) and still acknowledges the webhook when coexist payloads exist but no heavyQueue was provided", async () => { + handlePostMock.mockResolvedValue(200) + + const queueAdd = vi.fn().mockResolvedValue(undefined) + const queue = { add: queueAdd } as never + + const { logger } = await import("../src/lib/logger") + const loggerErrorSpy = vi + .spyOn(logger, "error") + .mockImplementation(() => undefined as unknown as undefined) + + const handlerPromise = webhookHandler({ + config: baseConfig, + req: makePostRequest(coexistBody), + queue, + // heavyQueue intentionally omitted — the footgun guard must fire. + }) + + await vi.advanceTimersByTimeAsync(400) + + await expect(handlerPromise).resolves.toBe("ok") + + expect(loggerErrorSpy).toHaveBeenCalledWith( + expect.objectContaining({ count: 1 }), + expect.stringContaining("no heavyQueue was provided"), + ) expect(queueAdd).not.toHaveBeenCalledWith( "coexistWhatsappBuffer", expect.anything(), ) + + loggerErrorSpy.mockRestore() }) }) diff --git a/integrations/whatsapp/src/handlers/webhook.ts b/integrations/whatsapp/src/handlers/webhook.ts index b29a1f8044..af355c8962 100644 --- a/integrations/whatsapp/src/handlers/webhook.ts +++ b/integrations/whatsapp/src/handlers/webhook.ts @@ -25,6 +25,7 @@ type AutomaticEventPayload = { payload: WhatsappAutomaticEventPayload } type WebhookQueue = HandleRequestProps["queue"] +type HeavyQueue = HandleRequestProps["heavyQueue"] /** * Per Meta docs, coexist payloads arrive under three distinct `field` values @@ -306,16 +307,30 @@ const capturePostResult = async (input: { } const enqueueCoexistPayloads = async ( - queue: WebhookQueue, + heavyQueue: HeavyQueue, coexistPayloads: CoexistPayload[], ): Promise => { - if (coexistPayloads.length > 0) { - for (const { phoneNumberId, value } of coexistPayloads) { - await queue?.add("coexistWhatsappBuffer", { - type: "coexistWhatsappBuffer", - data: { phoneNumberId, payload: value }, - }) - } + if (coexistPayloads.length === 0) { + return + } + + // [R5] Optional-prop footgun guard: coexist payloads exist but the caller + // never wired `heavyQueue` into `handleRequest`. Never drop this silently + // via `?.` — log loudly so a missing builder-route wire-up surfaces as an + // error instead of quietly stranding coexist history on the floor. + if (!heavyQueue) { + logger.error( + { count: coexistPayloads.length }, + "WhatsApp coexist payloads received but no heavyQueue was provided to handleRequest — dropping", + ) + return + } + + for (const { phoneNumberId, value } of coexistPayloads) { + await heavyQueue.add("coexistWhatsappBuffer", { + type: "coexistWhatsappBuffer", + data: { phoneNumberId, payload: value }, + }) } } @@ -430,7 +445,7 @@ export const webhookHandler = async ( rawBodyBuffer, middleware, }) - await enqueueCoexistPayloads(props.queue, coexistPayloads) + await enqueueCoexistPayloads(props.heavyQueue, coexistPayloads) await enqueueAutomaticEventPayloads(props.queue, automaticEventPayloads) await dispatchWebhookResult(props.queue, result) diff --git a/packages/business/__tests__/coexist.service.test.ts b/packages/business/__tests__/coexist.service.test.ts index 39feefb8a3..6ac81034bd 100644 --- a/packages/business/__tests__/coexist.service.test.ts +++ b/packages/business/__tests__/coexist.service.test.ts @@ -37,7 +37,7 @@ vi.mock("@chatbotx.io/redis", () => ({ })) vi.mock("@chatbotx.io/worker-config", () => ({ - IntegrationJobAction: { + HeavyJobAction: { coexistInstagramSync: "coexistInstagramSync", coexistMessengerSync: "coexistMessengerSync", coexistWhatsappFlush: "coexistWhatsappFlush", diff --git a/packages/business/src/coexist/service.ts b/packages/business/src/coexist/service.ts index 2d7c7a27fd..0f85278266 100644 --- a/packages/business/src/coexist/service.ts +++ b/packages/business/src/coexist/service.ts @@ -10,7 +10,7 @@ import { type PullCoexistChannel, } from "@chatbotx.io/database/repositories" import type { CoexistSyncRunModel } from "@chatbotx.io/database/types" -import { IntegrationJobAction } from "@chatbotx.io/worker-config" +import { HeavyJobAction } from "@chatbotx.io/worker-config" import { BaseService } from "../base.service" export type CoexistEnableInput = { @@ -33,26 +33,26 @@ export type CoexistJobStrategy = | { mode: "pull" action: - | typeof IntegrationJobAction.coexistMessengerSync - | typeof IntegrationJobAction.coexistInstagramSync + | typeof HeavyJobAction.coexistMessengerSync + | typeof HeavyJobAction.coexistInstagramSync } | { mode: "buffered" - action: typeof IntegrationJobAction.coexistWhatsappFlush + action: typeof HeavyJobAction.coexistWhatsappFlush } export const coexistJobStrategies = { messenger: { mode: "pull", - action: IntegrationJobAction.coexistMessengerSync, + action: HeavyJobAction.coexistMessengerSync, }, instagram: { mode: "pull", - action: IntegrationJobAction.coexistInstagramSync, + action: HeavyJobAction.coexistInstagramSync, }, whatsapp: { mode: "buffered", - action: IntegrationJobAction.coexistWhatsappFlush, + action: HeavyJobAction.coexistWhatsappFlush, }, } satisfies Record diff --git a/packages/sdk/src/lib/shared/index.ts b/packages/sdk/src/lib/shared/index.ts index 9eddb81a0d..bd557722ed 100644 --- a/packages/sdk/src/lib/shared/index.ts +++ b/packages/sdk/src/lib/shared/index.ts @@ -15,6 +15,14 @@ export type HandleRequestProps = { config: IConfig req: Request queue?: ContextQueue + /** + * Optional handle to the workload-class `heavy` queue (long-lock, + * throughput-oriented, latency-tolerant jobs — see + * docs/plans/2026-08-30-heavy-worker-coexist-split.md). Generic — not + * channel-specific — so any integration can route its own heavy jobs here; + * today only the WhatsApp integration's coexist buffer enqueue consumes it. + */ + heavyQueue?: ContextQueue } export type ReceivedMessageProps = { diff --git a/packages/worker-config/__tests__/heavy-job-data-schema.test.ts b/packages/worker-config/__tests__/heavy-job-data-schema.test.ts new file mode 100644 index 0000000000..790de923cb --- /dev/null +++ b/packages/worker-config/__tests__/heavy-job-data-schema.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, test } from "vitest" +import { heavyJobDataSchema } from "../src/queues/heavy" + +// `heavyJobDataSchema` is the runtime counterpart of `HeavyJobData`, consumed +// by the integration worker's forward-only shim to *parse* — not cast — an +// untyped legacy job payload still sitting in `bull:integration` after the +// coexist-to-heavy queue cutover. See +// docs/plans/2026-08-30-heavy-worker-coexist-split.md. + +describe("heavyJobDataSchema", () => { + test("accepts a valid coexistWhatsappBuffer payload", () => { + const result = heavyJobDataSchema.safeParse({ + type: "coexistWhatsappBuffer", + data: { phoneNumberId: "phone-1", payload: { entry: [] } }, + }) + + expect(result.success).toBe(true) + }) + + test("accepts a valid coexistWhatsappFlush payload without runId (webhook-driven buffer flush)", () => { + const result = heavyJobDataSchema.safeParse({ + type: "coexistWhatsappFlush", + data: { phoneNumberId: "phone-1" }, + }) + + expect(result.success).toBe(true) + }) + + test("accepts a valid coexistWhatsappFlush payload with runId (scheduler/self-continuation)", () => { + const result = heavyJobDataSchema.safeParse({ + type: "coexistWhatsappFlush", + data: { runId: "run-1", phoneNumberId: "phone-1" }, + }) + + expect(result.success).toBe(true) + }) + + test("accepts a valid coexistMessengerSync payload", () => { + const result = heavyJobDataSchema.safeParse({ + type: "coexistMessengerSync", + data: { runId: "run-1", integrationId: "int-1", workspaceId: "ws-1" }, + }) + + expect(result.success).toBe(true) + }) + + test("accepts a valid coexistInstagramSync payload", () => { + const result = heavyJobDataSchema.safeParse({ + type: "coexistInstagramSync", + data: { runId: "run-1", integrationId: "int-1", workspaceId: "ws-1" }, + }) + + expect(result.success).toBe(true) + }) + + test("accepts a valid coexistAttachmentDownload payload for each channel", () => { + for (const channel of ["messenger", "whatsapp", "instagram"] as const) { + const result = heavyJobDataSchema.safeParse({ + type: "coexistAttachmentDownload", + data: { + attachmentId: "att-1", + workspaceId: "ws-1", + channel, + integrationId: "int-1", + }, + }) + + expect(result.success).toBe(true) + } + }) + + test("rejects an unrecognized type string", () => { + const result = heavyJobDataSchema.safeParse({ + type: "incomingMessage", + data: {}, + }) + + expect(result.success).toBe(false) + }) + + test("rejects a coexistAttachmentDownload payload with an invalid channel", () => { + const result = heavyJobDataSchema.safeParse({ + type: "coexistAttachmentDownload", + data: { + attachmentId: "att-1", + workspaceId: "ws-1", + channel: "telegram", + integrationId: "int-1", + }, + }) + + expect(result.success).toBe(false) + }) + + test("rejects a payload missing a required field for its type", () => { + const result = heavyJobDataSchema.safeParse({ + type: "coexistMessengerSync", + data: { runId: "run-1", integrationId: "int-1" }, + }) + + expect(result.success).toBe(false) + }) + + test("rejects a non-object payload", () => { + expect(heavyJobDataSchema.safeParse("not-an-object").success).toBe(false) + expect(heavyJobDataSchema.safeParse(null).success).toBe(false) + expect(heavyJobDataSchema.safeParse(undefined).success).toBe(false) + }) +}) diff --git a/packages/worker-config/__tests__/no-redis-env.test.ts b/packages/worker-config/__tests__/no-redis-env.test.ts index e5b22de272..efef141e00 100644 --- a/packages/worker-config/__tests__/no-redis-env.test.ts +++ b/packages/worker-config/__tests__/no-redis-env.test.ts @@ -51,4 +51,27 @@ describe("isNoRedisEnv", () => { expect(typeof aiAgentQueue.add).toBe("function") expect(aiAgentQueue).not.toHaveProperty("opts") }) + + // The `heavy` queue (coexist historical sync) is imported at build time by + // the builder webhook routes (`heavyQueue` prop on `HandleRequestProps`), so + // it must resolve to the fake queue under a no-Redis env just like every + // other queue barrel — see docs/plans/2026-08-30-heavy-worker-coexist-split.md. + test("importing the heavy queue barrel under vitest yields the fake queue, not a BullMQ Queue", async () => { + vi.stubEnv("VITEST", "true") + + const { heavyQueue } = await import("../src/queues/heavy") + + expect(typeof heavyQueue.add).toBe("function") + expect(heavyQueue).not.toHaveProperty("opts") + }) + + test("does not dial Redis for the heavy queue when NEXT_PHASE is phase-production-build", async () => { + vi.stubEnv("VITEST", "") + vi.stubEnv("NEXT_PHASE", "phase-production-build") + + const { heavyQueue } = await import("../src/queues/heavy") + + expect(typeof heavyQueue.add).toBe("function") + expect(heavyQueue).not.toHaveProperty("opts") + }) }) diff --git a/packages/worker-config/src/index.ts b/packages/worker-config/src/index.ts index abf34940f9..edaa8a420d 100644 --- a/packages/worker-config/src/index.ts +++ b/packages/worker-config/src/index.ts @@ -7,6 +7,7 @@ export * from "./message-queue" export * from "./queues/ai-agent" export * from "./queues/chat" export * from "./queues/default" +export * from "./queues/heavy" export * from "./queues/integration" export * from "./queues/notification" export * from "./queues/quota" diff --git a/packages/worker-config/src/lib/types.ts b/packages/worker-config/src/lib/types.ts index 6f784a4554..1b98e1fa93 100644 --- a/packages/worker-config/src/lib/types.ts +++ b/packages/worker-config/src/lib/types.ts @@ -12,4 +12,5 @@ export const queueNames = z.enum([ "broadcast", "quota", "notification", + "heavy", ]) diff --git a/packages/worker-config/src/queues/heavy/index.ts b/packages/worker-config/src/queues/heavy/index.ts new file mode 100644 index 0000000000..6484bf031f --- /dev/null +++ b/packages/worker-config/src/queues/heavy/index.ts @@ -0,0 +1,166 @@ +import { Queue } from "bullmq" +import { z } from "zod" +import { + defaultJobOptions, + fakeQueue, + getRedisConnection, + isNoRedisEnv, +} from "../../lib/connection" +import { queueNames } from "../../lib/types" + +/** + * Workload-class queue: long-lock (10 min), throughput-oriented, + * latency-tolerant jobs. Coexist historical sync is its first tenant; future + * heavy actions (e.g. a contact-import backfill) join this queue instead of + * spawning a new worker. See docs/plans/2026-08-30-heavy-worker-coexist-split.md. + * + * Job action strings are unchanged from their former home on + * `IntegrationJobAction` — this keeps jobId/dedup schemes and the + * forward-only shim in `apps/worker/src/integration/worker.ts` trivial. + */ +export const HeavyJobAction = { + coexistWhatsappBuffer: "coexistWhatsappBuffer", + coexistWhatsappFlush: "coexistWhatsappFlush", + coexistMessengerSync: "coexistMessengerSync", + coexistInstagramSync: "coexistInstagramSync", + coexistAttachmentDownload: "coexistAttachmentDownload", +} as const + +/** Buffers a raw WhatsApp Coexistence history payload into the staging table. */ +export type HeavyJobCoexistWhatsappBuffer = { + type: typeof HeavyJobAction.coexistWhatsappBuffer + data: { + phoneNumberId: string + payload: unknown + } +} + +const heavyJobCoexistWhatsappBufferSchema = z.object({ + type: z.literal(HeavyJobAction.coexistWhatsappBuffer), + data: z.object({ + phoneNumberId: z.string(), + payload: z.unknown(), + }), +}) satisfies z.ZodType + +/** + * Flushes buffered WhatsApp staging rows into Contact/Message once enabled. + * `runId` is optional: the buffer (webhook-driven) omits it and the flush + * handler looks up the live run by phoneNumberId. Scheduler + self-continuation + * keep passing the explicit runId so they stay pinned to a specific run. + */ +export type HeavyJobCoexistWhatsappFlush = { + type: typeof HeavyJobAction.coexistWhatsappFlush + data: { + runId?: string + phoneNumberId: string + } +} + +const heavyJobCoexistWhatsappFlushSchema = z.object({ + type: z.literal(HeavyJobAction.coexistWhatsappFlush), + data: z.object({ + runId: z.string().optional(), + phoneNumberId: z.string(), + }), +}) satisfies z.ZodType + +/** Pulls historical Messenger conversations/messages via the Graph API. */ +export type HeavyJobCoexistMessengerSync = { + type: typeof HeavyJobAction.coexistMessengerSync + data: { + runId: string + integrationId: string + workspaceId: string + } +} + +const heavyJobCoexistMessengerSyncSchema = z.object({ + type: z.literal(HeavyJobAction.coexistMessengerSync), + data: z.object({ + runId: z.string(), + integrationId: z.string(), + workspaceId: z.string(), + }), +}) satisfies z.ZodType + +/** Pulls historical native Instagram conversations/messages via the Graph API. */ +export type HeavyJobCoexistInstagramSync = { + type: typeof HeavyJobAction.coexistInstagramSync + data: { + runId: string + integrationId: string + workspaceId: string + } +} + +const heavyJobCoexistInstagramSyncSchema = z.object({ + type: z.literal(HeavyJobAction.coexistInstagramSync), + data: z.object({ + runId: z.string(), + integrationId: z.string(), + workspaceId: z.string(), + }), +}) satisfies z.ZodType + +/** + * Downloads a Coexist attachment's bytes from the channel API (Facebook URL + * for Messenger; WhatsApp media-id for WhatsApp — both encoded into + * `Attachment.originPath` by the historical importer), uploads to object + * storage, and UPDATEs the row with the resulting S3 path. Dispatched per + * attachment after `bulkImportMessages` inserts the placeholder row. + * + * Idempotency: jobId `att-${attachmentId}` dedups concurrent enqueues; the + * handler additionally checks the originPath prefix to no-op on retries + * where a prior worker already finished the upload. + */ +export type HeavyJobCoexistAttachmentDownload = { + type: typeof HeavyJobAction.coexistAttachmentDownload + data: { + attachmentId: string + workspaceId: string + channel: "messenger" | "whatsapp" | "instagram" + integrationId: string + } +} + +const heavyJobCoexistAttachmentDownloadSchema = z.object({ + type: z.literal(HeavyJobAction.coexistAttachmentDownload), + data: z.object({ + attachmentId: z.string(), + workspaceId: z.string(), + channel: z.enum(["messenger", "whatsapp", "instagram"]), + integrationId: z.string(), + }), +}) satisfies z.ZodType + +export type HeavyJobData = + | HeavyJobCoexistWhatsappBuffer + | HeavyJobCoexistWhatsappFlush + | HeavyJobCoexistMessengerSync + | HeavyJobCoexistInstagramSync + | HeavyJobCoexistAttachmentDownload + +/** + * Runtime counterpart of `HeavyJobData`, consumed by the integration worker's + * forward-only shim (`apps/worker/src/integration/worker.ts`) to *parse* — + * not cast — an untyped legacy job payload still sitting in `bull:integration` + * after cutover. + */ +export const heavyJobDataSchema = z.discriminatedUnion("type", [ + heavyJobCoexistWhatsappBufferSchema, + heavyJobCoexistWhatsappFlushSchema, + heavyJobCoexistMessengerSyncSchema, + heavyJobCoexistInstagramSyncSchema, + heavyJobCoexistAttachmentDownloadSchema, +]) + +// No per-action JobsOptions today — the default retry (`attempts: 2`) is +// preserved for every coexist action; the scan-runs DB-side attempt tracking +// (`CoexistSyncRun.attempts`) is the real retry authority for sync runs. +export const heavyQueue = isNoRedisEnv() + ? fakeQueue + : new Queue(queueNames.enum.heavy, { + connection: getRedisConnection(), + defaultJobOptions, + }) diff --git a/packages/worker-config/src/queues/integration/index.ts b/packages/worker-config/src/queues/integration/index.ts index 52f8ec5143..ec952da676 100644 --- a/packages/worker-config/src/queues/integration/index.ts +++ b/packages/worker-config/src/queues/integration/index.ts @@ -42,11 +42,6 @@ export const IntegrationJobAction = { assignConversation: "assignConversation", createMessage: "createMessage", sendEmail: "sendEmail", - coexistWhatsappBuffer: "coexistWhatsappBuffer", - coexistWhatsappFlush: "coexistWhatsappFlush", - coexistMessengerSync: "coexistMessengerSync", - coexistInstagramSync: "coexistInstagramSync", - coexistAttachmentDownload: "coexistAttachmentDownload", adsAutomaticEvent: "adsAutomaticEvent", updateContactAvatar: "updateContactAvatar", channelLabelChange: "channelLabelChange", @@ -301,15 +296,6 @@ export type IntegrationJobSendSequenceFlow = { } } -/** Buffers a raw WhatsApp Coexistence history payload into the staging table. */ -export type IntegrationJobCoexistWhatsappBuffer = { - type: typeof IntegrationJobAction.coexistWhatsappBuffer - data: { - phoneNumberId: string - payload: unknown - } -} - export type IntegrationJobChannelLabelChange = { type: typeof IntegrationJobAction.channelLabelChange data: @@ -325,61 +311,6 @@ export type IntegrationJobChannelLabelChange = { } } -/** - * Flushes buffered WhatsApp staging rows into Contact/Message once enabled. - * `runId` is optional: the buffer (webhook-driven) omits it and the flush - * handler looks up the live run by phoneNumberId. Scheduler + self-continuation - * keep passing the explicit runId so they stay pinned to a specific run. - */ -export type IntegrationJobCoexistWhatsappFlush = { - type: typeof IntegrationJobAction.coexistWhatsappFlush - data: { - runId?: string - phoneNumberId: string - } -} - -/** Pulls historical Messenger conversations/messages via the Graph API. */ -export type IntegrationJobCoexistMessengerSync = { - type: typeof IntegrationJobAction.coexistMessengerSync - data: { - runId: string - integrationId: string - workspaceId: string - } -} - -/** Pulls historical native Instagram conversations/messages via the Graph API. */ -export type IntegrationJobCoexistInstagramSync = { - type: typeof IntegrationJobAction.coexistInstagramSync - data: { - runId: string - integrationId: string - workspaceId: string - } -} - -/** - * Downloads a Coexist attachment's bytes from the channel API (Facebook URL - * for Messenger; WhatsApp media-id for WhatsApp — both encoded into - * `Attachment.originPath` by the historical importer), uploads to object - * storage, and UPDATEs the row with the resulting S3 path. Dispatched per - * attachment after `bulkImportMessages` inserts the placeholder row. - * - * Idempotency: jobId `att-${attachmentId}` dedups concurrent enqueues; the - * handler additionally checks the originPath prefix to no-op on retries - * where a prior worker already finished the upload. - */ -export type IntegrationJobCoexistAttachmentDownload = { - type: typeof IntegrationJobAction.coexistAttachmentDownload - data: { - attachmentId: string - workspaceId: string - channel: "messenger" | "whatsapp" | "instagram" - integrationId: string - } -} - export type IntegrationJobAdsAutomaticEvent = { type: typeof IntegrationJobAction.adsAutomaticEvent data: { @@ -604,11 +535,6 @@ export type IntegrationJobData = | IntegrationJobCreateMessage | IntegrationJobProcessAutomatedResponse | IntegrationJobSendSequenceFlow - | IntegrationJobCoexistWhatsappBuffer - | IntegrationJobCoexistWhatsappFlush - | IntegrationJobCoexistMessengerSync - | IntegrationJobCoexistInstagramSync - | IntegrationJobCoexistAttachmentDownload | IntegrationJobAdsAutomaticEvent | IntegrationJobUpdateContactAvatar | IntegrationJobChannelLabelChange diff --git a/scripts/check-coexist-drain.mts b/scripts/check-coexist-drain.mts new file mode 100644 index 0000000000..a77de05a68 --- /dev/null +++ b/scripts/check-coexist-drain.mts @@ -0,0 +1,90 @@ +/** + * FORWARD-ONLY SHIM removal gate for the heavy-worker/coexist split + * (docs/plans/2026-08-30-heavy-worker-coexist-split.md, Phase 3). + * + * The 5 coexist actions (`coexistWhatsappBuffer`, `coexistWhatsappFlush`, + * `coexistMessengerSync`, `coexistInstagramSync`, + * `coexistAttachmentDownload`) moved from the `integration` queue to the new + * `heavy` queue/worker. `bull:integration` survives a deploy, so the + * integration worker keeps a FORWARD-ONLY SHIM block + * (`apps/worker/src/integration/worker.ts`) that forwards any already-queued + * coexist job it finds there into the `heavy` queue. This script is the + * removal gate for that block: it reports how many coexist jobs remain in + * `bull:integration`, and exits non-zero while any are left. + * + * NEVER writes anything — read-only via `Queue.getJobs`. + * + * Usage: + * pnpm tsx scripts/check-coexist-drain.mts [envFile] + * + * Defaults to `.env`. Requires REDIS_URL (no DB access needed). + */ + +const [envFile = ".env"] = process.argv.slice(2) +process.loadEnvFile(envFile) +process.env.SKIP_ENV_CHECK = "true" + +// Import AFTER env is loaded — package env schemas read process.env at import +// time. Imported by relative path (not the `@chatbotx.io/worker-config` bare +// specifier) because this script lives at the repo root, outside any +// workspace package's own dependency graph — Node resolves a relative +// import's bare specifiers against ITS OWN location, so this reaches +// `packages/worker-config`'s node_modules instead of the repo root's (which +// does not depend on `@chatbotx.io/worker-config` at all). Mirrors the same +// trick `scripts/debug-run-job.mts` and +// `scripts/audit-bot-field-reserved-names.mts` use. +const { HeavyJobAction, integrationQueue } = await import( + "../packages/worker-config/src/index.ts" +) + +const COEXIST_JOB_TYPES: ReadonlySet = new Set( + Object.values(HeavyJobAction), +) + +const DRAIN_STATUSES = ["delayed", "waiting", "active", "failed"] as const + +type JobCounts = Record + +async function countRemainingCoexistJobs(): Promise { + const jobs = await integrationQueue.getJobs([...DRAIN_STATUSES]) + const counts: JobCounts = {} + + for (const job of jobs) { + const type = + job.data && typeof job.data === "object" && "type" in job.data + ? String((job.data as { type: unknown }).type) + : undefined + if (!(type && COEXIST_JOB_TYPES.has(type))) { + continue + } + counts[type] = (counts[type] ?? 0) + 1 + } + + return counts +} + +console.log( + `Checking bull:integration for legacy coexist jobs (${DRAIN_STATUSES.join(", ")})...`, +) + +const counts = await countRemainingCoexistJobs() +const total = Object.values(counts).reduce((sum, count) => sum + count, 0) + +if (total > 0) { + console.error("Legacy coexist jobs still remain in bull:integration:") + for (const [type, count] of Object.entries(counts)) { + console.error(` ${type}: ${count}`) + } + console.error( + "\nDo NOT remove the FORWARD-ONLY SHIM block yet — re-run this script after the drain window.", + ) + process.exitCode = 1 +} else { + console.log( + "Clean: no legacy coexist jobs remain in bull:integration. Safe to remove the FORWARD-ONLY SHIM block and this script.", + ) + process.exitCode = 0 +} + +// Redis connection keeps the event loop alive — force exit. +setTimeout(() => process.exit(process.exitCode ?? 0), 500) From f8d63bd397951e0adc2a11dc3b5a349cc39ea4d9 Mon Sep 17 00:00:00 2001 From: Hung Phan Viet Date: Sun, 30 Aug 2026 19:57:41 +0700 Subject: [PATCH 2/2] chore(worker): correct stall recovery comments --- apps/worker/src/env.ts | 3 ++- apps/worker/src/integration/worker.ts | 10 ++++++---- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts index 6261bfef24..1e6f60e8b0 100644 --- a/apps/worker/src/env.ts +++ b/apps/worker/src/env.ts @@ -26,7 +26,8 @@ export const env = createEnv({ // otherwise BullMQ would treat the job as stalled and reprocess it (double // send). Validated so a bad value can't become NaN (= wait forever). Since // the coexist split (heavy worker), this max (9 min) is the ONLY reason the - // integration worker's lockDuration/stalledInterval stay at 10 min — see + // integration worker's lockDuration stays at 10 min (its stalledInterval + // reverted to the BullMQ default) — see // `apps/worker/src/integration/worker.ts`. CHAT_JOB_WAIT_TIMEOUT_MS: z.coerce .number() diff --git a/apps/worker/src/integration/worker.ts b/apps/worker/src/integration/worker.ts index 47d39ff40f..69172ad0ea 100644 --- a/apps/worker/src/integration/worker.ts +++ b/apps/worker/src/integration/worker.ts @@ -367,10 +367,12 @@ async function startIntegrationWorker() { // the max wait to tighten this lock (see plan §7, out of scope here). lockDuration: 10 * 60 * 1000, // No stalledInterval override: with coexist gone there are no - // long-running jobs here, so the BullMQ default (30s) applies — a - // crashed worker's jobs are re-picked in seconds instead of minutes. - // Live workers auto-renew their locks, so the frequent check cannot - // false-positive an in-progress job. + // long-running jobs here, so the BullMQ default (30s) applies. A + // crashed worker's job is reclaimed once its lock EXPIRES and the next + // scan runs — ~10–10.5 min after the crash (lock expiry + ≤30s scan), + // down from ~10–20 min with the old 10-min scan interval. Live workers + // auto-renew their locks (including through awaitChatJob waits), so the + // frequent check cannot false-positive an in-progress job. maxStalledCount: 1, }, )