import { EntityManager } from "@mikro-orm/postgresql"; import { Processor } from "@nestjs/bullmq"; import { Logger } from "@nestjs/common"; import { Job } from "bullmq"; import { WorkerProcessor } from "../../../common/queues/worker.processor"; import { NOTIFICATION } from "../constants"; import { NotifType } from "../enums/notif-settings.enum"; import { IAnnouncementNotificationData, IInvoiceNotificationData, IOtpNotificationData, IPaymentNotificationData, ITicketNotificationData, } from "../interfaces/ISendNotificationData"; import { NotificationsService } from "../providers/notifications.service"; type NotificationJobData = { type: NotifType; recipientId: string; data: | ITicketNotificationData | IInvoiceNotificationData | IAnnouncementNotificationData | IPaymentNotificationData | IOtpNotificationData; }; @Processor(NOTIFICATION.QUEUE_NAME, { concurrency: 5 }) export class NotificationProcessor extends WorkerProcessor { protected readonly logger = new Logger(NotificationProcessor.name); constructor( private readonly notificationsService: NotificationsService, private readonly em: EntityManager, ) { super(); } async process(job: Job, token?: string) { this.logger.log(`Processing notification job: ${job.id} ${token ? `with token: ${token}` : ""}`); const entityManager = this.em.fork(); try { await entityManager.begin(); const { type, recipientId, data } = job.data; switch (type) { // User Notifications case NotifType.OTP: await this.notificationsService.createOtpNotification(data as IOtpNotificationData); break; case NotifType.USER_LOGIN: await this.notificationsService.createLoginNotification(recipientId, data); break; case NotifType.ANNOUNCEMENT: await this.notificationsService.createAnnouncementNotification(recipientId, data as IAnnouncementNotificationData, entityManager); break; // Ticket Notifications case NotifType.ANSWER_TICKET: await this.notificationsService.createAnswerTicketNotification(recipientId, data as ITicketNotificationData, entityManager); break; case NotifType.CREATE_TICKET: await this.notificationsService.createTicketNotification(recipientId, data as ITicketNotificationData, entityManager); break; case NotifType.ASSIGN_TICKET: await this.notificationsService.createAssignTicketNotificationForAdmin( recipientId, data as ITicketNotificationData, entityManager, ); break; // Invoice Notifications case NotifType.CREATE_INVOICE: await this.notificationsService.createInvoiceCreationNotification(recipientId, data as IInvoiceNotificationData, entityManager); break; case NotifType.BILL_INVOICE_REMINDER: await this.notificationsService.createBillInvoiceReminderNotification( recipientId, data as IInvoiceNotificationData, entityManager, ); break; case NotifType.BILL_INVOICE: await this.notificationsService.createBillInvoiceNotification(recipientId, data as IInvoiceNotificationData, entityManager); break; case NotifType.APPROVE_INVOICE: await this.notificationsService.createApprovedInvoiceNotification(recipientId, data as IInvoiceNotificationData, entityManager); break; case NotifType.INVOICE_OVERDUE: await this.notificationsService.createInvoiceOverdueNotification(recipientId, data as IInvoiceNotificationData, entityManager); break; case NotifType.RECURRING_INVOICE: await this.notificationsService.createRecurringInvoiceNotification(recipientId, data as IInvoiceNotificationData, entityManager); break; // Payment Notifications case NotifType.PAYMENT_REMINDER: await this.notificationsService.createPaymentReminderNotification(recipientId, data as IPaymentNotificationData, entityManager); break; case NotifType.PAYMENT_CANCELLATION: await this.notificationsService.createPaymentCancellationNotification( recipientId, data as IPaymentNotificationData, entityManager, ); break; default: this.logger.warn(`Unknown notification type: ${type}`); } await entityManager.commit(); return true; } catch (error) { this.logger.error( `Failed to process notification: ${error instanceof Error ? error.message : "Unknown error"}`, error instanceof Error ? error.stack : undefined, ); await entityManager.rollback(); throw error; } } }