import { Processor } from "@nestjs/bullmq"; import { Logger } from "@nestjs/common"; import { Job } from "bullmq"; import { firstValueFrom } from "rxjs"; import { WorkerProcessor } from "../../../common/queues/worker.processor"; import { EMAIL_QUEUE_CONSTANTS } from "../../email/constants/email-events.constant"; import { EmailGateway } from "../../email-gateway/email.gateway"; import { MailServerService } from "../../mail-server/services/mail-server.service"; import { CacheService } from "../../utils/services/cache.service"; import { EmailPollingJobData } from "../interfaces/email-polling-job.interface"; import { EmailNotificationService } from "../services/email-notification.service"; import { MailboxResolverService } from "../services/mailbox-resolver.service"; @Processor(EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_QUEUE) export class EmailPollingProcessor extends WorkerProcessor { protected readonly logger = new Logger(EmailPollingProcessor.name); private readonly CACHE_KEY = "email-polling-notified-emails"; constructor( private readonly emailGateway: EmailGateway, private readonly emailNotificationService: EmailNotificationService, private readonly mailboxResolverService: MailboxResolverService, private readonly mailServerService: MailServerService, private readonly cacheService: CacheService, ) { super(); } async process(job: Job): Promise { const { wildduckUserId, userId, lastCheckTimestamp } = job.data; try { if (!this.emailGateway.isUserConnected(userId)) { await this.cacheService.del(`${this.CACHE_KEY}:${userId}`); return; } const mailboxIds = await this.mailboxResolverService.getUserMailboxIds(wildduckUserId); if (!mailboxIds.inbox) { this.logger.warn(`No inbox found for user ${userId}`); return; } const checkSince = lastCheckTimestamp ? new Date(lastCheckTimestamp) : new Date(Date.now() - 5 * 60 * 1000); const now = new Date(); const newEmails = await this.getRecentEmails(wildduckUserId, mailboxIds.inbox, checkSince); if (newEmails.length > 0) { this.logger.log(`Found ${newEmails.length} potential new emails for user ${wildduckUserId}`); // Get notified emails from cache (stored as array, converted to Set for operations) const notifiedEmailIds = await this.getNotifiedEmailIds(userId); const userNotifiedEmails = new Set(notifiedEmailIds); // Filter out emails that have already been notified about const trulyNewEmails = newEmails.filter((email) => !userNotifiedEmails.has(email.id)); if (trulyNewEmails.length > 0) { this.logger.log( `Found ${trulyNewEmails.length} truly new emails for user ${userId} (${newEmails.length - trulyNewEmails.length} already notified)`, ); // Process each truly new email for (const email of trulyNewEmails) { await this.emailNotificationService.notifyNewEmail({ userId, wildduckUserId, messageId: email.id, date: email.date || new Date().toISOString(), from: email.from, to: email.to || [], subject: email.subject, timestamp: email.date || new Date().toISOString(), mailboxId: email.mailbox, mailboxName: "Inbox", isRead: !!email.seen, hasAttachments: !!(email.attachments && Array.isArray(email.attachments) && email.attachments.length > 0), }); // Mark this email as notified userNotifiedEmails.add(email.id); } // Clean up old notified emails periodically (keep only last 500 per user to prevent memory/cache bloat) let finalNotifiedEmails = userNotifiedEmails; if (userNotifiedEmails.size > 1000) { const sortedIds = Array.from(userNotifiedEmails).sort((a, b) => b - a); const recentIds = sortedIds.slice(0, 500); // Keep most recent 500 finalNotifiedEmails = new Set(recentIds); this.logger.debug(`Cleaned up ${userNotifiedEmails.size - finalNotifiedEmails.size} old notification records for user ${userId}`); } // Update cache with new notified email IDs await this.setNotifiedEmailIds(userId, Array.from(finalNotifiedEmails)); this.logger.log(`Notified user ${userId} of ${trulyNewEmails.length} new emails`); } else { this.logger.debug(`No new emails to notify for user ${userId} - all ${newEmails.length} emails already notified`); } } else { this.logger.debug(`No new emails found for user ${userId}`); } // Update job data with new timestamp for next run job.updateData({ ...job.data, lastCheckTimestamp: now.toISOString(), }); } catch (error) { this.logger.error(`Error processing email polling job for user ${userId}:`, error); throw error; // Re-throw to trigger retry mechanism } } //======================================================= private async getRecentEmails(wildduckUserId: string, mailboxId: string, since: Date) { try { const response = await firstValueFrom( this.mailServerService.messages.searchMessages(wildduckUserId, { mailbox: mailboxId, limit: 50, order: "desc", unseen: true, }), ); if (!response.success || !response.results) { return []; } return response.results.filter((email) => { const emailDate = new Date(email.date || new Date().toISOString()); return emailDate >= since; }); } catch (error) { this.logger.error(`Error fetching recent emails for user ${wildduckUserId}:`, error); return []; } } //======================================================= private async getNotifiedEmailIds(userId: string): Promise { try { const cached = await this.cacheService.get(`${this.CACHE_KEY}:${userId}`); if (!cached) { return []; } return JSON.parse(cached) as number[]; } catch (error) { this.logger.error(`Error getting notified email IDs for user ${userId}:`, error); return []; } } //======================================================= private async setNotifiedEmailIds(userId: string, emailIds: number[]): Promise { try { // Set with 24 hour TTL to prevent indefinite cache growth await this.cacheService.set(`${this.CACHE_KEY}:${userId}`, JSON.stringify(emailIds), 24 * 60 * 60); } catch (error) { this.logger.error(`Error setting notified email IDs for user ${userId}:`, error); } } //======================================================= }