chore: websocket

src/modules/email/WEBSOCKET_USAGE.md

@@ -0,0 +1,318 @@
+# Email WebSocket Gateway Usage
+
+This document explains how to use the WebSocket gateway for real-time email notifications.
+
+## Overview
+
+The Email WebSocket Gateway provides real-time notifications for email events such as:
+
+- New message arrivals
+- Message status updates (read/unread, flagged/unflagged)
+- Message movements (archive, favorite, delete)
+- Connection status updates
+
+## Connection
+
+### WebSocket URL
+
+```
+ws://localhost:3000/email
+```
+
+### Authentication
+
+The gateway requires JWT authentication. Pass the token in one of these ways:
+
+**Method 1: Auth object**
+
+```javascript
+const socket = io("ws://localhost:3000/email", {
+  auth: {
+    token: "your-jwt-token",
+  },
+});
+```
+
+**Method 2: Query parameter**
+
+```javascript
+const socket = io("ws://localhost:3000/email?token=your-jwt-token");
+```
+
+## Client-Side Implementation
+
+### Basic Setup (JavaScript)
+
+```javascript
+import { io } from "socket.io-client";
+
+const socket = io("ws://localhost:3000/email", {
+  auth: {
+    token: localStorage.getItem("authToken"), // Your JWT token
+  },
+});
+
+// Connection events
+socket.on("connect", () => {
+  console.log("Connected to email gateway");
+});
+
+socket.on("connection_established", (data) => {
+  console.log("Connection established:", data);
+});
+
+socket.on("connection_error", (error) => {
+  console.error("Connection error:", error);
+});
+
+socket.on("disconnect", () => {
+  console.log("Disconnected from email gateway");
+});
+```
+
+### Listening for Email Events
+
+```javascript
+// New message notification
+socket.on("new_message", (message) => {
+  console.log("New message received:", message);
+  // Update UI with new message
+  updateInboxUI(message);
+  showNotification(`New email from ${message.from.name}: ${message.subject}`);
+});
+
+// Message status update
+socket.on("message_status_update", (update) => {
+  console.log("Message status updated:", update);
+  // Update message status in UI
+  updateMessageStatus(update.messageId, update.status);
+});
+
+// Message moved
+socket.on("message_moved", (moveData) => {
+  console.log("Message moved:", moveData);
+  // Update UI to reflect message movement
+  moveMessageInUI(moveData.messageId, moveData.fromMailboxName, moveData.toMailboxName);
+});
+
+// Message deleted
+socket.on("message_deleted", (deleteData) => {
+  console.log("Message deleted:", deleteData);
+  // Remove message from UI
+  removeMessageFromUI(deleteData.messageId);
+});
+```
+
+### Joining/Leaving User Rooms
+
+```javascript
+// Join user room (usually done automatically on connection)
+socket.emit("join_user_room", {
+  userId: "user-id",
+  clientId: socket.id,
+  timestamp: new Date().toISOString(),
+});
+
+// Leave user room
+socket.emit("leave_user_room", {
+  userId: "user-id",
+  clientId: socket.id,
+  timestamp: new Date().toISOString(),
+});
+```
+
+### React Hook Example
+
+```javascript
+import { useEffect, useState } from "react";
+import { io } from "socket.io-client";
+
+export function useEmailWebSocket(authToken) {
+  const [socket, setSocket] = useState(null);
+  const [isConnected, setIsConnected] = useState(false);
+  const [newMessages, setNewMessages] = useState([]);
+
+  useEffect(() => {
+    if (!authToken) return;
+
+    const newSocket = io("ws://localhost:3000/email", {
+      auth: { token: authToken },
+    });
+
+    newSocket.on("connect", () => {
+      setIsConnected(true);
+    });
+
+    newSocket.on("disconnect", () => {
+      setIsConnected(false);
+    });
+
+    newSocket.on("new_message", (message) => {
+      setNewMessages((prev) => [...prev, message]);
+    });
+
+    newSocket.on("message_status_update", (update) => {
+      // Handle message status updates
+      console.log("Message status update:", update);
+    });
+
+    setSocket(newSocket);
+
+    return () => {
+      newSocket.disconnect();
+    };
+  }, [authToken]);
+
+  return { socket, isConnected, newMessages };
+}
+```
+
+## Event Types
+
+### Server to Client Events
+
+#### `new_message`
+
+Emitted when a new message arrives.
+
+```typescript
+interface EmailNotificationPayload {
+  messageId: number;
+  userId: string;
+  from: { name?: string; address: string };
+  to: Array<{ name?: string; address: string }>;
+  subject: string;
+  preview?: string;
+  hasAttachments: boolean;
+  received: string;
+  mailboxId: string;
+  mailboxName: string;
+  isRead: boolean;
+  isFlagged: boolean;
+  size: number;
+}
+```
+
+#### `message_status_update`
+
+Emitted when a message's status changes.
+
+```typescript
+interface EmailStatusUpdatePayload {
+  messageId: number;
+  userId: string;
+  status: "read" | "unread" | "flagged" | "unflagged" | "archived" | "deleted";
+  mailboxId: string;
+  timestamp: string;
+}
+```
+
+#### `message_moved`
+
+Emitted when a message is moved between mailboxes.
+
+```typescript
+interface EmailMovePayload {
+  messageId: number;
+  userId: string;
+  fromMailboxId: string;
+  toMailboxId: string;
+  fromMailboxName: string;
+  toMailboxName: string;
+  timestamp: string;
+}
+```
+
+#### `message_deleted`
+
+Emitted when a message is deleted.
+
+```typescript
+interface EmailDeletePayload {
+  messageId: number;
+  userId: string;
+  mailboxId: string;
+  mailboxName: string;
+  timestamp: string;
+}
+```
+
+### Client to Server Events
+
+#### `join_user_room`
+
+Join a user's room to receive notifications.
+
+```typescript
+interface ClientJoinPayload {
+  userId: string;
+  clientId: string;
+  timestamp: string;
+}
+```
+
+#### `leave_user_room`
+
+Leave a user's room to stop receiving notifications.
+
+## Error Handling
+
+```javascript
+socket.on("connection_error", (error) => {
+  console.error("WebSocket connection error:", error);
+
+  switch (error.error) {
+    case "Authentication required":
+      // Redirect to login
+      break;
+    case "Invalid token":
+      // Refresh token and reconnect
+      break;
+    case "Authentication failed":
+      // Show error message
+      break;
+    default:
+      console.error("Unknown error:", error);
+  }
+});
+```
+
+## Best Practices
+
+1. **Authentication**: Always include a valid JWT token
+2. **Reconnection**: Handle connection drops gracefully
+3. **Error Handling**: Implement proper error handling for all events
+4. **Memory Management**: Clean up event listeners when components unmount
+5. **Rate Limiting**: Don't overwhelm the server with too many connections
+
+## Testing
+
+You can test the WebSocket gateway using the simulation endpoint:
+
+```javascript
+// Call the simulation endpoint from your backend
+await emailNotificationService.simulateNewMessage("user-id", {
+  subject: "Test Message",
+  from: { name: "Test User", address: "test@example.com" },
+});
+```
+
+## Troubleshooting
+
+### Common Issues
+
+1. **Connection Refused**: Check if the server is running and WebSocket dependencies are installed
+2. **Authentication Failed**: Verify JWT token is valid and not expired
+3. **No Events Received**: Ensure you're subscribed to the correct user room
+4. **CORS Issues**: Configure CORS settings in the gateway properly
+
+### Debug Mode
+
+Enable debug mode to see detailed logs:
+
+```javascript
+const socket = io("ws://localhost:3000/email", {
+  auth: { token: authToken },
+  debug: true,
+});
+```

src/modules/email/constants/email-events.constant.ts

@@ -0,0 +1,47 @@
+export const WEBSOCKET_EVENTS = {
+  // Connection Events
+  CONNECTION_ESTABLISHED: "connection_established",
+  CONNECTION_ERROR: "connection_error",
+  USER_JOINED: "user_joined",
+  USER_LEFT: "user_left",
+
+  // Email Events - Server to Client
+  NEW_MESSAGE: "new_message",
+  MESSAGE_STATUS_UPDATE: "message_status_update",
+  MESSAGE_MOVED: "message_moved",
+  MESSAGE_DELETED: "message_deleted",
+
+  // Room Management - Client to Server
+  JOIN_USER_ROOM: "join_user_room",
+  LEAVE_USER_ROOM: "leave_user_room",
+  JOIN_SESSION: "join_session",
+  LEAVE_SESSION: "leave_session",
+
+  // Authentication Events
+  AUTHENTICATE: "authenticate",
+  AUTHENTICATION_SUCCESS: "authentication_success",
+  AUTHENTICATION_FAILED: "authentication_failed",
+
+  // System Events
+  PING: "ping",
+  PONG: "pong",
+  RECONNECT: "reconnect",
+  FORCE_DISCONNECT: "force_disconnect",
+} as const;
+
+export const WEBSOCKET_ROOMS = {
+  USER: (userId: string) => `user_${userId}`,
+  SESSION: (sessionId: string) => `session_${sessionId}`,
+  BROADCAST: "broadcast",
+} as const;
+
+export const WEBSOCKET_NAMESPACES = {
+  EMAIL: "/email",
+  NOTIFICATIONS: "/notifications",
+  ADMIN: "/admin",
+} as const;
+
+// Type definitions for better TypeScript support
+export type WebSocketEvent = (typeof WEBSOCKET_EVENTS)[keyof typeof WEBSOCKET_EVENTS];
+export type WebSocketRoom = string;
+export type WebSocketNamespace = (typeof WEBSOCKET_NAMESPACES)[keyof typeof WEBSOCKET_NAMESPACES];

src/modules/email/email.gateway.ts

@@ -0,0 +1,227 @@
+import { Logger, UseFilters, UseGuards, UsePipes, ValidationPipe } from "@nestjs/common";
+import {
+  ConnectedSocket,
+  MessageBody,
+  OnGatewayConnection,
+  OnGatewayDisconnect,
+  OnGatewayInit,
+  SubscribeMessage,
+  WebSocketGateway,
+  WebSocketServer,
+} from "@nestjs/websockets";
+import { Server, Socket } from "socket.io";
+
+import { WEBSOCKET_EVENTS } from "./constants/email-events.constant";
+import {
+  ClientJoinPayload,
+  EmailDeletePayload,
+  EmailMovePayload,
+  EmailNotificationPayload,
+  EmailStatusUpdatePayload,
+} from "./interfaces/email-events.interface";
+import { ConnectedUserInfo } from "./interfaces/websocket.interface";
+import { WebSocketAuthService } from "./services/websocket-auth.service";
+import { WebSocketMessage } from "../../common/enums/message.enum";
+import { WsExceptionFilter } from "../../core/filters/ws-exception.filter";
+import { WsAuthGuard } from "../auth/guards/ws-auth.guard";
+
+@WebSocketGateway({
+  cors: { origin: "*", credentials: true },
+  namespace: "/email",
+})
+@UseFilters(WsExceptionFilter)
+@UseGuards(WsAuthGuard)
+@UsePipes(new ValidationPipe())
+export class EmailGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
+  private readonly logger = new Logger(EmailGateway.name);
+
+  // Connection state management
+  private readonly userConnections: Map<string, Set<string>> = new Map(); // userId -> Set of socketIds
+  private readonly connectedUsers = new Map<string, ConnectedUserInfo>();
+  private readonly userSessions = new Map<string, Set<string>>();
+
+  @WebSocketServer()
+  server: Server;
+  constructor(private readonly authService: WebSocketAuthService) {}
+
+  afterInit(server: Server): void {
+    this.logger.log("WebSocket Gateway initialized", {
+      namespace: "/email",
+      cors: true,
+      serverInstance: !!server,
+    });
+  }
+
+  /**
+   * Handles new WebSocket connections with authentication
+   * Implements fail-fast principle for security
+   */
+  async handleConnection(client: Socket): Promise<void> {
+    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);
+      this.authService.emitAuthenticationSuccess(client, user);
+
+      this.logger.log("WebSocket connection established", {
+        ...connectionContext,
+        userId: user.id,
+        authenticated: true,
+      });
+    } catch (error) {
+      this.logger.error("WebSocket connection error", {
+        ...connectionContext,
+        error: error instanceof Error ? error.message : String(error),
+      });
+
+      this.authService.handleAuthenticationFailure(client, WebSocketMessage.WS_CONNECTION_FAILED);
+    }
+  }
+
+  /**
+   * Handles client disconnections with cleanup
+   */
+  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) {
+      this.cleanupUserConnection(client, userInfo);
+    }
+
+    this.connectedUsers.delete(client.id);
+  }
+
+  @SubscribeMessage("join_user_room")
+  async handleJoinUserRoom(@ConnectedSocket() client: Socket, @MessageBody() data: ClientJoinPayload) {
+    try {
+      const userId = client.data.userId;
+
+      if (userId !== data.userId) {
+        client.emit("connection_error", { error: "Unauthorized", timestamp: new Date().toISOString() });
+        return;
+      }
+
+      await client.join(`user:${userId}`);
+      this.logger.log(`Client ${client.id} joined room for user ${userId}`);
+
+      return { success: true, message: "Joined user room successfully" };
+    } catch (error) {
+      this.logger.error(`Error joining user room: ${error instanceof Error ? error.message : "Unknown error"}`);
+      client.emit("connection_error", { error: "Failed to join room", timestamp: new Date().toISOString() });
+    }
+  }
+
+  @SubscribeMessage("leave_user_room")
+  async handleLeaveUserRoom(@ConnectedSocket() client: Socket, @MessageBody() data: ClientJoinPayload) {
+    try {
+      const userId = client.data.userId;
+
+      if (userId !== data.userId) {
+        client.emit("connection_error", { error: "Unauthorized", timestamp: new Date().toISOString() });
+        return;
+      }
+
+      await client.leave(`user:${userId}`);
+      this.logger.log(`Client ${client.id} left room for user ${userId}`);
+
+      return { success: true, message: "Left user room successfully" };
+    } catch (error) {
+      this.logger.error(`Error leaving user room: ${error instanceof Error ? error.message : "Unknown error"}`);
+      client.emit("connection_error", { error: "Failed to leave room", timestamp: new Date().toISOString() });
+    }
+  }
+
+  // Methods to emit events to clients
+  async notifyNewMessage(payload: EmailNotificationPayload) {
+    this.logger.log(`Emitting new message notification to user ${payload.userId}`);
+    this.server.to(`user:${payload.userId}`).emit("new_message", payload);
+  }
+
+  async notifyMessageStatusUpdate(payload: EmailStatusUpdatePayload) {
+    this.logger.log(`Emitting message status update to user ${payload.userId}`);
+    this.server.to(`user:${payload.userId}`).emit("message_status_update", payload);
+  }
+
+  async notifyMessageMoved(payload: EmailMovePayload) {
+    this.logger.log(`Emitting message moved notification to user ${payload.userId}`);
+    this.server.to(`user:${payload.userId}`).emit("message_moved", payload);
+  }
+
+  async notifyMessageDeleted(payload: EmailDeletePayload) {
+    this.logger.log(`Emitting message deleted notification to user ${payload.userId}`);
+    this.server.to(`user:${payload.userId}`).emit("message_deleted", payload);
+  }
+
+  // ============================================================================
+  // PRIVATE HELPER METHODS
+  // ============================================================================
+  /**
+   * Cleans up user connection on disconnect
+   */
+  private cleanupUserConnection(client: Socket, userInfo: ConnectedUserInfo): void {
+    // Remove from user sessions
+    const userSockets = this.userSessions.get(userInfo.userId);
+    if (userSockets) {
+      userSockets.delete(client.id);
+      if (userSockets.size === 0) {
+        this.userSessions.delete(userInfo.userId);
+      }
+    }
+
+    // Leave session room if joined
+    if (userInfo.sessionId) {
+      client.leave(`session_${userInfo.sessionId}`);
+
+      // Notify others in session
+      this.server.to(`session_${userInfo.sessionId}`).emit(WEBSOCKET_EVENTS.USER_LEFT, {
+        userId: userInfo.userId,
+        sessionId: userInfo.sessionId,
+      });
+    }
+  }
+
+  /**
+   * Registers a new user connection
+   */
+  private registerUserConnection(clientId: string, userId: string): void {
+    this.connectedUsers.set(clientId, { userId });
+
+    if (!this.userSessions.has(userId)) {
+      this.userSessions.set(userId, new Set());
+    }
+    this.userSessions.get(userId)!.add(clientId);
+  }
+
+  public getUserConnections(userId: string): number {
+    return this.userConnections.get(userId)?.size || 0;
+  }
+
+  public isUserConnected(userId: string): boolean {
+    return this.userConnections.has(userId) && this.userConnections.get(userId)!.size > 0;
+  }
+
+  public getAllConnectedUsers(): string[] {
+    return Array.from(this.userConnections.keys());
+  }
+}

src/modules/email/email.module.ts

@@ -1,15 +1,20 @@
 import { Module } from "@nestjs/common";
+import { JwtModule } from "@nestjs/jwt";
 
 import { EmailController } from "./email.controller";
+import { EmailGateway } from "./email.gateway";
+import { EmailNotificationService } from "./services/email-notification.service";
 import { EmailService } from "./services/email.service";
 import { MailboxResolverService } from "./services/mailbox-resolver.service";
+import { WebSocketAuthService } from "./services/websocket-auth.service";
+import { jwtConfig } from "../../configs/jwt.config";
 import { MailServerModule } from "../mail-server/mail-server.module";
 import { UsersModule } from "../users/users.module";
 
 @Module({
-  imports: [MailServerModule, UsersModule],
+  imports: [MailServerModule, UsersModule, JwtModule.registerAsync(jwtConfig())],
   controllers: [EmailController],
-  providers: [EmailService, MailboxResolverService],
-  exports: [EmailService, MailboxResolverService],
+  providers: [EmailService, MailboxResolverService, EmailGateway, EmailNotificationService, WebSocketAuthService],
+  exports: [EmailService, MailboxResolverService, EmailGateway, EmailNotificationService, WebSocketAuthService],
 })
 export class EmailModule {}

src/modules/email/interfaces/email-events.interface.ts

@@ -0,0 +1,69 @@
+export interface EmailNotificationPayload {
+  messageId: number;
+  userId: string;
+  from: {
+    name?: string;
+    address: string;
+  };
+  to: Array<{
+    name?: string;
+    address: string;
+  }>;
+  subject: string;
+  preview?: string;
+  hasAttachments: boolean;
+  received: string;
+  mailboxId: string;
+  mailboxName: string;
+  isRead: boolean;
+  isFlagged: boolean;
+  size: number;
+}
+
+export interface EmailStatusUpdatePayload {
+  messageId: number;
+  userId: string;
+  status: "read" | "unread" | "flagged" | "unflagged" | "archived" | "deleted";
+  mailboxId: string;
+  timestamp: string;
+}
+
+export interface EmailMovePayload {
+  messageId: number;
+  userId: string;
+  fromMailboxId: string;
+  toMailboxId: string;
+  fromMailboxName: string;
+  toMailboxName: string;
+  timestamp: string;
+}
+
+export interface EmailDeletePayload {
+  messageId: number;
+  userId: string;
+  mailboxId: string;
+  mailboxName: string;
+  timestamp: string;
+}
+
+export interface ClientJoinPayload {
+  userId: string;
+  clientId: string;
+  timestamp: string;
+}
+
+export interface EmailWebSocketEvents {
+  // Client to Server events
+  join_user_room: ClientJoinPayload;
+  leave_user_room: ClientJoinPayload;
+
+  // Server to Client events
+  new_message: EmailNotificationPayload;
+  message_status_update: EmailStatusUpdatePayload;
+  message_moved: EmailMovePayload;
+  message_deleted: EmailDeletePayload;
+  user_joined: ClientJoinPayload;
+  user_left: ClientJoinPayload;
+  connection_established: { userId: string; timestamp: string };
+  connection_error: { error: string; timestamp: string };
+}

src/modules/email/interfaces/websocket.interface.ts

@@ -0,0 +1,134 @@
+import { Socket } from "socket.io";
+
+/**
+ * Information about a connected user
+ */
+export interface ConnectedUserInfo {
+  userId: string;
+  sessionId?: string;
+  connectedAt?: Date;
+  lastActivity?: Date;
+  metadata?: Record<string, any>;
+}
+
+/**
+ * WebSocket authentication result
+ */
+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 {
+  clientId: string;
+  clientIP: string;
+  userAgent?: string;
+  timestamp: string;
+  userId?: string;
+  sessionId?: string;
+}
+
+/**
+ * WebSocket session information
+ */
+export interface WebSocketSession {
+  id: string;
+  userId: string;
+  createdAt: Date;
+  lastActivity: Date;
+  clientInfo: {
+    userAgent: string;
+    ip: string;
+  };
+  isActive: boolean;
+}
+
+/**
+ * WebSocket room subscription information
+ */
+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;
+}

src/modules/email/services/email-notification.service.ts

@@ -0,0 +1,158 @@
+import { Injectable, Logger } from "@nestjs/common";
+import { firstValueFrom } from "rxjs";
+
+import { MailServerService } from "../../mail-server/services/mail-server.service";
+import { EmailGateway } from "../email.gateway";
+import { MailboxResolverService } from "./mailbox-resolver.service";
+import { EmailNotificationPayload } from "../interfaces/email-events.interface";
+
+@Injectable()
+export class EmailNotificationService {
+  private readonly logger = new Logger(EmailNotificationService.name);
+
+  constructor(
+    private readonly emailGateway: EmailGateway,
+    private readonly mailServerService: MailServerService,
+    private readonly mailboxResolverService: MailboxResolverService,
+  ) {}
+
+  /**
+   * Handle new incoming message and emit WebSocket notification
+   * This method can be called by webhook handlers, polling services, etc.
+   */
+  async handleNewMessage(userId: string, messageId: number, mailboxId?: string) {
+    try {
+      this.logger.log(`Processing new message notification for user ${userId}, message ${messageId}`);
+
+      // Get the mailbox ID if not provided
+      const targetMailboxId = mailboxId || (await this.mailboxResolverService.getInboxMailboxId(userId));
+
+      // Fetch the message details
+      const message = await firstValueFrom(this.mailServerService.messages.getMessage(userId, targetMailboxId, messageId));
+
+      // Determine mailbox name
+      const mailboxName = await this.getMailboxName(userId, targetMailboxId);
+
+      // Create notification payload
+      const notificationPayload: EmailNotificationPayload = {
+        messageId,
+        userId,
+        from: {
+          name: message.from?.name || message.from?.address?.split("@")[0] || "Unknown",
+          address: message.from?.address || "unknown@example.com",
+        },
+        to:
+          message.to?.map((recipient) => ({
+            name: recipient.name || recipient.address?.split("@")[0] || "Unknown",
+            address: recipient.address || "unknown@example.com",
+          })) || [],
+        subject: message.subject || "No Subject",
+        preview: this.extractPreview(
+          typeof message.text === "string" ? message.text : Array.isArray(message.html) ? message.html.join("") : message.html || "",
+        ),
+        hasAttachments: Array.isArray(message.attachments) && message.attachments.length > 0,
+        received: message.idate || new Date().toISOString(),
+        mailboxId: targetMailboxId,
+        mailboxName,
+        isRead: message.seen || false,
+        isFlagged: message.flagged || false,
+        size: message.size || 0,
+      };
+
+      // Emit WebSocket notification
+      await this.emailGateway.notifyNewMessage(notificationPayload);
+
+      this.logger.log(`New message notification sent for user ${userId}, message ${messageId}`);
+
+      return notificationPayload;
+    } catch (error) {
+      this.logger.error(
+        `Failed to process new message notification for user ${userId}, message ${messageId}: ${
+          error instanceof Error ? error.message : "Unknown error"
+        }`,
+      );
+      throw error;
+    }
+  }
+
+  /**
+   * Handle bulk new messages (useful for initial sync or batch processing)
+   */
+  async handleBulkNewMessages(userId: string, messageIds: number[], mailboxId?: string) {
+    this.logger.log(`Processing bulk new message notifications for user ${userId}, ${messageIds.length} messages`);
+
+    const results = [];
+    for (const messageId of messageIds) {
+      try {
+        const result = await this.handleNewMessage(userId, messageId, mailboxId);
+        results.push(result);
+      } catch (error) {
+        this.logger.error(`Failed to process message ${messageId} in bulk notification: ${error}`);
+      }
+    }
+
+    return results;
+  }
+
+  /**
+   * Simulate new message arrival (for testing purposes)
+   */
+  async simulateNewMessage(userId: string, messageData: Partial<EmailNotificationPayload>) {
+    this.logger.log(`Simulating new message for user ${userId}`);
+
+    const simulatedPayload: EmailNotificationPayload = {
+      messageId: Math.floor(Math.random() * 10000),
+      userId,
+      from: {
+        name: "Test Sender",
+        address: "test@example.com",
+      },
+      to: [
+        {
+          name: "Test Recipient",
+          address: "recipient@example.com",
+        },
+      ],
+      subject: "Test Message",
+      preview: "This is a test message for WebSocket notification",
+      hasAttachments: false,
+      received: new Date().toISOString(),
+      mailboxId: "inbox",
+      mailboxName: "Inbox",
+      isRead: false,
+      isFlagged: false,
+      size: 1024,
+      ...messageData,
+    };
+
+    await this.emailGateway.notifyNewMessage(simulatedPayload);
+
+    return simulatedPayload;
+  }
+
+  private async getMailboxName(userId: string, mailboxId: string): Promise<string> {
+    try {
+      const mailboxIds = await this.mailboxResolverService.getUserMailboxIds(userId);
+
+      // Check which mailbox this ID corresponds to
+      if (mailboxIds.inbox === mailboxId) return "Inbox";
+      if (mailboxIds.sent === mailboxId) return "Sent";
+      if (mailboxIds.drafts === mailboxId) return "Drafts";
+      if (mailboxIds.trash === mailboxId) return "Trash";
+      if (mailboxIds.junk === mailboxId) return "Junk";
+      if (mailboxIds.archive === mailboxId) return "Archive";
+      if (mailboxIds.favorite === mailboxId) return "Favorite";
+
+      return "Unknown";
+    } catch (error) {
+      this.logger.error(`Failed to get mailbox name for ${mailboxId}: ${error}`);
+      return "Unknown";
+    }
+  }
+
+  private extractPreview(content: string): string {
+    // Remove HTML tags and get first 150 characters
+    const textContent = content.replace(/<[^>]*>/g, "").trim();
+    return textContent.length > 150 ? textContent.substring(0, 150) + "..." : textContent;
+  }
+}

src/modules/email/services/email.service.ts

@@ -6,6 +6,8 @@ import { EmailMessage } from "../../../common/enums/message.enum";
 import { MailServerService } from "../../mail-server/services/mail-server.service";
 import { MessageListQueryDto, SearchMessagesQueryDto } from "../DTO/email-query.dto";
 import { SendEmailDto, UpdateDraftDto } from "../DTO/send-email.dto";
+import { EmailGateway } from "../email.gateway";
+import { EmailDeletePayload, EmailMovePayload, EmailStatusUpdatePayload } from "../interfaces/email-events.interface";
 
 @Injectable()
 export class EmailService {
@@ -14,6 +16,7 @@ export class EmailService {
   constructor(
     private readonly mailServerService: MailServerService,
     private readonly mailboxResolverService: MailboxResolverService,
+    private readonly emailGateway: EmailGateway,
     // private readonly userService: UsersService,
   ) {}
 
@@ -131,6 +134,16 @@ export class EmailService {
       const inboxMailboxId = await this.mailboxResolverService.getInboxMailboxId(userId);
       const result = await firstValueFrom(this.mailServerService.messages.deleteMessage(userId, inboxMailboxId, messageId));
 
+      // Emit WebSocket notification
+      const payload: EmailDeletePayload = {
+        messageId,
+        userId,
+        mailboxId: inboxMailboxId,
+        mailboxName: "Inbox",
+        timestamp: new Date().toISOString(),
+      };
+      await this.emailGateway.notifyMessageDeleted(payload);
+
       return {
         success: true,
         message: EmailMessage.MESSAGE_DELETED_SUCCESSFULLY,
@@ -148,6 +161,16 @@ export class EmailService {
       const inboxMailboxId = await this.mailboxResolverService.getInboxMailboxId(userId);
       const result = await firstValueFrom(this.mailServerService.messages.updateMessage(userId, inboxMailboxId, messageId, { seen: true }));
 
+      // Emit WebSocket notification
+      const payload: EmailStatusUpdatePayload = {
+        messageId,
+        userId,
+        status: "read",
+        mailboxId: inboxMailboxId,
+        timestamp: new Date().toISOString(),
+      };
+      await this.emailGateway.notifyMessageStatusUpdate(payload);
+
       return {
         success: true,
         message: EmailMessage.MESSAGE_MARKED_AS_SEEN_SUCCESSFULLY,
@@ -380,6 +403,18 @@ export class EmailService {
         }),
       );
 
+      // Emit WebSocket notification
+      const payload: EmailMovePayload = {
+        messageId,
+        userId,
+        fromMailboxId: inboxMailboxId,
+        toMailboxId: archiveMailboxId,
+        fromMailboxName: "Inbox",
+        toMailboxName: "Archive",
+        timestamp: new Date().toISOString(),
+      };
+      await this.emailGateway.notifyMessageMoved(payload);
+
       this.logger.log(`Message ${messageId} moved to archive successfully`);
 
       return {
@@ -409,6 +444,18 @@ export class EmailService {
         }),
       );
 
+      // Emit WebSocket notification
+      const payload: EmailMovePayload = {
+        messageId,
+        userId,
+        fromMailboxId: inboxMailboxId,
+        toMailboxId: favoriteMailboxId,
+        fromMailboxName: "Inbox",
+        toMailboxName: "Favorite",
+        timestamp: new Date().toISOString(),
+      };
+      await this.emailGateway.notifyMessageMoved(payload);
+
       this.logger.log(`Message ${messageId} moved to favorite successfully`);
 
       return {

src/modules/email/services/websocket-auth.service.ts

@@ -0,0 +1,240 @@
+import { Injectable, Logger } from "@nestjs/common";
+import { JwtService } from "@nestjs/jwt";
+import { Socket } from "socket.io";
+
+import { WebSocketMessage } from "../../../common/enums/message.enum";
+import { WEBSOCKET_EVENTS, WEBSOCKET_ROOMS } from "../constants/email-events.constant";
+import { AuthenticatedSocket, WebSocketAuthResult, WebSocketConnectionContext, WebSocketUser } from "../interfaces/websocket.interface";
+
+@Injectable()
+export class WebSocketAuthService {
+  private readonly logger = new Logger(WebSocketAuthService.name);
+  private readonly connectionTimeouts = new Map<string, NodeJS.Timeout>();
+
+  constructor(private readonly jwtService: JwtService) {}
+
+  /**
+   * Authenticates a WebSocket client using JWT token
+   */
+  async authenticateClient(client: Socket): Promise<WebSocketAuthResult> {
+    try {
+      const token = this.extractToken(client);
+
+      if (!token) {
+        return {
+          success: false,
+          error: WebSocketMessage.WS_AUTHENTICATION_REQUIRED,
+        };
+      }
+
+      const decoded = await this.verifyToken(token);
+      if (!decoded) {
+        return {
+          success: false,
+          error: WebSocketMessage.WS_INVALID_TOKEN,
+        };
+      }
+
+      const user = this.extractUserFromToken(decoded);
+      if (!user) {
+        return {
+          success: false,
+          error: WebSocketMessage.WS_INVALID_TOKEN,
+        };
+      }
+
+      // Set user data on socket
+      client.data = {
+        user,
+        connectedAt: new Date(),
+        lastActivity: new Date(),
+        sessionId: user.sessionId || this.generateSessionId(),
+      };
+
+      // Join user to their personal room
+      await client.join(WEBSOCKET_ROOMS.USER(user.id));
+
+      return {
+        success: true,
+        user,
+      };
+    } catch (error) {
+      this.logger.error("WebSocket authentication error", {
+        clientId: client.id,
+        error: error instanceof Error ? error.message : String(error),
+      });
+
+      return {
+        success: false,
+        error: WebSocketMessage.WS_AUTHENTICATION_FAILED,
+      };
+    }
+  }
+
+  /**
+   * Handles authentication failure by emitting error and disconnecting client
+   */
+  handleAuthenticationFailure(client: Socket, errorMessage: string): void {
+    const context: WebSocketConnectionContext = {
+      clientId: client.id,
+      clientIP: client.handshake.address,
+      userAgent: client.handshake.headers["user-agent"],
+      timestamp: new Date().toISOString(),
+    };
+
+    this.logger.warn("WebSocket authentication failed", {
+      ...context,
+      errorMessage,
+    });
+
+    // Emit error to client
+    client.emit(WEBSOCKET_EVENTS.CONNECTION_ERROR, {
+      error: errorMessage,
+      timestamp: new Date().toISOString(),
+      retryAllowed: this.shouldAllowRetry(client),
+    });
+
+    // Set timeout before disconnecting to allow client to handle error
+    const timeout = setTimeout(() => {
+      if (client.connected) {
+        client.disconnect(true);
+      }
+      this.connectionTimeouts.delete(client.id);
+    }, 1000);
+
+    this.connectionTimeouts.set(client.id, timeout);
+  }
+
+  /**
+   * Emits successful authentication event to client
+   */
+  emitAuthenticationSuccess(client: Socket, user: WebSocketUser): void {
+    const authSocket = client as AuthenticatedSocket;
+
+    client.emit(WEBSOCKET_EVENTS.CONNECTION_ESTABLISHED, {
+      userId: user.id,
+      sessionId: authSocket.data.sessionId,
+      timestamp: new Date().toISOString(),
+      permissions: user.permissions || [],
+    });
+
+    this.logger.log("WebSocket authentication successful", {
+      clientId: client.id,
+      userId: user.id,
+      sessionId: authSocket.data.sessionId,
+    });
+  }
+
+  /**
+   * Verifies if user has permission to access a specific room
+   */
+  canAccessRoom(user: WebSocketUser, roomId: string): boolean {
+    // Basic room access control
+    if (roomId.startsWith("user_")) {
+      const targetUserId = roomId.replace("user_", "");
+      return user.id === targetUserId || user.isAdmin === true;
+    }
+
+    if (roomId.startsWith("session_")) {
+      return true; // Sessions are generally accessible
+    }
+
+    if (roomId === "broadcast") {
+      return user.isAdmin === true;
+    }
+
+    // Default deny for unknown room patterns
+    return false;
+  }
+
+  /**
+   * Updates last activity timestamp for authenticated socket
+   */
+  updateLastActivity(client: Socket): void {
+    if (client.data?.user) {
+      client.data.lastActivity = new Date();
+    }
+  }
+
+  /**
+   * Cleans up connection timeouts on disconnect
+   */
+  cleanupConnection(client: Socket): void {
+    const timeout = this.connectionTimeouts.get(client.id);
+    if (timeout) {
+      clearTimeout(timeout);
+      this.connectionTimeouts.delete(client.id);
+    }
+  }
+
+  // Private helper methods
+
+  private extractToken(client: Socket): string | null {
+    // Try to get token from auth object first (recommended)
+    const authToken = client.handshake.auth?.token;
+    if (authToken && typeof authToken === "string") {
+      return authToken;
+    }
+
+    // Fallback to query parameter
+    const queryToken = client.handshake.query?.token;
+    if (queryToken && typeof queryToken === "string") {
+      return queryToken;
+    }
+
+    // Try headers as last resort
+    const headerToken = client.handshake.headers.authorization;
+    if (headerToken && typeof headerToken === "string") {
+      return headerToken.replace("Bearer ", "");
+    }
+
+    return null;
+  }
+
+  private async verifyToken(token: string): Promise<any> {
+    try {
+      return await this.jwtService.verifyAsync(token);
+    } catch (error) {
+      if (error instanceof Error) {
+        if (error.name === "TokenExpiredError") {
+          throw new Error(WebSocketMessage.WS_TOKEN_EXPIRED);
+        }
+        if (error.name === "JsonWebTokenError") {
+          throw new Error(WebSocketMessage.WS_INVALID_TOKEN);
+        }
+        if (error.name === "NotBeforeError") {
+          throw new Error(WebSocketMessage.WS_INVALID_TOKEN);
+        }
+      }
+      throw new Error(WebSocketMessage.WS_INVALID_TOKEN);
+    }
+  }
+
+  private extractUserFromToken(decoded: any): WebSocketUser | null {
+    try {
+      return {
+        id: decoded.wildduckUserId || decoded.sub || decoded.id,
+        email: decoded.email,
+        permissions: decoded.permissions || [],
+        isAdmin: decoded.isAdmin || false,
+        businessId: decoded.businessId,
+        sessionId: decoded.sessionId,
+      };
+    } catch (error) {
+      this.logger.error("Failed to extract user from token", {
+        error: error instanceof Error ? error.message : String(error),
+      });
+      return null;
+    }
+  }
+
+  private shouldAllowRetry(_client: Socket): boolean {
+    // Implement retry logic based on client IP or other factors
+    // For now, always allow retry
+    return true;
+  }
+
+  private generateSessionId(): string {
+    return `ws_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
+  }
+}