From e921b00da28a2a1dac3aabf631a2ff50de165026 Mon Sep 17 00:00:00 2001 From: mahyargdz Date: Sat, 26 Jul 2025 15:40:48 +0330 Subject: [PATCH] chore: add bulk action processor --- src/common/enums/message.enum.ts | 3 + .../email/constants/email-events.constant.ts | 11 + src/modules/email/email.controller.ts | 26 +++ src/modules/email/email.module.ts | 17 +- .../email/interfaces/email.interface.ts | 4 + .../queue/email-bulk-action.processor.ts | 192 ++++++++++++++++++ src/modules/email/services/email.service.ts | 56 +++++ 7 files changed, 306 insertions(+), 3 deletions(-) create mode 100644 src/modules/email/interfaces/email.interface.ts create mode 100644 src/modules/email/queue/email-bulk-action.processor.ts diff --git a/src/common/enums/message.enum.ts b/src/common/enums/message.enum.ts index e2a0ba1..d3eba93 100755 --- a/src/common/enums/message.enum.ts +++ b/src/common/enums/message.enum.ts @@ -225,6 +225,7 @@ export const enum EmailMessage { // Message operations MESSAGE_MARKED_AS_SEEN_SUCCESSFULLY = "پیام با موفقیت به عنوان خوانده شده علامت گذاری شد", + ALL_MESSAGES_MARKED_AS_READ_SUCCESSFULLY = "تمام پیام ها با موفقیت به عنوان خوانده شده علامت گذاری شدند", MESSAGE_MARK_AS_SEEN_FAILED = "علامت گذاری پیام با خطا مواجه شد", SENT_MESSAGES_RETRIEVED_SUCCESSFULLY = "پیام های ارسالی با موفقیت دریافت شد", SENT_MESSAGES_RETRIEVAL_FAILED = "دریافت پیام های ارسالی با خطا مواجه شد", @@ -273,6 +274,8 @@ export const enum EmailMessage { // Favorite/Star messages (Gmail-like behavior) MESSAGE_MARKED_AS_FAVORITE_SUCCESSFULLY = "پیام با موفقیت به عنوان مورد علاقه علامت گذاری شد", MESSAGE_REMOVED_FROM_FAVORITE_SUCCESSFULLY = "پیام با موفقیت از مورد علاقه حذف شد", + TRASH_EMPTYED_SUCCESSFULLY = "سطل زباله با موفقیت خالی شد", + SPAM_EMPTYED_SUCCESSFULLY = "همه پیام های اسپم با موفقیت حذف شدند", } export const enum WebSocketMessage { diff --git a/src/modules/email/constants/email-events.constant.ts b/src/modules/email/constants/email-events.constant.ts index 69379a2..673fc64 100644 --- a/src/modules/email/constants/email-events.constant.ts +++ b/src/modules/email/constants/email-events.constant.ts @@ -39,6 +39,17 @@ export type WebSocketNamespace = (typeof WEBSOCKET_NAMESPACES)[keyof typeof WEBS export const EMAIL_QUEUE_CONSTANTS = { EMAIL_QUEUE: "email_queue", + EMAIL_BULK_ACTION_QUEUE: "email_bulk_action_queue", + EMAIL_BULK_ACTION_SEEN_ALL: "email_bulk_action_seen_all", + EMAIL_BULK_ACTION_EMPTY_TRASH: "email_bulk_action_empty_trash", + EMAIL_BULK_ACTION_EMPTY_SPAM: "email_bulk_action_empty_spam", + // + EMAIL_BULK_ACTION_DELAY: 1000, + EMAIL_BULK_ACTION_BATCH_SIZE: 1000, + EMAIL_BULK_ACTION_ATTEMPTS: 3, + EMAIL_BULK_ACTION_BACKOFF_DELAY: 2000, // 2 seconds + EMAIL_BULK_ACTION_PRIORITY: 1, + // EMAIL_QUEUE_PROCESSING: "email_queue_processing", EMAIL_SCHEDULER_JOB: "email_monitoring_scheduler", EMAIL_MONITORING_ACTION_NAME: "schedule_email_monitoring", diff --git a/src/modules/email/email.controller.ts b/src/modules/email/email.controller.ts index 830b661..978295c 100644 --- a/src/modules/email/email.controller.ts +++ b/src/modules/email/email.controller.ts @@ -96,6 +96,32 @@ export class EmailController { return this.emailService.markMessageAsUnseen(userId, Number(params.messageId), query.mailbox); } + @Patch("messages/mark-all-read") + @ApiOperation({ summary: "Mark all messages as read in a mailbox" }) + @ApiResponse({ status: 200, description: "All messages marked as read successfully" }) + @ApiResponse({ status: 206, description: "Some messages marked as read, some failed" }) + @ApiResponse({ status: 404, description: "Mailbox not found" }) + @ApiResponse({ status: 403, description: "Not authorized to mark messages as read" }) + markAllMessagesAsRead(@UserDec("wildduckUserId") userId: string, @Query() query: MailboxIdQueryDto) { + return this.emailService.markAllMessagesAsRead(userId, query.mailbox); + } + + @Patch("messages/empty-trash") + @ApiOperation({ summary: "Empty trash" }) + @ApiResponse({ status: 200, description: "Trash emptied successfully" }) + @ApiResponse({ status: 404, description: "Trash not found" }) + emptyTrash(@UserDec("wildduckUserId") userId: string, @Query() query: MailboxIdQueryDto) { + return this.emailService.emptyTrash(userId, query.mailbox); + } + + @Patch("messages/empty-spam") + @ApiOperation({ summary: "Empty spam" }) + @ApiResponse({ status: 200, description: "Spam emptied successfully" }) + @ApiResponse({ status: 404, description: "Spam not found" }) + emptySpam(@UserDec("wildduckUserId") userId: string, @Query() query: MailboxIdQueryDto) { + return this.emailService.emptySpam(userId, query.mailbox); + } + @Get("messages/inbox") @ApiOperation({ summary: "List messages in a inbox" }) @ApiResponse({ status: 200, description: "List of messages" }) diff --git a/src/modules/email/email.module.ts b/src/modules/email/email.module.ts index 2bdfa4e..450fa96 100644 --- a/src/modules/email/email.module.ts +++ b/src/modules/email/email.module.ts @@ -1,4 +1,5 @@ import { MikroOrmModule } from "@mikro-orm/nestjs"; +import { BullModule } from "@nestjs/bullmq"; import { Module } from "@nestjs/common"; import { EmailController } from "./email.controller"; @@ -8,12 +9,22 @@ import { EmailService } from "./services/email.service"; import { EmailUtilsModule } from "../email-utils/email-utils.module"; import { MailServerModule } from "../mail-server/mail-server.module"; import { TemplatesModule } from "../templates/templates.module"; +import { EMAIL_QUEUE_CONSTANTS } from "./constants/email-events.constant"; +import { EmailBulkActionProcessor } from "./queue/email-bulk-action.processor"; import { User } from "../users/entities/user.entity"; @Module({ - imports: [MailServerModule, TemplatesModule, EmailUtilsModule, MikroOrmModule.forFeature([User])], + imports: [ + BullModule.registerQueue({ + name: EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_QUEUE, + }), + MailServerModule, + TemplatesModule, + EmailUtilsModule, + MikroOrmModule.forFeature([User]), + ], controllers: [EmailController], - providers: [EmailService, EmailSpamService, EmailHeadersService], - exports: [EmailService, EmailSpamService, EmailHeadersService], + providers: [EmailService, EmailSpamService, EmailHeadersService, EmailBulkActionProcessor], + exports: [EmailService, EmailSpamService, EmailHeadersService, EmailBulkActionProcessor], }) export class EmailModule {} diff --git a/src/modules/email/interfaces/email.interface.ts b/src/modules/email/interfaces/email.interface.ts new file mode 100644 index 0000000..c55494d --- /dev/null +++ b/src/modules/email/interfaces/email.interface.ts @@ -0,0 +1,4 @@ +export interface IEmailBulkActionJob { + wildduckUserId: string; + mailboxId: string; +} diff --git a/src/modules/email/queue/email-bulk-action.processor.ts b/src/modules/email/queue/email-bulk-action.processor.ts new file mode 100644 index 0000000..332a2a5 --- /dev/null +++ b/src/modules/email/queue/email-bulk-action.processor.ts @@ -0,0 +1,192 @@ +import { Processor, WorkerHost } from "@nestjs/bullmq"; +import { Logger } from "@nestjs/common"; +import { Job } from "bullmq"; +import { firstValueFrom } from "rxjs"; + +import { EmailMessage } from "../../../common/enums/message.enum"; +import { MailServerService } from "../../mail-server/services/mail-server.service"; +import { EMAIL_QUEUE_CONSTANTS } from "../constants/email-events.constant"; +import { IEmailBulkActionJob } from "../interfaces/email.interface"; +import { EmailService } from "../services/email.service"; + +@Processor(EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_QUEUE) +export class EmailBulkActionProcessor extends WorkerHost { + private readonly logger = new Logger(EmailBulkActionProcessor.name); + + constructor( + private readonly mailServerService: MailServerService, + private readonly emailService: EmailService, + ) { + super(); + } + + async process(job: Job) { + const { name, data } = job; + + try { + switch (name) { + case EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_SEEN_ALL: + this.logger.debug(`Processing bulk action for: ${data.wildduckUserId}`); + await this.markAllMessagesAsRead(data.wildduckUserId, data.mailboxId); + break; + + case EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_EMPTY_TRASH: + this.logger.debug(`Processing bulk action for: ${data.wildduckUserId}`); + await this.emptyTrash(data.wildduckUserId, data.mailboxId); + break; + + case EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_EMPTY_SPAM: + this.logger.debug(`Processing bulk action for: ${data.wildduckUserId}`); + await this.emptySpam(data.wildduckUserId, data.mailboxId); + break; + + default: + this.logger.warn(`Unknown job type: ${name}`); + throw new Error(`Unknown job type: ${name}`); + } + } catch (error) { + this.logger.error(`Failed to process bulk action: ${error}`); + throw error; + } + } + + //============================================== + private async markAllMessagesAsRead(wildduckUserId: string, mailboxId: string) { + this.logger.log(`Marking all messages as read for user ${wildduckUserId} in mailbox ${mailboxId}`); + + try { + // Get all unread messages from the mailbox with a reasonable limit + const limit = EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_BATCH_SIZE; // Process in batches to avoid memory issues + let totalMarked = 0; + let hasMore = true; + let next: string | undefined; + + const results = { + successful: [] as number[], + failed: [] as { messageId: number; error: string }[], + totalMarked: 0, + }; + + while (hasMore) { + // Fetch unread messages in batches + const query: Record = { + limit, + order: "desc", + unseen: true, // Only get unread messages + }; + + if (next) { + query.next = next; + } + + const response = await firstValueFrom(this.mailServerService.messages.listMessages(wildduckUserId, mailboxId, query)); + + if (!response.results || response.results.length === 0) { + hasMore = false; + break; + } + + // Mark each message as seen + for (const message of response.results) { + try { + await this.emailService.markMessageAsSeen(wildduckUserId, message.id, mailboxId); + results.successful.push(message.id); + totalMarked++; + } catch (error) { + this.logger.error(`Failed to mark message ${message.id} as read: ${error}`); + results.failed.push({ + messageId: message.id, + error: error instanceof Error ? error.message : "Unknown error", + }); + } + } + + // Check if there are more messages + next = typeof response.nextCursor === "string" ? response.nextCursor : undefined; + hasMore = !!next && response.results.length === limit; + } + + results.totalMarked = totalMarked; + + this.logger.log(`Marked ${totalMarked} messages as read for user ${wildduckUserId} in mailbox ${mailboxId}, ${results.failed.length} failed`); + + return { + success: true, + message: EmailMessage.ALL_MESSAGES_MARKED_AS_READ_SUCCESSFULLY, + ...results, + }; + } catch (error) { + this.logger.error(`Failed to mark all messages as read for user ${wildduckUserId} in mailbox ${mailboxId}:`, error); + throw error; + } + } + + //============================================== + + private async emptyTrash(wildduckUserId: string, mailboxId: string) { + this.logger.log(`Emptying trash for user ${wildduckUserId} in mailbox ${mailboxId}`); + + try { + const response = await firstValueFrom( + this.mailServerService.messages.listMessages(wildduckUserId, mailboxId, { + limit: 1000, + }), + ); + + if (!response.results || response.results.length === 0) { + this.logger.log(`No messages found in trash for user ${wildduckUserId} in mailbox ${mailboxId}`); + return; + } + + const { result } = await this.emailService.deleteAllMessages(wildduckUserId, mailboxId); + + if (!result.success) { + this.logger.error(`Failed to empty trash for user ${wildduckUserId} in mailbox ${mailboxId}:`, result); + throw new Error(result.success); + } + + return { + success: true, + message: EmailMessage.TRASH_EMPTYED_SUCCESSFULLY, + result, + }; + } catch (error) { + this.logger.error(`Failed to empty trash for user ${wildduckUserId} in mailbox ${mailboxId}:`, error); + throw error; + } + } + + //============================================== + private async emptySpam(wildduckUserId: string, mailboxId: string) { + this.logger.log(`Emptying spam for user ${wildduckUserId} in mailbox ${mailboxId}`); + + try { + const response = await firstValueFrom( + this.mailServerService.messages.listMessages(wildduckUserId, mailboxId, { + limit: 1000, + }), + ); + + if (!response.results || response.results.length === 0) { + this.logger.log(`No messages found in spam for user ${wildduckUserId} in mailbox ${mailboxId}`); + return; + } + + const { result } = await this.emailService.deleteAllMessages(wildduckUserId, mailboxId); + + if (!result.success) { + this.logger.error(`Failed to empty spam for user ${wildduckUserId} in mailbox ${mailboxId}:`, result); + throw new Error(result.success); + } + + return { + success: true, + message: EmailMessage.SPAM_EMPTYED_SUCCESSFULLY, + result, + }; + } catch (error) { + this.logger.error(`Failed to empty spam for user ${wildduckUserId} in mailbox ${mailboxId}:`, error); + throw error; + } + } +} diff --git a/src/modules/email/services/email.service.ts b/src/modules/email/services/email.service.ts index bfdb3f2..be8e678 100644 --- a/src/modules/email/services/email.service.ts +++ b/src/modules/email/services/email.service.ts @@ -1,4 +1,6 @@ +import { InjectQueue } from "@nestjs/bullmq"; import { BadRequestException, Injectable, Logger } from "@nestjs/common"; +import { Queue } from "bullmq"; import { FastifyReply } from "fastify"; import { firstValueFrom } from "rxjs"; @@ -12,6 +14,7 @@ import { MailServerService } from "../../mail-server/services/mail-server.servic import { TemplateProcessorService } from "../../templates/services/template-processor.service"; import { User } from "../../users/entities/user.entity"; import { UserRepository } from "../../users/repositories/user.repository"; +import { EMAIL_QUEUE_CONSTANTS } from "../constants/email-events.constant"; import { BulkActionDto, BulkActionType } from "../DTO/bulk-actions.dto"; import { MessageListQueryDto, SearchMessagesQueryDto } from "../DTO/email-query.dto"; import { EmailRecipientDto, EmailType, SendEmailDto, UpdateDraftDto } from "../DTO/send-email.dto"; @@ -23,6 +26,7 @@ export class EmailService { private readonly logger = new Logger(EmailService.name); constructor( + @InjectQueue(EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_QUEUE) private readonly emailBulkActionQueue: Queue, private readonly mailServerService: MailServerService, private readonly mailboxResolverService: MailboxResolverService, private readonly emailNotificationService: EmailNotificationService, @@ -246,6 +250,58 @@ export class EmailService { updated, }; } + + //============================================== + async markAllMessagesAsRead(wildduckUserId: string, mailboxId: string) { + await this.emailBulkActionQueue.add( + EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_SEEN_ALL, + { + wildduckUserId, + mailboxId, + }, + { + delay: EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_DELAY, + priority: EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_PRIORITY, + attempts: EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_ATTEMPTS, + backoff: { + type: "exponential", + delay: EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_BACKOFF_DELAY, + }, + }, + ); + + return { + success: true, + message: EmailMessage.ALL_MESSAGES_MARKED_AS_READ_SUCCESSFULLY, + }; + } + + //============================================== + async emptyTrash(wildduckUserId: string, mailboxId: string) { + await this.emailBulkActionQueue.add(EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_EMPTY_TRASH, { + wildduckUserId, + mailboxId, + }); + + return { + success: true, + message: EmailMessage.TRASH_EMPTYED_SUCCESSFULLY, + }; + } + + //============================================== + async emptySpam(wildduckUserId: string, mailboxId: string) { + await this.emailBulkActionQueue.add(EMAIL_QUEUE_CONSTANTS.EMAIL_BULK_ACTION_EMPTY_SPAM, { + wildduckUserId, + mailboxId, + }); + + return { + success: true, + message: EmailMessage.SPAM_EMPTYED_SUCCESSFULLY, + }; + } + //============================================== async getSentMessages(wildduckUserId: string, query: MessageListQueryDto) { const sentMailboxId = await this.mailboxResolverService.getSentMailboxId(wildduckUserId);