chore: add websocket for real time
This commit is contained in:
mahyargdz
@@ -15,7 +15,7 @@ export class PaginationDto {
|
||||
@IsNotEmpty()
|
||||
@IsNumber()
|
||||
@Min(1)
|
||||
@Max(50)
|
||||
@Max(100)
|
||||
@Type(() => Number)
|
||||
@ApiPropertyOptional({ type: "number", required: false })
|
||||
limit?: number;
|
||||
|
||||
@@ -218,7 +218,8 @@ export const enum EmailMessage {
|
||||
BULK_ACTION_PARTIALLY_COMPLETED = "عملیات گروهی تا حدی انجام شد",
|
||||
BULK_ACTION_FAILED = "عملیات گروهی با خطا مواجه شد",
|
||||
USER_NOT_FOUND = "کاربر یافت نشد",
|
||||
PRIORITY_MUST_BE_ONE_OF_HIGH_NORMAL_LOW = "Priority must be one of: high, normal, low",
|
||||
PRIORITY_MUST_BE_ONE_OF_HIGH_NORMAL_LOW = "اولویت باید یکی از مقادیر high, normal, low باشد",
|
||||
SUBJECT_REQUIRED = "موضوع ایمیل مورد نیاز است",
|
||||
}
|
||||
|
||||
export const enum WebSocketMessage {
|
||||
|
||||
@@ -1,11 +1,15 @@
|
||||
import { HttpModuleAsyncOptions } from "@nestjs/axios";
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
|
||||
export const wildduckConfig = () => ({
|
||||
export const wildduckConfig = (): HttpModuleAsyncOptions => ({
|
||||
inject: [ConfigService],
|
||||
useFactory: (configService: ConfigService) => ({
|
||||
baseUrl: configService.getOrThrow<string>("WILDDUCK_BASE_URL"),
|
||||
accessToken: configService.getOrThrow<string>("WILDDUCK_ACCESS_TOKEN"),
|
||||
baseURL: configService.getOrThrow<string>("WILDDUCK_BASE_URL"),
|
||||
timeout: configService.getOrThrow<number>("WILDDUCK_TIMEOUT"),
|
||||
retries: configService.getOrThrow<number>("WILDDUCK_RETRIES"),
|
||||
headers: {
|
||||
Accept: "application/json",
|
||||
"X-Access-Token": configService.getOrThrow<string>("WILDDUCK_ACCESS_TOKEN"),
|
||||
},
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import { ArgumentsHost, Catch, ExceptionFilter, Logger, UnauthorizedException } from "@nestjs/common";
|
||||
import { WsException } from "@nestjs/websockets";
|
||||
import { Socket } from "socket.io";
|
||||
|
||||
import { WebSocketMessage } from "../../common/enums/message.enum";
|
||||
import { WEBSOCKET_EVENTS } from "../../modules/email/constants/email-events.constant";
|
||||
|
||||
@Catch(WsException)
|
||||
export class WsExceptionFilter implements ExceptionFilter {
|
||||
private readonly logger = new Logger(WsExceptionFilter.name);
|
||||
|
||||
catch(exception: WsException, host: ArgumentsHost) {
|
||||
const client = host.switchToWs().getClient<Socket>();
|
||||
const data = host.switchToWs().getData();
|
||||
|
||||
this.logger.error(`WebSocket Error: ${exception.message}`);
|
||||
this.logger.error(`Client ID: ${client.id}`);
|
||||
this.logger.error(`Data: ${JSON.stringify(data)}`);
|
||||
|
||||
if (exception instanceof UnauthorizedException) {
|
||||
client.emit(WEBSOCKET_EVENTS.CONNECTION_ERROR, {
|
||||
message: WebSocketMessage.INVALID_TOKEN,
|
||||
});
|
||||
client.disconnect();
|
||||
return;
|
||||
}
|
||||
|
||||
const errorResponse = {
|
||||
status: "error",
|
||||
message: exception.message || WebSocketMessage.WEBSOCKET_ERROR,
|
||||
};
|
||||
|
||||
client.emit(WEBSOCKET_EVENTS.CONNECTION_ERROR, errorResponse);
|
||||
}
|
||||
}
|
||||
@@ -86,12 +86,12 @@ export class PaginationInterceptor implements NestInterceptor {
|
||||
if (!page) return false;
|
||||
|
||||
const { protocol, hostname, url } = request;
|
||||
const baseUrl = url.split("?")[0];
|
||||
const baseURL = url.split("?")[0];
|
||||
const queryParams = new URLSearchParams({
|
||||
page: page.toString(),
|
||||
limit: limit.toString(),
|
||||
});
|
||||
|
||||
return `${protocol}://${hostname}${baseUrl}?${queryParams.toString()}`;
|
||||
return `${protocol}://${hostname}${baseURL}?${queryParams.toString()}`;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -151,10 +151,11 @@ export class SendEmailDto {
|
||||
// })
|
||||
headers?: EmailHeaderDto[];
|
||||
|
||||
@IsOptional()
|
||||
// @IsOptional()
|
||||
@IsNotEmpty({ message: EmailMessage.SUBJECT_REQUIRED })
|
||||
@IsString({ message: EmailMessage.SUBJECT_MUST_BE_STRING })
|
||||
@ApiPropertyOptional({ description: "Message subject. If not then resolved from Reference message (optional)", example: "Test email" })
|
||||
subject?: string;
|
||||
subject: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsString({ message: EmailMessage.TEXT_CONTENT_MUST_BE_STRING })
|
||||
|
||||
@@ -2,7 +2,9 @@ export const WEBSOCKET_EVENTS = {
|
||||
// Connection Events
|
||||
CONNECTION_ESTABLISHED: "connection_established",
|
||||
CONNECTION_ERROR: "connection_error",
|
||||
|
||||
CONNECTION_CLOSED: "connection_closed",
|
||||
USER_LEFT: "user_left",
|
||||
UNAUTHORIZED: "unauthorized",
|
||||
// Email Events - Server to Client
|
||||
NEW_EMAIL: "new_email",
|
||||
EMAIL_READ: "email_read",
|
||||
@@ -34,3 +36,10 @@ export const WEBSOCKET_NAMESPACES = {
|
||||
export type WebSocketEvent = (typeof WEBSOCKET_EVENTS)[keyof typeof WEBSOCKET_EVENTS];
|
||||
export type WebSocketRoom = string;
|
||||
export type WebSocketNamespace = (typeof WEBSOCKET_NAMESPACES)[keyof typeof WEBSOCKET_NAMESPACES];
|
||||
|
||||
export const EMAIL_QUEUE_CONSTANTS = {
|
||||
EMAIL_QUEUE: "email_queue",
|
||||
EMAIL_QUEUE_PROCESSING: "email_queue_processing",
|
||||
EMAIL_QUEUE_FAILED: "email_queue_failed",
|
||||
EMAIL_QUEUE_COMPLETED: "email_queue_completed",
|
||||
};
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
// import { Body, Controller, Get, Param, Post } from "@nestjs/common";
|
||||
// import { ApiOperation, ApiResponse, ApiTags } from "@nestjs/swagger";
|
||||
|
||||
// import { AuthGuards } from "../../../common/decorators/auth-guard.decorator";
|
||||
// import { UserDec } from "../../../common/decorators/user.decorator";
|
||||
// import { EmailMonitoringService } from "../services/email-monitoring.service";
|
||||
// import { EmailNotificationService } from "../services/email-notification.service";
|
||||
|
||||
// @ApiTags("Email Gateway")
|
||||
// @AuthGuards()
|
||||
// @Controller("email/gateway")
|
||||
// export class EmailGatewayController {
|
||||
// constructor(
|
||||
// private readonly emailMonitoringService: EmailMonitoringService,
|
||||
// private readonly emailNotificationService: EmailNotificationService,
|
||||
// ) {}
|
||||
|
||||
// @Get("status")
|
||||
// @ApiOperation({ summary: "Get email gateway status" })
|
||||
// @ApiResponse({ status: 200, description: "Gateway status retrieved successfully" })
|
||||
// async getGatewayStatus() {
|
||||
// const [gatewayStats, queueStatus] = await Promise.all([
|
||||
// this.emailNotificationService.getGatewayStats(),
|
||||
// this.emailMonitoringService.getQueueStatus(),
|
||||
// ]);
|
||||
|
||||
// return {
|
||||
// message: "Email gateway status retrieved successfully",
|
||||
// data: {
|
||||
// gateway: gatewayStats,
|
||||
// queue: queueStatus,
|
||||
// },
|
||||
// };
|
||||
// }
|
||||
|
||||
// @Post("check-emails")
|
||||
// @ApiOperation({ summary: "Manually trigger email check for current user" })
|
||||
// @ApiResponse({ status: 200, description: "Email check initiated successfully" })
|
||||
// async checkUserEmails(@UserDec("id") userId: string) {
|
||||
// await this.emailMonitoringService.checkUserEmails(userId);
|
||||
|
||||
// return {
|
||||
// message: "Email check initiated successfully",
|
||||
// userId,
|
||||
// };
|
||||
// }
|
||||
|
||||
// @Get("connected-users")
|
||||
// @ApiOperation({ summary: "Get connected users count" })
|
||||
// @ApiResponse({ status: 200, description: "Connected users count retrieved successfully" })
|
||||
// async getConnectedUsers() {
|
||||
// const stats = await this.emailNotificationService.getGatewayStats();
|
||||
|
||||
// return {
|
||||
// message: "Connected users retrieved successfully",
|
||||
// data: stats,
|
||||
// };
|
||||
// }
|
||||
|
||||
// @Post("test-notification")
|
||||
// @ApiOperation({ summary: "Send test notification to current user" })
|
||||
// @ApiResponse({ status: 200, description: "Test notification sent successfully" })
|
||||
// async sendTestNotification(@UserDec("id") userId: string, @Body() body: { message?: string }) {
|
||||
// await this.emailNotificationService.broadcastSystemNotification("test", {
|
||||
// message: body.message || "Test notification from Email Gateway",
|
||||
// userId,
|
||||
// });
|
||||
|
||||
// return {
|
||||
// message: "Test notification sent successfully",
|
||||
// userId,
|
||||
// };
|
||||
// }
|
||||
|
||||
// @Get("business/:businessId/users")
|
||||
// @ApiOperation({ summary: "Get connected users for a business" })
|
||||
// @ApiResponse({ status: 200, description: "Business connected users retrieved successfully" })
|
||||
// async getBusinessConnectedUsers(@Param("businessId") businessId: string) {
|
||||
// const connectedUsers = await this.emailNotificationService.getBusinessConnectedUsers(businessId);
|
||||
|
||||
// return {
|
||||
// message: "Business connected users retrieved successfully",
|
||||
// businessId,
|
||||
// data: {
|
||||
// connectedUsers,
|
||||
// count: connectedUsers.length,
|
||||
// },
|
||||
// };
|
||||
// }
|
||||
|
||||
// @Post("cleanup")
|
||||
// @ApiOperation({ summary: "Clean up old monitoring jobs" })
|
||||
// @ApiResponse({ status: 200, description: "Cleanup completed successfully" })
|
||||
// async cleanupJobs() {
|
||||
// await this.emailMonitoringService.cleanupJobs();
|
||||
|
||||
// return {
|
||||
// message: "Cleanup completed successfully",
|
||||
// };
|
||||
// }
|
||||
// }
|
||||
@@ -1,9 +1,16 @@
|
||||
import { MikroOrmModule } from "@mikro-orm/nestjs";
|
||||
import { BullModule } from "@nestjs/bullmq";
|
||||
import { Module } from "@nestjs/common";
|
||||
import { JwtModule } from "@nestjs/jwt";
|
||||
|
||||
// import { EmailGatewayController } from "./controllers/email-gateway.controller";
|
||||
import { EmailController } from "./email.controller";
|
||||
import { EmailGateway } from "./gateways/email.gateway";
|
||||
import { EmailMonitoringProcessor } from "./queue/email-monitoring.processor";
|
||||
import { EmailHeadersService } from "./services/email-headers.service";
|
||||
import { EmailMonitoringSchedulerService } from "./services/email-monitoring-scheduler.service";
|
||||
import { EmailMonitoringService } from "./services/email-monitoring.service";
|
||||
import { EmailNotificationService } from "./services/email-notification.service";
|
||||
import { EmailService } from "./services/email.service";
|
||||
import { MailboxResolverService } from "./services/mailbox-resolver.service";
|
||||
import { jwtConfig } from "../../configs/jwt.config";
|
||||
@@ -11,11 +18,32 @@ import { MailServerModule } from "../mail-server/mail-server.module";
|
||||
import { TemplatesModule } from "../templates/templates.module";
|
||||
import { User } from "../users/entities/user.entity";
|
||||
import { UsersModule } from "../users/users.module";
|
||||
import { EMAIL_QUEUE_CONSTANTS } from "./constants/email-events.constant";
|
||||
import { WebSocketAuthService } from "./services/websocket-auth.service";
|
||||
|
||||
@Module({
|
||||
imports: [MailServerModule, UsersModule, TemplatesModule, JwtModule.registerAsync(jwtConfig()), MikroOrmModule.forFeature([User])],
|
||||
imports: [
|
||||
MailServerModule,
|
||||
UsersModule,
|
||||
TemplatesModule,
|
||||
JwtModule.registerAsync(jwtConfig()),
|
||||
MikroOrmModule.forFeature([User]),
|
||||
BullModule.registerQueue({
|
||||
name: EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE,
|
||||
}),
|
||||
],
|
||||
controllers: [EmailController],
|
||||
providers: [EmailService, MailboxResolverService, EmailHeadersService],
|
||||
exports: [EmailService, MailboxResolverService, EmailHeadersService],
|
||||
providers: [
|
||||
EmailService,
|
||||
MailboxResolverService,
|
||||
EmailHeadersService,
|
||||
EmailGateway,
|
||||
EmailMonitoringService,
|
||||
EmailMonitoringSchedulerService,
|
||||
EmailNotificationService,
|
||||
EmailMonitoringProcessor,
|
||||
WebSocketAuthService,
|
||||
],
|
||||
exports: [EmailService, MailboxResolverService, EmailHeadersService, EmailGateway, EmailMonitoringService, EmailNotificationService],
|
||||
})
|
||||
export class EmailModule {}
|
||||
|
||||
@@ -0,0 +1,268 @@
|
||||
import { Logger, UseFilters, UseGuards, UsePipes, ValidationPipe } from "@nestjs/common";
|
||||
import {
|
||||
ConnectedSocket,
|
||||
OnGatewayConnection,
|
||||
OnGatewayDisconnect,
|
||||
OnGatewayInit,
|
||||
SubscribeMessage,
|
||||
WebSocketGateway,
|
||||
WebSocketServer,
|
||||
} from "@nestjs/websockets";
|
||||
import { Server, Socket } from "socket.io";
|
||||
|
||||
import { WebSocketMessage } from "../../../common/enums/message.enum";
|
||||
import { WsExceptionFilter } from "../../../core/filters/ws-exception.filter";
|
||||
import { WEBSOCKET_EVENTS, WEBSOCKET_NAMESPACES, WEBSOCKET_ROOMS, WebSocketEvent } from "../constants/email-events.constant";
|
||||
import { WebSocketAuthGuard } from "../guards/websocket-auth.guard";
|
||||
import {
|
||||
EmailDeletePayload,
|
||||
EmailPayload,
|
||||
EmailSentPayload,
|
||||
EmailStatusPayload,
|
||||
MailboxUpdatePayload,
|
||||
UnreadCountPayload,
|
||||
} from "../interfaces/email-events.interface";
|
||||
import { ConnectedUserInfo } from "../interfaces/websocket.interface";
|
||||
import { WebSocketAuthService } from "../services/websocket-auth.service";
|
||||
|
||||
@UsePipes(new ValidationPipe({ transform: true }))
|
||||
@UseFilters(WsExceptionFilter)
|
||||
@WebSocketGateway({ namespace: WEBSOCKET_NAMESPACES.EMAIL, cors: { origin: true, credentials: true } })
|
||||
export class EmailGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
|
||||
@WebSocketServer()
|
||||
server: Server;
|
||||
|
||||
private readonly logger = new Logger(EmailGateway.name);
|
||||
|
||||
private readonly connectedUsers = new Map<string, ConnectedUserInfo>();
|
||||
private readonly userSessions = new Map<string, Set<string>>();
|
||||
|
||||
constructor(private readonly authService: WebSocketAuthService) {}
|
||||
|
||||
afterInit(server: Server) {
|
||||
this.logger.log("Email WebSocket Gateway initialized");
|
||||
this.logger.log(`Namespace: ${WEBSOCKET_NAMESPACES.EMAIL}`);
|
||||
|
||||
setInterval(() => {
|
||||
server.emit(WEBSOCKET_EVENTS.PING, { timestamp: new Date().toISOString() });
|
||||
}, 30000);
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async handleConnection(client: Socket) {
|
||||
const connectionContext = {
|
||||
clientId: client.id,
|
||||
clientIP: client.handshake.address,
|
||||
userAgent: client.handshake.headers["user-agent"],
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
this.logger.log("New WebSocket connection attempt", connectionContext);
|
||||
|
||||
try {
|
||||
const authResult = await this.authService.authenticateClient(client);
|
||||
|
||||
if (!authResult.success) {
|
||||
this.authService.handleAuthenticationFailure(client, authResult.error!);
|
||||
return;
|
||||
}
|
||||
|
||||
const user = authResult.user!;
|
||||
this.registerUserConnection(client.id, user.id);
|
||||
|
||||
const userRoom = WEBSOCKET_ROOMS.USER(user.id);
|
||||
await client.join(userRoom);
|
||||
|
||||
const userInfo = this.connectedUsers.get(client.id);
|
||||
this.authService.emitAuthenticationSuccess(client, user);
|
||||
|
||||
this.logger.log("WebSocket connection established", {
|
||||
...connectionContext,
|
||||
userId: user.id,
|
||||
sessionId: userInfo?.sessionId,
|
||||
userRoom,
|
||||
authenticated: true,
|
||||
});
|
||||
} catch (error) {
|
||||
this.logger.error("WebSocket connection error", {
|
||||
...connectionContext,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
|
||||
this.authService.handleAuthenticationFailure(client, WebSocketMessage.CONNECTION_FAILED);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
handleDisconnect(client: Socket): void {
|
||||
const userInfo = this.connectedUsers.get(client.id);
|
||||
|
||||
this.logger.log("WebSocket client disconnected", {
|
||||
clientId: client.id,
|
||||
userId: userInfo?.userId,
|
||||
sessionId: userInfo?.sessionId,
|
||||
});
|
||||
|
||||
if (userInfo) {
|
||||
// Leave user room
|
||||
const userRoom = WEBSOCKET_ROOMS.USER(userInfo.userId);
|
||||
client.leave(userRoom);
|
||||
|
||||
this.cleanupUserConnection(client, userInfo);
|
||||
}
|
||||
|
||||
this.connectedUsers.delete(client.id);
|
||||
}
|
||||
|
||||
//=======================================================
|
||||
//=======================================================
|
||||
|
||||
@SubscribeMessage(WEBSOCKET_EVENTS.PING)
|
||||
@UseGuards(WebSocketAuthGuard)
|
||||
handlePing(@ConnectedSocket() client: Socket) {
|
||||
client.emit(WEBSOCKET_EVENTS.PONG, { timestamp: new Date().toISOString() });
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyNewEmail(userId: string, emailData: EmailPayload) {
|
||||
return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.NEW_EMAIL, emailData, `New email: ${emailData.subject}`);
|
||||
}
|
||||
//************************************************************************ */
|
||||
async notifyEmailRead(userId: string, data: EmailStatusPayload) {
|
||||
return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_READ, data, `Message ${data.messageId} marked as read`);
|
||||
}
|
||||
//************************************************************************ */
|
||||
async notifyEmailDeleted(userId: string, data: EmailDeletePayload) {
|
||||
return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_DELETED, data, `Message ${data.messageId} deleted`);
|
||||
}
|
||||
//************************************************************************ */
|
||||
async notifyEmailSent(userId: string, data: EmailSentPayload) {
|
||||
return await this.sendUserNotification(
|
||||
userId,
|
||||
WEBSOCKET_EVENTS.EMAIL_SENT,
|
||||
data,
|
||||
`Email ${data.messageId} sent to ${data.to.map((t) => t.address).join(", ")}`,
|
||||
);
|
||||
}
|
||||
//************************************************************************ */
|
||||
async notifyUnreadCountUpdate(userId: string, data: UnreadCountPayload) {
|
||||
return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.UNREAD_COUNT_UPDATED, data, `Unread count: ${data.count}`);
|
||||
}
|
||||
//************************************************************************ */
|
||||
|
||||
async notifyMailboxUpdate(userId: string, data: MailboxUpdatePayload) {
|
||||
return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.MAILBOX_UPDATED, data, `Mailbox updated: ${data.mailboxName}`);
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
|
||||
async broadcastSystemMessage(event: WebSocketEvent, data: Record<string, unknown>) {
|
||||
try {
|
||||
const enhancedData = {
|
||||
...data,
|
||||
timestamp: data.timestamp || new Date().toISOString(),
|
||||
};
|
||||
|
||||
this.server.emit(event, enhancedData);
|
||||
this.logger.log(`System message broadcasted: ${event}`);
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to broadcast system message ${event}:`, error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async getConnectedUsersCount(): Promise<number> {
|
||||
const sockets = await this.server.fetchSockets();
|
||||
return sockets.length;
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
getConnectedUserIds(): string[] {
|
||||
return Array.from(this.userSessions.keys());
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
isUserConnected(userId: string): boolean {
|
||||
const userSockets = this.userSessions.get(userId);
|
||||
return userSockets !== undefined && userSockets.size > 0;
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
getConnectedUsersInList(userIds: string[]): string[] {
|
||||
return userIds.filter((userId) => this.isUserConnected(userId));
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async getBusinessConnectedUsers(businessId: string): Promise<string[]> {
|
||||
try {
|
||||
// Note: Currently returning all connected users since business filtering
|
||||
// requires additional user data lookup. This can be enhanced later.
|
||||
this.logger.warn(`getBusinessConnectedUsers called for business ${businessId} - returning all connected users for now`);
|
||||
return this.getConnectedUserIds();
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to get business connected users for ${businessId}:`, error);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// PRIVATE HELPER METHODS
|
||||
// ============================================================================
|
||||
|
||||
private async sendUserNotification(userId: string, event: WebSocketEvent, data: Record<string, any>, context?: string) {
|
||||
try {
|
||||
const userRoom = WEBSOCKET_ROOMS.USER(userId);
|
||||
const enhancedData = {
|
||||
...data,
|
||||
timestamp: data.timestamp || new Date().toISOString(),
|
||||
};
|
||||
|
||||
this.server.to(userRoom).emit(event, enhancedData);
|
||||
|
||||
this.logger.log(`${event} notification sent to user ${userId}${context ? ` - ${context}` : ""}`);
|
||||
return true;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to send ${event} notification to user ${userId}:`, error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
private registerUserConnection(clientId: string, userId: string): void {
|
||||
const sessionId = this.generateSessionId();
|
||||
this.connectedUsers.set(clientId, { userId, sessionId });
|
||||
|
||||
if (!this.userSessions.has(userId)) {
|
||||
this.userSessions.set(userId, new Set());
|
||||
}
|
||||
this.userSessions.get(userId)!.add(clientId);
|
||||
|
||||
this.logger.debug(`User ${userId} registered with session ${sessionId}`);
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
private generateSessionId(): string {
|
||||
return `session_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
private cleanupUserConnection(client: Socket, userInfo: ConnectedUserInfo): void {
|
||||
const userSockets = this.userSessions.get(userInfo.userId);
|
||||
if (userSockets) {
|
||||
userSockets.delete(client.id);
|
||||
if (userSockets.size === 0) {
|
||||
this.userSessions.delete(userInfo.userId);
|
||||
}
|
||||
}
|
||||
|
||||
if (userInfo.sessionId) {
|
||||
client.leave(`session_${userInfo.sessionId}`);
|
||||
|
||||
this.server.to(`session_${userInfo.sessionId}`).emit(WEBSOCKET_EVENTS.CONNECTION_CLOSED, {
|
||||
userId: userInfo.userId,
|
||||
sessionId: userInfo.sessionId,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
import { CanActivate, ExecutionContext, Injectable, Logger } from "@nestjs/common";
|
||||
import { JwtService } from "@nestjs/jwt";
|
||||
import { WsException } from "@nestjs/websockets";
|
||||
import { Socket } from "socket.io";
|
||||
|
||||
import { WebSocketMessage } from "../../../common/enums/message.enum";
|
||||
import { WEBSOCKET_EVENTS } from "../constants/email-events.constant";
|
||||
|
||||
@Injectable()
|
||||
export class WebSocketAuthGuard implements CanActivate {
|
||||
private readonly logger = new Logger(WebSocketAuthGuard.name);
|
||||
|
||||
constructor(private jwtService: JwtService) {}
|
||||
|
||||
async canActivate(context: ExecutionContext): Promise<boolean> {
|
||||
const client: Socket = context.switchToWs().getClient<Socket>();
|
||||
|
||||
try {
|
||||
// Get token from auth object or handshake query
|
||||
const token = this.extractTokenFromHeader(client);
|
||||
|
||||
if (!token) {
|
||||
this.logger.warn("No token provided in WebSocket connection");
|
||||
throw new WsException({
|
||||
event: WEBSOCKET_EVENTS.UNAUTHORIZED,
|
||||
message: WebSocketMessage.AUTHENTICATION_REQUIRED,
|
||||
});
|
||||
}
|
||||
|
||||
// Verify JWT token
|
||||
const payload = await this.jwtService.verifyAsync(token);
|
||||
|
||||
client.data.user = payload;
|
||||
this.logger.log(`WebSocket authentication successful for user: ${payload.id}`);
|
||||
|
||||
return true;
|
||||
} catch (error) {
|
||||
this.logger.error("WebSocket authentication failed:", error);
|
||||
|
||||
client.emit(WEBSOCKET_EVENTS.UNAUTHORIZED, {
|
||||
message: WebSocketMessage.INVALID_TOKEN,
|
||||
});
|
||||
|
||||
throw new WsException({
|
||||
event: WEBSOCKET_EVENTS.UNAUTHORIZED,
|
||||
message: WebSocketMessage.INVALID_TOKEN,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private extractTokenFromHeader(client: Socket): string | undefined {
|
||||
const authToken = client.handshake?.auth?.token || client.handshake?.query?.token;
|
||||
return authToken;
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,26 @@
|
||||
export interface EmailPayload {
|
||||
messageId: number;
|
||||
// Base interface with common properties
|
||||
export interface BaseEventPayload {
|
||||
userId: string;
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
// Base interface for message-related events
|
||||
export interface BaseMessageEventPayload extends BaseEventPayload {
|
||||
messageId: number;
|
||||
}
|
||||
|
||||
// Base interface for mailbox-related events
|
||||
export interface BaseMailboxEventPayload extends BaseEventPayload {
|
||||
mailboxId: string;
|
||||
}
|
||||
|
||||
// Base interface for message events that also involve mailboxes
|
||||
export interface BaseMessageMailboxEventPayload extends BaseMessageEventPayload {
|
||||
mailboxId: string;
|
||||
}
|
||||
|
||||
// Specific email event interfaces
|
||||
export interface EmailPayload extends BaseMessageMailboxEventPayload {
|
||||
from: {
|
||||
name?: string;
|
||||
address: string;
|
||||
@@ -12,64 +32,46 @@ export interface EmailPayload {
|
||||
subject: string;
|
||||
preview?: string;
|
||||
hasAttachments: boolean;
|
||||
timestamp: string;
|
||||
mailboxId: string;
|
||||
mailboxName: string;
|
||||
isRead: boolean;
|
||||
isFlagged: boolean;
|
||||
size: number;
|
||||
}
|
||||
|
||||
export interface EmailStatusPayload {
|
||||
messageId: number;
|
||||
userId: string;
|
||||
mailboxId: string;
|
||||
timestamp: string;
|
||||
}
|
||||
export type EmailStatusPayload = BaseMessageMailboxEventPayload;
|
||||
|
||||
export interface EmailMovePayload {
|
||||
messageId: number;
|
||||
userId: string;
|
||||
export interface EmailMovePayload extends BaseMessageEventPayload {
|
||||
fromMailboxId: string;
|
||||
toMailboxId: string;
|
||||
fromMailboxName: string;
|
||||
toMailboxName: string;
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
export interface EmailDeletePayload {
|
||||
messageId: number;
|
||||
userId: string;
|
||||
mailboxId: string;
|
||||
timestamp: string;
|
||||
}
|
||||
export type EmailDeletePayload = BaseMessageMailboxEventPayload;
|
||||
|
||||
export interface EmailSentPayload {
|
||||
messageId: number;
|
||||
userId: string;
|
||||
export interface EmailSentPayload extends BaseMessageMailboxEventPayload {
|
||||
from: {
|
||||
name?: string;
|
||||
address: string;
|
||||
};
|
||||
to: Array<{
|
||||
name?: string;
|
||||
address: string;
|
||||
}>;
|
||||
subject: string;
|
||||
timestamp: string;
|
||||
mailboxId: string;
|
||||
hasAttachments: boolean;
|
||||
mailboxName: string;
|
||||
isRead: boolean;
|
||||
}
|
||||
|
||||
export interface MailboxUpdatePayload {
|
||||
userId: string;
|
||||
mailboxId: string;
|
||||
export interface MailboxUpdatePayload extends BaseMailboxEventPayload {
|
||||
mailboxName: string;
|
||||
unreadCount: number;
|
||||
totalCount: number;
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
export interface UnreadCountPayload {
|
||||
userId: string;
|
||||
export interface UnreadCountPayload extends BaseEventPayload {
|
||||
totalUnread: number;
|
||||
mailboxCounts: Record<string, number>;
|
||||
timestamp: string;
|
||||
count: number;
|
||||
}
|
||||
|
||||
export interface EmailWebSocketEvents {
|
||||
@@ -93,3 +95,39 @@ export interface EmailWebSocketEvents {
|
||||
ping: { timestamp: string };
|
||||
pong: { timestamp: string };
|
||||
}
|
||||
|
||||
export interface EmailMonitoringJob {
|
||||
userId: string;
|
||||
lastCheckTimestamp?: Date;
|
||||
}
|
||||
|
||||
export interface NewEmailData {
|
||||
userId: string;
|
||||
messageId: number;
|
||||
wildduckUserId: string;
|
||||
mailboxId: string;
|
||||
}
|
||||
|
||||
// Helper interfaces for notification service parameters
|
||||
export interface EmailSentNotificationParams {
|
||||
userId: string;
|
||||
messageId: number;
|
||||
mailboxId: string;
|
||||
recipients: Array<{ name?: string; address: string }>;
|
||||
subject: string;
|
||||
from: { name?: string; address: string };
|
||||
}
|
||||
|
||||
export interface EmailStatusNotificationParams {
|
||||
userId: string;
|
||||
messageId: number;
|
||||
mailboxId: string;
|
||||
}
|
||||
|
||||
export interface MailboxUpdateNotificationParams {
|
||||
userId: string;
|
||||
mailboxId: string;
|
||||
mailboxName: string;
|
||||
unreadCount: number;
|
||||
totalCount: number;
|
||||
}
|
||||
|
||||
@@ -1,134 +1,52 @@
|
||||
import { Socket } from "socket.io";
|
||||
|
||||
import { ITokenPayload } from "../../auth/interfaces/IToken-payload";
|
||||
|
||||
/**
|
||||
* Information about a connected user
|
||||
* Extended Socket interface with authenticated user data
|
||||
*/
|
||||
export interface AuthenticatedSocket extends Socket {
|
||||
data: {
|
||||
user: ITokenPayload;
|
||||
sessionId?: string;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* User connection information stored in gateway
|
||||
*/
|
||||
export interface ConnectedUserInfo {
|
||||
userId: string;
|
||||
sessionId?: string;
|
||||
connectedAt?: Date;
|
||||
lastActivity?: Date;
|
||||
metadata?: Record<string, any>;
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket authentication result
|
||||
* WebSocket error context for logging and debugging
|
||||
*/
|
||||
export interface WebSocketAuthResult {
|
||||
success: boolean;
|
||||
user?: WebSocketUser;
|
||||
error?: string;
|
||||
errorCode?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* User information extracted from JWT token
|
||||
*/
|
||||
export interface WebSocketUser {
|
||||
id: string;
|
||||
email?: string;
|
||||
permissions?: string[];
|
||||
isAdmin?: boolean;
|
||||
businessId?: string;
|
||||
sessionId?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket connection context for logging and debugging
|
||||
*/
|
||||
export interface WebSocketConnectionContext {
|
||||
export interface WebSocketErrorContext {
|
||||
clientId: string;
|
||||
clientIP: string;
|
||||
userAgent?: string;
|
||||
timestamp: string;
|
||||
userId?: string;
|
||||
sessionId?: string;
|
||||
event?: string;
|
||||
data?: unknown;
|
||||
timestamp: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket session information
|
||||
* WebSocket event response structure
|
||||
*/
|
||||
export interface WebSocketSession {
|
||||
id: string;
|
||||
userId: string;
|
||||
createdAt: Date;
|
||||
lastActivity: Date;
|
||||
clientInfo: {
|
||||
userAgent: string;
|
||||
ip: string;
|
||||
};
|
||||
isActive: boolean;
|
||||
export interface WebSocketResponse<T = unknown> {
|
||||
status: "success" | "error";
|
||||
message?: string;
|
||||
data?: T;
|
||||
timestamp?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket room subscription information
|
||||
* Authentication result for WebSocket connections
|
||||
*/
|
||||
export interface RoomSubscription {
|
||||
roomId: string;
|
||||
userId: string;
|
||||
subscribedAt: Date;
|
||||
permissions?: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket error information
|
||||
*/
|
||||
export interface WebSocketError {
|
||||
code: string;
|
||||
message: string;
|
||||
timestamp: string;
|
||||
context?: Record<string, any>;
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket authentication payload from client
|
||||
*/
|
||||
export interface WebSocketAuthPayload {
|
||||
token: string;
|
||||
sessionId?: string;
|
||||
clientInfo?: {
|
||||
userAgent?: string;
|
||||
deviceId?: string;
|
||||
platform?: string;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Extended Socket interface with user data
|
||||
*/
|
||||
export interface AuthenticatedSocket extends Socket {
|
||||
data: {
|
||||
user: WebSocketUser;
|
||||
sessionId?: string;
|
||||
connectedAt: Date;
|
||||
lastActivity: Date;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket health check information
|
||||
*/
|
||||
export interface WebSocketHealthInfo {
|
||||
totalConnections: number;
|
||||
activeUsers: number;
|
||||
activeSessions: number;
|
||||
uptime: number;
|
||||
memoryUsage?: {
|
||||
used: number;
|
||||
total: number;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* WebSocket metrics for monitoring
|
||||
*/
|
||||
export interface WebSocketMetrics {
|
||||
connectionsPerSecond: number;
|
||||
messagesPerSecond: number;
|
||||
errorsPerSecond: number;
|
||||
averageResponseTime: number;
|
||||
activeConnections: number;
|
||||
peakConnections: number;
|
||||
totalMessages: number;
|
||||
totalErrors: number;
|
||||
export interface AuthenticationResult {
|
||||
success: boolean;
|
||||
user?: ITokenPayload;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import { Processor, WorkerHost } from "@nestjs/bullmq";
|
||||
import { Logger } from "@nestjs/common";
|
||||
import { Job } from "bullmq";
|
||||
|
||||
import { EMAIL_QUEUE_CONSTANTS } from "../constants/email-events.constant";
|
||||
import { EmailMonitoringJob } from "../interfaces/email-events.interface";
|
||||
import { EmailMonitoringService } from "../services/email-monitoring.service";
|
||||
|
||||
@Processor(EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE)
|
||||
export class EmailMonitoringProcessor extends WorkerHost {
|
||||
private readonly logger = new Logger(EmailMonitoringProcessor.name);
|
||||
|
||||
constructor(private readonly emailMonitoringService: EmailMonitoringService) {
|
||||
super();
|
||||
}
|
||||
|
||||
async process(job: Job<EmailMonitoringJob>) {
|
||||
const { name, data } = job;
|
||||
|
||||
try {
|
||||
switch (name) {
|
||||
case EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE_PROCESSING:
|
||||
await this.emailMonitoringService.processUserEmailCheck(data);
|
||||
break;
|
||||
|
||||
default:
|
||||
this.logger.warn(`Unknown job type: ${name}`);
|
||||
throw new Error(`Unknown job type: ${name}`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to process email monitoring job ${name} for user ${data.userId}:`, error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
import { Injectable, Logger, OnModuleInit } from "@nestjs/common";
|
||||
|
||||
import { EmailMonitoringService } from "./email-monitoring.service";
|
||||
|
||||
@Injectable()
|
||||
export class EmailMonitoringSchedulerService implements OnModuleInit {
|
||||
private readonly logger = new Logger(EmailMonitoringSchedulerService.name);
|
||||
private monitoringInterval: NodeJS.Timeout | null = null;
|
||||
|
||||
constructor(private readonly emailMonitoringService: EmailMonitoringService) {}
|
||||
|
||||
async onModuleInit() {
|
||||
this.logger.log("Initializing Email Monitoring Scheduler");
|
||||
await this.startMonitoring();
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async startMonitoring() {
|
||||
if (this.monitoringInterval) {
|
||||
this.logger.warn("Email monitoring is already running");
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.log("Starting email monitoring scheduler - checking every 5 seconds");
|
||||
|
||||
await this.runMonitoringCheck();
|
||||
|
||||
this.monitoringInterval = setInterval(async () => {
|
||||
try {
|
||||
await this.runMonitoringCheck();
|
||||
} catch (error) {
|
||||
this.logger.error("Error in scheduled email monitoring:", error);
|
||||
}
|
||||
}, 5000);
|
||||
|
||||
this.logger.log("Email monitoring scheduler started successfully");
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
stopMonitoring() {
|
||||
if (this.monitoringInterval) {
|
||||
clearInterval(this.monitoringInterval);
|
||||
this.monitoringInterval = null;
|
||||
this.logger.log("Email monitoring scheduler stopped");
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
private async runMonitoringCheck() {
|
||||
try {
|
||||
this.logger.log("Running scheduled email monitoring check");
|
||||
await this.emailMonitoringService.scheduleEmailCheck();
|
||||
} catch (error) {
|
||||
this.logger.error("Failed to run email monitoring check:", error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
getStatus() {
|
||||
return {
|
||||
isRunning: this.monitoringInterval !== null,
|
||||
checkInterval: 5000,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async restartMonitoring() {
|
||||
this.logger.log("Restarting email monitoring scheduler");
|
||||
this.stopMonitoring();
|
||||
await this.startMonitoring();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,212 @@
|
||||
import { EntityManager } from "@mikro-orm/core";
|
||||
import { InjectQueue } from "@nestjs/bullmq";
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { Queue } from "bullmq";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import { MailboxResolverService } from "./mailbox-resolver.service";
|
||||
import { MailServerService } from "../../mail-server/services/mail-server.service";
|
||||
import { MailboxEnum } from "../../mailbox/enums/mailbox.enum";
|
||||
import { User } from "../../users/entities/user.entity";
|
||||
import { EMAIL_QUEUE_CONSTANTS } from "../constants/email-events.constant";
|
||||
import { EmailGateway } from "../gateways/email.gateway";
|
||||
import { EmailMonitoringJob, EmailPayload } from "../interfaces/email-events.interface";
|
||||
|
||||
@Injectable()
|
||||
export class EmailMonitoringService {
|
||||
private readonly logger = new Logger(EmailMonitoringService.name);
|
||||
private lastGlobalCheck: Date = new Date();
|
||||
|
||||
constructor(
|
||||
private readonly emailGateway: EmailGateway,
|
||||
private readonly mailboxResolver: MailboxResolverService,
|
||||
private readonly mailServerService: MailServerService,
|
||||
private readonly em: EntityManager,
|
||||
@InjectQueue(EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE) private readonly monitoringQueue: Queue,
|
||||
) {}
|
||||
|
||||
//****************************************************** */
|
||||
async scheduleEmailCheck() {
|
||||
try {
|
||||
// Get only connected users from the websocket gateway
|
||||
const connectedUserIds = this.emailGateway.getConnectedUserIds();
|
||||
|
||||
if (connectedUserIds.length === 0) {
|
||||
return; // No connected users, skip monitoring
|
||||
}
|
||||
|
||||
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,
|
||||
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) {
|
||||
// Handle case where BullMQ serializes Date to string
|
||||
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.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);
|
||||
|
||||
// Get all messages to properly count unread ones
|
||||
const { results } = await firstValueFrom(
|
||||
this.mailServerService.messages.listMessages(wildduckUserId, inboxMailboxId, {
|
||||
limit: 250,
|
||||
unseen: true,
|
||||
}),
|
||||
);
|
||||
|
||||
const unreadCount = results.length;
|
||||
|
||||
// Always send unread count notification
|
||||
this.emailGateway.notifyUnreadCountUpdate(userId, {
|
||||
count: unreadCount,
|
||||
timestamp: new Date().toISOString(),
|
||||
totalUnread: unreadCount,
|
||||
mailboxCounts: {
|
||||
[MailboxEnum.INBOX]: unreadCount,
|
||||
},
|
||||
userId,
|
||||
});
|
||||
|
||||
// Log the unread count if there are unread messages
|
||||
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),
|
||||
});
|
||||
}
|
||||
|
||||
//****************************************************** */
|
||||
async getQueueStatus() {
|
||||
const waiting = await this.monitoringQueue.getWaiting();
|
||||
const active = await this.monitoringQueue.getActive();
|
||||
const completed = await this.monitoringQueue.getCompleted();
|
||||
const failed = await this.monitoringQueue.getFailed();
|
||||
|
||||
return {
|
||||
waiting: waiting.length,
|
||||
active: active.length,
|
||||
completed: completed.length,
|
||||
failed: failed.length,
|
||||
};
|
||||
}
|
||||
|
||||
//****************************************************** */
|
||||
async cleanupJobs() {
|
||||
await this.monitoringQueue.clean(1000 * 60 * 60, 100); // Clean jobs older than 1 hour
|
||||
this.logger.log("Cleaned up old monitoring jobs");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
import { EntityManager } from "@mikro-orm/core";
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import { MailboxResolverService } from "./mailbox-resolver.service";
|
||||
import { MailServerService } from "../../mail-server/services/mail-server.service";
|
||||
import { MailboxEnum } from "../../mailbox/enums/mailbox.enum";
|
||||
import { User } from "../../users/entities/user.entity";
|
||||
import { WebSocketEvent } from "../constants/email-events.constant";
|
||||
import { EmailGateway } from "../gateways/email.gateway";
|
||||
import {
|
||||
EmailDeletePayload,
|
||||
EmailSentNotificationParams,
|
||||
EmailSentPayload,
|
||||
EmailStatusNotificationParams,
|
||||
EmailStatusPayload,
|
||||
MailboxUpdateNotificationParams,
|
||||
MailboxUpdatePayload,
|
||||
UnreadCountPayload,
|
||||
} from "../interfaces/email-events.interface";
|
||||
|
||||
@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 notifyEmailRead(params: EmailStatusNotificationParams) {
|
||||
try {
|
||||
const payload: EmailStatusPayload = {
|
||||
userId: params.userId,
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailRead(params.userId, payload);
|
||||
await this.updateUnreadCount(params.userId);
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email read notification dispatched for user ${params.userId}: message ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email read notification for user ${params.userId}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email read for user ${params.userId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
//************************************************************************ */
|
||||
async notifyEmailDeleted(params: EmailStatusNotificationParams) {
|
||||
try {
|
||||
const payload: EmailDeletePayload = {
|
||||
userId: params.userId,
|
||||
messageId: params.messageId,
|
||||
mailboxId: params.mailboxId,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyEmailDeleted(params.userId, payload);
|
||||
await this.updateUnreadCount(params.userId);
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`Email deleted notification dispatched for user ${params.userId}: message ${params.messageId}`);
|
||||
} else {
|
||||
this.logger.warn(`Failed to dispatch email deleted notification for user ${params.userId}: user not connected`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to notify email deleted 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, {
|
||||
id: 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,
|
||||
count: unreadCount,
|
||||
timestamp: new Date().toISOString(),
|
||||
totalUnread: unreadCount,
|
||||
mailboxCounts: {
|
||||
[inboxMailboxId]: unreadCount,
|
||||
},
|
||||
};
|
||||
|
||||
const success = await this.emailGateway.notifyUnreadCountUpdate(userId, payload);
|
||||
|
||||
if (!success) {
|
||||
this.logger.warn(`Failed to update unread count for user ${userId}: 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(),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ import { BadRequestException, Injectable, Logger } from "@nestjs/common";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import { EmailHeadersService } from "./email-headers.service";
|
||||
import { EmailNotificationService } from "./email-notification.service";
|
||||
import { MailboxResolverService } from "./mailbox-resolver.service";
|
||||
import { EmailMessage } from "../../../common/enums/message.enum";
|
||||
import { MailServerService } from "../../mail-server/services/mail-server.service";
|
||||
@@ -21,6 +22,7 @@ export class EmailService {
|
||||
constructor(
|
||||
private readonly mailServerService: MailServerService,
|
||||
private readonly mailboxResolverService: MailboxResolverService,
|
||||
private readonly emailNotificationService: EmailNotificationService,
|
||||
private readonly templateProcessorService: TemplateProcessorService,
|
||||
private readonly userRepository: UserRepository,
|
||||
private readonly emailHeadersService: EmailHeadersService,
|
||||
@@ -66,6 +68,17 @@ export class EmailService {
|
||||
|
||||
this.logger.log(`Email sent successfully: ${response.message.id} (Queue: ${response.message.queueId})`);
|
||||
|
||||
const sentMailboxId = await this.mailboxResolverService.getSentMailboxId(userId);
|
||||
|
||||
await this.emailNotificationService.notifyEmailSent({
|
||||
userId,
|
||||
messageId: parseInt(response.message.id),
|
||||
mailboxId: sentMailboxId,
|
||||
recipients: sendEmailDto.to,
|
||||
subject: sendEmailDto.subject,
|
||||
from: sendEmailDto.from,
|
||||
});
|
||||
|
||||
return {
|
||||
message: EmailMessage.EMAIL_SENDING_SUCCESS,
|
||||
messageId: response.message.id,
|
||||
@@ -101,15 +114,35 @@ export class EmailService {
|
||||
this.logger.log(`Searching messages for user ${userId} with query: ${query.q}`);
|
||||
|
||||
try {
|
||||
const response = await firstValueFrom(
|
||||
this.mailServerService.messages.searchMessages(userId, { query: query.q || "", limit: query.limit, page: query.page }),
|
||||
);
|
||||
// Transform query parameters for WildDuck API compatibility
|
||||
const wildDuckQuery = {
|
||||
query: query.q || "",
|
||||
limit: query.limit || 20,
|
||||
page: query.page || 1,
|
||||
order: "desc" as const,
|
||||
};
|
||||
|
||||
this.logger.log(`Found ${response.results.length} messages matching search for user: ${userId}`);
|
||||
// Ensure limit is within bounds
|
||||
if (wildDuckQuery.limit > 250) wildDuckQuery.limit = 250;
|
||||
if (wildDuckQuery.limit < 1) wildDuckQuery.limit = 1;
|
||||
if (wildDuckQuery.page < 1) wildDuckQuery.page = 1;
|
||||
|
||||
this.logger.debug("Search query for WildDuck:", wildDuckQuery);
|
||||
|
||||
const response = await firstValueFrom(this.mailServerService.messages.searchMessages(userId, wildDuckQuery));
|
||||
|
||||
this.logger.log(
|
||||
`Found ${response.results.length} messages (${response.total} total, page ${response.page}) matching search for user: ${userId}`,
|
||||
);
|
||||
|
||||
return {
|
||||
message: EmailMessage.MESSAGES_SEARCHED_SUCCESSFULLY,
|
||||
...response,
|
||||
results: response.results,
|
||||
total: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
query: response.query,
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to search messages for user ${userId}: ${error instanceof Error ? error.message : "Unknown error"}`);
|
||||
@@ -123,14 +156,21 @@ export class EmailService {
|
||||
const inboxMailboxId = await this.mailboxResolverService.getInboxMailboxId(userId);
|
||||
this.logger.log(`Listing messages in mailbox ${inboxMailboxId} for user: ${userId}`);
|
||||
|
||||
const { results, total: count } = await firstValueFrom(this.mailServerService.messages.listMessages(userId, inboxMailboxId, query));
|
||||
// Transform query parameters for WildDuck API compatibility
|
||||
const wildDuckQuery = this.transformPaginationQuery(query);
|
||||
this.logger.debug("Transformed query for WildDuck:", wildDuckQuery);
|
||||
|
||||
this.logger.log(`Found ${count} messages in mailbox ${inboxMailboxId} for user: ${userId}`);
|
||||
const response = await firstValueFrom(this.mailServerService.messages.listMessages(userId, inboxMailboxId, wildDuckQuery));
|
||||
|
||||
this.logger.log(`Found ${response.total} total messages (page ${response.page}) in mailbox ${inboxMailboxId} for user: ${userId}`);
|
||||
|
||||
return {
|
||||
message: EmailMessage.MESSAGES_LISTED_SUCCESSFULLY,
|
||||
results,
|
||||
count,
|
||||
results: response.results,
|
||||
count: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
paginate: true,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -201,6 +241,12 @@ export class EmailService {
|
||||
|
||||
this.logger.log(`Message ${messageId} marked as seen successfully in ${messageLocation.mailboxName}`);
|
||||
|
||||
await this.emailNotificationService.notifyEmailRead({
|
||||
userId,
|
||||
messageId,
|
||||
mailboxId: messageLocation.mailboxId,
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
message: EmailMessage.MESSAGE_MARKED_AS_SEEN_SUCCESSFULLY,
|
||||
@@ -218,11 +264,17 @@ export class EmailService {
|
||||
async getSentMessages(userId: string, query: MessageListQueryDto) {
|
||||
try {
|
||||
const sentMailboxId = await this.mailboxResolverService.getSentMailboxId(userId);
|
||||
const { results: sentMails, total: count } = await firstValueFrom(this.mailServerService.messages.listMessages(userId, sentMailboxId, query));
|
||||
const wildDuckQuery = this.transformPaginationQuery(query);
|
||||
const response = await firstValueFrom(this.mailServerService.messages.listMessages(userId, sentMailboxId, wildDuckQuery));
|
||||
|
||||
this.logger.log(`Found ${response.total} sent messages (page ${response.page}) for user: ${userId}`);
|
||||
|
||||
return {
|
||||
sentMails,
|
||||
count,
|
||||
sentMails: response.results,
|
||||
count: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
paginate: true,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -235,10 +287,17 @@ export class EmailService {
|
||||
async getTrashMessages(userId: string, query: MessageListQueryDto) {
|
||||
try {
|
||||
const trashMailboxId = await this.mailboxResolverService.getTrashMailboxId(userId);
|
||||
const { results: trashMails, total: count } = await firstValueFrom(this.mailServerService.messages.listMessages(userId, trashMailboxId, query));
|
||||
const wildDuckQuery = this.transformPaginationQuery(query);
|
||||
const response = await firstValueFrom(this.mailServerService.messages.listMessages(userId, trashMailboxId, wildDuckQuery));
|
||||
|
||||
this.logger.log(`Found ${response.total} trash messages (page ${response.page}) for user: ${userId}`);
|
||||
|
||||
return {
|
||||
trashMails,
|
||||
count,
|
||||
trashMails: response.results,
|
||||
count: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
paginate: true,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -251,11 +310,17 @@ export class EmailService {
|
||||
async getJunkMessages(userId: string, query: MessageListQueryDto) {
|
||||
try {
|
||||
const junkMailboxId = await this.mailboxResolverService.getJunkMailboxId(userId);
|
||||
const { results: junkMails, total: count } = await firstValueFrom(this.mailServerService.messages.listMessages(userId, junkMailboxId, query));
|
||||
const wildDuckQuery = this.transformPaginationQuery(query);
|
||||
const response = await firstValueFrom(this.mailServerService.messages.listMessages(userId, junkMailboxId, wildDuckQuery));
|
||||
|
||||
this.logger.log(`Found ${response.total} junk messages (page ${response.page}) for user: ${userId}`);
|
||||
|
||||
return {
|
||||
junkMails,
|
||||
count,
|
||||
junkMails: response.results,
|
||||
count: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
paginate: true,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -268,13 +333,17 @@ export class EmailService {
|
||||
async getArchivedMessages(userId: string, query: MessageListQueryDto) {
|
||||
try {
|
||||
const archiveMailboxId = await this.mailboxResolverService.getArchiveMailboxId(userId);
|
||||
const { results: archiveMails, total: count } = await firstValueFrom(
|
||||
this.mailServerService.messages.listMessages(userId, archiveMailboxId, query),
|
||||
);
|
||||
const wildDuckQuery = this.transformPaginationQuery(query);
|
||||
const response = await firstValueFrom(this.mailServerService.messages.listMessages(userId, archiveMailboxId, wildDuckQuery));
|
||||
|
||||
this.logger.log(`Found ${response.total} archived messages (page ${response.page}) for user: ${userId}`);
|
||||
|
||||
return {
|
||||
archiveMails,
|
||||
count,
|
||||
archiveMails: response.results,
|
||||
count: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
paginate: true,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -287,17 +356,21 @@ export class EmailService {
|
||||
async getFavoriteMessages(userId: string, query: MessageListQueryDto) {
|
||||
try {
|
||||
const favoriteMailboxId = await this.mailboxResolverService.getFavoriteMailboxId(userId);
|
||||
const { results: favoriteMails, total: count } = await firstValueFrom(
|
||||
this.mailServerService.messages.listMessages(userId, favoriteMailboxId, query),
|
||||
);
|
||||
const wildDuckQuery = this.transformPaginationQuery(query);
|
||||
const response = await firstValueFrom(this.mailServerService.messages.listMessages(userId, favoriteMailboxId, wildDuckQuery));
|
||||
|
||||
this.logger.log(`Found ${response.total} favorite messages (page ${response.page}) for user: ${userId}`);
|
||||
|
||||
return {
|
||||
favoriteMails,
|
||||
count,
|
||||
favoriteMails: response.results,
|
||||
count: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
paginate: true,
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to get junk messages for user ${userId}: ${error instanceof Error ? error.message : "Unknown error"}`);
|
||||
this.logger.error(`Failed to get favorite messages for user ${userId}: ${error instanceof Error ? error.message : "Unknown error"}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
@@ -307,13 +380,17 @@ export class EmailService {
|
||||
async getDraftsMessages(userId: string, query: MessageListQueryDto) {
|
||||
try {
|
||||
const draftsMailboxId = await this.mailboxResolverService.getDraftsMailboxId(userId);
|
||||
const wildDuckQuery = this.transformPaginationQuery(query);
|
||||
const response = await firstValueFrom(this.mailServerService.messages.listMessages(userId, draftsMailboxId, wildDuckQuery));
|
||||
|
||||
this.logger.log(`Found ${response.total} draft messages (page ${response.page}) for user: ${userId}`);
|
||||
|
||||
const { results: draftsMail, total: count } = await firstValueFrom(
|
||||
this.mailServerService.messages.listMessages(userId, draftsMailboxId, query),
|
||||
);
|
||||
return {
|
||||
draftsMail,
|
||||
count,
|
||||
draftsMail: response.results,
|
||||
count: response.total,
|
||||
page: response.page,
|
||||
previousCursor: response.previousCursor,
|
||||
nextCursor: response.nextCursor,
|
||||
paginate: true,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -743,6 +820,38 @@ export class EmailService {
|
||||
|
||||
//########################################################
|
||||
|
||||
/**
|
||||
* Transform pagination query for WildDuck API compatibility
|
||||
*/
|
||||
private transformPaginationQuery(query: MessageListQueryDto): Record<string, unknown> {
|
||||
const transformedQuery: Record<string, any> = { ...query };
|
||||
|
||||
// Set default values if not provided
|
||||
if (!transformedQuery.page) transformedQuery.page = 1;
|
||||
if (!transformedQuery.limit) transformedQuery.limit = 20;
|
||||
|
||||
// Ensure limit is within WildDuck's bounds (1-250)
|
||||
if (transformedQuery.limit > 250) transformedQuery.limit = 250;
|
||||
if (transformedQuery.limit < 1) transformedQuery.limit = 1;
|
||||
|
||||
// Ensure page is at least 1
|
||||
if (transformedQuery.page < 1) transformedQuery.page = 1;
|
||||
|
||||
// Set default order if not provided
|
||||
if (!transformedQuery.order) transformedQuery.order = "desc";
|
||||
|
||||
// Remove undefined/null values
|
||||
Object.keys(transformedQuery).forEach((key) => {
|
||||
if (transformedQuery[key] === undefined || transformedQuery[key] === null) {
|
||||
delete transformedQuery[key];
|
||||
}
|
||||
});
|
||||
|
||||
return transformedQuery;
|
||||
}
|
||||
|
||||
//########################################################
|
||||
|
||||
private async findMessageMailbox(userId: string, messageId: number): Promise<{ mailboxId: string; mailboxName: string } | null> {
|
||||
try {
|
||||
const mailboxIds = await this.mailboxResolverService.getUserMailboxIds(userId);
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { JsonWebTokenError, JwtService, NotBeforeError, TokenExpiredError } from "@nestjs/jwt";
|
||||
import { Socket } from "socket.io";
|
||||
|
||||
import { WebSocketMessage } from "../../../common/enums/message.enum";
|
||||
import { ITokenPayload } from "../../auth/interfaces/IToken-payload";
|
||||
import { extractTokenFromClient } from "../../utils/services/extract-token.utils";
|
||||
import { WEBSOCKET_EVENTS, WebSocketEvent } from "../constants/email-events.constant";
|
||||
import { AuthenticationResult, WebSocketErrorContext } from "../interfaces/websocket.interface";
|
||||
|
||||
/**
|
||||
* Service responsible for WebSocket authentication logic
|
||||
* Follows single responsibility principle and provides reusable authentication methods
|
||||
*/
|
||||
@Injectable()
|
||||
export class WebSocketAuthService {
|
||||
private readonly logger = new Logger(WebSocketAuthService.name);
|
||||
|
||||
constructor(private readonly jwtService: JwtService) {}
|
||||
|
||||
/**
|
||||
* Authenticates a WebSocket client and returns the result
|
||||
*
|
||||
* @param client - Socket.IO client to authenticate
|
||||
* @returns Promise with authentication result
|
||||
*/
|
||||
async authenticateClient(client: Socket): Promise<AuthenticationResult> {
|
||||
try {
|
||||
const token = extractTokenFromClient(client);
|
||||
|
||||
if (!token) {
|
||||
this.logAuthenticationAttempt(client, false, "No token provided");
|
||||
return {
|
||||
success: false,
|
||||
error: WebSocketMessage.AUTHENTICATION_REQUIRED,
|
||||
};
|
||||
}
|
||||
|
||||
const payload = await this.jwtService.verifyAsync<ITokenPayload>(token);
|
||||
|
||||
if (!payload?.id) {
|
||||
this.logAuthenticationAttempt(client, false, "Invalid token payload");
|
||||
return {
|
||||
success: false,
|
||||
error: WebSocketMessage.INVALID_TOKEN,
|
||||
};
|
||||
}
|
||||
|
||||
// Store user data in client
|
||||
client.data.user = payload;
|
||||
|
||||
this.logAuthenticationAttempt(client, true, undefined, payload.id);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
user: payload,
|
||||
};
|
||||
} catch (error) {
|
||||
const errorMessage = this.getAuthErrorMessage(error);
|
||||
this.logAuthenticationAttempt(client, false, errorMessage);
|
||||
|
||||
return {
|
||||
success: false,
|
||||
error: errorMessage,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Emits authentication failure event to client and disconnects
|
||||
*
|
||||
* @param client - Socket.IO client
|
||||
* @param error - Error message to send
|
||||
*/
|
||||
handleAuthenticationFailure(client: Socket, error: string): void {
|
||||
client.emit(WEBSOCKET_EVENTS.CONNECTION_ERROR, {
|
||||
message: error,
|
||||
timestamp: new Date().toISOString(),
|
||||
});
|
||||
|
||||
// Graceful disconnect with delay to ensure message is received
|
||||
setTimeout(() => {
|
||||
client.disconnect(true);
|
||||
}, 100);
|
||||
}
|
||||
|
||||
/**
|
||||
* Emits successful authentication event to client
|
||||
*
|
||||
* @param client - Socket.IO client
|
||||
* @param user - Authenticated user payload
|
||||
*/
|
||||
emitAuthenticationSuccess(client: Socket, user: ITokenPayload): void {
|
||||
client.emit(WEBSOCKET_EVENTS.CONNECTION_ESTABLISHED, {
|
||||
message: WebSocketMessage.AUTHENTICATED,
|
||||
userId: user.id,
|
||||
timestamp: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates error context object for logging and monitoring
|
||||
*
|
||||
* @param client - Socket.IO client
|
||||
* @param event - Event name (optional)
|
||||
* @param data - Event data (optional)
|
||||
* @returns WebSocket error context
|
||||
*/
|
||||
createErrorContext(client: Socket, event?: WebSocketEvent, data?: unknown): WebSocketErrorContext {
|
||||
return {
|
||||
clientId: client.id,
|
||||
userId: client.data?.user?.id,
|
||||
sessionId: client.data?.sessionId,
|
||||
event,
|
||||
data,
|
||||
timestamp: new Date(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Validates if client is authenticated
|
||||
*
|
||||
* @param client - Socket.IO client to validate
|
||||
* @returns True if client has valid user data
|
||||
*/
|
||||
isClientAuthenticated(client: Socket): boolean {
|
||||
return !!client.data?.user?.id;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets user ID from authenticated client
|
||||
*
|
||||
* @param client - Socket.IO client
|
||||
* @returns User ID or undefined if not authenticated
|
||||
*/
|
||||
getUserId(client: Socket): string | undefined {
|
||||
return client.data?.user?.id;
|
||||
}
|
||||
|
||||
/**
|
||||
* Logs authentication attempts with structured data
|
||||
*/
|
||||
private logAuthenticationAttempt(client: Socket, success: boolean, error?: string, userId?: string): void {
|
||||
const logData = {
|
||||
clientId: client.id,
|
||||
clientIP: client.handshake.address,
|
||||
success,
|
||||
userId,
|
||||
error,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
if (success) {
|
||||
this.logger.log(`WebSocket authentication successful`, logData);
|
||||
} else {
|
||||
this.logger.warn(`WebSocket authentication failed`, logData);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Maps JWT errors to user-friendly messages
|
||||
*/
|
||||
private getAuthErrorMessage(error: unknown): string {
|
||||
if (error instanceof TokenExpiredError) {
|
||||
return WebSocketMessage.TOKEN_EXPIRED;
|
||||
}
|
||||
|
||||
if (error instanceof JsonWebTokenError) {
|
||||
return WebSocketMessage.INVALID_TOKEN;
|
||||
}
|
||||
|
||||
if (error instanceof NotBeforeError) {
|
||||
return WebSocketMessage.INVALID_TOKEN;
|
||||
}
|
||||
|
||||
return WebSocketMessage.INVALID_TOKEN;
|
||||
}
|
||||
}
|
||||
@@ -32,6 +32,11 @@ export class ListMessagesQueryDto {
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
previous?: string;
|
||||
|
||||
@ApiPropertyOptional({ description: "Unseen messages only" })
|
||||
@IsOptional()
|
||||
@IsBoolean()
|
||||
unseen?: boolean;
|
||||
}
|
||||
|
||||
export class UpdateMessageDto {
|
||||
|
||||
@@ -1,199 +0,0 @@
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import { MailServerService } from "./services/mail-server.service";
|
||||
|
||||
/**
|
||||
* Example service showing how to use the Mail Server Gateway
|
||||
* in other modules of your NestJS application
|
||||
*/
|
||||
@Injectable()
|
||||
export class MailServerGatewayExample {
|
||||
private readonly logger = new Logger(MailServerGatewayExample.name);
|
||||
|
||||
constructor(private readonly mailServerService: MailServerService) {}
|
||||
|
||||
/**
|
||||
* Example: Create a complete email account
|
||||
*/
|
||||
async createUserAccount(username: string, password: string, email?: string) {
|
||||
try {
|
||||
const result = await firstValueFrom(
|
||||
this.mailServerService.createEmailAccount({
|
||||
username,
|
||||
password,
|
||||
email,
|
||||
name: username,
|
||||
quota: 1024 * 1024 * 1024, // 1GB
|
||||
}),
|
||||
);
|
||||
|
||||
this.logger.log(`Created user account: ${result.user.id}`);
|
||||
return result;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to create user account: ${(error as Error).message}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Send an email through the gateway
|
||||
*/
|
||||
async sendWelcomeEmail(userId: string, recipientEmail: string, recipientName: string) {
|
||||
try {
|
||||
const result = await firstValueFrom(
|
||||
this.mailServerService.sendMessage(userId, {
|
||||
from: {
|
||||
name: "Support Team",
|
||||
address: "support@yourcompany.com",
|
||||
},
|
||||
to: [
|
||||
{
|
||||
name: recipientName,
|
||||
address: recipientEmail,
|
||||
},
|
||||
],
|
||||
subject: "Welcome to Our Service!",
|
||||
text: `Hello ${recipientName},\n\nWelcome to our email service!`,
|
||||
html: `<h1>Hello ${recipientName}!</h1><p>Welcome to our email service!</p>`,
|
||||
}),
|
||||
);
|
||||
|
||||
this.logger.log(`Welcome email sent: ${result.id}`);
|
||||
return result;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to send welcome email: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Setup 2FA for a user
|
||||
*/
|
||||
async enable2FAForUser(userId: string) {
|
||||
try {
|
||||
const result = await firstValueFrom(this.mailServerService.setup2FAForUser(userId, "Your Company"));
|
||||
|
||||
this.logger.log(`2FA setup completed for user: ${userId}`);
|
||||
return {
|
||||
qrCode: result.qrcode,
|
||||
backupCodes: result.backupCodes,
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to setup 2FA: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Get user's recent emails
|
||||
*/
|
||||
async getUserRecentEmails(userId: string, mailboxId: string) {
|
||||
try {
|
||||
const messages = await firstValueFrom(this.mailServerService.getRecentMessages(userId, mailboxId, 10));
|
||||
|
||||
this.logger.log(`Retrieved ${messages.length} recent messages for user: ${userId}`);
|
||||
return messages;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to get recent emails: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Search user's emails
|
||||
*/
|
||||
async searchUserEmails(userId: string, searchQuery: string) {
|
||||
try {
|
||||
const results = await firstValueFrom(this.mailServerService.searchAllMessages(userId, searchQuery, 50));
|
||||
|
||||
this.logger.log(`Found ${results.length} matching messages for query: ${searchQuery}`);
|
||||
return results;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to search emails: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Get system health status
|
||||
*/
|
||||
async getSystemHealth() {
|
||||
try {
|
||||
const status = await firstValueFrom(this.mailServerService.getSystemStatus());
|
||||
|
||||
this.logger.log("Retrieved system status");
|
||||
return status;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to get system status: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Direct access to specific services
|
||||
*/
|
||||
async directServiceAccess() {
|
||||
try {
|
||||
// Direct access to users service
|
||||
const users = await firstValueFrom(this.mailServerService.users.listUsers({ limit: 10 }));
|
||||
|
||||
// Direct access to health service
|
||||
const health = await firstValueFrom(this.mailServerService.health.getHealthStatus());
|
||||
|
||||
// Direct access to storage service
|
||||
const storageInfo = await firstValueFrom(this.mailServerService.storage.getStorageInfo());
|
||||
|
||||
// Direct access to DKIM service
|
||||
const dkimKeys = await firstValueFrom(this.mailServerService.dkim.listDKIMKeys());
|
||||
|
||||
return {
|
||||
users: users.results,
|
||||
health,
|
||||
storage: storageInfo,
|
||||
dkimKeys: dkimKeys.results,
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to access services directly: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Setup DKIM for a domain
|
||||
*/
|
||||
async setupDKIMForDomain(domain: string) {
|
||||
try {
|
||||
const result = await firstValueFrom(this.mailServerService.createDKIMForDomain(domain));
|
||||
|
||||
this.logger.log(`DKIM key created for domain: ${domain}`);
|
||||
this.logger.log(`Add this DNS TXT record: ${result.dnsTxt.name} = ${result.dnsTxt.value}`);
|
||||
|
||||
return result;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to setup DKIM: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Create webhook for message events
|
||||
*/
|
||||
async setupMessageWebhook(webhookUrl: string, userId?: string) {
|
||||
try {
|
||||
const result = await firstValueFrom(
|
||||
this.mailServerService.webhooks.createWebhook({
|
||||
type: ["messageNew", "messageDeleted"],
|
||||
url: webhookUrl,
|
||||
user: userId,
|
||||
}),
|
||||
);
|
||||
|
||||
this.logger.log(`Webhook created: ${result.id}`);
|
||||
return result;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to create webhook: ${error instanceof Error ? error.message : String(error)}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,4 @@
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { Observable, catchError, map, switchMap, throwError } from "rxjs";
|
||||
import { Injectable } from "@nestjs/common";
|
||||
|
||||
import { WildDuckAddressesService } from "./wildduck-addresses.service";
|
||||
import { WildDuckApplicationPasswordsService } from "./wildduck-application-passwords.service";
|
||||
@@ -24,8 +23,6 @@ import { DANAK_IMAPS_SERVER } from "../../../common/constants";
|
||||
|
||||
@Injectable()
|
||||
export class MailServerService {
|
||||
private readonly logger = new Logger(MailServerService.name);
|
||||
|
||||
constructor(
|
||||
private readonly usersService: WildDuckUsersService,
|
||||
private readonly messagesService: WildDuckMessagesService,
|
||||
@@ -136,229 +133,4 @@ export class MailServerService {
|
||||
get twoFactor() {
|
||||
return this.twoFactorService;
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to create a complete email account
|
||||
*/
|
||||
createEmailAccount(data: { username: string; password: string; email?: string; name?: string; quota?: number }): Observable<{
|
||||
user: { id: string };
|
||||
address?: { id: string };
|
||||
}> {
|
||||
this.logger.log(`Creating email account for user: ${data.username}`);
|
||||
|
||||
return this.usersService
|
||||
.createUser({
|
||||
username: data.username,
|
||||
password: data.password,
|
||||
address: data.email,
|
||||
name: data.name,
|
||||
quota: data.quota,
|
||||
})
|
||||
.pipe(
|
||||
map((userResult) => {
|
||||
this.logger.log(`User created with ID: ${userResult.id}`);
|
||||
return { user: { id: userResult.id } };
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Failed to create email account: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to get user's full mailbox structure with message counts
|
||||
*/
|
||||
getUserMailboxes(userId: string): Observable<any> {
|
||||
this.logger.log(`Getting mailboxes for user: ${userId}`);
|
||||
|
||||
return this.mailboxesService.listMailboxes(userId, { counters: true }).pipe(
|
||||
map((result) => {
|
||||
this.logger.log(`Found ${result.results.length} mailboxes for user: ${userId}`);
|
||||
return result.results;
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Failed to get mailboxes: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to authenticate user and get account info
|
||||
*/
|
||||
authenticateAndGetUser(
|
||||
username: string,
|
||||
password: string,
|
||||
): Observable<{
|
||||
auth: any;
|
||||
user: any;
|
||||
}> {
|
||||
this.logger.log(`Authenticating user: ${username}`);
|
||||
|
||||
return this.authService.authenticate({ username, password }).pipe(
|
||||
switchMap((authResult) => {
|
||||
if (!authResult.success) {
|
||||
throw new Error("Authentication failed");
|
||||
}
|
||||
|
||||
this.logger.log(`User authenticated: ${authResult.id}`);
|
||||
|
||||
// Get user details after successful authentication
|
||||
return this.usersService.getUser(authResult.id).pipe(
|
||||
map((userDetails) => ({
|
||||
auth: authResult,
|
||||
user: userDetails,
|
||||
})),
|
||||
);
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Authentication failed: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to send a message
|
||||
*/
|
||||
sendMessage(
|
||||
userId: string,
|
||||
messageData: {
|
||||
from: { name?: string; address: string };
|
||||
to: Array<{ name?: string; address: string }>;
|
||||
subject: string;
|
||||
text?: string;
|
||||
html?: string;
|
||||
},
|
||||
): Observable<{ id: string; queueId: string }> {
|
||||
this.logger.log(`Sending message for user: ${userId}`);
|
||||
|
||||
return this.submissionService.submitMessage(userId, messageData).pipe(
|
||||
map((result) => {
|
||||
this.logger.log(`Message submitted with ID: ${result.message.id}`);
|
||||
return { id: result.message.id, queueId: result.message.queueId };
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Failed to send message: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to get user's recent messages
|
||||
*/
|
||||
getRecentMessages(userId: string, mailboxId: string, limit: number = 20): Observable<any[]> {
|
||||
this.logger.log(`Getting recent messages for user: ${userId}`);
|
||||
|
||||
return this.messagesService.listMessages(userId, mailboxId, { limit, order: "desc" }).pipe(
|
||||
map((result) => {
|
||||
this.logger.log(`Found ${result.results.length} messages`);
|
||||
return result.results;
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Failed to get messages: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to search across all user's messages
|
||||
*/
|
||||
searchAllMessages(userId: string, searchQuery: string, limit: number = 50): Observable<any[]> {
|
||||
this.logger.log(`Searching messages for user: ${userId}, query: ${searchQuery}`);
|
||||
|
||||
return this.messagesService.searchMessages(userId, { query: searchQuery, limit }).pipe(
|
||||
map((result) => {
|
||||
this.logger.log(`Found ${result.results.length} matching messages`);
|
||||
return result.results;
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Search failed: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to check server health and get system status
|
||||
*/
|
||||
getSystemStatus(): Observable<{
|
||||
health: any;
|
||||
storage: any;
|
||||
}> {
|
||||
this.logger.log("Getting system status");
|
||||
|
||||
return this.healthService.getHealthStatus().pipe(
|
||||
switchMap((healthStatus) => {
|
||||
return this.storageService.getStorageInfo().pipe(
|
||||
map((storageInfo) => ({
|
||||
health: healthStatus,
|
||||
storage: storageInfo,
|
||||
})),
|
||||
);
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Failed to get system status: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to setup complete 2FA for user
|
||||
*/
|
||||
setup2FAForUser(
|
||||
userId: string,
|
||||
issuer?: string,
|
||||
): Observable<{
|
||||
qrcode: string;
|
||||
backupCodes: string[];
|
||||
}> {
|
||||
this.logger.log(`Setting up 2FA for user: ${userId}`);
|
||||
|
||||
return this.twoFactorService.setupTOTP(userId, { issuer }).pipe(
|
||||
switchMap((totpResult) => {
|
||||
return this.twoFactorService.generateBackupCodes(userId).pipe(
|
||||
map((backupResult) => ({
|
||||
qrcode: totpResult.qrcode,
|
||||
backupCodes: backupResult.codes,
|
||||
})),
|
||||
);
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Failed to setup 2FA: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level method to create DKIM key and get DNS settings
|
||||
*/
|
||||
createDKIMForDomain(
|
||||
domain: string,
|
||||
selector: string = "default",
|
||||
): Observable<{
|
||||
id: string;
|
||||
dnsTxt: { name: string; value: string };
|
||||
}> {
|
||||
this.logger.log(`Creating DKIM key for domain: ${domain}`);
|
||||
|
||||
return this.dkimService.createDKIMKey({ domain, selector, description: `DKIM key for ${domain}` }).pipe(
|
||||
map((result) => {
|
||||
this.logger.log(`DKIM key created for domain: ${domain}`);
|
||||
return {
|
||||
id: result.id,
|
||||
dnsTxt: result.dnsTxt,
|
||||
};
|
||||
}),
|
||||
catchError((error) => {
|
||||
this.logger.error(`Failed to create DKIM key: ${error.message}`);
|
||||
return throwError(() => error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,31 +4,24 @@ import { ConfigService } from "@nestjs/config";
|
||||
import { AxiosResponse } from "axios";
|
||||
import { Observable, catchError, map, throwError } from "rxjs";
|
||||
|
||||
import { MailServerException } from "../../../core/exceptions/mail-server.exceptions";
|
||||
import { WildDuckErrorResponse, WildDuckSuccessResponse } from "../interfaces/wildduck-response.interface";
|
||||
|
||||
@Injectable()
|
||||
export class WildDuckBaseService {
|
||||
protected readonly logger = new Logger(this.constructor.name);
|
||||
protected readonly baseUrl: string;
|
||||
protected readonly accessToken?: string;
|
||||
protected readonly baseURL: string;
|
||||
|
||||
constructor(
|
||||
protected readonly httpService: HttpService,
|
||||
protected readonly configService: ConfigService,
|
||||
) {
|
||||
this.baseUrl = this.configService.get<string>("WILDDUCK_BASE_URL", "http://localhost:8080");
|
||||
this.accessToken = this.configService.get<string>("WILDDUCK_ACCESS_TOKEN");
|
||||
this.baseURL = this.configService.getOrThrow<string>("WILDDUCK_BASE_URL");
|
||||
}
|
||||
|
||||
protected buildUrl(path: string, params?: Record<string, any>): string {
|
||||
const url = new URL(path, this.baseUrl);
|
||||
protected buildUrl(path: string, params?: Record<string, unknown>): string {
|
||||
const url = new URL(path, this.baseURL);
|
||||
|
||||
// Add access token if configured
|
||||
if (this.accessToken) {
|
||||
url.searchParams.set("accessToken", this.accessToken);
|
||||
}
|
||||
|
||||
// Add other parameters
|
||||
if (params) {
|
||||
Object.entries(params).forEach(([key, value]) => {
|
||||
if (value !== undefined && value !== null) {
|
||||
@@ -48,12 +41,10 @@ export class WildDuckBaseService {
|
||||
const data = response.data;
|
||||
|
||||
if ("success" in data && data.success === false) {
|
||||
throw new Error(data.error || "WildDuck API error");
|
||||
throw new MailServerException(data.error || "WildDuck API error");
|
||||
}
|
||||
|
||||
if ("success" in data && data.success === true) {
|
||||
// Return the entire response data for paginated responses
|
||||
// or extract specific data field for simple responses
|
||||
return data as T;
|
||||
}
|
||||
|
||||
@@ -63,7 +54,7 @@ export class WildDuckBaseService {
|
||||
this.logger.error(`WildDuck API error: ${error.message}`, error.stack);
|
||||
|
||||
if (error.response?.data?.error) {
|
||||
return throwError(() => new Error(error.response.data.error));
|
||||
return throwError(() => new MailServerException(error.response.data.error));
|
||||
}
|
||||
|
||||
return throwError(() => error);
|
||||
@@ -73,6 +64,10 @@ export class WildDuckBaseService {
|
||||
|
||||
protected get<T>(path: string, params?: Record<string, any>): Observable<T> {
|
||||
const url = this.buildUrl(path, params);
|
||||
this.logger.debug(`WildDuck GET: ${url}`);
|
||||
if (params) {
|
||||
this.logger.debug(`WildDuck GET Params:`, params);
|
||||
}
|
||||
return this.handleResponse<T>(this.httpService.get(url));
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
import { Socket } from "socket.io";
|
||||
|
||||
/**
|
||||
* Extracts authentication token from WebSocket client handshake
|
||||
* Supports multiple token sources for flexibility
|
||||
*
|
||||
* @param client - Socket.IO client instance
|
||||
* @returns Authentication token string or undefined if not found
|
||||
*/
|
||||
export function extractTokenFromClient(client: Socket): string | undefined {
|
||||
// Priority order: auth.token > query.token > headers.authorization
|
||||
const authToken = client.handshake?.auth?.token;
|
||||
if (authToken) {
|
||||
return authToken;
|
||||
}
|
||||
|
||||
const queryToken = client.handshake?.query?.token;
|
||||
if (queryToken && typeof queryToken === "string") {
|
||||
return queryToken;
|
||||
}
|
||||
|
||||
const authHeader = client.handshake?.headers?.authorization;
|
||||
if (authHeader && typeof authHeader === "string") {
|
||||
// Remove 'Bearer ' prefix if present
|
||||
return authHeader.replace(/^Bearer\s+/i, "");
|
||||
}
|
||||
|
||||
return undefined;
|
||||
}
|
||||
Reference in New Issue
Block a user