chore: update2
This commit is contained in:
@@ -14,6 +14,10 @@ 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);
|
||||
|
||||
// Track notified emails to prevent duplicates - key: userId, value: Set of messageIds
|
||||
private notifiedEmails = new Map<string, Set<number>>();
|
||||
|
||||
constructor(
|
||||
private readonly emailGateway: EmailGateway,
|
||||
private readonly emailNotificationService: EmailNotificationService,
|
||||
@@ -30,7 +34,9 @@ export class EmailPollingProcessor extends WorkerProcessor {
|
||||
|
||||
try {
|
||||
if (!this.emailGateway.isUserConnected(userId)) {
|
||||
this.logger.debug(`User ${userId} no longer connected, skipping email check`);
|
||||
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);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -48,25 +54,55 @@ export class EmailPollingProcessor extends WorkerProcessor {
|
||||
const newEmails = await this.getRecentEmails(wildduckUserId, mailboxIds.inbox, checkSince);
|
||||
|
||||
if (newEmails.length > 0) {
|
||||
this.logger.log(`Found ${newEmails.length} new emails for user ${wildduckUserId}`);
|
||||
this.logger.log(`Found ${newEmails.length} potential new emails for user ${wildduckUserId}`);
|
||||
|
||||
const lastEmail = newEmails[newEmails.length - 1];
|
||||
await this.emailNotificationService.notifyNewEmail({
|
||||
userId,
|
||||
wildduckUserId,
|
||||
messageId: lastEmail.id,
|
||||
date: lastEmail.date || new Date().toISOString(),
|
||||
from: lastEmail.from,
|
||||
to: lastEmail.to || [],
|
||||
subject: lastEmail.subject,
|
||||
timestamp: lastEmail.date || new Date().toISOString(),
|
||||
mailboxId: lastEmail.mailbox,
|
||||
mailboxName: "Inbox",
|
||||
isRead: !!lastEmail.seen,
|
||||
hasAttachments: !!(lastEmail.attachments && Array.isArray(lastEmail.attachments) && lastEmail.attachments.length > 0),
|
||||
});
|
||||
// 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)!;
|
||||
|
||||
this.logger.log(`Notified user ${userId} of ${newEmails.length} new emails`);
|
||||
// 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);
|
||||
}
|
||||
|
||||
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}`);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user