notification queue

src/modules/notifications/constants/queue.ts

@@ -1,4 +1,5 @@
 export enum NotificationQueueNameEnum {
   SMS = 'sms',
   PUSH = 'push',
+  IN_APP = 'in-app',
 }

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;

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<void> {
     const guard = this.moduleRef.get(WsAdminAuthGuard);
     const context = {

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,
   ],

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<InAppNotificationQueueJob>): Promise<void> {
+    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);
+  }
+}

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<void> {
@@ -112,4 +117,26 @@ export class NotificationQueueService {
       throw error;
     }
   }
+  async addBulkInAppNotifications(jobs: InAppNotificationQueueJob[]): Promise<void> {
+    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;
+    }
+  }
 }

src/modules/notifications/services/notification.service.ts

@@ -41,19 +41,15 @@ export class NotificationService {
 
     // send in app notification
     if (preference?.channels?.includes(NotifChannelEnum.IN_APP)) {
-      notifications.forEach(notif => {
+      await this.queueService.addBulkInAppNotifications(
-        if (notif.admin) {
+        notifications.map(notification => ({
-          this.notificationGateway.sendInAppNotification(
+          recipient: { adminId: notification.admin?.id || '', restaurantId },
-            { adminId: notif.admin.id, restaurantId },
-            {
-              notificationId: notif.id,
           subject: message.title,
           body: message.content,
-            },
+          notificationId: notification.id,
+        })),
       );
     }
-      });
-    }
     // add sms notifications to queue
     if (preference?.channels?.includes(NotifChannelEnum.SMS)) {
       await this.queueService.addBulkSmsNotifications(