add new sms-queue

database/migrations/Migration20260701170000_updateSmsLogsEntity.ts

@@ -0,0 +1,40 @@
+import { Migration } from '@mikro-orm/migrations';
+
+export class Migration20260701170000_updateSmsLogsEntity extends Migration {
+
+  override async up(): Promise<void> {
+    this.addSql(`alter table "sms_logs" add column "count" int not null default 0;`);
+
+    this.addSql(`
+      with aggregated as (
+        select restaurant_id, count(*)::int as total_count
+        from sms_logs
+        group by restaurant_id
+      )
+      update sms_logs sl
+      set count = a.total_count
+      from aggregated a
+      where sl.restaurant_id = a.restaurant_id
+        and sl.id = (
+          select min(id) from sms_logs where restaurant_id = sl.restaurant_id
+        );
+    `);
+
+    this.addSql(`
+      delete from sms_logs sl1
+      using sms_logs sl2
+      where sl1.restaurant_id = sl2.restaurant_id
+        and sl1.id > sl2.id;
+    `);
+
+    this.addSql(`alter table "sms_logs" drop column "phone";`);
+    this.addSql(`alter table "sms_logs" add constraint "sms_logs_restaurant_id_unique" unique ("restaurant_id");`);
+  }
+
+  override async down(): Promise<void> {
+    this.addSql(`alter table "sms_logs" drop constraint if exists "sms_logs_restaurant_id_unique";`);
+    this.addSql(`alter table "sms_logs" add column "phone" varchar(255) not null default '';`);
+    this.addSql(`alter table "sms_logs" drop column "count";`);
+  }
+
+}

src/modules/notifications/constants/queue.ts

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

src/modules/notifications/entities/smsLogs.entity.ts

@@ -6,12 +6,9 @@ export class SmsLog {
   @PrimaryKey()
   id!: number;
 
-  @ManyToOne(() => Restaurant)
+  @ManyToOne(() => Restaurant, { unique: true })
   restaurant!: Restaurant;
 
-  @Property()
-  phone!: string;
-
-  @Property()
-  createdAt: Date = new Date();
+  @Property({ default: 0 })
+  count: number = 0;
 }

src/modules/notifications/interfaces/jobs-queue.interface.ts

@@ -6,6 +6,13 @@ export interface SmsNotificationQueueJob {
   restaurantId: string;
   parameters?: Record<string, string>;
 }
+
+export interface DirectSmsQueueJob {
+  phone: string;
+  templateId: string;
+  params?: Record<string, unknown>;
+  restaurantId?: string;
+}
 export interface PushNotificationQueueJob {
   title: string;
   content: string;

src/modules/notifications/listeners/sms.listeners.ts

@@ -21,10 +21,9 @@ export class SmsListeners {
   async handleSmsSent(event: SmsSentEvent) {
     try {
       this.logger.log(
-        `SMS sent event received: phone ${event.phoneNumber} for restaurant: ${event.restaurantId}`,
+        `SMS sent event received for restaurant: ${event.restaurantId}`,
       );
 
-      // Get the restaurant entity
       const restaurant = await this.restRepository.findOne({ id: event.restaurantId });
 
       if (!restaurant) {
@@ -34,17 +33,22 @@ export class SmsListeners {
         return;
       }
 
-      // Create SMS log record
-      const smsLog = this.em.create(SmsLog, {
-        restaurant: restaurant,
-        phone: event.phoneNumber,
+      let smsLog = await this.em.findOne(SmsLog, { restaurant: event.restaurantId });
+
+      if (smsLog) {
+        smsLog.count += 1;
+      } else {
+        smsLog = this.em.create(SmsLog, {
+          restaurant,
+          count: 1,
           createdAt: new Date(),
         });
+      }
 
       await this.em.flush();
 
       this.logger.log(
-        `SMS log created successfully: ${smsLog.id} for phone ${event.phoneNumber}`,
+        `SMS log updated successfully: ${smsLog.id} for restaurant ${event.restaurantId}, count: ${smsLog.count}`,
       );
     } catch (error) {
       this.logger.error(

src/modules/notifications/notifications.module.ts

@@ -9,6 +9,7 @@ import { NotificationPreferenceService } from './services/notification-preferenc
 import { NotificationQueueService } from './services/notification-queue.service';
 import { PushNotificationService } from './services/push-notification.service';
 import { SmsProcessor } from './processors/sms.processor';
+import { DirectSmsProcessor } from './processors/direct-sms.processor';
 import { PushProcessor } from './processors/push.processor';
 import { NotificationsController } from './controllers/notifications.controller';
 import { User } from '../users/entities/user.entity';
@@ -24,6 +25,7 @@ import { AdminModule } from '../admin/admin.module';
 import { UserModule } from '../users/user.module';
 import { NotificationCrone } from './crone/notification.crone';
 import { SmsService } from './services/sms.service';
+import { SmsQueueService } from './services/sms-queue.service';
 import { SmsLog } from './entities/smsLogs.entity';
 import { SmsLogRepository } from './repositories/sms-log.repository';
 import { RestaurantsModule } from '../restaurants/restaurants.module';
@@ -36,6 +38,7 @@ import { SmsListeners } from './listeners/sms.listeners';
     MikroOrmModule.forFeature([Notification, NotificationPreference, User, Restaurant, SmsLog]),
     BullModule.registerQueue(
       { name: NotificationQueueNameEnum.SMS },
+      { name: NotificationQueueNameEnum.DIRECT_SMS },
       { name: NotificationQueueNameEnum.PUSH },
       { name: NotificationQueueNameEnum.IN_APP },
     ),
@@ -62,15 +65,17 @@ import { SmsListeners } from './listeners/sms.listeners';
     PushNotificationService,
     PushProcessor,
     SmsProcessor,
+    DirectSmsProcessor,
     InAppProcessor,
     NotificationsGateway,
     WsAdminAuthGuard,
     NotificationCrone,
     SmsService,
+    SmsQueueService,
     SmsLogRepository,
     SmsListeners,
   ],
   exports: [NotificationService, NotificationPreferenceService,
-    NotificationQueueService, PushNotificationService, SmsService],
+    NotificationQueueService, PushNotificationService, SmsService, SmsQueueService],
 })
 export class NotificationsModule { }

src/modules/notifications/processors/direct-sms.processor.ts

@@ -0,0 +1,55 @@
+import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq';
+import { Logger } from '@nestjs/common';
+import { Job } from 'bullmq';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { DirectSmsQueueJob } from '../interfaces/jobs-queue.interface';
+import { NotificationQueueNameEnum } from '../constants/queue';
+import { SmsService } from '../services/sms.service';
+import { SmsSentEvent } from '../events/sms.events';
+
+@Processor(NotificationQueueNameEnum.DIRECT_SMS)
+export class DirectSmsProcessor extends WorkerHost {
+  private readonly logger = new Logger(DirectSmsProcessor.name);
+
+  constructor(
+    private readonly smsService: SmsService,
+    private readonly eventEmitter: EventEmitter2,
+  ) {
+    super();
+  }
+
+  async process(job: Job<DirectSmsQueueJob>) {
+    const { phone, templateId, params, restaurantId } = job.data;
+
+    this.logger.log(`Processing direct SMS - phone: ${phone}, templateId: ${templateId}`);
+
+    try {
+      await this.smsService.sendSms({
+        phone,
+        templateId,
+        parameters: params as Record<string, string | number | Date>,
+      });
+
+      this.logger.log(`Direct SMS sent successfully to ${phone}`);
+
+      if (restaurantId) {
+        this.eventEmitter.emit(SmsSentEvent.name, new SmsSentEvent(phone, restaurantId));
+      }
+
+      return { success: true };
+    } catch (error) {
+      this.logger.error(`Error processing direct SMS job ${job.id}:`, error);
+      throw error;
+    }
+  }
+
+  @OnWorkerEvent('completed')
+  onCompleted(job: Job) {
+    this.logger.log(`Direct SMS job ${job.id} completed`);
+  }
+
+  @OnWorkerEvent('failed')
+  onFailed(job: Job, error: Error) {
+    this.logger.error(`Direct SMS job ${job.id} failed:`, error);
+  }
+}

src/modules/notifications/repositories/sms-log.repository.ts

@@ -15,26 +15,23 @@ export class SmsLogRepository extends EntityRepository<SmsLog> {
   ): Promise<PaginatedResult<{ restaurantId: string; restaurantName: string; smsCount: number }>> {
     const offset = (page - 1) * limit;
 
-    // Get total count of unique restaurants with SMS logs
     const totalResult = await this.em.execute(
       `
-      SELECT COUNT(DISTINCT sl.restaurant_id) as total
+      SELECT COUNT(*) as total
       FROM sms_logs sl
       WHERE sl.restaurant_id IS NOT NULL
       `,
     );
     const total = parseInt(totalResult[0]?.total || '0', 10);
 
-    // Get paginated results with SMS counts grouped by restaurant
     const results = await this.em.execute(
       `
       SELECT 
         r.id as "restaurantId",
         r.name as "restaurantName",
-        COUNT(sl.id)::int as "smsCount"
+        sl.count as "smsCount"
       FROM sms_logs sl
       INNER JOIN restaurants r ON sl.restaurant_id = r.id
-      GROUP BY r.id, r.name
       ORDER BY "smsCount" DESC, r.name ASC
       LIMIT ? OFFSET ?
       `,
@@ -61,16 +58,8 @@ export class SmsLogRepository extends EntityRepository<SmsLog> {
   }
 
   async getSmsCountByRestaurantId(restaurantId: string): Promise<number> {
-    const result = await this.em.execute(
-      `
-      SELECT COUNT(sl.id)::int as "smsCount"
-      FROM sms_logs sl
-      WHERE sl.restaurant_id = ?
-      `,
-      [restaurantId],
-    );
-
-    return parseInt(result[0]?.smsCount || '0', 10);
+    const smsLog = await this.findOne({ restaurant: restaurantId });
+    return smsLog?.count ?? 0;
   }
 }
 

src/modules/notifications/services/sms-queue.service.ts

@@ -0,0 +1,68 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectQueue } from '@nestjs/bullmq';
+import { Queue } from 'bullmq';
+import { NotificationQueueNameEnum } from '../constants/queue';
+import { DirectSmsQueueJob } from '../interfaces/jobs-queue.interface';
+
+@Injectable()
+export class SmsQueueService {
+  private readonly logger = new Logger(SmsQueueService.name);
+
+  constructor(
+    @InjectQueue(NotificationQueueNameEnum.DIRECT_SMS) private readonly directSmsQueue: Queue,
+  ) {}
+
+  async enqueue(job: DirectSmsQueueJob): Promise<void> {
+    try {
+      const jobId = `${job.templateId}-${this.generateRandomInt()}-${Date.now()}`;
+
+      await this.directSmsQueue.add('direct-sms', job, {
+        jobId,
+        attempts: 3,
+        backoff: {
+          type: 'exponential',
+          delay: 2000,
+        },
+        removeOnComplete: {
+          age: 24 * 3600,
+          count: 1000,
+        },
+        removeOnFail: {
+          age: 7 * 24 * 3600,
+        },
+      });
+
+      this.logger.log(`Direct SMS job added to queue: ${jobId}`);
+    } catch (error) {
+      this.logger.error('Failed to add direct SMS to queue:', error);
+      throw error;
+    }
+  }
+
+  async enqueueBulk(jobs: DirectSmsQueueJob[]): Promise<void> {
+    try {
+      const queueJobs = jobs.map(job => ({
+        name: 'direct-sms',
+        data: job,
+        opts: {
+          jobId: `${job.templateId}-${this.generateRandomInt()}-${Date.now()}`,
+          attempts: 3,
+          backoff: {
+            type: 'exponential',
+            delay: 2000,
+          },
+        },
+      }));
+
+      await this.directSmsQueue.addBulk(queueJobs);
+      this.logger.log(`Added ${jobs.length} direct SMS jobs to queue`);
+    } catch (error) {
+      this.logger.error('Failed to add bulk direct SMS to queue:', error);
+      throw error;
+    }
+  }
+
+  private generateRandomInt(): string {
+    return Math.random().toString(36).substring(2, 15);
+  }
+}