danak/dmail-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: update webscoket connetion and polling service

Browse Source
This commit is contained in:
mahyargdz
2025-07-30 09:37:11 +03:30
parent 5315fae914
commit f46c2ced99
13 changed files with 140 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/email-utils/queue/email-polling.processor.ts
+18 -26
View File
@@ -23,47 +23,41 @@ export class EmailPollingProcessor extends WorkerProcessor {
}
async process(job: Job<EmailPollingJobData>): Promise<void> {
const { userId, lastCheckTimestamp, connectionId } = job.data;
const { wildduckUserId, userId, lastCheckTimestamp, connectionId } = job.data;
this.logger.debug(`Processing email polling job for user ${userId} (connection: ${connectionId})`);
try {
// Check if user is still connected before processing
if (!this.emailGateway.isUserConnected(userId)) {
this.logger.debug(`User ${userId} no longer connected, skipping email check`);
return;
}
// Get user's mailbox information
const mailboxIds = await this.mailboxResolverService.getUserMailboxIds(userId);
const mailboxIds = await this.mailboxResolverService.getUserMailboxIds(wildduckUserId);
if (!mailboxIds.inbox) {
this.logger.warn(`No inbox found for user ${userId}`);
return;
}
// Calculate time range for checking new emails
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()}`);
// Get new emails from inbox since last check
const newEmails = await this.getRecentEmails(userId, mailboxIds.inbox, checkSince);
const newEmails = await this.getRecentEmails(wildduckUserId, mailboxIds.inbox, checkSince);
if (newEmails.length > 0) {
this.logger.log(`Found ${newEmails.length} new emails for user ${userId}`);
// Notify user of each new email via websocket
this.logger.log(`Found ${newEmails.length} new emails for user ${wildduckUserId}`);
const lastEmail = newEmails[newEmails.length - 1];
await this.emailGateway.notifyNewEmail(userId, {
wildduckUserId: userId,
wildduckUserId,
userId,
messageId: lastEmail.id,
date: lastEmail.date || new Date().toISOString(),
from: lastEmail.from,
to: lastEmail.to || [],
subject: lastEmail.subject,
userId,
timestamp: lastEmail.date || new Date().toISOString(),
mailboxId: lastEmail.mailbox,
mailboxName: "Inbox",
@@ -72,7 +66,7 @@ export class EmailPollingProcessor extends WorkerProcessor {
});
// Update unread count
const unreadStats = await this.getUnreadCounts(userId, mailboxIds);
const unreadStats = await this.getUnreadCounts(wildduckUserId, mailboxIds);
await this.emailGateway.notifyUnreadCountUpdate(userId, {
userId,
totalUnread: unreadStats.total,
@@ -96,15 +90,15 @@ export class EmailPollingProcessor extends WorkerProcessor {
throw error; // Re-throw to trigger retry mechanism
}
}
// Helper method to get recent emails from a mailbox
private async getRecentEmails(userId: string, mailboxId: string, since: Date) {
//=======================================================
private async getRecentEmails(wildduckUserId: string, mailboxId: string, since: Date) {
try {
const response = await firstValueFrom(
this.mailServerService.messages.searchMessages(userId, {
this.mailServerService.messages.searchMessages(wildduckUserId, {
mailbox: mailboxId,
limit: 50, // Limit to avoid too many results
order: "desc", // Most recent first
limit: 50,
order: "desc",
unseen: true,
}),
);
@@ -117,22 +111,20 @@ export class EmailPollingProcessor extends WorkerProcessor {
return emailDate >= since;
});
} catch (error) {
this.logger.error(`Error fetching recent emails for user ${userId}:`, error);
this.logger.error(`Error fetching recent emails for user ${wildduckUserId}:`, error);
return [];
}
}
// Helper method to get unread counts from all mailboxes
private async getUnreadCounts(userId: string, mailboxIds: UserMailboxIds) {
//=======================================================
private async getUnreadCounts(wildduckUserId: string, mailboxIds: UserMailboxIds) {
try {
const mailboxCounts: Record<string, number> = {};
let total = 0;
// Get unread count from each mailbox
for (const [mailboxName, mailboxId] of Object.entries(mailboxIds)) {
if (mailboxId) {
try {
const response = await firstValueFrom(this.mailServerService.mailboxes.getMailbox(userId, mailboxId));
const response = await firstValueFrom(this.mailServerService.mailboxes.getMailbox(wildduckUserId, mailboxId));
if (response.success && response.unseen !== undefined) {
mailboxCounts[mailboxName] = response.unseen;
@@ -150,7 +142,7 @@ export class EmailPollingProcessor extends WorkerProcessor {
mailboxCounts,
};
} catch (error) {
this.logger.error(`Error calculating unread counts for user ${userId}:`, error);
this.logger.error(`Error calculating unread counts for user ${wildduckUserId}:`, error);
return {
total: 0,
mailboxCounts: {},