danak/dsc-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add all notification sewrvice to the queue

Browse Source
This commit is contained in:
mahyargdz
2025-05-04 16:06:14 +03:30
parent d50b18b8aa
commit c3ad2a1733
16 changed files with 539 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app.module.ts
+3 -3
View File
@@ -7,13 +7,13 @@ import { ConfigModule } from "@nestjs/config";
import { ThrottlerModule } from "@nestjs/throttler";
import { TypeOrmModule } from "@nestjs/typeorm";
import { MailerModule } from "@nestjs-modules/mailer";
// import { TelegrafModule } from "nestjs-telegraf";
import { TelegrafModule } from "nestjs-telegraf";
import { bullMqConfig } from "./configs/bullmq.config";
import { cacheConfig } from "./configs/cache.config";
import { mailerConfig } from "./configs/mailer.config";
import { rateLimitConfig } from "./configs/rateLimit.config";
// import { telegrafConfig } from "./configs/telegraf.config";
import { telegrafConfig } from "./configs/telegraf.config";
import { databaseConfigs } from "./configs/typeorm.config";
import { HTTPLogger } from "./core/middlewares/logger.middleware";
import { AddressModule } from "./modules/address/address.module";
@@ -43,7 +43,7 @@ import { MonitoringModule } from "./monitoring/monitoring.module";
@Module({
imports: [
// TelegrafModule.forRootAsync(telegrafConfig()),
TelegrafModule.forRootAsync(telegrafConfig()),
MailerModule.forRootAsync(mailerConfig()),
BullModule.forRootAsync(bullMqConfig()),
ThrottlerModule.forRootAsync(rateLimitConfig()),
src/common/enums/message.enum.ts
+1 -1
View File
@@ -420,7 +420,7 @@ export const enum NotificationMessage {
NEW_SERVICE_REVIEW = "نقد و بررسی جدید سرویس",
NEW_SERVICE_REVIEW_MESSAGE = "یک نقد و بررسی جدید برای سرویس [serviceName] ثبت شده است",
NEW_CUSTOMER = "مشتری جدید",
NEW_CUSTOMER_MESSAGE = "یک مشتری جدید با نام [customerName] ثبت نام کرده است",
NEW_CUSTOMER_MESSAGE = "یک مشتری جدید با نام [fullName] ثبت نام کرده است",
NEW_SUBSCRIPTION = "اشتراک جدید",
NEW_SUBSCRIPTION_MESSAGE = "یک اشتراک جدید برای پلن [serviceName] برای [fullName] ثبت شد",
NEW_TICKET = "تیکت جدید",
src/modules/auth/providers/auth.service.ts
+6 -10
View File
@@ -3,7 +3,7 @@ import { DataSource } from "typeorm";
import { TokensService } from "./tokens.service";
import { AuthMessage, UserMessage } from "../../../common/enums/message.enum";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { ReferralsService } from "../../referrals/providers/referrals.service";
import { Role } from "../../users/entities/role.entity";
import { AdminsService } from "../../users/providers/admins.service";
@@ -25,7 +25,7 @@ export class AuthService {
private readonly otpService: OTPService,
private readonly tokensService: TokensService,
private readonly dataSource: DataSource,
private readonly notificationService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
private readonly smsService: SmsService,
private readonly referralsService: ReferralsService,
) {}
@@ -76,16 +76,12 @@ export class AuthService {
const superAdmins = await this.adminsService.getSuperAdmins(queryRunner);
for (const admin of superAdmins) {
await this.notificationService.createNewCustomerNotification(
admin.id,
{
await this.notificationQueue.addNewCustomerNotification(admin.id, {
userPhone: admin.phone,
userEmail: admin.email,
fullName: `${user.firstName} ${user.lastName}`,
mobile: user.phone,
},
queryRunner,
);
});
}
await queryRunner.commitTransaction();
@@ -112,7 +108,7 @@ export class AuthService {
const tokens = await this.tokensService.generateTokens(user);
await this.notificationService.createLoginNotification(user.id, { userPhone: user.phone });
await this.notificationQueue.addLoginNotification(user.id, { userPhone: user.phone });
return {
message: AuthMessage.PASSWORD_LOGIN_SUCCESS,
@@ -193,7 +189,7 @@ export class AuthService {
const tokens = await this.tokensService.generateTokens(user);
await this.notificationService.createLoginNotification(user.id, { userPhone: user.phone });
await this.notificationQueue.addLoginNotification(user.id, { userPhone: user.phone });
return {
message: AuthMessage.LOGIN_SUCCESS,
...tokens,
src/modules/blogs/providers/blogs.service.ts
+4 -8
View File
@@ -3,7 +3,7 @@ import { DataSource, IsNull, Not } from "typeorm";
import { AdminMessage, BlogMessage, CommonMessage } from "../../../common/enums/message.enum";
import { CategoryListSearchQueryDto } from "../../danak-services/DTO/category-search-query.dto";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { AdminsService } from "../../users/providers/admins.service";
import { UsersService } from "../../users/providers/users.service";
import { BlogCommentQueryDto } from "../DTO/blog-comment-query.dto";
@@ -27,7 +27,7 @@ export class BlogsService {
private readonly usersService: UsersService,
private readonly adminsService: AdminsService,
private readonly dataSource: DataSource,
private readonly notificationsService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
) {}
//*********************************** */
@@ -261,16 +261,12 @@ export class BlogsService {
const superAdmin = await this.adminsService.getSuperAdmins(queryRunner);
for (const admin of superAdmin) {
await this.notificationsService.createNewBlogCommentNotification(
admin.id,
{
await this.notificationQueue.addNewBlogCommentNotification(admin.id, {
userPhone: admin.phone,
userEmail: admin.email,
blogTitle: blog.title,
fullName: `${user.firstName} ${user.lastName}`,
},
queryRunner,
);
});
}
await queryRunner.commitTransaction();
src/modules/criticisms/providers/criticisms.service.ts
+4 -8
View File
@@ -2,7 +2,7 @@ import { BadRequestException, Injectable } from "@nestjs/common";
import { DataSource, QueryRunner } from "typeorm";
import { CommonMessage, CriticismMessage } from "../../../common/enums/message.enum";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { AdminsService } from "../../users/providers/admins.service";
import { UsersService } from "../../users/providers/users.service";
import { PaginationUtils } from "../../utils/providers/pagination.utils";
@@ -17,7 +17,7 @@ export class CriticismsService {
private readonly usersService: UsersService,
private readonly dataSource: DataSource,
private readonly adminsService: AdminsService,
private readonly notificationsService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
) {}
async create(createDto: CreateCriticismDto, userId: string) {
@@ -101,16 +101,12 @@ export class CriticismsService {
const superAdmins = await this.adminsService.getSuperAdmins(queryRunner);
for (const admin of superAdmins) {
await this.notificationsService.createNewCriticismNotification(
admin.id,
{
await this.notificationQueue.addNewCriticismNotification(admin.id, {
fullName: `${criticism.user.firstName} ${criticism.user.lastName}`,
userPhone: admin.phone,
userEmail: admin.email,
title: criticism.title,
},
queryRunner,
);
});
}
}
}
src/modules/danak-services/providers/danak-services.service.ts
+4 -8
View File
@@ -3,7 +3,7 @@ import { DataSource, FindOptionsWhere, In, IsNull, Not, QueryRunner } from "type
import { ParamDto } from "../../../common/DTO/param.dto";
import { CategoryMessage, CommonMessage, ServiceMessage, SubscriptionMessage } from "../../../common/enums/message.enum";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { SubscriptionStatus } from "../../subscriptions/enums/subscription-status.enum";
import { AdminsService } from "../../users/providers/admins.service";
import { UsersService } from "../../users/providers/users.service";
@@ -34,8 +34,8 @@ export class DanakServicesService {
private readonly danakServicesImageRepository: DanakServicesImageRepository,
private readonly dataSource: DataSource,
private readonly adminsService: AdminsService,
private readonly notificationsService: NotificationsService,
private readonly usersService: UsersService,
private readonly notificationQueue: NotificationQueue,
) {}
/******************************************** */
@@ -525,16 +525,12 @@ export class DanakServicesService {
const superAdmins = await this.adminsService.getSuperAdmins(queryRunner);
for (const admin of superAdmins) {
await this.notificationsService.createNewServiceReviewNotification(
admin.id,
{
await this.notificationQueue.addNewServiceReviewNotification(admin.id, {
serviceName: service.name,
userPhone: admin.phone,
userEmail: admin.email,
fullName: `${user.firstName} ${user.lastName}`,
},
queryRunner,
);
});
}
await queryRunner.commitTransaction();
src/modules/invoices/providers/invoices.service.ts
+10 -27
View File
@@ -9,7 +9,7 @@ import { AuthMessage, InvoiceMessage, WalletMessage } from "../../../common/enum
import { VerifyOtpWithUserId } from "../../auth/DTO/verify-otp.dto";
import { Discount } from "../../discounts/entities/discount.entity";
import { DiscountType } from "../../discounts/enums/discount-type.enum";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { SubscriptionPlan } from "../../subscriptions/entities/subscription.entity";
import { UserSubscription } from "../../subscriptions/entities/user-subscription.entity";
import { SubscriptionStatus } from "../../subscriptions/enums/subscription-status.enum";
@@ -29,14 +29,13 @@ import { RecurringPeriodEnum } from "../enums/invoice-recurring-period.enum";
import { InvoiceStatus } from "../enums/invoice-status.enum";
import { InvoiceItemsRepository } from "../repositories/invoice-items.repository";
import { InvoicesRepository } from "../repositories/invoices.repository";
@Injectable()
export class InvoicesService {
private readonly logger = new Logger(InvoicesService.name);
constructor(
@InjectQueue(INVOICE.QUEUE_NAME) private readonly invoiceQueue: Queue,
private readonly notificationsService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
private readonly invoiceRepository: InvoicesRepository,
private readonly invoiceItemsRepository: InvoiceItemsRepository,
private readonly usersService: UsersService,
@@ -116,9 +115,7 @@ export class InvoicesService {
const user = await this.usersService.findOneByIdWithQueryRunner(createDto.userId, queryRunner);
await this.notificationsService.createInvoiceCreationNotification(
createDto.userId,
{
await this.notificationQueue.addInvoiceCreationNotification(createDto.userId, {
invoiceId: invoice.numericId.toString(),
dueDate: invoice.dueDate,
createDate: invoice.createdAt,
@@ -127,9 +124,7 @@ export class InvoicesService {
userEmail: user.email,
items: invoiceItems.map((item) => item.name).join(", "),
paidAt: invoice.paidAt,
},
queryRunner,
);
});
await this.scheduleInvoiceJobs(invoice, createDto);
await queryRunner.commitTransaction();
@@ -279,9 +274,7 @@ export class InvoicesService {
invoice.status = InvoiceStatus.WAIT_PAYMENT;
await queryRunner.manager.save(Invoice, invoice);
await this.notificationsService.createApprovedInvoiceNotification(
userId,
{
await this.notificationQueue.addApprovedInvoiceNotification(userId, {
invoiceId: invoice.numericId.toString(),
dueDate: invoice.dueDate,
createDate: invoice.createdAt,
@@ -289,9 +282,7 @@ export class InvoicesService {
userPhone: user.phone,
userEmail: user.email,
items: invoice.items.map((item) => item.name).join(", "),
},
queryRunner,
);
});
await this.otpService.delOtpFormCache(user.phone, "INVOICE_VERIFY");
@@ -365,9 +356,7 @@ export class InvoicesService {
await qryRnr.manager.save(Invoice, invoice);
await this.notificationsService.createInvoiceCreationNotification(
user.id,
{
await this.notificationQueue.addInvoiceCreationNotification(user.id, {
invoiceId: invoice.numericId.toString(),
dueDate: invoice.dueDate,
createDate: invoice.createdAt,
@@ -375,9 +364,7 @@ export class InvoicesService {
userPhone: user.phone,
userEmail: user.email,
items: plan.service.name,
},
qryRnr,
);
});
await this.invoiceQueue.add(
INVOICE.REMINDER_JOB_NAME,
@@ -507,9 +494,7 @@ export class InvoicesService {
await queryRunner.manager.save(Invoice, invoice);
await this.notificationsService.createBillInvoiceNotification(
userId,
{
await this.notificationQueue.addBillInvoiceNotification(userId, {
invoiceId: invoice.numericId.toString(),
dueDate: invoice.dueDate,
createDate: invoice.createdAt,
@@ -518,9 +503,7 @@ export class InvoicesService {
userEmail: user.email,
items: invoice.items.map((item) => item.name).join(", "),
paidAt: invoice.paidAt,
},
queryRunner,
);
});
await queryRunner.commitTransaction();
return {
src/modules/invoices/queue/invoice.processor.ts
+14 -33
View File
@@ -6,7 +6,7 @@ import { DataSource, QueryRunner } from "typeorm";
import { WorkerProcessor } from "../../../common/queues/worker.processor";
import { LoggerService } from "../../logger/logger.service";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { SubscriptionStatus } from "../../subscriptions/enums/subscription-status.enum";
import { User } from "../../users/entities/user.entity";
import { RoleEnum } from "../../users/enums/role.enum";
@@ -14,14 +14,13 @@ import { INVOICE } from "../constants";
import { Invoice } from "../entities/invoice.entity";
import { InvoiceStatus } from "../enums/invoice-status.enum";
import { InvoicesService } from "../providers/invoices.service";
@Processor(INVOICE.QUEUE_NAME, { concurrency: 5 })
@Processor(INVOICE.QUEUE_NAME, { concurrency: 2 })
export class InvoiceProcessor extends WorkerProcessor {
constructor(
@InjectQueue(INVOICE.QUEUE_NAME) private readonly invoiceQueue: Queue,
private readonly invoicesService: InvoicesService,
private readonly dataSource: DataSource,
private readonly notificationService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
protected readonly logger: LoggerService,
) {
super();
@@ -161,7 +160,7 @@ export class InvoiceProcessor extends WorkerProcessor {
await this.applyLateFeeIfOverdue(invoice, queryRunner);
await this.sendReminderNotification(invoice, queryRunner);
await this.sendReminderNotification(invoice);
await this.scheduleNextReminder(invoice.id);
@@ -204,16 +203,12 @@ export class InvoiceProcessor extends WorkerProcessor {
await queryRunner.manager.save(userSubscription);
await this.notificationService.createBlockServiceNotification(
invoice.user.id,
{
await this.notificationQueue.addBlockServiceNotification(invoice.user.id, {
userPhone: invoice.user.phone,
userEmail: invoice.user.email,
planName: userSubscription.plan.name,
invoiceId: invoice.numericId.toString(),
},
queryRunner,
);
});
}
this.logger.log(`Invoice ${invoice.id} has been cancelled as it is more than 10 days overdue.`);
}
@@ -253,9 +248,7 @@ export class InvoiceProcessor extends WorkerProcessor {
await queryRunner.manager.save(Invoice, invoice);
// Send overdue notification to the user
await this.notificationService.createInvoiceOverdueNotification(
invoice.user.id,
{
await this.notificationQueue.addInvoiceOverdueNotification(invoice.user.id, {
userPhone: invoice.user.phone,
userEmail: invoice.user.email,
invoiceId: invoice.numericId.toString(),
@@ -264,19 +257,15 @@ export class InvoiceProcessor extends WorkerProcessor {
dueDate: invoice.dueDate,
createDate: invoice.createdAt,
items: invoice.items.map((item) => item.name).join(", "),
},
queryRunner,
);
});
}
}
}
//********************************** */
private async sendReminderNotification(invoice: Invoice, queryRunner: QueryRunner) {
private async sendReminderNotification(invoice: Invoice) {
this.logger.log(`Sending invoice reminder to user: ${invoice.user.id}`);
await this.notificationService.createBillInvoiceReminderNotification(
invoice.user.id,
{
await this.notificationQueue.addBillInvoiceReminderNotification(invoice.user.id, {
userPhone: invoice.user.phone,
userEmail: invoice.user.email,
invoiceId: invoice.numericId.toString(),
@@ -284,9 +273,7 @@ export class InvoiceProcessor extends WorkerProcessor {
dueDate: invoice.dueDate,
createDate: invoice.createdAt,
items: invoice.items.map((item) => item.name).join(", "),
},
queryRunner,
);
});
}
//********************************** */
@@ -324,9 +311,7 @@ export class InvoiceProcessor extends WorkerProcessor {
items: invoice.items.map((item) => item.name).join(", "),
};
await Promise.all(
admins.map((admin) => this.notificationService.createRecurringInvoiceNotification(admin.id, notificationPayload, queryRunner)),
);
await Promise.all(admins.map((admin) => this.notificationQueue.addRecurringInvoiceNotification(admin.id, notificationPayload)));
}
//********************************** */
@@ -344,17 +329,13 @@ export class InvoiceProcessor extends WorkerProcessor {
const superAdmins = await this.fetchAdmin(queryRunner);
for (const admin of superAdmins) {
await this.notificationService.createNewSubscriptionNotification(
admin.id,
{
await this.notificationQueue.addNewSubscriptionNotification(admin.id, {
date: invoice.dueDate,
userPhone: admin.phone,
userEmail: admin.email,
serviceName: invoice.items[0]?.subscriptionPlan?.plan.service.name ?? "",
fullName: invoice.user.firstName + " " + invoice.user.lastName,
},
queryRunner,
);
});
}
await queryRunner.commitTransaction();
src/modules/notifications/constants/index.ts
+10
View File
@@ -0,0 +1,10 @@
export const NOTIFICATION = Object.freeze({
QUEUE_NAME: "notification",
SEND_NOTIFICATION_JOB_NAME: "sendNotification",
SEND_NOTIFICATION_JOB_DELAY: 1000, // delay after 1 second
SEND_NOTIFICATION_JOB_PRIORITY: 1, // high priority
SEND_NOTIFICATION_JOB_ATTEMPTS: 3, // retry 3 times
SEND_NOTIFICATION_JOB_BACKOFF: 5 * 1000, // retry after 5 seconds
SEND_NOTIFICATION_JOB_TIMEOUT: 10000, // timeout after 10 seconds
});
src/modules/notifications/notifications.module.ts
+33 -5
View File
@@ -1,17 +1,45 @@
import { BullModule } from "@nestjs/bullmq";
import { Module } from "@nestjs/common";
import { ConfigModule } from "@nestjs/config";
import { TypeOrmModule } from "@nestjs/typeorm";
import { NOTIFICATION } from "./constants";
import { LoggerModule } from "../logger/logger.module";
import { SettingModule } from "../settings/settings.module";
import { UtilsModule } from "../utils/utils.module";
import { Notification } from "./entities/notification.entity";
import { NotificationController } from "./notifications.controller";
import { NotificationsService } from "./providers/notifications.service";
import { NotificationProcessor } from "./queue/notification.processor";
import { NotificationQueue } from "./queue/notification.queue";
import { NotificationRepository } from "./repositories/notifications.repository";
import { SettingModule } from "../settings/settings.module";
import { UtilsModule } from "../utils/utils.module";
import { NotificationSetting } from "../settings/entities/notification-setting.entity";
import { EmailService } from "../utils/providers/email.service";
import { SmsService } from "../utils/providers/sms.service";
@Module({
imports: [TypeOrmModule.forFeature([Notification]), UtilsModule, SettingModule],
providers: [NotificationRepository, NotificationsService],
imports: [
ConfigModule,
LoggerModule,
TypeOrmModule.forFeature([Notification, NotificationSetting]),
BullModule.registerQueue({
name: NOTIFICATION.QUEUE_NAME,
defaultJobOptions: {
attempts: NOTIFICATION.SEND_NOTIFICATION_JOB_ATTEMPTS,
backoff: {
type: "exponential",
delay: NOTIFICATION.SEND_NOTIFICATION_JOB_BACKOFF,
},
removeOnComplete: true,
removeOnFail: false,
delay: NOTIFICATION.SEND_NOTIFICATION_JOB_DELAY,
},
}),
UtilsModule,
SettingModule,
],
providers: [NotificationRepository, NotificationsService, NotificationQueue, NotificationProcessor, SmsService, EmailService],
controllers: [NotificationController],
exports: [NotificationRepository, NotificationsService],
exports: [NotificationRepository, NotificationsService, NotificationQueue],
})
export class NotificationModule {}
src/modules/notifications/queue/notification.processor.ts
+173
View File
@@ -0,0 +1,173 @@
import { Processor } from "@nestjs/bullmq";
import { Logger } from "@nestjs/common";
import { Job } from "bullmq";
import { DataSource } from "typeorm";
import { WorkerProcessor } from "../../../common/queues/worker.processor";
import { LoggerService } from "../../logger/logger.service";
import { NotifType } from "../../settings/enums/notif-settings.enum";
import { NOTIFICATION } from "../constants";
import {
IAnnouncementNotificationData,
IBlogCommentNotificationData,
IInvoiceNotificationData,
INewCriticismNotificationData,
INewCustomerNotificationData,
INewSubscriptionNotificationData,
INewTicketNotificationData,
IPaymentNotificationData,
IServiceReviewNotificationData,
ISubscriptionNotificationData,
ITicketNotificationData,
IWalletNotificationData,
} from "../interfaces/ISendNotificationData";
import { NotificationsService } from "../providers/notifications.service";
type NotificationJobData = {
type: NotifType;
recipientId: string;
data:
| IBlogCommentNotificationData
| IServiceReviewNotificationData
| INewCustomerNotificationData
| INewSubscriptionNotificationData
| ITicketNotificationData
| INewCriticismNotificationData
| IInvoiceNotificationData
| IAnnouncementNotificationData
| IWalletNotificationData
| ISubscriptionNotificationData
| IPaymentNotificationData
| INewTicketNotificationData;
};
@Processor(NOTIFICATION.QUEUE_NAME, { concurrency: 5 })
export class NotificationProcessor extends WorkerProcessor {
protected readonly logger = new Logger(NotificationProcessor.name);
constructor(
private readonly notificationsService: NotificationsService,
private readonly dataSource: DataSource,
protected readonly loggerService: LoggerService,
) {
super();
}
async process(job: Job<NotificationJobData>, token?: string) {
this.logger.log(`Processing notification job: ${job.id} ${token ? `with token: ${token}` : ""}`);
const queryRunner = this.dataSource.createQueryRunner();
try {
await queryRunner.connect();
await queryRunner.startTransaction();
const { type, recipientId, data } = job.data;
switch (type) {
// Admin Notifications
case NotifType.NEW_BLOG_COMMENT:
await this.notificationsService.createNewBlogCommentNotification(recipientId, data as IBlogCommentNotificationData, queryRunner);
break;
case NotifType.NEW_SERVICE_REVIEW:
await this.notificationsService.createNewServiceReviewNotification(
recipientId,
data as IServiceReviewNotificationData,
queryRunner,
);
break;
case NotifType.NEW_CUSTOMER:
await this.notificationsService.createNewCustomerNotification(recipientId, data as INewCustomerNotificationData, queryRunner);
break;
case NotifType.NEW_SUBSCRIPTION:
await this.notificationsService.createNewSubscriptionNotification(
recipientId,
data as INewSubscriptionNotificationData,
queryRunner,
);
break;
case NotifType.NEW_CRITICISM:
await this.notificationsService.createNewCriticismNotification(recipientId, data as INewCriticismNotificationData, queryRunner);
break;
case NotifType.NEW_TICKET:
await this.notificationsService.createNewTicketGlobalNotification(recipientId, data as INewTicketNotificationData, queryRunner);
break;
// User Notifications
case NotifType.USER_LOGIN:
await this.notificationsService.createLoginNotification(recipientId, data);
break;
case NotifType.ANNOUNCEMENT:
await this.notificationsService.createAnnouncementNotification(recipientId, data as IAnnouncementNotificationData, queryRunner);
break;
// Wallet Notifications
case NotifType.WALLET_CHARGE:
await this.notificationsService.createWalletChargeNotification(recipientId, data as IWalletNotificationData, queryRunner);
break;
case NotifType.WALLET_DEDUCTION:
await this.notificationsService.createWalletDeductionNotification(recipientId, data as IWalletNotificationData, queryRunner);
break;
// Ticket Notifications
case NotifType.ANSWER_TICKET:
await this.notificationsService.createAnswerTicketNotification(recipientId, data as ITicketNotificationData, queryRunner);
break;
case NotifType.CREATE_TICKET:
await this.notificationsService.createTicketNotification(recipientId, data as ITicketNotificationData, queryRunner);
break;
case NotifType.ASSIGN_TICKET:
await this.notificationsService.createAssignTicketNotificationForAdmin(recipientId, data as ITicketNotificationData, queryRunner);
break;
// Invoice Notifications
case NotifType.CREATE_INVOICE:
await this.notificationsService.createInvoiceCreationNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
break;
case NotifType.BILL_INVOICE_REMINDER:
await this.notificationsService.createBillInvoiceReminderNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
break;
case NotifType.BILL_INVOICE:
await this.notificationsService.createBillInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
break;
case NotifType.APPROVE_INVOICE:
await this.notificationsService.createApprovedInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
break;
case NotifType.INVOICE_OVERDUE:
await this.notificationsService.createInvoiceOverdueNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
break;
case NotifType.RECURRING_INVOICE:
await this.notificationsService.createRecurringInvoiceNotification(recipientId, data as IInvoiceNotificationData, queryRunner);
break;
// Service Notifications
case NotifType.BLOCK_SERVICE:
await this.notificationsService.createBlockServiceNotification(recipientId, data as ISubscriptionNotificationData, queryRunner);
break;
// Payment Notifications
case NotifType.PAYMENT_REMINDER:
await this.notificationsService.createPaymentReminderNotification(recipientId, data as IPaymentNotificationData, queryRunner);
break;
case NotifType.PAYMENT_CANCELLATION:
await this.notificationsService.createPaymentCancellationNotification(recipientId, data as IPaymentNotificationData, queryRunner);
break;
default:
this.logger.warn(`Unknown notification type: ${type}`);
}
await queryRunner.commitTransaction();
return true;
} catch (error) {
this.logger.error(
`Failed to process notification: ${error instanceof Error ? error.message : "Unknown error"}`,
error instanceof Error ? error.stack : undefined,
);
await queryRunner.rollbackTransaction();
throw error;
} finally {
await queryRunner.release();
}
}
}
src/modules/notifications/queue/notification.queue.ts
+136
View File
@@ -0,0 +1,136 @@
import { InjectQueue } from "@nestjs/bullmq";
import { Injectable } from "@nestjs/common";
import { Queue } from "bullmq";
import { NotifType } from "../../settings/enums/notif-settings.enum";
import { NOTIFICATION } from "../constants";
import {
IAnnouncementNotificationData,
IBaseNotificationData,
IBlogCommentNotificationData,
IInvoiceNotificationData,
INewCriticismNotificationData,
INewCustomerNotificationData,
INewSubscriptionNotificationData,
INewTicketNotificationData,
IPaymentNotificationData,
IServiceReviewNotificationData,
ISubscriptionNotificationData,
ITicketNotificationData,
IWalletNotificationData,
} from "../interfaces/ISendNotificationData";
@Injectable()
export class NotificationQueue {
constructor(@InjectQueue(NOTIFICATION.QUEUE_NAME) private readonly notificationsQueue: Queue) {}
async addNotificationJob(type: NotifType, recipientId: string, data: IBaseNotificationData) {
await this.notificationsQueue.add(
NOTIFICATION.SEND_NOTIFICATION_JOB_NAME,
{
type,
recipientId,
data,
},
{
delay: NOTIFICATION.SEND_NOTIFICATION_JOB_DELAY,
attempts: NOTIFICATION.SEND_NOTIFICATION_JOB_ATTEMPTS,
backoff: {
type: "exponential",
delay: NOTIFICATION.SEND_NOTIFICATION_JOB_BACKOFF,
},
priority: NOTIFICATION.SEND_NOTIFICATION_JOB_PRIORITY,
},
);
}
// Admin notification methods
async addNewBlogCommentNotification(recipientId: string, data: IBlogCommentNotificationData) {
await this.addNotificationJob(NotifType.NEW_BLOG_COMMENT, recipientId, data);
}
async addNewServiceReviewNotification(recipientId: string, data: IServiceReviewNotificationData) {
await this.addNotificationJob(NotifType.NEW_SERVICE_REVIEW, recipientId, data);
}
async addNewCustomerNotification(recipientId: string, data: INewCustomerNotificationData) {
await this.addNotificationJob(NotifType.NEW_CUSTOMER, recipientId, data);
}
async addNewSubscriptionNotification(recipientId: string, data: INewSubscriptionNotificationData) {
await this.addNotificationJob(NotifType.NEW_SUBSCRIPTION, recipientId, data);
}
async addNewTicketGlobalNotification(recipientId: string, data: INewTicketNotificationData) {
await this.addNotificationJob(NotifType.NEW_TICKET, recipientId, data);
}
async addNewCriticismNotification(recipientId: string, data: INewCriticismNotificationData) {
await this.addNotificationJob(NotifType.NEW_CRITICISM, recipientId, data);
}
// User notification methods
async addLoginNotification(recipientId: string, data: IBaseNotificationData) {
await this.addNotificationJob(NotifType.USER_LOGIN, recipientId, data);
}
async addInvoiceCreationNotification(recipientId: string, data: IInvoiceNotificationData) {
await this.addNotificationJob(NotifType.CREATE_INVOICE, recipientId, data);
}
async addAnnouncementNotification(recipientId: string, data: IAnnouncementNotificationData) {
await this.addNotificationJob(NotifType.ANNOUNCEMENT, recipientId, data);
}
async addWalletChargeNotification(recipientId: string, data: IWalletNotificationData) {
await this.addNotificationJob(NotifType.WALLET_CHARGE, recipientId, data);
}
async addWalletDeductionNotification(recipientId: string, data: IWalletNotificationData) {
await this.addNotificationJob(NotifType.WALLET_DEDUCTION, recipientId, data);
}
async addAnswerTicketNotification(recipientId: string, data: ITicketNotificationData) {
await this.addNotificationJob(NotifType.ANSWER_TICKET, recipientId, data);
}
async addTicketNotification(recipientId: string, data: ITicketNotificationData) {
await this.addNotificationJob(NotifType.CREATE_TICKET, recipientId, data);
}
async addAssignTicketNotificationForAdmin(recipientId: string, data: ITicketNotificationData) {
await this.addNotificationJob(NotifType.ASSIGN_TICKET, recipientId, data);
}
async addBillInvoiceReminderNotification(recipientId: string, data: IInvoiceNotificationData) {
await this.addNotificationJob(NotifType.BILL_INVOICE_REMINDER, recipientId, data);
}
async addBillInvoiceNotification(recipientId: string, data: IInvoiceNotificationData) {
await this.addNotificationJob(NotifType.BILL_INVOICE, recipientId, data);
}
async addApprovedInvoiceNotification(recipientId: string, data: IInvoiceNotificationData) {
await this.addNotificationJob(NotifType.APPROVE_INVOICE, recipientId, data);
}
async addInvoiceOverdueNotification(recipientId: string, data: IInvoiceNotificationData) {
await this.addNotificationJob(NotifType.INVOICE_OVERDUE, recipientId, data);
}
async addRecurringInvoiceNotification(recipientId: string, data: IInvoiceNotificationData) {
await this.addNotificationJob(NotifType.RECURRING_INVOICE, recipientId, data);
}
async addBlockServiceNotification(recipientId: string, data: ISubscriptionNotificationData) {
await this.addNotificationJob(NotifType.BLOCK_SERVICE, recipientId, data);
}
async addPaymentReminderNotification(recipientId: string, data: IPaymentNotificationData) {
await this.addNotificationJob(NotifType.PAYMENT_REMINDER, recipientId, data);
}
async addPaymentCancellationNotification(recipientId: string, data: IPaymentNotificationData) {
await this.addNotificationJob(NotifType.PAYMENT_CANCELLATION, recipientId, data);
}
}
src/modules/payments/providers/payments.service.ts
+4 -8
View File
@@ -9,7 +9,7 @@ import { DataSource, Not, QueryRunner } from "typeorm";
import { PaginationDto } from "../../../common/DTO/pagination.dto";
import { CommonMessage, PaymentMessage, UserMessage, WalletMessage } from "../../../common/enums/message.enum";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { ReferralRewardStatus } from "../../referrals/enums/referral-reward-status.enum";
import { ReferralStatus } from "../../referrals/enums/referral-status.enum";
import { ReferralsService } from "../../referrals/providers/referrals.service";
@@ -43,7 +43,7 @@ export class PaymentsService {
private readonly logger = new Logger(PaymentsService.name);
constructor(
@InjectQueue(PAYMENT.QUEUE_NAME) private readonly paymentQueue: Queue,
private readonly notificationsService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
private readonly configService: ConfigService,
private readonly gatewayFactory: PaymentGatewayFactory,
private readonly paymentGatewaysRepository: PaymentGatewaysRepository,
@@ -185,18 +185,14 @@ export class PaymentsService {
const additionalParams: Record<string, string> = {};
//
await this.notificationsService.createWalletChargeNotification(
payment.user.id,
{
await this.notificationQueue.addWalletChargeNotification(payment.user.id, {
amount: new Decimal(payment.amount).toNumber(),
balance: new Decimal(wallet.balance).toNumber(),
date: payment.createdAt,
reason: WalletMessage.DEPOSIT_WALLET_IPG,
userPhone: payment.user.phone,
userEmail: payment.user.email,
},
queryRunner,
);
});
return {
status: PaymentStatus.COMPLETED,
src/modules/payments/queue/payment.processor.ts
+11 -20
View File
@@ -5,14 +5,13 @@ import Decimal from "decimal.js";
import { DataSource, QueryRunner } from "typeorm";
import { WorkerProcessor } from "../../../common/queues/worker.processor";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { PAYMENT } from "../constants";
import { Payment } from "../entities/payment.entity";
import { PaymentStatus } from "../enums/payment-status.enum";
import { PaymentGatewayFactory } from "../factories/payment.factory";
import { PaymentsService } from "../providers/payments.service";
import { PaymentJobData } from "../types/payment-job.type";
@Processor(PAYMENT.QUEUE_NAME)
export class PaymentProcessor extends WorkerProcessor {
protected readonly logger = new Logger(PaymentProcessor.name);
@@ -21,7 +20,7 @@ export class PaymentProcessor extends WorkerProcessor {
@InjectQueue(PAYMENT.QUEUE_NAME) private readonly paymentQueue: Queue,
private readonly gatewayFactory: PaymentGatewayFactory,
private readonly dataSource: DataSource,
private readonly notificationsService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
private readonly paymentsService: PaymentsService,
) {
super();
@@ -83,7 +82,7 @@ export class PaymentProcessor extends WorkerProcessor {
return payment;
}
await this.sendPaymentReminderNotification(payment, "first", queryRunner);
await this.sendPaymentReminderNotification(payment, "first");
await this.scheduleSecondReminder(payment.id);
return payment;
@@ -98,7 +97,7 @@ export class PaymentProcessor extends WorkerProcessor {
return payment;
}
await this.sendPaymentReminderNotification(payment, "second", queryRunner);
await this.sendPaymentReminderNotification(payment, "second");
await this.schedulePaymentCancellation(payment.id);
return payment;
@@ -162,7 +161,7 @@ export class PaymentProcessor extends WorkerProcessor {
private async cancelPayment(payment: Payment, queryRunner: QueryRunner) {
await queryRunner.manager.update(Payment, { id: payment.id }, { status: PaymentStatus.CANCELLED });
await this.sendPaymentCancellationNotification(payment, queryRunner);
await this.sendPaymentCancellationNotification(payment);
}
//===============================================
@@ -206,34 +205,26 @@ export class PaymentProcessor extends WorkerProcessor {
}
//===============================================
private async sendPaymentReminderNotification(payment: Payment, reminderType: "first" | "second", queryRunner: QueryRunner) {
private async sendPaymentReminderNotification(payment: Payment, reminderType: "first" | "second") {
this.logger.log(`Sending ${reminderType} payment reminder for payment ${payment.id}`);
await this.notificationsService.createPaymentReminderNotification(
payment.user.id,
{
await this.notificationQueue.addPaymentReminderNotification(payment.user.id, {
paymentId: payment.id,
amount: new Decimal(payment.amount).toNumber(),
userPhone: payment.user.phone,
userEmail: payment.user.email,
},
queryRunner,
);
});
}
//===============================================
private async sendPaymentCancellationNotification(payment: Payment, queryRunner: QueryRunner) {
private async sendPaymentCancellationNotification(payment: Payment) {
this.logger.log(`Sending payment cancellation notification for payment ${payment.id}`);
await this.notificationsService.createPaymentCancellationNotification(
payment.user.id,
{
await this.notificationQueue.addPaymentCancellationNotification(payment.user.id, {
paymentId: payment.id,
amount: new Decimal(payment.amount).toNumber(),
userPhone: payment.user.phone,
userEmail: payment.user.email,
},
queryRunner,
);
});
}
}
src/modules/tickets/providers/tickets.service.ts
+16 -36
View File
@@ -5,7 +5,7 @@ import { ReferralService } from "./referral.service";
import { ParamDto } from "../../../common/DTO/param.dto";
import { CommonMessage, TicketMessageEnum, UserMessage } from "../../../common/enums/message.enum";
import { DanakService } from "../../danak-services/entities/danak-service.entity";
import { NotificationsService } from "../../notifications/providers/notifications.service";
import { NotificationQueue } from "../../notifications/queue/notification.queue";
import { User } from "../../users/entities/user.entity";
import { AdminsService } from "../../users/providers/admins.service";
import { UsersService } from "../../users/providers/users.service";
@@ -28,7 +28,7 @@ import { TicketsRepository } from "../repositories/tickets.repository";
export class TicketsService {
private readonly logger = new Logger(TicketsService.name);
constructor(
private readonly notificationsService: NotificationsService,
private readonly notificationQueue: NotificationQueue,
private readonly ticketsCategoryRepository: TicketCategoryRepository,
private readonly ticketsRepository: TicketsRepository,
private readonly ticketMessagesRepository: TicketMessagesRepository,
@@ -219,7 +219,7 @@ export class TicketsService {
await queryRunner.manager.save(Ticket, ticket);
if (isAdmin) {
await this.sendNotificationForAdminResponse(queryRunner, ticket, user);
await this.sendNotificationForAdminResponse(ticket, user);
}
await queryRunner.commitTransaction();
@@ -295,18 +295,14 @@ export class TicketsService {
ticket.assignedTo = admin;
await queryRunner.manager.save(this.ticketsRepository.target, ticket);
await this.notificationsService.createAssignTicketNotificationForAdmin(
admin.id,
{
await this.notificationQueue.addAssignTicketNotificationForAdmin(admin.id, {
ticketId: ticket.numericId.toString(),
subject: ticket.subject,
date: ticket.createdAt,
userPhone: ticket.user.phone,
userEmail: ticket.user.email,
userPhone: admin.phone,
userEmail: admin.email,
user: `${ticket.user.firstName} ${ticket.user.lastName}`,
},
queryRunner,
);
});
await queryRunner.commitTransaction();
return {
@@ -345,30 +341,22 @@ export class TicketsService {
ticket.assignedTo = assignedUser;
await queryRunner.manager.save(Ticket, ticket);
await this.notificationsService.createAssignTicketNotificationForAdmin(
assignedUser.id,
{
await this.notificationQueue.addAssignTicketNotificationForAdmin(assignedUser.id, {
ticketId: ticket.numericId.toString(),
subject: ticket.subject,
date: ticket.createdAt,
userPhone: user.phone,
userEmail: user.email,
user: `${user.firstName} ${user.lastName}`,
},
queryRunner,
);
});
await this.notificationsService.createTicketNotification(
user.id,
{
await this.notificationQueue.addTicketNotification(user.id, {
ticketId: ticket.numericId.toString(),
subject: ticket.subject,
date: ticket.createdAt,
userPhone: user.phone,
userEmail: user.email,
},
queryRunner,
);
});
await this.notifySuperAdminsForTicket(queryRunner, ticket);
}
//***************************** */
@@ -409,18 +397,14 @@ export class TicketsService {
}
//***************************** */
private async sendNotificationForAdminResponse(queryRunner: QueryRunner, ticket: Ticket, admin: User) {
await this.notificationsService.createAnswerTicketNotification(
ticket.user.id,
{
private async sendNotificationForAdminResponse(ticket: Ticket, admin: User) {
await this.notificationQueue.addAnswerTicketNotification(ticket.user.id, {
ticketId: ticket.numericId.toString(),
subject: ticket.subject,
date: ticket.createdAt,
userPhone: admin.phone,
userEmail: admin.email,
},
queryRunner,
);
});
}
//***************************** */
@@ -428,17 +412,13 @@ export class TicketsService {
const superAdmins = await this.adminsService.getSuperAdmins(queryRunner);
for (const admin of superAdmins) {
await this.notificationsService.createNewTicketGlobalNotification(
admin.id,
{
await this.notificationQueue.addNewTicketGlobalNotification(admin.id, {
ticketId: ticket.numericId.toString(),
ticketSubject: ticket.subject,
userPhone: admin.phone,
userEmail: admin.email,
fullName: `${ticket.user.firstName} ${ticket.user.lastName}`,
},
queryRunner,
);
});
}
}
}
src/modules/utils/providers/email.service.ts
+1 -1
View File
@@ -169,7 +169,7 @@ export class EmailService {
await this.mailerService.sendMail(emailData);
} catch (error) {
this.logger.error("error in sending admin notification email", error);
throw new InternalServerErrorException("error in sending admin notification email");
// throw new InternalServerErrorException("error in sending admin notification email");
}
}
}