diff --git a/src/modules/notifications/constants/queue.ts b/src/modules/notifications/constants/queue.ts index 777a59e..644e72f 100644 --- a/src/modules/notifications/constants/queue.ts +++ b/src/modules/notifications/constants/queue.ts @@ -1,4 +1,5 @@ export enum NotificationQueueNameEnum { SMS = 'sms', PUSH = 'push', + IN_APP = 'in-app', } diff --git a/src/modules/notifications/interfaces/jobs-queue.interface.ts b/src/modules/notifications/interfaces/jobs-queue.interface.ts index 14ac48c..0cdd80f 100644 --- a/src/modules/notifications/interfaces/jobs-queue.interface.ts +++ b/src/modules/notifications/interfaces/jobs-queue.interface.ts @@ -1,4 +1,4 @@ -import type { recipientType } from './notification.interface'; +import type { NotifTitleEnum, recipientType } from './notification.interface'; export interface SmsNotificationQueueJob { recipient: recipientType; @@ -16,6 +16,12 @@ export interface PushNotificationQueueJob { pushToken?: string; // FCM token for push notifications pushTokens?: string[]; // Multiple FCM tokens for bulk push notifications } +export interface InAppNotificationQueueJob { + recipient: { adminId: string; restaurantId: string }; + subject: NotifTitleEnum; + body: string; + notificationId: string; +} export interface SmsNotificationQueueJobResult { success: boolean; diff --git a/src/modules/notifications/notifications.gateway.ts b/src/modules/notifications/notifications.gateway.ts index 4107cba..2bbd80d 100644 --- a/src/modules/notifications/notifications.gateway.ts +++ b/src/modules/notifications/notifications.gateway.ts @@ -71,9 +71,13 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco } sendInAppNotification(repipient: { adminId: string; restaurantId: string }, payload: IInAppNotificationPayload) { - this.logger.log(`Sending in app notification to admin: ${repipient.adminId}`); const room = this.getRoom(repipient.adminId, repipient.restaurantId); - this.server.to(room).emit('notifications', payload); + + this.logger.log(`Sending in app notification to admin: ${repipient.adminId}`); + this.server.to(room).emit('notifications', payload, async () => { + // 3. ACK (delivered) + // await this.notificationService.markAsDelivered(payload.notificationId); + }); this.logger.log(`In app notification sent to admin: ${repipient.adminId}`); } @@ -87,12 +91,7 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco this.logger.log(`Admin ${client.adminId} (Client ${client.id}) left room: ${room}`); void client.emit('left', { room, message: 'Successfully Admin left room' }); } - /** - * Join a room for a specific restaurant (admin/staff) - * Requires admin authentication via JWT token - * Restaurant ID is extracted from the authenticated admin's token - * Room format: `restaurant:${restId}` - */ + handleJoinRoom(client: AuthenticatedSocket) { const room = this.getRoom(client.adminId!, client.restId!); void client.join(room); @@ -100,192 +99,6 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco void client.emit('joined', { room, message: 'Successfully Admin joined room' }); } - /** - * Join a room for a specific cookie/session (public user) - * Room format: `cookie:${cookieId}` - */ - // @SubscribeMessage('join:session') - // handleJoinSession(@ConnectedSocket() client: Socket, @MessageBody() data: { cookieId: string }) { - // if (!data.cookieId) { - // void client.emit('error', { message: 'Cookie ID is required' }); - // return; - // } - - // const room = `cookie:${data.cookieId}`; - // void client.join(room); - // this.logger.log(`Client ${client.id} joined room: ${room}`); - // void client.emit('joined', { room, message: 'Successfully joined session room' }); - // } - - /** - * Leave a room - */ - // @SubscribeMessage('leave:room') - // handleLeaveRoom(@ConnectedSocket() client: Socket, @MessageBody() data: { room: string }) { - // if (data.room) { - // void client.leave(data.room); - // this.logger.log(`Client ${client.id} left room: ${data.room}`); - // void client.emit('left', { room: data.room, message: 'Successfully left room' }); - // } - // } - - /** - * Get list of pagers for the restaurant (admin only) - * Requires admin authentication via JWT token - * Restaurant ID is extracted from the authenticated admin's token - */ - // @UseGuards(WsAdminAuthGuard) - // @SubscribeMessage('get:pagers') - // async handleGetPagers(@ConnectedSocket() client: AuthenticatedSocket, @WsRestId() restId: string) { - // try { - // const pagers = await this.pagerService.findAllByRestaurantId(restId); - - // const pagersData = pagers.map(pager => ({ - // id: pager.id, - // tableNumber: pager.tableNumber, - // message: pager.message, - // status: pager.status, - // cookieId: pager.cookieId, - // createdAt: pager.createdAt, - // updatedAt: pager.updatedAt, - // user: pager.user - // ? { - // id: pager.user.id, - // firstName: pager.user.firstName, - // lastName: pager.user.lastName, - // } - // : null, - // staff: pager.staff - // ? { - // id: pager.staff.id, - // firstName: pager.staff.firstName, - // lastName: pager.staff.lastName, - // } - // : null, - // })); - - // void client.emit('pagers:list', { pagers: pagersData }); - // this.logger.log(`Admin ${client.adminId} requested pagers list for restaurant ${restId}`); - // } catch (error) { - // this.logger.error(`Error getting pagers list: ${error instanceof Error ? error.message : 'Unknown error'}`); - // void client.emit('error', { - // message: 'Failed to get pagers list', - // code: 'GET_PAGERS_ERROR', - // details: error instanceof Error ? error.message : 'Unknown error', - // }); - // } - // } - - /** - * Update pager status (admin only) - * Requires admin authentication via JWT token - * Restaurant ID and Admin ID are extracted from the authenticated admin's token - */ - // @UseGuards(WsAdminAuthGuard) - // @SubscribeMessage('update:pager:status') - // async handleUpdatePagerStatus( - // @ConnectedSocket() client: AuthenticatedSocket, - // @WsRestId() restId: string, - // @WsAdminId() adminId: string, - // @MessageBody() data: { pagerId: string; status: PagerStatus }, - // ) { - // try { - // if (!data.pagerId || !data.status) { - // void client.emit('error', { - // message: 'Pager ID and status are required', - // code: 'INVALID_PAYLOAD', - // }); - // return; - // } - - // const pager = await this.pagerService.updateStatus(data.pagerId, restId, adminId, data.status); - - // const pagerData = { - // id: pager.id, - // tableNumber: pager.tableNumber, - // message: pager.message, - // status: pager.status, - // updatedAt: pager.updatedAt, - // staff: pager.staff - // ? { - // id: pager.staff.id, - // firstName: pager.staff.firstName, - // lastName: pager.staff.lastName, - // } - // : null, - // }; - - // void client.emit('pager:status:updated:response', { pager: pagerData }); - // this.logger.log(`Admin ${adminId} updated pager ${data.pagerId} status to ${data.status}`); - // } catch (error) { - // this.logger.error(`Error updating pager status: ${error instanceof Error ? error.message : 'Unknown error'}`); - // void client.emit('error', { - // message: error instanceof Error ? error.message : 'Failed to update pager status', - // code: 'UPDATE_PAGER_STATUS_ERROR', - // details: error instanceof Error ? error.message : 'Unknown error', - // }); - // } - // } - - // /** - // * Emit pager created event to restaurant room - // */ - // emitPagerCreated(pager: Pager) { - // const room = `restaurant:${pager.restaurant.id}`; - // this.server.to(room).emit('pager:created', { - // pager: { - // id: pager.id, - // tableNumber: pager.tableNumber, - // message: pager.message, - // status: pager.status, - // createdAt: pager.createdAt, - // user: pager.user - // ? { - // id: pager.user.id, - // firstName: pager.user.firstName, - // lastName: pager.user.lastName, - // } - // : null, - // }, - // }); - // this.logger.log(`Emitted pager:created to room: ${room}`); - // } - - // /** - // * Emit pager status updated event to both restaurant and cookie rooms - // */ - // emitPagerStatusUpdated(pager: Pager) { - // const restaurantRoom = `restaurant:${pager.restaurant.id}`; - // const cookieRoom = `cookie:${pager.cookieId}`; - - // const pagerData = { - // id: pager.id, - // tableNumber: pager.tableNumber, - // message: pager.message, - // status: pager.status, - // updatedAt: pager.updatedAt, - // staff: pager.staff - // ? { - // id: pager.staff.id, - // firstName: pager.staff.firstName, - // lastName: pager.staff.lastName, - // } - // : null, - // }; - - // // Emit to restaurant room (admin/staff) - // this.server.to(restaurantRoom).emit('pager:status:updated', { - // pager: pagerData, - // }); - // this.logger.log(`Emitted pager:status:updated to room: ${restaurantRoom}`); - - // // Emit to cookie room (public user) - // this.server.to(cookieRoom).emit('pager:status:updated', { - // pager: pagerData, - // }); - // this.logger.log(`Emitted pager:status:updated to room: ${cookieRoom}`); - // } - private async authenticateConnection(client: Socket): Promise { const guard = this.moduleRef.get(WsAdminAuthGuard); const context = { diff --git a/src/modules/notifications/notifications.module.ts b/src/modules/notifications/notifications.module.ts index d3a49fb..fbb9c73 100644 --- a/src/modules/notifications/notifications.module.ts +++ b/src/modules/notifications/notifications.module.ts @@ -19,6 +19,7 @@ import { NotificationQueueNameEnum } from './constants/queue'; import { UtilsModule } from '../utils/utils.module'; import { NotificationsGateway } from './notifications.gateway'; import { WsAdminAuthGuard } from './guards/ws-admin-auth.guard'; +import { InAppProcessor } from './processors/in-app.processor'; @Module({ imports: [ @@ -26,7 +27,11 @@ import { WsAdminAuthGuard } from './guards/ws-admin-auth.guard'; UtilsModule, JwtModule, MikroOrmModule.forFeature([Notification, NotificationPreference, User, Restaurant]), - BullModule.registerQueue({ name: NotificationQueueNameEnum.SMS }, { name: NotificationQueueNameEnum.PUSH }), + BullModule.registerQueue( + { name: NotificationQueueNameEnum.SMS }, + { name: NotificationQueueNameEnum.PUSH }, + { name: NotificationQueueNameEnum.IN_APP }, + ), BullModule.forRootAsync({ inject: [ConfigService], useFactory: (config: ConfigService) => ({ @@ -46,6 +51,7 @@ import { WsAdminAuthGuard } from './guards/ws-admin-auth.guard'; PushNotificationService, PushProcessor, SmsProcessor, + InAppProcessor, NotificationsGateway, WsAdminAuthGuard, ], diff --git a/src/modules/notifications/processors/in-app.processor.ts b/src/modules/notifications/processors/in-app.processor.ts new file mode 100644 index 0000000..45c23df --- /dev/null +++ b/src/modules/notifications/processors/in-app.processor.ts @@ -0,0 +1,50 @@ +import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq'; +import { Logger } from '@nestjs/common'; +import { Job } from 'bullmq'; +import { InjectRepository } from '@mikro-orm/nestjs'; +import { EntityManager } from '@mikro-orm/postgresql'; +import { Notification } from '../entities/notification.entity'; +import { InAppNotificationQueueJob } from '../interfaces/jobs-queue.interface'; +import { NotificationQueueNameEnum } from '../constants/queue'; +import { NotificationsGateway } from '../notifications.gateway'; + +@Processor(NotificationQueueNameEnum.IN_APP) +export class InAppProcessor extends WorkerHost { + private readonly logger = new Logger(InAppProcessor.name); + + constructor( + @InjectRepository(Notification) + private readonly em: EntityManager, + private readonly notificationsGateway: NotificationsGateway, + ) { + super(); + } + + async process(job: Job): Promise { + const { recipient, subject, body, notificationId } = job.data; + + this.logger.log(`Processing InApp notification - Recipient: ${JSON.stringify(recipient)}, subject: ${subject}`); + + try { + this.notificationsGateway.sendInAppNotification( + { adminId: recipient.adminId, restaurantId: recipient.restaurantId }, + { subject, body, notificationId }, + ); + + this.logger.log(`InApp notification sent successfully to ${recipient.adminId}`); + } catch (error) { + this.logger.error(`Error processing InApp 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); + } +} diff --git a/src/modules/notifications/services/notification-queue.service.ts b/src/modules/notifications/services/notification-queue.service.ts index 2336aec..0989ac7 100644 --- a/src/modules/notifications/services/notification-queue.service.ts +++ b/src/modules/notifications/services/notification-queue.service.ts @@ -2,7 +2,11 @@ import { Injectable, Logger } from '@nestjs/common'; import { InjectQueue } from '@nestjs/bullmq'; import { Queue } from 'bullmq'; import { NotificationQueueNameEnum } from '../constants/queue'; -import { PushNotificationQueueJob, SmsNotificationQueueJob } from '../interfaces/jobs-queue.interface'; +import { + InAppNotificationQueueJob, + PushNotificationQueueJob, + SmsNotificationQueueJob, +} from '../interfaces/jobs-queue.interface'; @Injectable() export class NotificationQueueService { @@ -11,6 +15,7 @@ export class NotificationQueueService { constructor( @InjectQueue(NotificationQueueNameEnum.SMS) private readonly smsQueue: Queue, @InjectQueue(NotificationQueueNameEnum.PUSH) private readonly pushQueue: Queue, + @InjectQueue(NotificationQueueNameEnum.IN_APP) private readonly inAppQueue: Queue, ) {} async addSmsNotification(job: SmsNotificationQueueJob): Promise { @@ -112,4 +117,26 @@ export class NotificationQueueService { throw error; } } + async addBulkInAppNotifications(jobs: InAppNotificationQueueJob[]): Promise { + try { + const queueJobs = jobs.map(job => ({ + name: 'in-app-notification', + data: job, + opts: { + jobId: `${job.subject}-${job.notificationId}-${Date.now()}`, + attempts: 3, + backoff: { + type: 'exponential', + delay: 2000, + }, + }, + })); + + await this.inAppQueue.addBulk(queueJobs); + this.logger.log(`Added ${jobs.length} InApp notification jobs to queue`); + } catch (error) { + this.logger.error(`Failed to add bulk notifications to queue:`, error); + throw error; + } + } } diff --git a/src/modules/notifications/services/notification.service.ts b/src/modules/notifications/services/notification.service.ts index 90890fe..a3b9e0c 100644 --- a/src/modules/notifications/services/notification.service.ts +++ b/src/modules/notifications/services/notification.service.ts @@ -41,18 +41,14 @@ export class NotificationService { // send in app notification if (preference?.channels?.includes(NotifChannelEnum.IN_APP)) { - notifications.forEach(notif => { - if (notif.admin) { - this.notificationGateway.sendInAppNotification( - { adminId: notif.admin.id, restaurantId }, - { - notificationId: notif.id, - subject: message.title, - body: message.content, - }, - ); - } - }); + await this.queueService.addBulkInAppNotifications( + notifications.map(notification => ({ + recipient: { adminId: notification.admin?.id || '', restaurantId }, + subject: message.title, + body: message.content, + notificationId: notification.id, + })), + ); } // add sms notifications to queue if (preference?.channels?.includes(NotifChannelEnum.SMS)) { diff --git a/src/modules/pager/listeners/notification.listeners.ts b/src/modules/pager/listeners/notification.listeners.ts index 0d8a1cd..1ae8bfd 100644 --- a/src/modules/pager/listeners/notification.listeners.ts +++ b/src/modules/pager/listeners/notification.listeners.ts @@ -48,7 +48,7 @@ export class PagerListeners { recipients, metadata: { priority: 1, - }, + }, }); } catch (error) { this.logger.error(