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

fix: notification service

Browse Source
This commit is contained in:
mahyargdz
2025-07-29 12:47:28 +03:30
parent 1a056a330e
commit 6b1a9dd7a9
8 changed files with 41 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/notifications/handlers/base-notification.handler.ts
-31
View File
@@ -1,31 +0,0 @@
import { EntityManager } from "@mikro-orm/postgresql";
import { Logger } from "@nestjs/common";
export abstract class BaseNotificationHandler<TData = unknown> {
protected abstract readonly logger: Logger;
/**
* Process the notification with automatic transaction management
*/
async processWithTransaction(recipientId: string, data: TData, em: EntityManager): Promise<void> {
try {
await this.handle(recipientId, data, em);
} catch (error) {
this.logger.error(
`Failed to process ${this.getHandlerName()} notification: ${error instanceof Error ? error.message : "Unknown error"}`,
error instanceof Error ? error.stack : undefined,
);
throw error;
}
}
/**
* Handle the specific notification logic - to be implemented by each handler
*/
protected abstract handle(recipientId: string, data: TData, em: EntityManager): Promise<void>;
/**
* Get the handler name for logging purposes
*/
protected abstract getHandlerName(): string;
}
src/modules/notifications/handlers/change-password.handler.ts
-23
View File
@@ -1,23 +0,0 @@
import { EntityManager } from "@mikro-orm/postgresql";
import { Injectable, Logger } from "@nestjs/common";
import { BaseNotificationHandler } from "./base-notification.handler";
import { IChangePasswordNotificationData } from "../interfaces/ISendNotificationData";
import { NotificationsService } from "../services/notifications.service";
@Injectable()
export class ChangePasswordHandler extends BaseNotificationHandler<IChangePasswordNotificationData> {
protected readonly logger = new Logger(ChangePasswordHandler.name);
constructor(private readonly notificationsService: NotificationsService) {
super();
}
protected async handle(recipientId: string, data: IChangePasswordNotificationData, em: EntityManager): Promise<void> {
await this.notificationsService.changePasswordNotification(recipientId, data, em);
}
protected getHandlerName(): string {
return "ChangePassword";
}
}
src/modules/notifications/handlers/new-email.handler.ts
-29
View File
@@ -1,29 +0,0 @@
import { EntityManager } from "@mikro-orm/postgresql";
import { Injectable, Logger } from "@nestjs/common";
import { BaseNotificationHandler } from "./base-notification.handler";
import { INewEmailNotificationData } from "../interfaces/ISendNotificationData";
import { NotificationsService } from "../services/notifications.service";
@Injectable()
export class NewEmailHandler extends BaseNotificationHandler<INewEmailNotificationData> {
protected readonly logger = new Logger(NewEmailHandler.name);
constructor(private readonly notificationsService: NotificationsService) {
super();
}
protected async handle(recipientId: string, data: INewEmailNotificationData, em: EntityManager): Promise<void> {
try {
await this.notificationsService.createNewEmailNotification(recipientId, data, em);
this.logger.log(`New email notifications (in-app + push) processed for user: ${recipientId}`);
} catch (error) {
this.logger.error(`Error processing new email notifications for user ${recipientId}:`, error);
throw error;
}
}
protected getHandlerName(): string {
return "NewEmail";
}
}
src/modules/notifications/handlers/notification-handler.factory.ts
-47
View File
@@ -1,47 +0,0 @@
import { EntityManager } from "@mikro-orm/postgresql";
import { Injectable, Logger } from "@nestjs/common";
import { BaseNotificationHandler } from "./base-notification.handler";
import { ChangePasswordHandler } from "./change-password.handler";
import { NewEmailHandler } from "./new-email.handler";
import { NotifType } from "../../settings/enums/notif-settings.enum";
@Injectable()
export class NotificationHandlerFactory {
private readonly logger = new Logger(NotificationHandlerFactory.name);
private readonly handlers: Map<NotifType, BaseNotificationHandler> = new Map();
constructor(
private readonly newEmailHandler: NewEmailHandler,
private readonly changePasswordHandler: ChangePasswordHandler,
) {
this.registerHandlers();
}
private registerHandlers(): void {
// User notifications
this.handlers.set(NotifType.NEW_EMAIL, this.newEmailHandler);
this.handlers.set(NotifType.CHANGE_PASSWORD, this.changePasswordHandler);
this.logger.log(`Registered ${this.handlers.size} notification handlers`);
}
async processNotification(type: NotifType, recipientId: string, data: unknown, em: EntityManager): Promise<void> {
const handler = this.handlers.get(type);
if (!handler) {
this.logger.warn(`No handler found for notification type: ${type}`);
throw new Error(`No handler registered for notification type: ${type}`);
}
await handler.processWithTransaction(recipientId, data, em);
}
getRegisteredTypes(): NotifType[] {
return Array.from(this.handlers.keys());
}
isTypeSupported(type: NotifType): boolean {
return this.handlers.has(type);
}
}
src/modules/notifications/handlers/user-login.handler.ts
-24
View File
@@ -1,24 +0,0 @@
import { Injectable, Logger } from "@nestjs/common";
import { IBaseNotificationData } from "../interfaces/ISendNotificationData";
import { NotificationsService } from "../services/notifications.service";
@Injectable()
export class UserLoginHandler {
private readonly logger = new Logger(UserLoginHandler.name);
constructor(private readonly notificationsService: NotificationsService) {}
async processUserLogin(recipientId: string, data: IBaseNotificationData): Promise<void> {
try {
await this.notificationsService.createLoginNotification(recipientId, data);
this.logger.log(`User login notification sent successfully to ${recipientId}`);
} catch (error) {
this.logger.error(
`Failed to send user login notification: ${error instanceof Error ? error.message : "Unknown error"}`,
error instanceof Error ? error.stack : undefined,
);
throw error;
}
}
}
src/modules/notifications/notifications.module.ts
+2 -28
View File
@@ -5,10 +5,6 @@ import { Module } from "@nestjs/common";
import { NOTIFICATION } from "./constants"; import { NOTIFICATION } from "./constants";
import { Notification } from "./entities/notification.entity"; import { Notification } from "./entities/notification.entity";
import { ChangePasswordHandler } from "./handlers/change-password.handler";
import { NewEmailHandler } from "./handlers/new-email.handler";
import { NotificationHandlerFactory } from "./handlers/notification-handler.factory";
import { UserLoginHandler } from "./handlers/user-login.handler";
import { NotificationController } from "./notifications.controller"; import { NotificationController } from "./notifications.controller";
import { NotificationProcessor } from "./queue/notification.processor"; import { NotificationProcessor } from "./queue/notification.processor";
import { NotificationQueue } from "./queue/notification.queue"; import { NotificationQueue } from "./queue/notification.queue";
@@ -24,33 +20,11 @@ import { UtilsModule } from "../utils/utils.module";
imports: [ imports: [
HttpModule, HttpModule,
MikroOrmModule.forFeature([Notification, NotificationSetting, User]), MikroOrmModule.forFeature([Notification, NotificationSetting, User]),
BullModule.registerQueue({ BullModule.registerQueue({ name: NOTIFICATION.QUEUE_NAME }),
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, UtilsModule,
SettingModule, SettingModule,
], ],
providers: [ providers: [NotificationsService, NajvaPushService, NotificationQueue, NotificationProcessor, najvaConfigProvider],
NotificationsService,
NajvaPushService,
NotificationQueue,
NotificationProcessor,
NotificationHandlerFactory,
UserLoginHandler,
NewEmailHandler,
ChangePasswordHandler,
najvaConfigProvider,
],
controllers: [NotificationController], controllers: [NotificationController],
exports: [NotificationsService, NotificationQueue, NajvaPushService], exports: [NotificationsService, NotificationQueue, NajvaPushService],
}) })
src/modules/notifications/queue/notification.processor.ts
+25 -21
View File
@@ -1,4 +1,3 @@
import { EntityManager } from "@mikro-orm/postgresql";
import { Processor } from "@nestjs/bullmq"; import { Processor } from "@nestjs/bullmq";
import { Logger } from "@nestjs/common"; import { Logger } from "@nestjs/common";
import { Job } from "bullmq"; import { Job } from "bullmq";
@@ -6,10 +5,9 @@ import { Job } from "bullmq";
import { WorkerProcessor } from "../../../common/queues/worker.processor"; import { WorkerProcessor } from "../../../common/queues/worker.processor";
import { NotifType } from "../../settings/enums/notif-settings.enum"; import { NotifType } from "../../settings/enums/notif-settings.enum";
import { NOTIFICATION } from "../constants"; import { NOTIFICATION } from "../constants";
import { NotificationHandlerFactory } from "../handlers/notification-handler.factory";
import { UserLoginHandler } from "../handlers/user-login.handler";
import { NotificationJobData, OtpJobData } from "../interfaces/INotification-job-data"; import { NotificationJobData, OtpJobData } from "../interfaces/INotification-job-data";
import { IBaseNotificationData } from "../interfaces/ISendNotificationData"; import { IBaseNotificationData, IChangePasswordNotificationData, INewEmailNotificationData } from "../interfaces/ISendNotificationData";
import { NotificationsService } from "../services/notifications.service";
@Processor(NOTIFICATION.QUEUE_NAME, { @Processor(NOTIFICATION.QUEUE_NAME, {
concurrency: NOTIFICATION.SEND_NOTIFICATION_JOB_CONCURRENCY, concurrency: NOTIFICATION.SEND_NOTIFICATION_JOB_CONCURRENCY,
@@ -20,11 +18,7 @@ import { IBaseNotificationData } from "../interfaces/ISendNotificationData";
export class NotificationProcessor extends WorkerProcessor { export class NotificationProcessor extends WorkerProcessor {
protected readonly logger = new Logger(NotificationProcessor.name); protected readonly logger = new Logger(NotificationProcessor.name);
constructor( constructor(private readonly notificationsService: NotificationsService) {
private readonly em: EntityManager,
private readonly notificationHandlerFactory: NotificationHandlerFactory,
private readonly userLoginHandler: UserLoginHandler,
) {
super(); super();
} }
@@ -34,12 +28,7 @@ export class NotificationProcessor extends WorkerProcessor {
try { try {
if (job.name === NOTIFICATION.SEND_NOTIFICATION_JOB_NAME) { if (job.name === NOTIFICATION.SEND_NOTIFICATION_JOB_NAME) {
const { type, recipientId, data } = job.data as NotificationJobData; const { type, recipientId, data } = job.data as NotificationJobData;
await this.processNotification(type, recipientId, data);
if (type === NotifType.USER_LOGIN) {
await this.userLoginHandler.processUserLogin(recipientId, data as IBaseNotificationData);
} else {
await this.processWithTransaction(type, recipientId, data);
}
} else { } else {
this.logger.warn(`Unknown job name: ${job.name}`); this.logger.warn(`Unknown job name: ${job.name}`);
} }
@@ -54,17 +43,32 @@ export class NotificationProcessor extends WorkerProcessor {
} }
} }
private async processWithTransaction(type: NotifType, recipientId: string, data: IBaseNotificationData): Promise<void> { //*********************************** */
const em = this.em.fork();
private async processNotification(type: NotifType, recipientId: string, data: IBaseNotificationData): Promise<void> {
try { try {
await em.begin(); this.logger.log(`Processing ${type} notification for user: ${recipientId}`);
await this.notificationHandlerFactory.processNotification(type, recipientId, data, em); switch (type) {
case NotifType.NEW_EMAIL:
await this.notificationsService.createNewEmailNotification(recipientId, data as INewEmailNotificationData);
break;
await em.commit(); case NotifType.CHANGE_PASSWORD:
await this.notificationsService.changePasswordNotification(recipientId, data as IChangePasswordNotificationData);
break;
case NotifType.USER_LOGIN:
await this.notificationsService.createLoginNotification(recipientId, data);
return;
default:
throw new Error(`Unsupported notification type: ${type}`);
}
this.logger.log(`Successfully processed ${type} notification for user: ${recipientId}`);
} catch (error) { } catch (error) {
await em.rollback(); this.logger.error(`Failed to process ${type} notification for user ${recipientId}:`, error);
throw error; throw error;
} }
} }
src/modules/notifications/services/notifications.service.ts
+14 -57
View File
@@ -29,53 +29,16 @@ export class NotificationsService {
//************************ */ //************************ */
async createNotification(createNotificationDto: CreateNotificationDto, em?: EntityManager) { async createNotification(createNotificationDto: CreateNotificationDto) {
if (!em) { const em = this.em.fork();
const localEm = this.em.fork();
await localEm.begin(); try {
try { const user = await em.findOne(User, { id: createNotificationDto.recipientId });
//
const user = await localEm.findOne(User, { id: createNotificationDto.recipientId });
if (!user) throw new NotFoundException(UserMessage.USER_NOT_FOUND);
const notification = localEm.create(Notification, {
...createNotificationDto,
recipient: user,
});
await localEm.persistAndFlush(notification);
// Send push notification to all user devices if user has push tokens
if (user.pushTokens && user.pushTokens.length > 0) {
await this.sendPushNotificationToUserDevices(user.id, user.pushTokens, {
title: createNotificationDto.title,
body: createNotificationDto.message,
notification_click: {
click_url: `${this.configService.getOrThrow("FRONTEND_URL")}/notifications`,
},
});
}
await localEm.commit();
return notification;
//
} catch (error) {
this.logger.error("error in createNotification", error);
await localEm.rollback();
throw error;
//
}
//
} else {
//
const user = await em.findOne(User, { id: createNotificationDto.recipientId, deletedAt: null });
if (!user) throw new NotFoundException(UserMessage.USER_NOT_FOUND); if (!user) throw new NotFoundException(UserMessage.USER_NOT_FOUND);
const notification = em.create(Notification, { const notification = em.create(Notification, {
...createNotificationDto, ...createNotificationDto,
recipient: user, recipient: user,
}); });
await em.persistAndFlush(notification); await em.persistAndFlush(notification);
// Send push notification to all user devices if user has push tokens // Send push notification to all user devices if user has push tokens
@@ -89,24 +52,18 @@ export class NotificationsService {
}); });
} }
await em.commit();
return notification; return notification;
} catch (error) {
this.logger.error("error in createNotification", error);
await em.rollback();
throw error;
} }
} }
//=============================================== //===============================================
async sendPushNotificationToUser(userToken: string, message: INajvaNotificationMessage) {
return this.najvaPushService.sendNotificationToUser(userToken, message);
}
//===============================================
async sendPushNotificationToMultipleUsers(userTokens: string[], message: INajvaNotificationMessage) {
return this.najvaPushService.sendNotificationToMultipleUsers(userTokens, message);
}
//===============================================
async sendPushNotificationToUserDevices(userId: string, userTokens: string[], message: INajvaNotificationMessage) { async sendPushNotificationToUserDevices(userId: string, userTokens: string[], message: INajvaNotificationMessage) {
try { try {
const result = await this.najvaPushService.sendNotificationToMultipleUsers(userTokens, message); const result = await this.najvaPushService.sendNotificationToMultipleUsers(userTokens, message);
@@ -184,20 +141,20 @@ export class NotificationsService {
} }
//=============================================== //===============================================
async createNewEmailNotification(recipientId: string, data: INewEmailNotificationData, em: EntityManager) { async createNewEmailNotification(recipientId: string, data: INewEmailNotificationData) {
const message = NotificationMessage.NEW_EMAIL_MESSAGE.replace("[subject]", data.subject); const message = NotificationMessage.NEW_EMAIL_MESSAGE.replace("[subject]", data.subject);
const { isActive } = await this.userSettingsService.getUserNotificationSettingWithType(NotifType.NEW_EMAIL, recipientId); const { isActive } = await this.userSettingsService.getUserNotificationSettingWithType(NotifType.NEW_EMAIL, recipientId);
if (isActive) { if (isActive) {
return this.createNotification({ title: NotificationMessage.NEW_EMAIL, type: NotifType.NEW_EMAIL, message, recipientId }, em); return this.createNotification({ title: NotificationMessage.NEW_EMAIL, type: NotifType.NEW_EMAIL, message, recipientId });
} }
} }
//=============================================== //===============================================
async changePasswordNotification(recipientId: string, _data: IChangePasswordNotificationData, em: EntityManager) { async changePasswordNotification(recipientId: string, _data: IChangePasswordNotificationData) {
const message = NotificationMessage.CHANGE_PASSWORD_MESSAGE; const message = NotificationMessage.CHANGE_PASSWORD_MESSAGE;
const { isActive } = await this.userSettingsService.getUserNotificationSettingWithType(NotifType.CHANGE_PASSWORD, recipientId); const { isActive } = await this.userSettingsService.getUserNotificationSettingWithType(NotifType.CHANGE_PASSWORD, recipientId);
if (isActive) { if (isActive) {
return this.createNotification({ title: NotificationMessage.CHANGE_PASSWORD, type: NotifType.CHANGE_PASSWORD, message, recipientId }, em); return this.createNotification({ title: NotificationMessage.CHANGE_PASSWORD, type: NotifType.CHANGE_PASSWORD, message, recipientId });
} }
} }