feat: change the email structure into seprate module
This commit is contained in:
mahyargdz
@@ -0,0 +1,381 @@
|
||||
import { EntityManager } from "@mikro-orm/core";
|
||||
import { InjectQueue } from "@nestjs/bullmq";
|
||||
import { Injectable, Logger, OnModuleInit } from "@nestjs/common";
|
||||
import { Queue } from "bullmq";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import { MailboxResolverService } from "./mailbox-resolver.service";
|
||||
import { EMAIL_QUEUE_CONSTANTS } from "../../email/constants/email-events.constant";
|
||||
import { EmailMonitoringJob, EmailPayload, EmailSchedulerJob } from "../../email/interfaces/email-events.interface";
|
||||
import { EmailGateway } from "../../email-gateway/email.gateway";
|
||||
import { MailServerService } from "../../mail-server/services/mail-server.service";
|
||||
import { MailboxEnum } from "../../mailbox/enums/mailbox.enum";
|
||||
import { NotificationQueue } from "../../notifications/queue/notification.queue";
|
||||
import { User } from "../../users/entities/user.entity";
|
||||
|
||||
@Injectable()
|
||||
export class EmailMonitoringService implements OnModuleInit {
|
||||
private readonly logger = new Logger(EmailMonitoringService.name);
|
||||
private lastGlobalCheck: Date = new Date();
|
||||
private isMonitoringActive = false;
|
||||
|
||||
constructor(
|
||||
@InjectQueue(EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE) private readonly monitoringQueue: Queue,
|
||||
private readonly emailGateway: EmailGateway,
|
||||
private readonly mailboxResolver: MailboxResolverService,
|
||||
private readonly mailServerService: MailServerService,
|
||||
private readonly notificationQueue: NotificationQueue,
|
||||
private readonly em: EntityManager,
|
||||
) {}
|
||||
|
||||
async onModuleInit() {
|
||||
this.logger.log("Initializing integrated email monitoring service");
|
||||
await this.startMonitoring();
|
||||
}
|
||||
|
||||
//****************************************************** */
|
||||
// MONITORING SCHEDULER FUNCTIONALITY
|
||||
//****************************************************** */
|
||||
|
||||
async startMonitoring() {
|
||||
if (this.isMonitoringActive) {
|
||||
this.logger.warn("Email monitoring is already running");
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.stopMonitoring();
|
||||
|
||||
this.logger.log("Starting integrated email monitoring with BullMQ cron scheduler");
|
||||
|
||||
await this.monitoringQueue.add(
|
||||
EMAIL_QUEUE_CONSTANTS.EMAIL_SCHEDULER_JOB,
|
||||
{
|
||||
action: EMAIL_QUEUE_CONSTANTS.EMAIL_MONITORING_ACTION_NAME,
|
||||
timestamp: new Date().toISOString(),
|
||||
} as EmailSchedulerJob,
|
||||
{
|
||||
removeOnComplete: true,
|
||||
removeOnFail: false,
|
||||
repeat: {
|
||||
pattern: EMAIL_QUEUE_CONSTANTS.EMAIL_MONITORING_JOB_PATTERN,
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
this.isMonitoringActive = true;
|
||||
this.logger.log("Integrated BullMQ email monitoring scheduler started successfully");
|
||||
|
||||
// Run initial check immediately
|
||||
await this.scheduleEmailCheck();
|
||||
} catch (error) {
|
||||
this.logger.error("Failed to start email monitoring scheduler:", error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async stopMonitoring() {
|
||||
try {
|
||||
// Remove all repeatable jobs with this name
|
||||
const repeatableJobs = await this.monitoringQueue.getJobSchedulers();
|
||||
const existingJobs = repeatableJobs.filter((job) => job.name === EMAIL_QUEUE_CONSTANTS.EMAIL_SCHEDULER_JOB);
|
||||
|
||||
if (existingJobs.length > 0) {
|
||||
for (const job of existingJobs) {
|
||||
await this.monitoringQueue.removeJobScheduler(job.key);
|
||||
this.logger.debug(`Removed repeatable job: ${job.key}`);
|
||||
}
|
||||
this.logger.log("Email monitoring scheduler stopped and cleaned up");
|
||||
} else {
|
||||
this.logger.debug("No existing scheduler jobs found to remove");
|
||||
}
|
||||
|
||||
// Clean up old jobs
|
||||
await this.cleanupJobs();
|
||||
this.isMonitoringActive = false;
|
||||
} catch (error) {
|
||||
this.logger.error("Error stopping email monitoring scheduler:", error);
|
||||
}
|
||||
}
|
||||
|
||||
async restartMonitoring() {
|
||||
this.logger.log("Restarting integrated email monitoring");
|
||||
await this.stopMonitoring();
|
||||
await this.startMonitoring();
|
||||
}
|
||||
|
||||
//****************************************************** */
|
||||
// CORE MONITORING FUNCTIONALITY
|
||||
//****************************************************** */
|
||||
|
||||
async scheduleEmailCheck() {
|
||||
try {
|
||||
const connectedUserIds = this.emailGateway.getConnectedUserIds();
|
||||
|
||||
if (connectedUserIds.length === 0) {
|
||||
this.logger.debug("No connected users - skipping email check");
|
||||
return;
|
||||
}
|
||||
|
||||
const em = this.em.fork();
|
||||
|
||||
const activeUsers = await em.find(User, {
|
||||
id: { $in: connectedUserIds },
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
wildduckUserId: { $ne: null },
|
||||
});
|
||||
|
||||
this.logger.log(`Monitoring ${activeUsers.length} connected users`);
|
||||
|
||||
for (const user of activeUsers) {
|
||||
await this.monitoringQueue.add(
|
||||
EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE_PROCESSING,
|
||||
{
|
||||
userId: user.id,
|
||||
wildduckUserId: user.wildduckUserId,
|
||||
lastCheckTimestamp: this.lastGlobalCheck,
|
||||
},
|
||||
{
|
||||
delay: Math.random() * 10000, // Random delay to prevent thundering herd
|
||||
removeOnComplete: true,
|
||||
removeOnFail: false,
|
||||
attempts: 3,
|
||||
backoff: {
|
||||
type: "exponential",
|
||||
delay: 2000,
|
||||
},
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
this.lastGlobalCheck = new Date();
|
||||
} catch (error) {
|
||||
this.logger.error("Error scheduling email monitoring checks:", error);
|
||||
}
|
||||
}
|
||||
|
||||
async processUserEmailCheck(job: EmailMonitoringJob) {
|
||||
const { userId, lastCheckTimestamp } = job;
|
||||
|
||||
try {
|
||||
const em = this.em.fork();
|
||||
|
||||
const user = await em.findOne(User, {
|
||||
id: userId,
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
});
|
||||
|
||||
if (!user || !user.wildduckUserId) {
|
||||
this.logger.warn(`User ${userId} not found or email not enabled`);
|
||||
return;
|
||||
}
|
||||
|
||||
const inboxMailboxId = await this.mailboxResolver.getInboxMailboxId(user.wildduckUserId);
|
||||
|
||||
const query: Record<string, unknown> = {
|
||||
limit: 100,
|
||||
order: "desc",
|
||||
unseen: true,
|
||||
};
|
||||
|
||||
if (lastCheckTimestamp) {
|
||||
const timestamp = lastCheckTimestamp instanceof Date ? lastCheckTimestamp : new Date(lastCheckTimestamp);
|
||||
query.datestart = timestamp.toISOString();
|
||||
}
|
||||
|
||||
const { results } = await firstValueFrom(this.mailServerService.messages.listMessages(user.wildduckUserId, inboxMailboxId, query));
|
||||
|
||||
if (results.length > 0) {
|
||||
this.logger.log(`${user.emailAddress}: ${results.length} new messages`);
|
||||
|
||||
for (const message of results) {
|
||||
await this.processNewEmailMessage(user.id, user.wildduckUserId, message);
|
||||
}
|
||||
}
|
||||
|
||||
await this.updateUnreadCount(userId, user.wildduckUserId, user.emailAddress);
|
||||
} catch (error) {
|
||||
this.logger.error(`Error checking emails for user ${userId}:`, error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async processNewEmailMessage(userId: string, _wildduckUserId: string, message: Record<string, any>) {
|
||||
try {
|
||||
if (message.seen) {
|
||||
return;
|
||||
}
|
||||
|
||||
const emailPayload: EmailPayload = {
|
||||
messageId: message.id,
|
||||
userId,
|
||||
from: message.from,
|
||||
to: message.to,
|
||||
subject: message.subject,
|
||||
hasAttachments: message.hasAttachments,
|
||||
mailboxName: message.mailboxName,
|
||||
isRead: message.seen,
|
||||
mailboxId: message.mailboxId,
|
||||
timestamp: message.timestamp,
|
||||
};
|
||||
|
||||
await this.notificationQueue.addNewEmailNotification(userId, {
|
||||
fromEmail: message.from[0].address,
|
||||
fromName: message.from[0].name,
|
||||
subject: message.subject,
|
||||
});
|
||||
|
||||
await this.emailGateway.notifyNewEmail(userId, emailPayload);
|
||||
} catch (error) {
|
||||
this.logger.error(`Error processing new email message for user ${userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
private async updateUnreadCount(userId: string, wildduckUserId: string, userEmailAddress?: string) {
|
||||
try {
|
||||
const inboxMailboxId = await this.mailboxResolver.getInboxMailboxId(wildduckUserId);
|
||||
|
||||
const { results } = await firstValueFrom(
|
||||
this.mailServerService.messages.listMessages(wildduckUserId, inboxMailboxId, {
|
||||
limit: 250,
|
||||
unseen: true,
|
||||
}),
|
||||
);
|
||||
|
||||
const unreadCount = results.length;
|
||||
|
||||
this.emailGateway.notifyUnreadCountUpdate(userId, {
|
||||
count: unreadCount,
|
||||
timestamp: new Date().toISOString(),
|
||||
totalUnread: unreadCount,
|
||||
mailboxCounts: {
|
||||
[MailboxEnum.INBOX]: unreadCount,
|
||||
},
|
||||
userId,
|
||||
});
|
||||
|
||||
if (unreadCount > 0 && userEmailAddress) {
|
||||
this.logger.log(`${userEmailAddress}: ${unreadCount} unread total`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Error updating unread count for user ${userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
async checkUserEmails(userId: string): Promise<void> {
|
||||
await this.monitoringQueue.add(
|
||||
EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE_PROCESSING,
|
||||
{
|
||||
userId,
|
||||
lastCheckTimestamp: new Date(Date.now() - 60000), // Check emails from last minute
|
||||
},
|
||||
{
|
||||
attempts: 3,
|
||||
removeOnComplete: true,
|
||||
removeOnFail: false,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
//****************************************************** */
|
||||
// STATUS AND MANAGEMENT FUNCTIONALITY
|
||||
//****************************************************** */
|
||||
|
||||
async getMonitoringStatus() {
|
||||
try {
|
||||
const repeatableJobs = await this.monitoringQueue.getJobSchedulers();
|
||||
const schedulerJob = repeatableJobs.find((job) => job.name === EMAIL_QUEUE_CONSTANTS.EMAIL_SCHEDULER_JOB);
|
||||
const queueStatus = await this.getQueueStatus();
|
||||
|
||||
return {
|
||||
isRunning: this.isMonitoringActive && !!schedulerJob,
|
||||
cronPattern: schedulerJob?.pattern || null,
|
||||
nextRunTime: schedulerJob?.next ? new Date(schedulerJob.next) : null,
|
||||
checkInterval: EMAIL_QUEUE_CONSTANTS.EMAIL_MONITORING_JOB_PATTERN,
|
||||
lastGlobalCheck: this.lastGlobalCheck,
|
||||
queueStatus,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.error("Error getting monitoring status:", error);
|
||||
return {
|
||||
isRunning: false,
|
||||
error: error instanceof Error ? error.message : "Unknown error",
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
async getQueueStatus() {
|
||||
const [waiting, active, completed, failed] = await Promise.all([
|
||||
this.monitoringQueue.getWaiting(),
|
||||
this.monitoringQueue.getActive(),
|
||||
this.monitoringQueue.getCompleted(),
|
||||
this.monitoringQueue.getFailed(),
|
||||
]);
|
||||
|
||||
return {
|
||||
waiting: waiting.length,
|
||||
active: active.length,
|
||||
completed: completed.length,
|
||||
failed: failed.length,
|
||||
};
|
||||
}
|
||||
|
||||
async getDetailedQueueInfo() {
|
||||
try {
|
||||
const [waiting, active, completed, failed, repeatableJobs] = await Promise.all([
|
||||
this.monitoringQueue.getWaiting(),
|
||||
this.monitoringQueue.getActive(),
|
||||
this.monitoringQueue.getCompleted(0, 10),
|
||||
this.monitoringQueue.getFailed(0, 10),
|
||||
this.monitoringQueue.getJobSchedulers(),
|
||||
]);
|
||||
|
||||
return {
|
||||
counts: {
|
||||
waiting: waiting.length,
|
||||
active: active.length,
|
||||
completed: completed.length,
|
||||
failed: failed.length,
|
||||
repeatable: repeatableJobs.length,
|
||||
},
|
||||
repeatableJobs: repeatableJobs.map((job) => ({
|
||||
name: job.name,
|
||||
pattern: job.pattern,
|
||||
next: job.next ? new Date(job.next) : null,
|
||||
})),
|
||||
recentCompleted: completed.map((job) => ({
|
||||
id: job.id,
|
||||
processedOn: job.processedOn ? new Date(job.processedOn) : null,
|
||||
finishedOn: job.finishedOn ? new Date(job.finishedOn) : null,
|
||||
})),
|
||||
recentFailed: failed.map((job) => ({
|
||||
id: job.id,
|
||||
failedReason: job.failedReason,
|
||||
processedOn: job.processedOn ? new Date(job.processedOn) : null,
|
||||
})),
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.error("Error getting detailed queue info:", error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async cleanupJobs() {
|
||||
try {
|
||||
// Clean completed jobs older than 30 minutes
|
||||
await this.monitoringQueue.clean(1000 * 60 * 30, 50, "completed");
|
||||
// Clean failed jobs older than 2 hours
|
||||
await this.monitoringQueue.clean(1000 * 60 * 120, 20, "failed");
|
||||
// Clean active jobs older than 10 minutes (stuck jobs)
|
||||
await this.monitoringQueue.clean(1000 * 60 * 10, 10, "active");
|
||||
|
||||
this.logger.log("Successfully cleaned up old monitoring jobs");
|
||||
} catch (error) {
|
||||
this.logger.error("Error cleaning up jobs:", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,339 @@
|
||||
import { EntityManager } from "@mikro-orm/core";
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import { MailboxResolverService } from "./mailbox-resolver.service";
|
||||
import { WebSocketEvent } from "../../email/constants/email-events.constant";
|
||||
import {
|
||||
EmailDeletePayload,
|
||||
EmailSentNotificationParams,
|
||||
EmailSentPayload,
|
||||
EmailStatusNotificationParams,
|
||||
EmailStatusPayload,
|
||||
MailboxUpdateNotificationParams,
|
||||
MailboxUpdatePayload,
|
||||
UnreadCountPayload,
|
||||
} from "../../email/interfaces/email-events.interface";
|
||||
import { EmailGateway } from "../../email-gateway/email.gateway";
|
||||
import { MailServerService } from "../../mail-server/services/mail-server.service";
|
||||
import { MailboxEnum } from "../../mailbox/enums/mailbox.enum";
|
||||
import { User } from "../../users/entities/user.entity";
|
||||
|
||||
@Injectable()
|
||||
export class EmailNotificationService {
|
||||
private readonly logger = new Logger(EmailNotificationService.name);
|
||||
|
||||
constructor(
|
||||
private readonly emailGateway: EmailGateway,
|
||||
private readonly mailboxResolver: MailboxResolverService,
|
||||
private readonly mailServerService: MailServerService,
|
||||
private readonly em: EntityManager,
|
||||
) {}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyEmailSent(params: EmailSentNotificationParams) {
|
||||
try {
|
||||
const payload: EmailSentPayload = {
|
||||
userId: params.userId,
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
from: params.from,
|
||||
to: params.recipients,
|
||||
subject: params.subject,
|
||||
hasAttachments: false,
|
||||
mailboxName: MailboxEnum.SENT,
|
||||
isRead: true,
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailSent(params.userId, payload);
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email sent notification dispatched for user ${params.userId}: ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email sent notification for user ${params.userId}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email sent for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyEmailUnread(params: EmailStatusNotificationParams) {
|
||||
try {
|
||||
const em = this.em.fork();
|
||||
const user = await em.findOne(User, {
|
||||
wildduckUserId: params.userId,
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
});
|
||||
|
||||
if (!user) {
|
||||
this.logger.warn(`User not found with wildduckUserId: ${params.userId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const payload: EmailStatusPayload = {
|
||||
userId: user.id,
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailUnread(user.id, payload);
|
||||
await this.updateUnreadCount(params.userId);
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email unread notification dispatched for user ${user.id}: message ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email unread notification for user ${user.id}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email unread for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyEmailRead(params: EmailStatusNotificationParams) {
|
||||
try {
|
||||
// Find the user to get the correct User entity ID for websocket notifications
|
||||
const em = this.em.fork();
|
||||
const user = await em.findOne(User, {
|
||||
wildduckUserId: params.userId,
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
});
|
||||
|
||||
if (!user) {
|
||||
this.logger.warn(`User not found with wildduckUserId: ${params.userId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const payload: EmailStatusPayload = {
|
||||
userId: user.id, // Use User entity ID for websocket
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailRead(user.id, payload);
|
||||
await this.updateUnreadCount(params.userId); // Pass wildduckUserId to updateUnreadCount
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email read notification dispatched for user ${user.id}: message ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email read notification for user ${user.id}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email read for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyEmailDeleted(params: EmailStatusNotificationParams) {
|
||||
try {
|
||||
// Find the user to get the correct User entity ID for websocket notifications
|
||||
const em = this.em.fork();
|
||||
const user = await em.findOne(User, {
|
||||
wildduckUserId: params.userId,
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
});
|
||||
|
||||
if (!user) {
|
||||
this.logger.warn(`User not found with wildduckUserId: ${params.userId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const payload: EmailDeletePayload = {
|
||||
userId: user.id, // Use User entity ID for websocket
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailDeleted(user.id, payload);
|
||||
await this.updateUnreadCount(params.userId); // Pass wildduckUserId to updateUnreadCount
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email deleted notification dispatched for user ${user.id}: message ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email deleted notification for user ${user.id}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email deleted for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyEmailFlagged(params: EmailStatusNotificationParams) {
|
||||
try {
|
||||
// Find the user to get the correct User entity ID for websocket notifications
|
||||
const em = this.em.fork();
|
||||
const user = await em.findOne(User, {
|
||||
wildduckUserId: params.userId,
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
});
|
||||
|
||||
if (!user) {
|
||||
this.logger.warn(`User not found with wildduckUserId: ${params.userId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const payload: EmailStatusPayload = {
|
||||
userId: user.id, // Use User entity ID for websocket
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailFlagged(user.id, payload);
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email flagged notification dispatched for user ${user.id}: message ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email flagged notification for user ${user.id}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email flagged for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyEmailUnflagged(params: EmailStatusNotificationParams) {
|
||||
try {
|
||||
// Find the user to get the correct User entity ID for websocket notifications
|
||||
const em = this.em.fork();
|
||||
const user = await em.findOne(User, {
|
||||
wildduckUserId: params.userId,
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
});
|
||||
|
||||
if (!user) {
|
||||
this.logger.warn(`User not found with wildduckUserId: ${params.userId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const payload: EmailStatusPayload = {
|
||||
userId: user.id, // Use User entity ID for websocket
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailUnflagged(user.id, payload);
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email unflagged notification dispatched for user ${user.id}: message ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email unflagged notification for user ${user.id}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email unflagged for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyMailboxUpdate(params: MailboxUpdateNotificationParams) {
|
||||
try {
|
||||
const payload: MailboxUpdatePayload = {
|
||||
userId: params.userId,
|
||||
mailboxId: params.mailboxId,
|
||||
mailboxName: params.mailboxName,
|
||||
unreadCount: params.unreadCount,
|
||||
totalCount: params.totalCount,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyMailboxUpdate(params.userId, payload);
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Mailbox update notification dispatched for user ${params.userId}: ${params.mailboxName}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch mailbox update notification for user ${params.userId}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify mailbox update for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
private async updateUnreadCount(userId: string) {
|
||||
try {
|
||||
const em = this.em.fork();
|
||||
|
||||
const user = await em.findOne(User, {
|
||||
wildduckUserId: userId,
|
||||
emailEnabled: true,
|
||||
deletedAt: null,
|
||||
});
|
||||
|
||||
if (!user || !user.wildduckUserId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const inboxMailboxId = await this.mailboxResolver.getInboxMailboxId(user.wildduckUserId);
|
||||
|
||||
const { results } = await firstValueFrom(
|
||||
this.mailServerService.messages.listMessages(user.wildduckUserId, inboxMailboxId, {
|
||||
limit: 50,
|
||||
unseen: true,
|
||||
}),
|
||||
);
|
||||
|
||||
const unreadCount = results.length;
|
||||
|
||||
const payload: UnreadCountPayload = {
|
||||
userId: user.id, // Use the actual User entity ID for the notification payload
|
||||
count: unreadCount,
|
||||
timestamp: new Date().toISOString(),
|
||||
totalUnread: unreadCount,
|
||||
mailboxCounts: {
|
||||
[inboxMailboxId]: unreadCount,
|
||||
},
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyUnreadCountUpdate(user.id, payload); // Use User entity ID here too
|
||||
|
||||
if (!success) {
|
||||
this.logger.warn(`Failed to update unread count for user ${user.id}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to update unread count for user ${userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async broadcastSystemNotification(event: WebSocketEvent, data: Record<string, unknown>) {
|
||||
try {
|
||||
this.emailGateway.broadcastSystemMessage(event, {
|
||||
...data,
|
||||
timestamp: new Date().toISOString(),
|
||||
});
|
||||
|
||||
this.logger.log(`System notification broadcasted: ${event}`);
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to broadcast system notification:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async getGatewayStats() {
|
||||
try {
|
||||
const connectedUsers = await this.emailGateway.getConnectedUsersCount();
|
||||
|
||||
return {
|
||||
connectedUsers,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.error("Failed to get gateway statistics:", error);
|
||||
return {
|
||||
connectedUsers: 0,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
import { Injectable, Logger, NotFoundException } from "@nestjs/common";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import { UserMailboxIds } from "../../email/interfaces/user-mailbox.interface";
|
||||
import { MailboxInfo } from "../../mail-server/interfaces/mailboxes-response.interface";
|
||||
import { MailServerService } from "../../mail-server/services/mail-server.service";
|
||||
import { MailboxEnum } from "../../mailbox/enums/mailbox.enum";
|
||||
|
||||
@Injectable()
|
||||
export class MailboxResolverService {
|
||||
private readonly logger = new Logger(MailboxResolverService.name);
|
||||
private readonly mailboxCache = new Map<string, UserMailboxIds>();
|
||||
private readonly cacheExpiry = new Map<string, number>();
|
||||
private readonly CACHE_DURATION = 5 * 60 * 1000; // 5 minutes
|
||||
|
||||
constructor(private readonly mailServerService: MailServerService) {}
|
||||
|
||||
/**
|
||||
* Get user mailbox IDs with caching
|
||||
*/
|
||||
async getUserMailboxIds(userId: string): Promise<UserMailboxIds> {
|
||||
const cached = this.getCachedMailboxIds(userId);
|
||||
if (cached) return cached;
|
||||
|
||||
this.logger.log(`Fetching mailbox IDs for user: ${userId}`);
|
||||
|
||||
try {
|
||||
const mailboxes = await firstValueFrom(this.mailServerService.mailboxes.listMailboxes(userId));
|
||||
|
||||
const mailboxIds: UserMailboxIds = {
|
||||
inbox: this.findMailboxId(mailboxes.results, MailboxEnum.INBOX),
|
||||
sent: this.findMailboxId(mailboxes.results, MailboxEnum.SENT),
|
||||
drafts: this.findMailboxId(mailboxes.results, MailboxEnum.DRAFTS),
|
||||
trash: this.findMailboxId(mailboxes.results, MailboxEnum.TRASH),
|
||||
junk: this.findMailboxId(mailboxes.results, MailboxEnum.Junk, false),
|
||||
archive: this.findMailboxId(mailboxes.results, MailboxEnum.ARCHIVE, false),
|
||||
// favorite: this.findMailboxId(mailboxes.results, MailboxEnum.FAVORITE, false),
|
||||
};
|
||||
|
||||
this.cacheMailboxIds(userId, mailboxIds);
|
||||
|
||||
this.logger.log(`Mailbox IDs cached for user: ${userId}`);
|
||||
return mailboxIds;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to fetch mailbox IDs for user ${userId}: ${error instanceof Error ? error.message : "Unknown error"}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get specific mailbox ID
|
||||
*/
|
||||
async getMailboxId(userId: string, mailboxType: keyof UserMailboxIds): Promise<string> {
|
||||
const mailboxIds = await this.getUserMailboxIds(userId);
|
||||
const mailboxId = mailboxIds[mailboxType];
|
||||
|
||||
if (!mailboxId) {
|
||||
throw new NotFoundException(`${mailboxType} mailbox not found for user ${userId}`);
|
||||
}
|
||||
|
||||
return mailboxId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get mailbox name by ID
|
||||
*/
|
||||
async getMailboxName(wildduckUserId: string, mailboxId: string): Promise<string> {
|
||||
const mailboxes = await firstValueFrom(this.mailServerService.mailboxes.listMailboxes(wildduckUserId));
|
||||
const mailbox = mailboxes.results.find((mb) => mb.id === mailboxId);
|
||||
return mailbox?.name || "";
|
||||
}
|
||||
|
||||
/**
|
||||
* Get drafts mailbox ID
|
||||
*/
|
||||
async getDraftsMailboxId(userId: string): Promise<string> {
|
||||
return this.getMailboxId(userId, "drafts");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get inbox mailbox ID
|
||||
*/
|
||||
async getInboxMailboxId(userId: string): Promise<string> {
|
||||
return this.getMailboxId(userId, "inbox");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get sent mailbox ID
|
||||
*/
|
||||
async getSentMailboxId(userId: string): Promise<string> {
|
||||
return this.getMailboxId(userId, "sent");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get trash mailbox ID
|
||||
*/
|
||||
async getTrashMailboxId(userId: string): Promise<string> {
|
||||
return this.getMailboxId(userId, "trash");
|
||||
}
|
||||
|
||||
async getJunkMailboxId(userId: string): Promise<string> {
|
||||
return this.getMailboxId(userId, "junk");
|
||||
}
|
||||
|
||||
async getArchiveMailboxId(userId: string): Promise<string> {
|
||||
return this.getMailboxId(userId, "archive");
|
||||
}
|
||||
|
||||
// async getFavoriteMailboxId(userId: string): Promise<string> {
|
||||
// return this.getMailboxId(userId, "favorite");
|
||||
// }
|
||||
|
||||
/**
|
||||
* Clear cache for a specific user
|
||||
*/
|
||||
clearUserCache(userId: string): void {
|
||||
this.mailboxCache.delete(userId);
|
||||
this.cacheExpiry.delete(userId);
|
||||
this.logger.log(`Cache cleared for user: ${userId}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all cache
|
||||
*/
|
||||
clearAllCache(): void {
|
||||
this.mailboxCache.clear();
|
||||
this.cacheExpiry.clear();
|
||||
this.logger.log("All mailbox cache cleared");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get cached mailbox IDs if not expired
|
||||
*/
|
||||
private getCachedMailboxIds(userId: string): UserMailboxIds | null {
|
||||
const cached = this.mailboxCache.get(userId);
|
||||
const expiry = this.cacheExpiry.get(userId);
|
||||
|
||||
if (cached && expiry && Date.now() < expiry) {
|
||||
this.logger.debug(`Using cached mailbox IDs for user: ${userId}`);
|
||||
return cached;
|
||||
}
|
||||
|
||||
if (cached) {
|
||||
this.logger.debug(`Cache expired for user: ${userId}`);
|
||||
this.clearUserCache(userId);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Cache mailbox IDs with expiry
|
||||
*/
|
||||
private cacheMailboxIds(userId: string, mailboxIds: UserMailboxIds): void {
|
||||
this.mailboxCache.set(userId, mailboxIds);
|
||||
this.cacheExpiry.set(userId, Date.now() + this.CACHE_DURATION);
|
||||
}
|
||||
|
||||
/**
|
||||
* Find mailbox ID by name
|
||||
*/
|
||||
private findMailboxId(mailboxes: MailboxInfo[], mailboxName: string, required: boolean = true): string {
|
||||
const mailbox = mailboxes.find((mb) => mb.name === mailboxName || mb.path === mailboxName || mb.specialUse === mailboxName);
|
||||
|
||||
if (!mailbox && required) {
|
||||
throw new NotFoundException(`${mailboxName} mailbox not found`);
|
||||
}
|
||||
|
||||
return mailbox?.id || "";
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user