dsc-api/src/modules/notifications/queue/notification.processor.ts

import { Processor } from "@nestjs/bullmq";
import { Logger } from "@nestjs/common";
import { Job } from "bullmq";
import { DataSource } from "typeorm";

import { WorkerProcessor } from "../../../common/queues/worker.processor";
import { LoggerService } from "../../logger/logger.service";
import { NotifType } from "../../settings/enums/notif-settings.enum";
import { SmsService } from "../../utils/providers/sms.service";
import { NOTIFICATION } from "../constants";
import { NotificationJobData, OtpJobData } from "../interfaces/INotification-job-data";
import {
  IAnnouncementNotificationData,
  IBlogCommentNotificationData,
  IInvoiceNotificationData,
  INewCriticismNotificationData,
  INewCustomerNotificationData,
  INewSubscriptionNotificationData,
  INewTicketNotificationData,
  IPaymentNotificationData,
  IServiceReviewNotificationData,
  ISubscriptionNotificationData,
  ITicketNotificationData,
  IWalletNotificationData,
} from "../interfaces/ISendNotificationData";
import { NotificationsService } from "../providers/notifications.service";

@Processor(NOTIFICATION.QUEUE_NAME, {
  concurrency: NOTIFICATION.SEND_NOTIFICATION_JOB_CONCURRENCY,
  lockDuration: NOTIFICATION.SEND_NOTIFICATION_JOB_LOCK_DURATION,
  stalledInterval: NOTIFICATION.SEND_NOTIFICATION_JOB_STALLED_INTERVAL,
  maxStalledCount: NOTIFICATION.SEND_NOTIFICATION_JOB_MAX_STALLED_COUNT,
})
export class NotificationProcessor extends WorkerProcessor {
  protected readonly logger = new Logger(NotificationProcessor.name);

  constructor(
    private readonly notificationsService: NotificationsService,
    private readonly dataSource: DataSource,
    protected readonly loggerService: LoggerService,
    private readonly smsService: SmsService,
  ) {
    super();
  }

  async process(job: Job<NotificationJobData | OtpJobData>, token?: string) {
    this.logger.log(`Processing notification job: ${job.id} ${token ? `with token: ${token}` : ""}`);
    const queryRunner = this.dataSource.createQueryRunner();
    let heartbeat: NodeJS.Timeout | undefined;

    //

    try {
      await queryRunner.connect();
      await queryRunner.startTransaction();
      heartbeat = setInterval(() => job.updateProgress(50), 10000);

      //
      if (job.name === NOTIFICATION.SEND_LOGIN_OTP_JOB_NAME) {
        const { phone, code } = job.data as OtpJobData;
        await this.smsService.sendSmsVerifyCode(phone, code);
        //
      } else if (job.name === NOTIFICATION.SEND_NOTIFICATION_JOB_NAME) {
        const { type, recipientId, data } = job.data as NotificationJobData;
        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;
          // 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;
          // 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;
          // 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}`);
        }
      } else {
        this.logger.warn(`Unknown job name: ${job.name}`);
      }
      await queryRunner.commitTransaction();
      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 queryRunner.rollbackTransaction();
      throw error;
    } finally {
      if (heartbeat) clearInterval(heartbeat);
      await queryRunner.release();
    }
  }
}