feat: timeout xmtp messages

Author: marian2js

apps/api/src/chat/services/broadcast.consumer.ts

@@ -2,7 +2,6 @@ import { JobNonRetriableError } from '@app/common/errors/job-non-retriable-error
 import { wait } from '@app/common/utils/async.utils'
 import { XmtpLib } from '@app/definitions/integration-definitions/xmtp/xmtp.lib'
 import { getWalletName } from '@app/definitions/utils/address.utils'
-import { sendXmtpMessage } from '@chainjet/tools/dist/messages'
 import { InjectQueue, Process, Processor } from '@nestjs/bull'
 import { Logger } from '@nestjs/common'
 import { Interval } from '@nestjs/schedule'
@@ -108,7 +107,12 @@ export class BroadcastConsumer {
         continue
       }
       try {
-        const message = await sendXmtpMessage(client, sendTo, campaign.message + '\n\n' + unsubscribeMessage)
+        const message = await XmtpLib.sendDirectMessageWithTimeout(
+          client,
+          sendTo,
+          campaign.message + '\n\n' + unsubscribeMessage,
+          10 * 1000,
+        )
         await this.campaignMessageService.createOne({
           campaign: campaign._id,
           address: contact.address,

libs/definitions/src/integration-definitions/xmtp/xmtp.lib.ts

@@ -1,4 +1,5 @@
-import { Client, Conversation } from '@xmtp/xmtp-js'
+import { sendXmtpMessage } from '@chainjet/tools/dist/messages'
+import { Client, Conversation, DecodedMessage } from '@xmtp/xmtp-js'
 import { isAddress } from 'ethers/lib/utils'
 
 const clientsCache = new Map<string, Client>()
@@ -48,4 +49,29 @@ export const XmtpLib = {
     }
     return xmtpMessage
   },
+
+  sendDirectMessageWithTimeout: async (
+    client: Client,
+    sendTo: string,
+    message: string,
+    timeoutMs: number,
+  ): Promise<DecodedMessage> => {
+    let timeoutHandle: NodeJS.Timeout | undefined
+
+    const timeoutPromise = new Promise<DecodedMessage>((_, reject) => {
+      timeoutHandle = setTimeout(() => {
+        reject(new Error('Message sending timed out'))
+      }, timeoutMs)
+    })
+
+    const sendMessagePromise: Promise<DecodedMessage> = sendXmtpMessage(client, sendTo, message)
+
+    try {
+      return await Promise.race([sendMessagePromise, timeoutPromise])
+    } catch (error) {
+      throw error
+    } finally {
+      clearTimeout(timeoutHandle)
+    }
+  },
 }