fix(coexist): handle oversized attachments and message id collisions in historical import

Author: viethungpv91

apps/worker/__tests__/bulk-import-messages.test.ts

@@ -340,6 +340,52 @@ describe("bulkImportMessages", () => {
     expect(result.importedMessages).toBe(1)
   })
 
+  test("converges a multi-row PK collision by splitting and re-minting only the colliding row", async () => {
+    // Batch of two distinct messages; only src-2 collides with an existing DB
+    // row on (id, createdAt). The bulk insert fails, the batch is split, the
+    // src-1 half inserts cleanly, and only src-2 is re-minted to a free slot.
+    const pkError = Object.assign(new Error("duplicate key value"), {
+      cause: { code: "23505", constraint: "173_Message_pkey" },
+    })
+
+    let call = 0
+    mockBulkCreate.mockImplementation(
+      (rows: { id: string; sourceId: string | null }[]) => {
+        call++
+        // call 1: the full batch [src-1, src-2] — collides.
+        // call 3: the isolated original src-2 — still collides.
+        if (call === 1 || call === 3) {
+          throw pkError
+        }
+        // call 2: [src-1] inserts; call 4: [re-minted src-2] inserts.
+        return rows.map((r) => ({ id: r.id, sourceId: r.sourceId }))
+      },
+    )
+
+    const result = await bulkImportMessages({
+      ...BASE_PROPS,
+      messages: [makeMessage("src-1"), makeMessage("src-2")],
+    })
+
+    expect(mockBulkCreate).toHaveBeenCalledTimes(4)
+    // Both messages land — no data loss from the collision.
+    expect(result.importedMessages).toBe(2)
+
+    const originalSrc2Id = (
+      mockBulkCreate.mock.calls[2][0] as { id: string; sourceId: string }[]
+    )[0].id
+    const remintedSrc2Id = (
+      mockBulkCreate.mock.calls[3][0] as { id: string; sourceId: string }[]
+    )[0].id
+    // The isolated collider is re-minted to a different id; src-1 is untouched.
+    expect(remintedSrc2Id).not.toBe(originalSrc2Id)
+    const src1Call = mockBulkCreate.mock.calls[1][0] as {
+      id: string
+      sourceId: string
+    }[]
+    expect(src1Call[0].sourceId).toBe("src-1")
+  })
+
   test("does not bump activity itself — returns message timestamps for the caller to batch", async () => {
     // Activity bumps (lastMessageAt / lastActivityAt) are the caller's job now;
     // bulkImportMessages only inserts and reports the newest message time.
@@ -421,6 +467,22 @@ describe("bulkImportMessages", () => {
     expect(result.newestIncomingMessageAt).toEqual(incomingTimestamp)
   })
 
+  test("fails loudly when a single row cannot find a free slot after the re-mint cap", async () => {
+    // A row that keeps colliding on every re-mint must eventually give up with a
+    // clear error rather than looping forever.
+    const pkError = Object.assign(new Error("duplicate key value"), {
+      cause: { code: "23505", constraint: "173_Message_pkey" },
+    })
+    mockBulkCreate.mockRejectedValue(pkError)
+
+    await expect(
+      bulkImportMessages({ ...BASE_PROPS, messages: [makeMessage("src-1")] }),
+    ).rejects.toThrow("unresolved after")
+    // Initial bulk insert plus the bounded re-mint attempts — proves it looped
+    // and gave up, not that it tried once.
+    expect(mockBulkCreate.mock.calls.length).toBeGreaterThan(2)
+  })
+
   test("does NOT retry on non-PK errors — they propagate", async () => {
     const fkError = Object.assign(new Error("fk violation"), {
       cause: { code: "23503", constraint: "Message_conversationId_fkey" },

apps/worker/__tests__/coexist-attachment-download.test.ts

@@ -122,7 +122,10 @@ vi.mock("../src/lib/logger", () => ({
 // Import the handler AFTER mocks
 // ---------------------------------------------------------------------------
 
-import { coexistAttachmentDownload } from "../src/integration/handlers/coexist/attachment-download"
+import {
+  coexistAttachmentDownload,
+  MAX_ATTACHMENT_BYTES,
+} from "../src/integration/handlers/coexist/attachment-download"
 
 // ---------------------------------------------------------------------------
 // Fixtures
@@ -358,30 +361,41 @@ describe("coexistAttachmentDownload", () => {
   })
 
   // ─────────────────────────────────────────────────────────────────────────
-  // SECURITY: size cap
+  // SIZE CAP: an oversized attachment is a PERMANENT condition, so the handler
+  // must skip it terminally (log a warning and return) rather than throw —
+  // throwing would burn all BullMQ retry attempts on a job that can never
+  // succeed. See AttachmentTooLargeError.
   // ─────────────────────────────────────────────────────────────────────────
 
-  it("rejects Messenger response with content-length exceeding MAX_ATTACHMENT_BYTES", async () => {
+  it("skips (no throw) a Messenger response whose content-length exceeds the cap", async () => {
     wireSelectChain({
       id: "att-001",
       originPath: "https://example.com/media/huge.mp4",
       mimeType: "video/mp4",
     })
     wireIntegrationLookup()
 
-    const OVER_LIMIT = String(51 * 1024 * 1024) // 51 MB
+    const OVER_LIMIT = String(MAX_ATTACHMENT_BYTES + 1)
     const fetchMock = vi.fn(() =>
       Promise.resolve(makeFetchResponse({ contentLength: OVER_LIMIT })),
     )
     vi.stubGlobal("fetch", fetchMock)
 
-    await expect(coexistAttachmentDownload(BASE_DATA)).rejects.toThrow()
+    // Must resolve (terminal skip), never reject — a reject would trigger retry.
+    await expect(coexistAttachmentDownload(BASE_DATA)).resolves.toBeUndefined()
     expect(mockPutObject).not.toHaveBeenCalled()
+    expect(mockUpdateAttachment).not.toHaveBeenCalled()
+    expect(mockLoggerWarn).toHaveBeenCalledWith(
+      expect.objectContaining({ attachmentId: BASE_DATA.attachmentId }),
+      expect.stringContaining("exceeds size cap"),
+    )
+    // Terminal skip must NOT be logged as an error (that reads as a failure).
+    expect(mockLoggerError).not.toHaveBeenCalled()
 
     vi.unstubAllGlobals()
   })
 
-  it("rejects Messenger response whose body bytes exceed MAX_ATTACHMENT_BYTES (no content-length)", async () => {
+  it("skips (no throw) a Messenger response whose streamed body exceeds the cap (no content-length)", async () => {
     wireSelectChain({
       id: "att-001",
       originPath: "https://example.com/media/huge.mp4",
@@ -393,34 +407,60 @@ describe("coexistAttachmentDownload", () => {
       Promise.resolve(
         makeFetchResponse({
           contentLength: undefined, // no header — must be caught by streaming cap
-          totalBytes: 51 * 1024 * 1024, // 51 MB streamed
+          totalBytes: MAX_ATTACHMENT_BYTES + 1, // one byte past the cap
+          chunkSize: 10 * 1024 * 1024,
         }),
       ),
     )
     vi.stubGlobal("fetch", fetchMock)
 
-    await expect(coexistAttachmentDownload(BASE_DATA)).rejects.toThrow()
+    await expect(coexistAttachmentDownload(BASE_DATA)).resolves.toBeUndefined()
     expect(mockPutObject).not.toHaveBeenCalled()
+    expect(mockLoggerWarn).toHaveBeenCalledWith(
+      expect.objectContaining({ attachmentId: BASE_DATA.attachmentId }),
+      expect.stringContaining("exceeds size cap"),
+    )
 
     vi.unstubAllGlobals()
   })
 
-  it("rejects WhatsApp response with content-length exceeding MAX_ATTACHMENT_BYTES", async () => {
+  it("skips (no throw) a WhatsApp response whose content-length exceeds the cap", async () => {
     wireSelectChain({
       id: "att-001",
       originPath: "wa-media:media-id-xyz",
       mimeType: "video/mp4",
     })
     wireIntegrationLookup({ ...FAKE_INTEGRATION_ROW, channel: "whatsapp" })
 
-    const OVER_LIMIT = String(51 * 1024 * 1024)
+    const OVER_LIMIT = String(MAX_ATTACHMENT_BYTES + 1)
     const fetchMock = vi.fn(() =>
       Promise.resolve(makeFetchResponse({ contentLength: OVER_LIMIT })),
     )
     vi.stubGlobal("fetch", fetchMock)
 
-    await expect(coexistAttachmentDownload(WA_DATA)).rejects.toThrow()
+    await expect(coexistAttachmentDownload(WA_DATA)).resolves.toBeUndefined()
+    expect(mockPutObject).not.toHaveBeenCalled()
+
+    vi.unstubAllGlobals()
+  })
+
+  it("still throws (retryable) on a transient download failure — non-size errors keep retrying", async () => {
+    wireSelectChain({
+      id: "att-001",
+      originPath: "https://example.com/media/photo.jpg",
+      mimeType: "image/jpeg",
+    })
+    wireIntegrationLookup()
+
+    // A 5xx is transient: the handler must rethrow so BullMQ retries it.
+    const fetchMock = vi.fn(() =>
+      Promise.resolve(makeFetchResponse({ ok: false })),
+    )
+    vi.stubGlobal("fetch", fetchMock)
+
+    await expect(coexistAttachmentDownload(BASE_DATA)).rejects.toThrow()
     expect(mockPutObject).not.toHaveBeenCalled()
+    expect(mockLoggerError).toHaveBeenCalled()
 
     vi.unstubAllGlobals()
   })

apps/worker/src/integration/handlers/coexist/attachment-download.ts

@@ -62,8 +62,23 @@ const getStorageExtension = (
   return MIME_EXTENSION_MAP[mimeType] ?? ""
 }
 
-/** Maximum allowed attachment size in bytes (50 MiB). */
-const MAX_ATTACHMENT_BYTES = 50 * 1024 * 1024
+/** Maximum allowed attachment size in bytes (100 MiB). */
+export const MAX_ATTACHMENT_BYTES = 100 * 1024 * 1024
+
+/**
+ * Thrown when an attachment is genuinely larger than `MAX_ATTACHMENT_BYTES`.
+ * This is a PERMANENT condition — the same bytes exceed the cap on every
+ * retry — so the download handler treats it as a terminal skip (logs and
+ * returns) instead of rethrowing, which would burn all 5 BullMQ attempts and
+ * spam identical error logs for a job that can never succeed. Transient
+ * failures (network, 5xx, timeout) keep throwing so they still retry.
+ */
+export class AttachmentTooLargeError extends Error {
+  constructor(message: string) {
+    super(message)
+    this.name = "AttachmentTooLargeError"
+  }
+}
 
 const isPendingOriginPath = (path: string): boolean =>
   path.startsWith("http://") ||
@@ -91,7 +106,7 @@ const readBodyWithCap = async (
   if (declaredHeader !== null) {
     const declared = Number.parseInt(declaredHeader, 10)
     if (!Number.isNaN(declared) && declared > MAX_ATTACHMENT_BYTES) {
-      throw new SdkException(
+      throw new AttachmentTooLargeError(
         `[coexist-attachment] ${label} exceeds size limit: ${declared} bytes (max ${MAX_ATTACHMENT_BYTES})`,
       )
     }
@@ -116,7 +131,7 @@ const readBodyWithCap = async (
     total += value.byteLength
     if (total > MAX_ATTACHMENT_BYTES) {
       await reader.cancel()
-      throw new SdkException(
+      throw new AttachmentTooLargeError(
         `[coexist-attachment] ${label} body exceeds size limit: >${MAX_ATTACHMENT_BYTES} bytes`,
       )
     }
@@ -293,6 +308,18 @@ export const coexistAttachmentDownload = async (
       row.mimeType,
     )
   } catch (err) {
+    // An oversized attachment is a permanent condition: retrying re-downloads
+    // the same too-large bytes and fails identically, burning every BullMQ
+    // attempt. Terminate the job cleanly (no rethrow → no retry) and leave the
+    // row pending; a future coexist sync may re-enqueue it, but it never enters
+    // a retry storm. All other failures stay retryable.
+    if (err instanceof AttachmentTooLargeError) {
+      logger.warn(
+        { attachmentId, channel, err },
+        "[coexist-attachment] attachment exceeds size cap — skipping (permanent)",
+      )
+      return
+    }
     logger.error(
       { err, attachmentId, channel },
       "[coexist-attachment] download failed",