From d2838f0976816afeeade94adfb57c3179e850101 Mon Sep 17 00:00:00 2001 From: morteza-mortezai Date: Wed, 1 Jul 2026 20:26:00 +0330 Subject: [PATCH] add new sms-queue --- ...ation20260701170000_updateSmsLogsEntity.ts | 40 +++++++++++ src/modules/notifications/constants/queue.ts | 1 + .../notifications/entities/smsLogs.entity.ts | 9 +-- .../interfaces/jobs-queue.interface.ts | 7 ++ .../notifications/listeners/sms.listeners.ts | 22 +++--- .../notifications/notifications.module.ts | 7 +- .../processors/direct-sms.processor.ts | 55 +++++++++++++++ .../repositories/sms-log.repository.ts | 19 ++---- .../services/sms-queue.service.ts | 68 +++++++++++++++++++ 9 files changed, 197 insertions(+), 31 deletions(-) create mode 100644 database/migrations/Migration20260701170000_updateSmsLogsEntity.ts create mode 100644 src/modules/notifications/processors/direct-sms.processor.ts create mode 100644 src/modules/notifications/services/sms-queue.service.ts diff --git a/database/migrations/Migration20260701170000_updateSmsLogsEntity.ts b/database/migrations/Migration20260701170000_updateSmsLogsEntity.ts new file mode 100644 index 0000000..6817e8e --- /dev/null +++ b/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 { + 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 { + 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";`); + } + +} diff --git a/src/modules/notifications/constants/queue.ts b/src/modules/notifications/constants/queue.ts index 644e72f..cd5d429 100644 --- a/src/modules/notifications/constants/queue.ts +++ b/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', } diff --git a/src/modules/notifications/entities/smsLogs.entity.ts b/src/modules/notifications/entities/smsLogs.entity.ts index edeb103..bf3a880 100644 --- a/src/modules/notifications/entities/smsLogs.entity.ts +++ b/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; } \ No newline at end of file diff --git a/src/modules/notifications/interfaces/jobs-queue.interface.ts b/src/modules/notifications/interfaces/jobs-queue.interface.ts index 470a72c..d468465 100644 --- a/src/modules/notifications/interfaces/jobs-queue.interface.ts +++ b/src/modules/notifications/interfaces/jobs-queue.interface.ts @@ -6,6 +6,13 @@ export interface SmsNotificationQueueJob { restaurantId: string; parameters?: Record; } + +export interface DirectSmsQueueJob { + phone: string; + templateId: string; + params?: Record; + restaurantId?: string; +} export interface PushNotificationQueueJob { title: string; content: string; diff --git a/src/modules/notifications/listeners/sms.listeners.ts b/src/modules/notifications/listeners/sms.listeners.ts index 566008f..50534e5 100644 --- a/src/modules/notifications/listeners/sms.listeners.ts +++ b/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, - createdAt: new Date(), - }); + 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( diff --git a/src/modules/notifications/notifications.module.ts b/src/modules/notifications/notifications.module.ts index c99a1ae..336c35c 100644 --- a/src/modules/notifications/notifications.module.ts +++ b/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 { } diff --git a/src/modules/notifications/processors/direct-sms.processor.ts b/src/modules/notifications/processors/direct-sms.processor.ts new file mode 100644 index 0000000..22d035e --- /dev/null +++ b/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) { + 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, + }); + + 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); + } +} diff --git a/src/modules/notifications/repositories/sms-log.repository.ts b/src/modules/notifications/repositories/sms-log.repository.ts index 6076a29..84c7c4a 100644 --- a/src/modules/notifications/repositories/sms-log.repository.ts +++ b/src/modules/notifications/repositories/sms-log.repository.ts @@ -15,26 +15,23 @@ export class SmsLogRepository extends EntityRepository { ): Promise> { 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 { } async getSmsCountByRestaurantId(restaurantId: string): Promise { - 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; } } diff --git a/src/modules/notifications/services/sms-queue.service.ts b/src/modules/notifications/services/sms-queue.service.ts new file mode 100644 index 0000000..6356569 --- /dev/null +++ b/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 { + 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 { + 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); + } +}