fix: change sqs

Author: hemicharly

app/src/infrastructure/queue/sqs/impl/sqs.consumer.queue.provider.impl.ts

@@ -36,13 +36,16 @@ export class SqsConsumerQueueProviderImpl implements OnModuleInit, OnModuleDestr
   }
 
   async onModuleInit() {
-    this.logger.log('SqsConsumerQueueProviderImpl initialized.');
+    this.logger.log('SQS Consumer initialized.');
   }
 
   async onModuleDestroy() {
     this.isShuttingDown = true;
-    this.logger.log('Shutting down SqsConsumerQueueProviderImpl...');
+    this.logger.log('Shutting down sqs consumer...');
     this.stopAllPolling();
+
+    await Promise.all([...this.pollingIntervals.values()].map(interval => clearTimeout(interval)));
+    this.logger.log('All polling operations stopped.');
   }
 
   registerQueueHandler(queueConfig: QueueConfig, handler: SqsHandler) {
@@ -87,14 +90,14 @@ export class SqsConsumerQueueProviderImpl implements OnModuleInit, OnModuleDestr
               await handler(message);
               await this.deleteMessage(url, message.ReceiptHandle!);
             } catch (error) {
-              this.logger.error(`[${name}] Error processing message: ${error.message}`);
+              this.logger.error(`[${name}] Error processing message: ${error.message}`, error.stack);
               await this.extendVisibilityTimeout(url, message.ReceiptHandle!, queueConfig.visibilityTimeout || 30);
             }
           }
           backoffTime = 1000;
         }
       } catch (error) {
-        this.logger.error(`[${name}] Error receiving messages: ${error.message}`);
+        this.logger.error(`[${name}] Unhandled error: ${error.message}`, error.stack);
       } finally {
         if (!this.isShuttingDown) {
           backoffTime = Math.min(backoffTime * 2, maxBackoffTime);
@@ -110,6 +113,11 @@ export class SqsConsumerQueueProviderImpl implements OnModuleInit, OnModuleDestr
   }
 
   private async deleteMessage(queueUrl: string, receiptHandle: string) {
+    if (!receiptHandle) {
+      this.logger.warn(`Invalid receipt handle for queue: ${queueUrl}`);
+      return;
+    }
+
     try {
       const command = new DeleteMessageCommand({
         QueueUrl: queueUrl,
@@ -123,6 +131,11 @@ export class SqsConsumerQueueProviderImpl implements OnModuleInit, OnModuleDestr
   }
 
   private async extendVisibilityTimeout(queueUrl: string, receiptHandle: string, visibilityTimeout: number) {
+    if (!receiptHandle) {
+      this.logger.warn(`Invalid receipt handle for queue: ${queueUrl}`);
+      return;
+    }
+
     try {
       const command = new ChangeMessageVisibilityCommand({
         QueueUrl: queueUrl,
@@ -132,7 +145,7 @@ export class SqsConsumerQueueProviderImpl implements OnModuleInit, OnModuleDestr
       await this.sqsClient.send(command);
       this.logger.debug(`Extended visibility timeout for message in queue: ${queueUrl}`);
     } catch (error) {
-      this.logger.error(`Error extending visibility timeout for queue: ${queueUrl}, Error: ${error.message}`);
+      this.logger.error(`Error extending visibility timeout for queue: ${queueUrl}, Error: ${error.message}`, error.stack);
     }
   }
 

app/src/modules/consumers/notification/notification.order.consumer.module.ts

@@ -1,4 +1,4 @@
-import { Inject, Module, OnModuleInit } from '@nestjs/common';
+import { Inject, Logger, Module, OnModuleInit } from '@nestjs/common';
 import { SqsConsumerQueueProviderImpl } from '@infrastructure/queue/sqs/impl';
 import { configEnv } from '@src/shared/config';
 import { Message } from '@aws-sdk/client-sqs';
@@ -13,6 +13,9 @@ import { WebhookIntegrationClientModule } from 'src/infrastructure/integrations/
   providers: [SqsConsumerQueueProviderImpl, UsecaseProviderConfig(NotificationSendWebhookUsecaseImpl, [WebhookIntegrationClientProviderImpl])],
 })
 export class NotificationOrderConsumerModule implements OnModuleInit {
+
+  private readonly logger = new Logger(SqsConsumerQueueProviderImpl.name);
+
   @Inject(NotificationSendWebhookUsecaseImpl.name)
   private readonly notificationSendWebhookUsecase: NotificationSendWebhookUsecase;
 
@@ -28,6 +31,10 @@ export class NotificationOrderConsumerModule implements OnModuleInit {
         visibilityTimeout: 60,
       },
       async (message: Message) => {
+        if (!message || !message.Body) {
+          this.logger.warn(`Empty message body received in queue: ${configEnv.aws.sqs.queues.queueNotificationOrder}`);
+          return;
+        }
         await this.notificationSendWebhookUsecase.execute(message.Body);
       },
     );