Merge pull request #207 from vernu/dev

vernu · web-flow · commit c5630b30e95a · 2026-03-24T17:55:44.000+03:00
enhance sms queue processing
diff --git a/api/src/gateway/gateway.module.ts b/api/src/gateway/gateway.module.ts
@@ -10,7 +10,7 @@ import { SMSBatch, SMSBatchSchema } from './schemas/sms-batch.schema'
 import { WebhookModule } from 'src/webhook/webhook.module'
 import { BillingModule } from 'src/billing/billing.module'
 import { BullModule } from '@nestjs/bull'
-import { ConfigModule } from '@nestjs/config'
+import { ConfigModule, ConfigService } from '@nestjs/config'
 import { SmsQueueService } from './queue/sms-queue.service'
 import { SmsQueueProcessor } from './queue/sms-queue.processor'
 import { SmsStatusUpdateTask } from './tasks/sms-status-update.task'
@@ -32,17 +32,25 @@ import { HeartbeatCheckTask } from './tasks/heartbeat-check.task'
         schema: SMSBatchSchema,
       },
     ]),
-    BullModule.registerQueue({
+    BullModule.registerQueueAsync({
       name: 'sms',
-      defaultJobOptions: {
-        attempts: 2,
-        backoff: {
-          type: 'exponential',
-          delay: 1000,
+      imports: [ConfigModule],
+      inject: [ConfigService],
+      useFactory: async (configService: ConfigService) => ({
+        limiter: {
+          max: configService.get<number>('SMS_QUEUE_LIMITER_MAX', 20),
+          duration: configService.get<number>('SMS_QUEUE_LIMITER_DURATION_MS', 1000),
         },
-        removeOnComplete: { age: 24 * 3600 }, // 24 hours
-        removeOnFail: { age: 72 * 3600 }, // 72 hours
-      },
+        defaultJobOptions: {
+          attempts: 2,
+          backoff: {
+            type: 'exponential',
+            delay: 1000,
+          },
+          removeOnComplete: { age: 24 * 3600 }, // 24 hours
+          removeOnFail: { age: 72 * 3600 }, // 72 hours
+        },
+      }),
     }),
     AuthModule,
     UsersModule,
diff --git a/api/src/gateway/gateway.service.ts b/api/src/gateway/gateway.service.ts
@@ -478,6 +478,13 @@ export class GatewayService {
 
     // Track FCM messages with their calculated delays for grouping
     const fcmMessagesWithDelays: Array<{ message: Message; delayMs?: number }> = []
+    const smsDocumentsToInsert: Array<Record<string, any>> = []
+    const smsToFcmMetadata: Array<{
+      recipient: string
+      message: string
+      simSubscriptionId?: number
+      delayMs?: number
+    }> = []
 
     for (const smsData of messages) {
       const message = smsData.message
@@ -495,8 +502,8 @@ export class GatewayService {
       const delayMs = this.calculateDelayFromScheduledAt(smsData.scheduledAt)
 
       for (let recipient of recipients) {
-        recipient =  recipient.replace(/\s+/g, "")
-        const sms = await this.smsModel.create({
+        recipient = recipient.replace(/\s+/g, "")
+        smsDocumentsToInsert.push({
           device: device._id,
           smsBatch: smsBatch._id,
           message: message,
@@ -508,32 +515,73 @@ export class GatewayService {
             simSubscriptionId: smsData.simSubscriptionId,
           }),
         })
-        const updatedSMSData = {
-          smsId: sms._id,
-          smsBatchId: smsBatch._id,
+        smsToFcmMetadata.push({
+          recipient,
           message,
-          recipients: [recipient],
           ...(smsData.simSubscriptionId !== undefined && {
             simSubscriptionId: smsData.simSubscriptionId,
           }),
+          delayMs,
+        })
+      }
+    }
 
-          // Legacy fields to be removed in the future
-          smsBody: message,
-          receivers: [recipient],
-        }
-        const stringifiedSMSData = JSON.stringify(updatedSMSData)
+    const insertChunkSize = 500
+    const insertedSmsDocs: any[] = []
+    const hasInsertMany = typeof (this.smsModel as any).insertMany === 'function'
+    for (let i = 0; i < smsDocumentsToInsert.length; i += insertChunkSize) {
+      const chunk = smsDocumentsToInsert.slice(i, i + insertChunkSize)
+      if (hasInsertMany) {
+        const insertedChunk = await (this.smsModel as any).insertMany(chunk, { ordered: true })
+        insertedSmsDocs.push(...insertedChunk)
+        continue
+      }
 
-        const fcmMessage: Message = {
-          data: {
-            smsData: stringifiedSMSData,
-          },
-          token: device.fcmToken,
-          android: {
-            priority: 'high',
-          },
-        }
-        fcmMessagesWithDelays.push({ message: fcmMessage, delayMs })
+      // Fallback for mocked/non-standard models that don't expose insertMany
+      for (const smsDocument of chunk) {
+        const createdSmsDoc = await this.smsModel.create(smsDocument)
+        insertedSmsDocs.push(createdSmsDoc)
+      }
+    }
+
+    if (insertedSmsDocs.length !== smsToFcmMetadata.length) {
+      throw new HttpException(
+        {
+          success: false,
+          error: 'Failed to map created SMS records to queue payload',
+        },
+        HttpStatus.INTERNAL_SERVER_ERROR,
+      )
+    }
+
+    for (let i = 0; i < insertedSmsDocs.length; i++) {
+      const sms = insertedSmsDocs[i]
+      const metadata = smsToFcmMetadata[i]
+      const updatedSMSData = {
+        smsId: sms._id,
+        smsBatchId: smsBatch._id,
+        message: metadata.message,
+        recipients: [metadata.recipient],
+        ...(metadata.simSubscriptionId !== undefined && {
+          simSubscriptionId: metadata.simSubscriptionId,
+        }),
+
+        // Legacy fields to be removed in the future
+        smsBody: metadata.message,
+        receivers: [metadata.recipient],
+      }
+      const stringifiedSMSData = JSON.stringify(updatedSMSData)
+
+      const fcmMessage: Message = {
+        data: {
+          smsData: stringifiedSMSData,
+        },
+        token: device.fcmToken,
+        android: {
+          priority: 'high',
+        },
       }
+      fcmMessagesWithDelays.push({ message: fcmMessage, delayMs: metadata.delayMs })
     }
 
     // Check if we should use the queue
@@ -673,7 +721,7 @@ export class GatewayService {
       !dto.sender ||
       !dto.message
     ) {
-      console.log('Invalid received SMS data')
+      console.error(`receiveSMS: Invalid received SMS data (sender: ${dto.sender}, message: ${dto.message}) (receivedAt: ${dto.receivedAt}, receivedAtInMillis: ${dto.receivedAtInMillis})`)
       throw new HttpException(
         {
           success: false,
diff --git a/api/src/gateway/queue/sms-queue.processor.ts b/api/src/gateway/queue/sms-queue.processor.ts
@@ -89,37 +89,23 @@ export class SmsQueueProcessor {
       // )
 
       // Mark individual SMS records as failed when their FCM push failed
+      const failedSmsIds: string[] = []
+      const failedUpdates: Array<{
+        smsId: string
+        errorCode: string
+        errorMessage: string
+      }> = []
       for (let i = 0; i < response.responses.length; i++) {
         if (!response.responses[i].success) {
           try {
             const smsData = JSON.parse(fcmMessages[i].data.smsData)
             const fcmError = response.responses[i].error
-            const updatedSms = await this.smsModel.findByIdAndUpdate(
-              smsData.smsId,
-              {
-                $set: {
-                  status: 'failed',
-                  failedAt: new Date(),
-                  errorCode: getFcmErrorCode(fcmError ?? undefined),
-                  errorMessage: getFcmErrorMessage(fcmError ?? undefined),
-                },
-              },
-              { new: true },
-            )
-            if (device?.user && updatedSms) {
-              await this.webhookService
-                .deliverNotification({
-                  sms: updatedSms,
-                  user: device.user as any,
-                  event: WebhookEvent.MESSAGE_FAILED,
-                })
-                .catch((e) =>
-                  this.logger.warn(
-                    `Webhook delivery failed for SMS ${updatedSms._id}`,
-                    e?.message,
-                  ),
-                )
-            }
+            failedSmsIds.push(String(smsData.smsId))
+            failedUpdates.push({
+              smsId: String(smsData.smsId),
+              errorCode: getFcmErrorCode(fcmError ?? undefined),
+              errorMessage: getFcmErrorMessage(fcmError ?? undefined),
+            })
           } catch (parseError) {
             this.logger.error(
               `Failed to mark SMS as failed for FCM message index ${i}`,
@@ -131,13 +117,12 @@ export class SmsQueueProcessor {
 
       // Mark individual SMS records as dispatched when FCM push succeeded
       const now = new Date()
+      const dispatchedSmsIds: string[] = []
       for (let i = 0; i < response.responses.length; i++) {
         if (response.responses[i].success) {
           try {
             const smsData = JSON.parse(fcmMessages[i].data.smsData)
-            await this.smsModel.findByIdAndUpdate(smsData.smsId, {
-              $set: { status: 'dispatched', dispatchedAt: now },
-            })
+            dispatchedSmsIds.push(String(smsData.smsId))
           } catch (parseError) {
             this.logger.error(
               `Failed to mark SMS as dispatched for FCM message index ${i}`,
@@ -147,6 +132,52 @@ export class SmsQueueProcessor {
         }
       }
 
+      if (failedUpdates.length > 0) {
+        const failedAt = new Date()
+        for (const failedUpdate of failedUpdates) {
+          await this.smsModel.updateOne(
+            { _id: failedUpdate.smsId as any },
+            {
+              $set: {
+                status: 'failed',
+                failedAt,
+                errorCode: failedUpdate.errorCode,
+                errorMessage: failedUpdate.errorMessage,
+              },
+            },
+          )
+        }
+      }
+
+      if (dispatchedSmsIds.length > 0) {
+        await this.smsModel.updateMany(
+          { _id: { $in: dispatchedSmsIds } as any },
+          {
+            $set: { status: 'dispatched', dispatchedAt: now },
+          },
+        )
+      }
+
+      if (device?.user && failedSmsIds.length > 0) {
+        const failedSmsDocuments = await this.smsModel.find({
+          _id: { $in: failedSmsIds },
+        })
+        for (const failedSms of failedSmsDocuments) {
+          await this.webhookService
+            .deliverNotification({
+              sms: failedSms,
+              user: device.user as any,
+              event: WebhookEvent.MESSAGE_FAILED,
+            })
+            .catch((e) =>
+              this.logger.warn(
+                `Webhook delivery failed for SMS ${failedSms._id}`,
+                e?.message,
+              ),
+            )
+        }
+      }
+
       // Update device SMS count
       await this.deviceModel
         .findByIdAndUpdate(deviceId, {
@@ -181,38 +212,11 @@ export class SmsQueueProcessor {
       this.logger.error(`Failed to process SMS job ${job.id}`, error)
 
       // Mark all individual SMS in this batch of FCM messages as failed
+      const failedSmsIds: string[] = []
       for (const fcmMessage of fcmMessages) {
         try {
           const smsData = JSON.parse(fcmMessage.data.smsData)
-          const updatedSms = await this.smsModel.findByIdAndUpdate(
-            smsData.smsId,
-            {
-              $set: {
-                status: 'failed',
-                failedAt: new Date(),
-                errorCode:
-                  (error as any)?.code != null
-                    ? getFcmErrorCode(error as any)
-                    : 'FCM_SEND_ERROR',
-                errorMessage: getFcmErrorMessage(error as any),
-              },
-            },
-            { new: true },
-          )
-          if (device?.user && updatedSms) {
-            await this.webhookService
-              .deliverNotification({
-                sms: updatedSms,
-                user: device.user as any,
-                event: WebhookEvent.MESSAGE_FAILED,
-              })
-              .catch((e) =>
-                this.logger.warn(
-                  `Webhook delivery failed for SMS ${updatedSms._id}`,
-                  e?.message,
-                ),
-              )
-          }
+          failedSmsIds.push(String(smsData.smsId))
         } catch (parseError) {
           this.logger.error(
             'Failed to mark SMS as failed after FCM error',
@@ -221,6 +225,44 @@ export class SmsQueueProcessor {
         }
       }
 
+      if (failedSmsIds.length > 0) {
+        const failedAt = new Date()
+        await this.smsModel.updateMany(
+          { _id: { $in: failedSmsIds } as any },
+          {
+            $set: {
+              status: 'failed',
+              failedAt,
+              errorCode:
+                (error as any)?.code != null
+                  ? getFcmErrorCode(error as any)
+                  : 'FCM_SEND_ERROR',
+              errorMessage: getFcmErrorMessage(error as any),
+            },
+          },
+        )
+      }
+
+      if (device?.user && failedSmsIds.length > 0) {
+        const failedSmsDocuments = await this.smsModel.find({
+          _id: { $in: failedSmsIds },
+        })
+        for (const failedSms of failedSmsDocuments) {
+          await this.webhookService
+            .deliverNotification({
+              sms: failedSms,
+              user: device.user as any,
+              event: WebhookEvent.MESSAGE_FAILED,
+            })
+            .catch((e) =>
+              this.logger.warn(
+                `Webhook delivery failed for SMS ${failedSms._id}`,
+                e?.message,
+              ),
+            )
+        }
+      }
+
       const smsBatch = await this.smsBatchModel.findByIdAndUpdate(
         smsBatchId,
         {
diff --git a/api/src/gateway/queue/sms-queue.service.ts b/api/src/gateway/queue/sms-queue.service.ts
