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 { 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 { 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; } } }