chore: add bulk action processor

2025-07-26 15:40:48 +03:30
parent 6b6566b982
commit e921b00da2
7 changed files with 306 additions and 3 deletions
@@ -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 {
@@ -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",
@@ -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" })
@@ -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 {}
@@ -0,0 +1,4 @@
+export interface IEmailBulkActionJob {
+  wildduckUserId: string;
+  mailboxId: string;
+}
@@ -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<IEmailBulkActionJob>) {
+    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<string, unknown> = {
+          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;
+    }
+  }
+}
@@ -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);