diff --git a/src/modules/email-utils/email-utils.module.ts b/src/modules/email-utils/email-utils.module.ts index d664ef1..b47d647 100644 --- a/src/modules/email-utils/email-utils.module.ts +++ b/src/modules/email-utils/email-utils.module.ts @@ -10,12 +10,14 @@ import { EMAIL_QUEUE_CONSTANTS } from "../email/constants/email-events.constant" import { EmailGatewayModule } from "../email-gateway/email-gateway.module"; import { MailServerModule } from "../mail-server/mail-server.module"; import { NotificationModule } from "../notifications/notifications.module"; +import { UtilsModule } from "../utils/utils.module"; @Module({ imports: [ NotificationModule, MailServerModule, EmailGatewayModule, + UtilsModule, BullModule.forRootAsync(bullMqConfig()), BullModule.registerQueue({ name: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_QUEUE, diff --git a/src/modules/email-utils/queue/email-polling.processor.ts b/src/modules/email-utils/queue/email-polling.processor.ts index f496d84..ed67f59 100644 --- a/src/modules/email-utils/queue/email-polling.processor.ts +++ b/src/modules/email-utils/queue/email-polling.processor.ts @@ -7,6 +7,7 @@ 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"; @@ -15,28 +16,24 @@ import { MailboxResolverService } from "../services/mailbox-resolver.service"; export class EmailPollingProcessor extends WorkerProcessor { protected readonly logger = new Logger(EmailPollingProcessor.name); - // Track notified emails to prevent duplicates - key: userId, value: Set of messageIds - private notifiedEmails = new Map>(); + 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, connectionId } = job.data; - - this.logger.debug(`Processing email polling job for user ${userId} (connection: ${connectionId})`); + const { wildduckUserId, userId, lastCheckTimestamp } = job.data; try { if (!this.emailGateway.isUserConnected(userId)) { - this.logger.debug(`User ${userId} no longer connected, skipping email check and cleaning up notified emails`); - // Clean up notified emails for disconnected user - this.notifiedEmails.delete(userId); + await this.cacheService.del(`${this.CACHE_KEY}:${userId}`); return; } @@ -49,18 +46,14 @@ export class EmailPollingProcessor extends WorkerProcessor { const checkSince = lastCheckTimestamp ? new Date(lastCheckTimestamp) : new Date(Date.now() - 5 * 60 * 1000); const now = new Date(); - this.logger.debug(`Checking emails for user ${userId} since ${checkSince.toISOString()}`); - 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 or initialize the set of notified emails for this user - if (!this.notifiedEmails.has(userId)) { - this.notifiedEmails.set(userId, new Set()); - } - const userNotifiedEmails = this.notifiedEmails.get(userId)!; + // 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)); @@ -91,18 +84,22 @@ export class EmailPollingProcessor extends WorkerProcessor { 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`); } - - // Clean up old notified emails periodically (keep only last 1000 per user to prevent memory leaks) - if (userNotifiedEmails.size > 1000) { - const sortedIds = Array.from(userNotifiedEmails).sort((a, b) => b - a); - const toRemove = sortedIds.slice(500); // Keep most recent 500 - toRemove.forEach((id) => userNotifiedEmails.delete(id)); - this.logger.debug(`Cleaned up ${toRemove.length} old notification records for user ${userId}`); - } } else { this.logger.debug(`No new emails found for user ${userId}`); } @@ -142,4 +139,30 @@ export class EmailPollingProcessor extends WorkerProcessor { 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); + } + } + + //======================================================= }