feat: handle notification using redis

package-lock.json

@@ -36,7 +36,6 @@
         "firebase-admin": "^13.0.2",
         "google-libphonenumber": "^3.2.39",
         "handlebars": "^4.7.8",
-        "ioredis": "^5.4.1",
         "jwk-to-pem": "^2.0.7",
         "lodash": "^4.17.21",
         "moment": "^2.30.1",
@@ -1187,7 +1186,9 @@
     },
     "node_modules/@ioredis/commands": {
       "version": "1.2.0",
-      "license": "MIT"
+      "license": "MIT",
+      "optional": true,
+      "peer": true
     },
     "node_modules/@isaacs/cliui": {
       "version": "8.0.2",
@@ -5239,6 +5240,8 @@
     "node_modules/denque": {
       "version": "2.1.0",
       "license": "Apache-2.0",
+      "optional": true,
+      "peer": true,
       "engines": {
         "node": ">=0.10"
       }
@@ -7381,6 +7384,8 @@
     "node_modules/ioredis": {
       "version": "5.4.1",
       "license": "MIT",
+      "optional": true,
+      "peer": true,
       "dependencies": {
         "@ioredis/commands": "^1.1.1",
         "cluster-key-slot": "^1.1.0",
@@ -9105,7 +9110,9 @@
     },
     "node_modules/lodash.defaults": {
       "version": "4.2.0",
-      "license": "MIT"
+      "license": "MIT",
+      "optional": true,
+      "peer": true
     },
     "node_modules/lodash.includes": {
       "version": "4.3.0",
@@ -9113,7 +9120,9 @@
     },
     "node_modules/lodash.isarguments": {
       "version": "3.1.0",
-      "license": "MIT"
+      "license": "MIT",
+      "optional": true,
+      "peer": true
     },
     "node_modules/lodash.isboolean": {
       "version": "3.0.3",
@@ -15708,6 +15717,8 @@
     "node_modules/redis-errors": {
       "version": "1.2.0",
       "license": "MIT",
+      "optional": true,
+      "peer": true,
       "engines": {
         "node": ">=4"
       }
@@ -15715,6 +15726,8 @@
     "node_modules/redis-parser": {
       "version": "3.0.0",
       "license": "MIT",
+      "optional": true,
+      "peer": true,
       "dependencies": {
         "redis-errors": "^1.0.0"
       },
@@ -16499,7 +16512,9 @@
     },
     "node_modules/standard-as-callback": {
       "version": "2.1.0",
-      "license": "MIT"
+      "license": "MIT",
+      "optional": true,
+      "peer": true
     },
     "node_modules/statuses": {
       "version": "2.0.1",

package.json

@@ -54,7 +54,6 @@
     "firebase-admin": "^13.0.2",
     "google-libphonenumber": "^3.2.39",
     "handlebars": "^4.7.8",
-    "ioredis": "^5.4.1",
     "jwk-to-pem": "^2.0.7",
     "lodash": "^4.17.21",
     "moment": "^2.30.1",

src/common/modules/notification/events/notification.event.ts

@@ -1,3 +0,0 @@
-export class NotificationEvent {
-  constructor(public readonly notification: Notification) {}
-}

src/common/modules/notification/listeners/index.ts

@@ -0,0 +1 @@
+export * from './notification-created.listener';

src/common/modules/notification/listeners/notification-created.listener.ts

@@ -0,0 +1,83 @@
+// notification-created.handler.ts
+import { MailerService } from '@nestjs-modules/mailer';
+import { Injectable, Logger } from '@nestjs/common';
+import { SendEmailRequestDto } from '~/common/modules/notification/dtos/request';
+import { EventType, NotificationChannel, NotificationScope } from '~/common/modules/notification/enums';
+import { FirebaseService, TwilioService } from '~/common/modules/notification/services';
+import { IEventInterface } from '~/common/redis/interface';
+import { DeviceService } from '~/user/services';
+
+@Injectable()
+export class NotificationCreatedListener {
+  private readonly logger = new Logger(NotificationCreatedListener.name);
+
+  constructor(
+    private readonly twilioService: TwilioService,
+    private readonly deviceService: DeviceService,
+    private readonly mailerService: MailerService,
+    private readonly firebaseService: FirebaseService,
+  ) {}
+
+  /**
+   * Handles the NOTIFICATION_CREATED event by calling the appropriate channel logic.
+   */
+  async handle(event: IEventInterface) {
+    this.logger.log(
+      `Handling ${EventType.NOTIFICATION_CREATED} event for notification ${event.id} (channel: ${event.channel})`,
+    );
+
+    switch (event.channel) {
+      case NotificationChannel.SMS:
+        return this.sendSMS(event.recipient!, event.message);
+
+      case NotificationChannel.PUSH:
+        return this.sendPushNotification(event.userId, event.title, event.message);
+
+      case NotificationChannel.EMAIL:
+        return this.sendEmail({
+          to: event.recipient!,
+          subject: event.title,
+          template: this.getTemplateFromNotification(event),
+          data: event.data,
+        });
+    }
+  }
+
+  private getTemplateFromNotification(notification: IEventInterface) {
+    switch (notification.scope) {
+      case NotificationScope.OTP:
+        return 'otp';
+      case NotificationScope.USER_INVITED:
+        return 'user-invite';
+      default:
+        return 'otp';
+    }
+  }
+
+  private async sendPushNotification(userId: string, title: string, body: string) {
+    this.logger.log(`Sending push notification to user ${userId}`);
+    const tokens = await this.deviceService.getTokens(userId);
+
+    if (!tokens.length) {
+      this.logger.log(`No device tokens found for user ${userId}, but notification was created in the DB.`);
+      return;
+    }
+    return this.firebaseService.sendNotification(tokens, title, body);
+  }
+
+  private async sendSMS(to: string, body: string) {
+    this.logger.log(`Sending SMS to ${to}`);
+    await this.twilioService.sendSMS(to, body);
+  }
+
+  private async sendEmail({ to, subject, data, template }: SendEmailRequestDto) {
+    this.logger.log(`Sending email to ${to}`);
+    await this.mailerService.sendMail({
+      to,
+      subject,
+      template,
+      context: { ...data },
+    });
+    this.logger.log(`Email sent to ${to}`);
+  }
+}

src/common/modules/notification/notification.module.ts

@@ -3,15 +3,19 @@ import { forwardRef, Module } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { TwilioModule } from 'nestjs-twilio';
+import { RedisModule } from '~/common/redis/redis.module';
 import { buildMailerOptions, buildTwilioOptions } from '~/core/module-options';
 import { UserModule } from '~/user/user.module';
 import { NotificationsController } from './controllers';
 import { Notification } from './entities';
+import { NotificationCreatedListener } from './listeners';
 import { NotificationsRepository } from './repositories';
 import { FirebaseService, NotificationsService, TwilioService } from './services';
 
 @Module({
   imports: [
+    forwardRef(() => RedisModule.register()),
+    forwardRef(() => UserModule),
     TypeOrmModule.forFeature([Notification]),
     TwilioModule.forRootAsync({
       useFactory: buildTwilioOptions,
@@ -21,10 +25,15 @@ import { FirebaseService, NotificationsService, TwilioService } from './services
       useFactory: buildMailerOptions,
       inject: [ConfigService],
     }),
-    forwardRef(() => UserModule),
   ],
-  providers: [NotificationsService, FirebaseService, NotificationsRepository, TwilioService],
-  exports: [NotificationsService],
+  providers: [
+    NotificationsService,
+    FirebaseService,
+    NotificationsRepository,
+    TwilioService,
+    NotificationCreatedListener,
+  ],
+  exports: [NotificationsService, NotificationCreatedListener],
   controllers: [NotificationsController],
 })
 export class NotificationModule {}

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

@@ -1,8 +1,6 @@
-import { MailerService } from '@nestjs-modules/mailer';
-import { Injectable, Logger } from '@nestjs/common';
-import { EventEmitter2, OnEvent } from '@nestjs/event-emitter';
+import { forwardRef, Inject, Injectable, Logger } from '@nestjs/common';
+import { RedisPubSubService } from '~/common/redis/services';
 import { PageOptionsRequestDto } from '~/core/dtos';
-import { DeviceService } from '~/user/services';
 import { OTP_BODY, OTP_TITLE } from '../../otp/constants';
 import { OtpType } from '../../otp/enums';
 import { ISendOtp } from '../../otp/interfaces';
@@ -10,19 +8,15 @@ import { SendEmailRequestDto } from '../dtos/request';
 import { Notification } from '../entities';
 import { EventType, NotificationChannel, NotificationScope } from '../enums';
 import { NotificationsRepository } from '../repositories';
-import { FirebaseService } from './firebase.service';
-import { TwilioService } from './twilio.service';
 
 @Injectable()
 export class NotificationsService {
   private readonly logger = new Logger(NotificationsService.name);
   constructor(
-    private readonly firebaseService: FirebaseService,
     private readonly notificationRepository: NotificationsRepository,
-    private readonly twilioService: TwilioService,
-    private readonly eventEmitter: EventEmitter2,
-    private readonly deviceService: DeviceService,
-    private readonly mailerService: MailerService,
+
+    @Inject(forwardRef(() => RedisPubSubService))
+    private readonly redisPubSubService: RedisPubSubService,
   ) {}
 
   async getNotifications(userId: string, pageOptionsDto: PageOptionsRequestDto) {
@@ -56,9 +50,11 @@ export class NotificationsService {
       scope: NotificationScope.USER_INVITED,
       channel: NotificationChannel.EMAIL,
     });
-    console.log('++++++++++++++++++++++++=');
-    console.log(data);
-    return this.eventEmitter.emit(EventType.NOTIFICATION_CREATED, notification, data.data);
+    // return this.redisPubSubService.emit(EventType.NOTIFICATION_CREATED, notification, data.data);
+    this.redisPubSubService.publishEvent(EventType.NOTIFICATION_CREATED, {
+      ...notification,
+      data,
+    });
   }
 
   async sendOtpNotification(sendOtpRequest: ISendOtp, otp: string) {
@@ -73,67 +69,9 @@ export class NotificationsService {
 
     this.logger.log(`emitting ${EventType.NOTIFICATION_CREATED} event`);
 
-    return this.eventEmitter.emit(EventType.NOTIFICATION_CREATED, notification, { otp });
-  }
-
-  private async sendPushNotification(userId: string, title: string, body: string) {
-    this.logger.log(`Sending push notification to user ${userId}`);
-    // Get the device tokens for the user
-
-    const tokens = await this.deviceService.getTokens(userId);
-
-    if (!tokens.length) {
-      this.logger.log(`No device tokens found for user ${userId} but notification created in the database`);
-      return;
-    }
-    // Send the notification
-    return this.firebaseService.sendNotification(tokens, title, body);
-  }
-
-  private async sendSMS(to: string, body: string) {
-    this.logger.log(`Sending SMS to ${to}`);
-    await this.twilioService.sendSMS(to, body);
-  }
-
-  private async sendEmail({ to, subject, data, template }: SendEmailRequestDto) {
-    this.logger.log(`Sending email to ${to}`);
-    await this.mailerService.sendMail({
-      to,
-      subject,
-      template,
-      context: { ...data },
+    return this.redisPubSubService.publishEvent(EventType.NOTIFICATION_CREATED, {
+      ...notification,
+      data: { otp },
     });
-    this.logger.log(`Email sent to ${to}`);
-  }
-
-  private getTemplateFromNotification(notification: Notification) {
-    switch (notification.scope) {
-      case NotificationScope.OTP:
-        return 'otp';
-      case NotificationScope.USER_INVITED:
-        return 'user-invite';
-      default:
-        return 'otp';
-    }
-  }
-
-  @OnEvent(EventType.NOTIFICATION_CREATED)
-  handleNotificationCreatedEvent(notification: Notification, data?: any) {
-    this.logger.log(
-      `Handling ${EventType.NOTIFICATION_CREATED} event for notification ${notification.id} and type ${notification.channel}`,
-    );
-    switch (notification.channel) {
-      case NotificationChannel.SMS:
-        return this.sendSMS(notification.recipient!, notification.message);
-      case NotificationChannel.PUSH:
-        return this.sendPushNotification(notification.userId, notification.title, notification.message);
-      case NotificationChannel.EMAIL:
-        return this.sendEmail({
-          to: notification.recipient!,
-          subject: notification.title,
-          template: this.getTemplateFromNotification(notification),
-          data,
-        });
-    }
   }
 }

src/common/modules/otp/services/otp.service.ts

@@ -43,9 +43,8 @@ export class OtpService {
       return false;
     }
 
-    const { affected } = await this.otpRepository.updateOtp(otp.id, { isUsed: true });
-    console.log('+++++++++++++++++++++++++++');
-    console.log(affected);
+    await this.otpRepository.updateOtp(otp.id, { isUsed: true });
+
     this.logger.log(`OTP verified successfully for ${verifyOtpRequest.userId}`);
 
     return !!otp;

src/common/redis/interface/event.interface.ts

@@ -0,0 +1,5 @@
+import { Notification } from '~/common/modules/notification/entities';
+
+export interface IEventInterface extends Notification {
+  data?: any;
+}

src/common/redis/interface/index.ts

@@ -0,0 +1 @@
+export * from './event.interface';

src/common/redis/redis.module.ts

@@ -0,0 +1,39 @@
+// redis.module.ts (NestJS)
+import { createClient } from '@keyv/redis';
+import { DynamicModule, Module } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { NotificationModule } from '../modules/notification/notification.module';
+import { RedisPubSubService } from './services';
+
+@Module({})
+export class RedisModule {
+  static register(): DynamicModule {
+    return {
+      module: RedisModule,
+      providers: [
+        {
+          provide: 'REDIS_PUBLISHER',
+          useFactory: async (configService: ConfigService) => {
+            const publisher = createClient({ url: configService.get<string>('REDIS_URL') });
+            await publisher.connect();
+            return publisher;
+          },
+          inject: [ConfigService],
+        },
+
+        {
+          provide: 'REDIS_SUBSCRIBER',
+          useFactory: async (configService: ConfigService) => {
+            const subscriber = createClient({ url: configService.get<string>('REDIS_URL') });
+            await subscriber.connect();
+            return subscriber;
+          },
+          inject: [ConfigService],
+        },
+        RedisPubSubService,
+      ],
+      exports: [RedisPubSubService],
+      imports: [NotificationModule],
+    };
+  }
+}

src/common/redis/services/index.ts

@@ -0,0 +1 @@
+export * from './redis-pubsub.service';

src/common/redis/services/redis-pubsub.service.ts

@@ -0,0 +1,29 @@
+// redis.pubsub.service.ts (NestJS)
+import { RedisClientType } from '@keyv/redis';
+import { Inject, Injectable, Logger, OnModuleInit } from '@nestjs/common';
+import { EventType } from '~/common/modules/notification/enums';
+import { NotificationCreatedListener } from '~/common/modules/notification/listeners';
+import { IEventInterface } from '../interface';
+
+@Injectable()
+export class RedisPubSubService implements OnModuleInit {
+  private readonly logger = new Logger(RedisPubSubService.name);
+  constructor(
+    @Inject('REDIS_PUBLISHER') private readonly publisher: RedisClientType,
+    @Inject('REDIS_SUBSCRIBER') private readonly subscriber: RedisClientType,
+    private readonly notificationCreatedListener: NotificationCreatedListener,
+  ) {}
+
+  onModuleInit() {
+    this.subscriber.subscribe(EventType.NOTIFICATION_CREATED, async (message) => {
+      const data = JSON.parse(message);
+      this.logger.log('Received message on NOTIFICATION_CREATED channel:', data);
+
+      await this.notificationCreatedListener.handle(data);
+    });
+  }
+
+  async publishEvent(channel: string, payload: IEventInterface) {
+    await this.publisher.publish(channel, JSON.stringify(payload));
+  }
+}

src/customer/repositories/customer.repository.ts

@@ -51,7 +51,6 @@ export class CustomerRepository {
 
     if (filters.name) {
       const nameParts = filters.name.trim().split(/\s+/);
-      console.log(nameParts);
       nameParts.length > 1
         ? query.andWhere('customer.firstName LIKE :firstName AND customer.lastName LIKE :lastName', {
             firstName: `%${nameParts[0]}%`,