feat(worker): move coexist media backfill jobs to a dedicated low-priority queue

apps/worker/__tests__/coexist-instagram-sync.test.ts

@@ -63,6 +63,14 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     add: mockQueueAdd,
     addBulk: mockQueueAddBulk,
   },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: {
+    add: mockQueueAdd,
+    addBulk: mockQueueAddBulk,
+  },
 }))
 
 vi.mock("../src/lib/logger", () => ({

apps/worker/__tests__/coexist-messenger-sync.test.ts

@@ -104,6 +104,14 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     add: mockQueueAdd,
     addBulk: vi.fn().mockResolvedValue(undefined),
   },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: {
+    add: vi.fn().mockResolvedValue(undefined),
+    addBulk: vi.fn().mockResolvedValue(undefined),
+  },
 }))
 
 vi.mock("@chatbotx.io/database/schema", () => ({

apps/worker/__tests__/coexist-whatsapp-flush-lifecycle.test.ts

@@ -151,6 +151,11 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     coexistAttachmentDownload: "coexistAttachmentDownload",
   },
   integrationQueue: { add: mockQueueAdd, addBulk: vi.fn() },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: { add: vi.fn(), addBulk: vi.fn() },
 }))
 
 // Carries the real schema forward and overrides only the models these tests

apps/worker/__tests__/coexist-whatsapp-flush.test.ts

@@ -190,6 +190,11 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     coexistMessengerSync: "coexistMessengerSync",
   },
   integrationQueue: { add: mockQueueAdd },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: { add: vi.fn(), addBulk: vi.fn() },
 }))
 
 // Carries the real schema forward and overrides only the models these tests

apps/worker/__tests__/enqueue-attachment-downloads.test.ts

@@ -0,0 +1,100 @@
+import { beforeEach, describe, expect, it, vi } from "vitest"
+
+const { mockAddBulk } = vi.hoisted(() => ({
+  mockAddBulk: vi.fn(),
+}))
+
+// Coexist attachment downloads are light but high-volume and low-priority, so
+// they must be enqueued on the dedicated `low` queue — never the
+// latency-sensitive `integration` queue that drives customer replies.
+vi.mock("@chatbotx.io/worker-config", () => ({
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: { addBulk: mockAddBulk },
+}))
+
+import { enqueueAttachmentDownloadJobs } from "../src/integration/handlers/coexist/enqueue-attachment-downloads"
+
+describe("enqueueAttachmentDownloadJobs", () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockAddBulk.mockResolvedValue(undefined)
+  })
+
+  it.each([
+    "messenger",
+    "whatsapp",
+    "instagram",
+  ] as const)("routes %s attachment jobs to the low queue with preserved options", async (channel) => {
+    await enqueueAttachmentDownloadJobs({
+      workspaceId: "ws-1",
+      integrationId: "int-1",
+      channel,
+      attachmentIds: ["a1", "a2"],
+    })
+
+    expect(mockAddBulk).toHaveBeenCalledTimes(1)
+    const jobs = mockAddBulk.mock.calls[0][0]
+    expect(jobs).toHaveLength(2)
+    expect(jobs[0]).toEqual({
+      name: "coexistAttachmentDownload",
+      data: {
+        type: "coexistAttachmentDownload",
+        data: {
+          attachmentId: "a1",
+          workspaceId: "ws-1",
+          channel,
+          integrationId: "int-1",
+        },
+      },
+      opts: {
+        jobId: "att-a1",
+        attempts: 5,
+        backoff: { type: "exponential", delay: 30_000 },
+        removeOnComplete: true,
+        removeOnFail: { count: 100 },
+      },
+    })
+  })
+
+  it("produces jobIds free of the ':' delimiter BullMQ forbids", async () => {
+    await enqueueAttachmentDownloadJobs({
+      workspaceId: "ws-1",
+      integrationId: "int-1",
+      channel: "messenger",
+      attachmentIds: ["a1", "a2", "a3"],
+    })
+
+    const jobs = mockAddBulk.mock.calls[0][0]
+    for (const job of jobs) {
+      expect(job.opts.jobId).not.toContain(":")
+    }
+  })
+
+  it("no-ops without touching the queue when there are no attachments", async () => {
+    await enqueueAttachmentDownloadJobs({
+      workspaceId: "ws-1",
+      integrationId: "int-1",
+      channel: "whatsapp",
+      attachmentIds: [],
+    })
+
+    expect(mockAddBulk).not.toHaveBeenCalled()
+  })
+
+  it("propagates an addBulk failure so callers control their own error policy", async () => {
+    const failure = new Error("redis down")
+    mockAddBulk.mockRejectedValueOnce(failure)
+
+    await expect(
+      enqueueAttachmentDownloadJobs({
+        workspaceId: "ws-1",
+        integrationId: "int-1",
+        channel: "instagram",
+        attachmentIds: ["a1"],
+      }),
+    ).rejects.toThrow(failure)
+  })
+})

apps/worker/__tests__/enqueue-avatar-jobs.test.ts

@@ -0,0 +1,87 @@
+import { beforeEach, describe, expect, it, vi } from "vitest"
+
+const { mockAddBulk } = vi.hoisted(() => ({
+  mockAddBulk: vi.fn(),
+}))
+
+// The avatar backfill jobs are light but high-volume and low-priority, so they
+// must be enqueued on the dedicated `low` queue — never the latency-sensitive
+// `integration` queue that drives customer replies.
+vi.mock("@chatbotx.io/worker-config", () => ({
+  LowJobAction: {
+    updateContactAvatar: "updateContactAvatar",
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+  },
+  lowQueue: { addBulk: mockAddBulk },
+}))
+
+vi.mock("../src/lib/logger", () => ({
+  logger: { error: vi.fn(), warn: vi.fn(), info: vi.fn() },
+}))
+
+import { enqueueContactAvatarJobs } from "../src/integration/handlers/contact/enqueue-avatar-jobs"
+
+describe("enqueueContactAvatarJobs", () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockAddBulk.mockResolvedValue(undefined)
+  })
+
+  it("enqueues one updateContactAvatar job per contact on the low queue", async () => {
+    await enqueueContactAvatarJobs({
+      workspaceId: "ws-1",
+      contactInboxIds: new Map([
+        ["source-a", { contactInboxId: "ci-a" }],
+        ["source-b", { contactInboxId: "ci-b" }],
+      ]),
+    })
+
+    expect(mockAddBulk).toHaveBeenCalledTimes(1)
+    const jobs = mockAddBulk.mock.calls[0][0]
+    expect(jobs).toHaveLength(2)
+    expect(jobs[0]).toMatchObject({
+      name: "updateContactAvatar",
+      data: {
+        type: "updateContactAvatar",
+        data: {
+          workspaceId: "ws-1",
+          contactInboxId: "ci-a",
+          sourceId: "source-a",
+        },
+      },
+      opts: { jobId: "update-avatar-ci-a" },
+    })
+  })
+
+  it("produces jobIds free of the ':' delimiter BullMQ forbids", async () => {
+    await enqueueContactAvatarJobs({
+      workspaceId: "ws-1",
+      contactInboxIds: new Map([["source-a", { contactInboxId: "ci-a" }]]),
+    })
+
+    const jobs = mockAddBulk.mock.calls[0][0]
+    for (const job of jobs) {
+      expect(job.opts.jobId).not.toContain(":")
+    }
+  })
+
+  it("no-ops without touching the queue when there are no contacts", async () => {
+    await enqueueContactAvatarJobs({
+      workspaceId: "ws-1",
+      contactInboxIds: new Map(),
+    })
+
+    expect(mockAddBulk).not.toHaveBeenCalled()
+  })
+
+  it("swallows an addBulk failure so the caller's run is never failed", async () => {
+    mockAddBulk.mockRejectedValueOnce(new Error("redis down"))
+
+    await expect(
+      enqueueContactAvatarJobs({
+        workspaceId: "ws-1",
+        contactInboxIds: new Map([["source-a", { contactInboxId: "ci-a" }]]),
+      }),
+    ).resolves.toBeUndefined()
+  })
+})