import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq'; import { Logger } from '@nestjs/common'; import { Job } from 'bullmq'; import { InjectRepository } from '@mikro-orm/nestjs'; import { EntityRepository, EntityManager } from '@mikro-orm/postgresql'; import { Notification } from '../entities/notification.entity'; import { NotificationQueueJob, NotificationQueueJobResult } from '../interfaces/notification-queue.interface'; import { NotificationQueueNameEnum } from '../constants/queue'; import { PushNotificationService } from '../services/push-notification.service'; @Processor(NotificationQueueNameEnum.PUSH) export class PushProcessor extends WorkerHost { private readonly logger = new Logger(PushProcessor.name); constructor( @InjectRepository(Notification) private readonly notificationRepository: EntityRepository, private readonly em: EntityManager, private readonly pushNotificationService: PushNotificationService, ) { super(); } async process(job: Job): Promise { const { restaurantId, userId, title, content, idempotencyKey, notificationId } = job.data; this.logger.log(`Processing push notification job: ${job.id} - Title: ${title}, Restaurant: ${restaurantId}`); try { // Find existing notification record (should already exist from NotificationService.sendNotification) let notification: Notification | null = null; if (notificationId) { notification = await this.notificationRepository.findOne( { id: notificationId }, { populate: ['user', 'restaurant'] }, ); if (!notification) { throw new Error(`Notification ${notificationId} not found`); } } else if (idempotencyKey) { // Fallback: find by idempotency key if notificationId not provided notification = await this.notificationRepository.findOne( { idempotencyKey }, { populate: ['user', 'restaurant'] }, ); if (notification && notification.sentAt) { this.logger.warn(`Duplicate notification skipped: ${idempotencyKey}`); return { success: true, notificationId: notification.id, }; } } if (!notification) { throw new Error(`Notification not found for job ${job.id}. notificationId or idempotencyKey required.`); } // Check if push notification service is configured if (!this.pushNotificationService.isConfigured()) { this.logger.warn(`Push notification service not configured. Skipping notification ${notification.id}`); // Mark as sent even though we didn't send it (service unavailable) notification.sentAt = new Date(); await this.em.persistAndFlush(notification); return { success: false, notificationId: notification.id, error: 'Push notification service not configured', }; } // For push notifications, we need a push token if (!userId || !notification.user) { this.logger.warn(`No user found for push notification ${notification.id}`); throw new Error('User is required for push notifications'); } // Get push token from job data const pushToken = job.data.pushToken; const pushTokens = job.data.pushTokens; if (!pushToken && (!pushTokens || pushTokens.length === 0)) { this.logger.warn(`Push token(s) not provided for notification ${notification.id}`); throw new Error('Push token or pushTokens array is required for push notifications'); } // Send push notification // Convert NotificationTitle enum to a readable string for the notification title const notificationTitle = String(title) .replace(/\./g, ' ') .replace(/\b\w/g, l => l.toUpperCase()); let pushResult; if (pushTokens && pushTokens.length > 0) { // Send to multiple tokens pushResult = await this.pushNotificationService.sendToMultipleTokens({ title: notificationTitle, body: content, tokens: pushTokens, data: { notificationId: notification.id, restaurantId, userId, title: String(title), }, }); } else if (pushToken) { // Send to single token pushResult = await this.pushNotificationService.sendToToken({ title: notificationTitle, body: content, token: pushToken, data: { notificationId: notification.id, restaurantId, userId, title: String(title), }, }); } else { throw new Error('Push token or pushTokens array is required'); } if (!pushResult.success) { throw new Error(pushResult.error || 'Failed to send push notification'); } // Mark notification as sent notification.sentAt = new Date(); await this.em.persistAndFlush(notification); this.logger.log(`Push notification ${notification.id} sent successfully`); return { success: true, notificationId: notification.id, providerResponse: { messageId: pushResult.messageId }, }; } catch (error) { this.logger.error(`Error processing push notification job ${job.id}:`, error); throw error; } } @OnWorkerEvent('completed') onCompleted(job: Job) { this.logger.log(`Notification job ${job.id} completed`); } @OnWorkerEvent('failed') onFailed(job: Job, error: Error) { this.logger.error(`Notification job ${job.id} failed:`, error); } }