refactor(worker): move coexist jobs to dedicated heavy worker

.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/<domain>/` 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

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

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,
 }))
 

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,

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

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] : []),

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)

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)

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

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"
 )
 
 // ---------------------------------------------------------------------------

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

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).

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

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

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