chore: update the processor of the notification

src/configs/typeorm.config.ts

@@ -13,7 +13,7 @@ export function databaseConfigs(): TypeOrmModuleAsyncOptions {
         username: configService.getOrThrow<string>("DB_USER"),
         password: configService.getOrThrow<string>("DB_PASS"),
         autoLoadEntities: true,
-        synchronize: configService.getOrThrow<string>("NODE_ENV") == "production" ? false : true,
+        synchronize: configService.getOrThrow<string>("NODE_ENV") == "production" ? false : false,
         logging: configService.getOrThrow<string>("NODE_ENV") == "production" ? false : true,
         migrationsTableName: "typeorm_migrations",
         migrationsRun: false,

src/modules/notifications/queue/notification.processor.ts

@@ -41,7 +41,12 @@ type NotificationJobData = {
     | INewTicketNotificationData;
 };
 
-@Processor(NOTIFICATION.QUEUE_NAME, { concurrency: 5 })
+@Processor(NOTIFICATION.QUEUE_NAME, {
+  concurrency: 5,
+  lockDuration: 30000, // 30 seconds lock duration
+  stalledInterval: 30000, // Check for stalled jobs every 30 seconds
+  maxStalledCount: 2, // Allow 2 stalls before failing
+})
 export class NotificationProcessor extends WorkerProcessor {
   protected readonly logger = new Logger(NotificationProcessor.name);
 
@@ -62,12 +67,21 @@ export class NotificationProcessor extends WorkerProcessor {
       await queryRunner.connect();
       await queryRunner.startTransaction();
 
+      const heartbeat = setInterval(() => {
+        job.updateProgress(50);
+      }, 10000); // Update progress every 10 seconds
+
       const { type, recipientId, data } = job.data;
 
+      try {
         switch (type) {
           // Admin Notifications
           case NotifType.NEW_BLOG_COMMENT:
-          await this.notificationsService.createNewBlogCommentNotification(recipientId, data as IBlogCommentNotificationData, queryRunner);
+            await this.notificationsService.createNewBlogCommentNotification(
+              recipientId,
+              data as IBlogCommentNotificationData,
+              queryRunner,
+            );
             break;
           case NotifType.NEW_SERVICE_REVIEW:
             await this.notificationsService.createNewServiceReviewNotification(
@@ -117,7 +131,11 @@ export class NotificationProcessor extends WorkerProcessor {
             await this.notificationsService.createTicketNotification(recipientId, data as ITicketNotificationData, queryRunner);
             break;
           case NotifType.ASSIGN_TICKET:
-          await this.notificationsService.createAssignTicketNotificationForAdmin(recipientId, data as ITicketNotificationData, queryRunner);
+            await this.notificationsService.createAssignTicketNotificationForAdmin(
+              recipientId,
+              data as ITicketNotificationData,
+              queryRunner,
+            );
             break;
 
           // Invoice Notifications
@@ -125,7 +143,11 @@ export class NotificationProcessor extends WorkerProcessor {
             await this.notificationsService.createInvoiceCreationNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
             break;
           case NotifType.BILL_INVOICE_REMINDER:
-          await this.notificationsService.createBillInvoiceReminderNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
+            await this.notificationsService.createBillInvoiceReminderNotification(
+              recipientId,
+              data as IInvoiceNotificationData,
+              queryRunner,
+            );
             break;
           case NotifType.BILL_INVOICE:
             await this.notificationsService.createBillInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
@@ -150,7 +172,11 @@ export class NotificationProcessor extends WorkerProcessor {
             await this.notificationsService.createPaymentReminderNotification(recipientId, data as IPaymentNotificationData, queryRunner);
             break;
           case NotifType.PAYMENT_CANCELLATION:
-          await this.notificationsService.createPaymentCancellationNotification(recipientId, data as IPaymentNotificationData, queryRunner);
+            await this.notificationsService.createPaymentCancellationNotification(
+              recipientId,
+              data as IPaymentNotificationData,
+              queryRunner,
+            );
             break;
 
           default:
@@ -158,7 +184,12 @@ export class NotificationProcessor extends WorkerProcessor {
         }
 
         await queryRunner.commitTransaction();
+        clearInterval(heartbeat);
         return true;
+      } catch (error) {
+        clearInterval(heartbeat);
+        throw error;
+      }
     } catch (error) {
       this.logger.error(
         `Failed to process notification: ${error instanceof Error ? error.message : "Unknown error"}`,