diff --git a/src/configs/typeorm.config.ts b/src/configs/typeorm.config.ts index 8d8378f..9391ab2 100755 --- a/src/configs/typeorm.config.ts +++ b/src/configs/typeorm.config.ts @@ -13,7 +13,7 @@ export function databaseConfigs(): TypeOrmModuleAsyncOptions { username: configService.getOrThrow("DB_USER"), password: configService.getOrThrow("DB_PASS"), autoLoadEntities: true, - synchronize: configService.getOrThrow("NODE_ENV") == "production" ? false : true, + synchronize: configService.getOrThrow("NODE_ENV") == "production" ? false : false, logging: configService.getOrThrow("NODE_ENV") == "production" ? false : true, migrationsTableName: "typeorm_migrations", migrationsRun: false, diff --git a/src/modules/notifications/queue/notification.processor.ts b/src/modules/notifications/queue/notification.processor.ts index a5d2e7b..0992ab4 100644 --- a/src/modules/notifications/queue/notification.processor.ts +++ b/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,103 +67,129 @@ 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; - switch (type) { - // Admin Notifications - case NotifType.NEW_BLOG_COMMENT: - await this.notificationsService.createNewBlogCommentNotification(recipientId, data as IBlogCommentNotificationData, queryRunner); - break; - case NotifType.NEW_SERVICE_REVIEW: - await this.notificationsService.createNewServiceReviewNotification( - recipientId, - data as IServiceReviewNotificationData, - queryRunner, - ); - break; - case NotifType.NEW_CUSTOMER: - await this.notificationsService.createNewCustomerNotification(recipientId, data as INewCustomerNotificationData, queryRunner); - break; - case NotifType.NEW_SUBSCRIPTION: - await this.notificationsService.createNewSubscriptionNotification( - recipientId, - data as INewSubscriptionNotificationData, - queryRunner, - ); - break; - case NotifType.NEW_CRITICISM: - await this.notificationsService.createNewCriticismNotification(recipientId, data as INewCriticismNotificationData, queryRunner); - break; - case NotifType.NEW_TICKET: - await this.notificationsService.createNewTicketGlobalNotification(recipientId, data as INewTicketNotificationData, queryRunner); - break; + try { + switch (type) { + // Admin Notifications + case NotifType.NEW_BLOG_COMMENT: + await this.notificationsService.createNewBlogCommentNotification( + recipientId, + data as IBlogCommentNotificationData, + queryRunner, + ); + break; + case NotifType.NEW_SERVICE_REVIEW: + await this.notificationsService.createNewServiceReviewNotification( + recipientId, + data as IServiceReviewNotificationData, + queryRunner, + ); + break; + case NotifType.NEW_CUSTOMER: + await this.notificationsService.createNewCustomerNotification(recipientId, data as INewCustomerNotificationData, queryRunner); + break; + case NotifType.NEW_SUBSCRIPTION: + await this.notificationsService.createNewSubscriptionNotification( + recipientId, + data as INewSubscriptionNotificationData, + queryRunner, + ); + break; + case NotifType.NEW_CRITICISM: + await this.notificationsService.createNewCriticismNotification(recipientId, data as INewCriticismNotificationData, queryRunner); + break; + case NotifType.NEW_TICKET: + await this.notificationsService.createNewTicketGlobalNotification(recipientId, data as INewTicketNotificationData, queryRunner); + break; - // User Notifications - case NotifType.USER_LOGIN: - await this.notificationsService.createLoginNotification(recipientId, data); - break; - case NotifType.ANNOUNCEMENT: - await this.notificationsService.createAnnouncementNotification(recipientId, data as IAnnouncementNotificationData, queryRunner); - break; + // User Notifications + case NotifType.USER_LOGIN: + await this.notificationsService.createLoginNotification(recipientId, data); + break; + case NotifType.ANNOUNCEMENT: + await this.notificationsService.createAnnouncementNotification(recipientId, data as IAnnouncementNotificationData, queryRunner); + break; - // Wallet Notifications - case NotifType.WALLET_CHARGE: - await this.notificationsService.createWalletChargeNotification(recipientId, data as IWalletNotificationData, queryRunner); - break; - case NotifType.WALLET_DEDUCTION: - await this.notificationsService.createWalletDeductionNotification(recipientId, data as IWalletNotificationData, queryRunner); - break; + // Wallet Notifications + case NotifType.WALLET_CHARGE: + await this.notificationsService.createWalletChargeNotification(recipientId, data as IWalletNotificationData, queryRunner); + break; + case NotifType.WALLET_DEDUCTION: + await this.notificationsService.createWalletDeductionNotification(recipientId, data as IWalletNotificationData, queryRunner); + break; - // Ticket Notifications - case NotifType.ANSWER_TICKET: - await this.notificationsService.createAnswerTicketNotification(recipientId, data as ITicketNotificationData, queryRunner); - break; - case NotifType.CREATE_TICKET: - await this.notificationsService.createTicketNotification(recipientId, data as ITicketNotificationData, queryRunner); - break; - case NotifType.ASSIGN_TICKET: - await this.notificationsService.createAssignTicketNotificationForAdmin(recipientId, data as ITicketNotificationData, queryRunner); - break; + // Ticket Notifications + case NotifType.ANSWER_TICKET: + await this.notificationsService.createAnswerTicketNotification(recipientId, data as ITicketNotificationData, queryRunner); + break; + case NotifType.CREATE_TICKET: + await this.notificationsService.createTicketNotification(recipientId, data as ITicketNotificationData, queryRunner); + break; + case NotifType.ASSIGN_TICKET: + await this.notificationsService.createAssignTicketNotificationForAdmin( + recipientId, + data as ITicketNotificationData, + queryRunner, + ); + break; - // Invoice Notifications - case NotifType.CREATE_INVOICE: - await this.notificationsService.createInvoiceCreationNotification(recipientId, data as IInvoiceNotificationData, queryRunner); - break; - case NotifType.BILL_INVOICE_REMINDER: - await this.notificationsService.createBillInvoiceReminderNotification(recipientId, data as IInvoiceNotificationData, queryRunner); - break; - case NotifType.BILL_INVOICE: - await this.notificationsService.createBillInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner); - break; - case NotifType.APPROVE_INVOICE: - await this.notificationsService.createApprovedInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner); - break; - case NotifType.INVOICE_OVERDUE: - await this.notificationsService.createInvoiceOverdueNotification(recipientId, data as IInvoiceNotificationData, queryRunner); - break; - case NotifType.RECURRING_INVOICE: - await this.notificationsService.createRecurringInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner); - break; + // Invoice Notifications + case NotifType.CREATE_INVOICE: + await this.notificationsService.createInvoiceCreationNotification(recipientId, data as IInvoiceNotificationData, queryRunner); + break; + case NotifType.BILL_INVOICE_REMINDER: + await this.notificationsService.createBillInvoiceReminderNotification( + recipientId, + data as IInvoiceNotificationData, + queryRunner, + ); + break; + case NotifType.BILL_INVOICE: + await this.notificationsService.createBillInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner); + break; + case NotifType.APPROVE_INVOICE: + await this.notificationsService.createApprovedInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner); + break; + case NotifType.INVOICE_OVERDUE: + await this.notificationsService.createInvoiceOverdueNotification(recipientId, data as IInvoiceNotificationData, queryRunner); + break; + case NotifType.RECURRING_INVOICE: + await this.notificationsService.createRecurringInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner); + break; - // Service Notifications - case NotifType.BLOCK_SERVICE: - await this.notificationsService.createBlockServiceNotification(recipientId, data as ISubscriptionNotificationData, queryRunner); - break; + // Service Notifications + case NotifType.BLOCK_SERVICE: + await this.notificationsService.createBlockServiceNotification(recipientId, data as ISubscriptionNotificationData, queryRunner); + break; - // Payment Notifications - case NotifType.PAYMENT_REMINDER: - await this.notificationsService.createPaymentReminderNotification(recipientId, data as IPaymentNotificationData, queryRunner); - break; - case NotifType.PAYMENT_CANCELLATION: - await this.notificationsService.createPaymentCancellationNotification(recipientId, data as IPaymentNotificationData, queryRunner); - break; + // Payment Notifications + case NotifType.PAYMENT_REMINDER: + await this.notificationsService.createPaymentReminderNotification(recipientId, data as IPaymentNotificationData, queryRunner); + break; + case NotifType.PAYMENT_CANCELLATION: + await this.notificationsService.createPaymentCancellationNotification( + recipientId, + data as IPaymentNotificationData, + queryRunner, + ); + break; - default: - this.logger.warn(`Unknown notification type: ${type}`); + default: + this.logger.warn(`Unknown notification type: ${type}`); + } + + await queryRunner.commitTransaction(); + clearInterval(heartbeat); + return true; + } catch (error) { + clearInterval(heartbeat); + throw error; } - - await queryRunner.commitTransaction(); - return true; } catch (error) { this.logger.error( `Failed to process notification: ${error instanceof Error ? error.message : "Unknown error"}`,