diff --git a/FRONTEND_WEBSOCKET_INTEGRATION.md b/FRONTEND_WEBSOCKET_INTEGRATION.md new file mode 100644 index 0000000..5691791 --- /dev/null +++ b/FRONTEND_WEBSOCKET_INTEGRATION.md @@ -0,0 +1,849 @@ +# 📧 Email WebSocket Integration Guide for React + +This guide will help you integrate real-time email notifications into your React application using WebSocket connections. + +## 🚀 Quick Start + +### Installation + +```bash +npm install socket.io-client +# or +yarn add socket.io-client +``` + +### Basic Setup + +```typescript +import io from "socket.io-client"; + +const socket = io("ws://localhost:3000/email", { + auth: { + token: localStorage.getItem("jwt-token"), // Your JWT authentication token + }, +}); +``` + +## 🔌 Connection Setup + +### 1. Basic Connection Hook + +```typescript +// hooks/useEmailWebSocket.ts +import { useEffect, useState } from "react"; +import io, { Socket } from "socket.io-client"; + +interface UseEmailWebSocketOptions { + token: string; + serverUrl?: string; + onConnect?: () => void; + onDisconnect?: () => void; + onError?: (error: string) => void; +} + +export const useEmailWebSocket = ({ token, serverUrl = "ws://localhost:3000/email", onConnect, onDisconnect, onError }: UseEmailWebSocketOptions) => { + const [socket, setSocket] = useState(null); + const [isConnected, setIsConnected] = useState(false); + + useEffect(() => { + if (!token) return; + + const newSocket = io(serverUrl, { + auth: { token }, + transports: ["websocket", "polling"], + }); + + // Connection events + newSocket.on("connect", () => { + console.log("✅ Connected to email WebSocket"); + setIsConnected(true); + onConnect?.(); + }); + + newSocket.on("disconnect", () => { + console.log("❌ Disconnected from email WebSocket"); + setIsConnected(false); + onDisconnect?.(); + }); + + newSocket.on("connection_established", (data) => { + console.log("🎉 Email WebSocket connection established:", data); + }); + + newSocket.on("connection_error", (error) => { + console.error("💥 Email WebSocket connection error:", error); + onError?.(error.error); + }); + + // Health monitoring + newSocket.on("ping", () => { + newSocket.emit("pong", { timestamp: new Date().toISOString() }); + }); + + setSocket(newSocket); + + return () => { + newSocket.close(); + setSocket(null); + }; + }, [token, serverUrl]); + + return { socket, isConnected }; +}; +``` + +### 2. Email Events Hook + +```typescript +// hooks/useEmailEvents.ts +import { useEffect, useState } from "react"; +import { Socket } from "socket.io-client"; + +// TypeScript interfaces (copy from your backend) +export interface EmailPayload { + messageId: number; + userId: string; + from: { + name?: string; + address: string; + }; + to: Array<{ + name?: string; + address: string; + }>; + 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 interface UnreadCountPayload { + userId: string; + totalUnread: number; + mailboxCounts: Record; + timestamp: string; + count: number; +} + +interface EmailEventHandlers { + onNewEmail?: (email: EmailPayload) => void; + onEmailRead?: (data: EmailStatusPayload) => void; + onEmailDeleted?: (data: EmailStatusPayload) => void; + onEmailSent?: (data: any) => void; + onUnreadCountUpdated?: (data: UnreadCountPayload) => void; + onMailboxUpdated?: (data: any) => void; +} + +export const useEmailEvents = (socket: Socket | null, handlers: EmailEventHandlers) => { + useEffect(() => { + if (!socket) return; + + // Register event handlers + if (handlers.onNewEmail) { + socket.on("new_email", handlers.onNewEmail); + } + + if (handlers.onEmailRead) { + socket.on("email_read", handlers.onEmailRead); + } + + if (handlers.onEmailDeleted) { + socket.on("email_deleted", handlers.onEmailDeleted); + } + + if (handlers.onEmailSent) { + socket.on("email_sent", handlers.onEmailSent); + } + + if (handlers.onUnreadCountUpdated) { + socket.on("unread_count_updated", handlers.onUnreadCountUpdated); + } + + if (handlers.onMailboxUpdated) { + socket.on("mailbox_updated", handlers.onMailboxUpdated); + } + + // Cleanup on unmount + return () => { + socket.off("new_email"); + socket.off("email_read"); + socket.off("email_deleted"); + socket.off("email_sent"); + socket.off("unread_count_updated"); + socket.off("mailbox_updated"); + }; + }, [socket, handlers]); +}; +``` + +## 📱 Complete React Integration + +### 1. Email Notification Component + +```typescript +// components/EmailNotifications.tsx +import React, { useState, useCallback } from 'react'; +import { useEmailWebSocket, useEmailEvents, EmailPayload, UnreadCountPayload } from '../hooks'; + +interface EmailNotificationsProps { + userToken: string; + onEmailClick?: (email: EmailPayload) => void; +} + +export const EmailNotifications: React.FC = ({ + userToken, + onEmailClick +}) => { + const [notifications, setNotifications] = useState([]); + const [unreadCount, setUnreadCount] = useState(0); + const [isVisible, setIsVisible] = useState(false); + + const { socket, isConnected } = useEmailWebSocket({ + token: userToken, + onConnect: () => console.log('Email notifications connected'), + onError: (error) => console.error('WebSocket error:', error), + }); + + const handleNewEmail = useCallback((email: EmailPayload) => { + console.log('📧 New email received:', email); + + // Add to notifications list + setNotifications(prev => [email, ...prev.slice(0, 9)]); // Keep only 10 latest + + // Show browser notification if permission granted + if (Notification.permission === 'granted') { + new Notification(`📧 New Email: ${email.subject}`, { + body: `From: ${email.from.address}\n${email.preview}`, + icon: '/email-icon.png', + tag: `email-${email.messageId}`, // Prevent duplicates + }); + } + + // Play notification sound (optional) + const audio = new Audio('/notification.mp3'); + audio.play().catch(() => {}); // Ignore if audio fails + }, []); + + const handleUnreadCountUpdate = useCallback((data: UnreadCountPayload) => { + setUnreadCount(data.count); + }, []); + + const handleEmailRead = useCallback((data: any) => { + // Remove from notifications if it was there + setNotifications(prev => + prev.filter(notification => notification.messageId !== data.messageId) + ); + }, []); + + useEmailEvents(socket, { + onNewEmail: handleNewEmail, + onUnreadCountUpdated: handleUnreadCountUpdate, + onEmailRead: handleEmailRead, + }); + + // Request notification permission on mount + React.useEffect(() => { + if (Notification.permission === 'default') { + Notification.requestPermission(); + } + }, []); + + return ( +
+ {/* Connection Status */} +
+ + {isConnected ? 'Connected' : 'Disconnected'} +
+ + {/* Notification Bell */} +
setIsVisible(!isVisible)} + > + 🔔 + {unreadCount > 0 && ( + {unreadCount > 99 ? '99+' : unreadCount} + )} +
+ + {/* Notifications Dropdown */} + {isVisible && ( +
+
+

📧 Email Notifications

+ {unreadCount} unread +
+ +
+ {notifications.length === 0 ? ( +
No new emails
+ ) : ( + notifications.map((email) => ( +
onEmailClick?.(email)} + > +
+ {email.from.name || email.from.address} + {email.hasAttachments && 📎} +
+
{email.subject}
+
{email.preview}
+
+ {new Date(email.timestamp).toLocaleString()} +
+
+ )) + )} +
+
+ )} +
+ ); +}; +``` + +### 2. Email Dashboard Component + +```typescript +// components/EmailDashboard.tsx +import React, { useState } from 'react'; +import { useEmailWebSocket, useEmailEvents } from '../hooks'; + +interface EmailDashboardProps { + userToken: string; +} + +export const EmailDashboard: React.FC = ({ userToken }) => { + const [emailStats, setEmailStats] = useState({ + totalUnread: 0, + newToday: 0, + sentToday: 0, + }); + + const { socket, isConnected } = useEmailWebSocket({ + token: userToken, + }); + + useEmailEvents(socket, { + onNewEmail: (email) => { + setEmailStats(prev => ({ + ...prev, + totalUnread: prev.totalUnread + 1, + newToday: prev.newToday + 1, + })); + }, + + onEmailSent: () => { + setEmailStats(prev => ({ + ...prev, + sentToday: prev.sentToday + 1, + })); + }, + + onUnreadCountUpdated: (data) => { + setEmailStats(prev => ({ + ...prev, + totalUnread: data.count, + })); + }, + }); + + return ( +
+

📧 Email Dashboard

+ +
+
+

{emailStats.totalUnread}

+

Unread Emails

+
+ +
+

{emailStats.newToday}

+

Received Today

+
+ +
+

{emailStats.sentToday}

+

Sent Today

+
+ +
+

+ {isConnected ? '🟢' : '🔴'} +

+

{isConnected ? 'Connected' : 'Disconnected'}

+
+
+
+ ); +}; +``` + +### 3. Context Provider (Advanced) + +```typescript +// context/EmailWebSocketContext.tsx +import React, { createContext, useContext, ReactNode } from 'react'; +import { useEmailWebSocket } from '../hooks/useEmailWebSocket'; +import { Socket } from 'socket.io-client'; + +interface EmailWebSocketContextValue { + socket: Socket | null; + isConnected: boolean; +} + +const EmailWebSocketContext = createContext(null); + +interface EmailWebSocketProviderProps { + children: ReactNode; + token: string; + serverUrl?: string; +} + +export const EmailWebSocketProvider: React.FC = ({ + children, + token, + serverUrl, +}) => { + const { socket, isConnected } = useEmailWebSocket({ token, serverUrl }); + + return ( + + {children} + + ); +}; + +export const useEmailWebSocketContext = () => { + const context = useContext(EmailWebSocketContext); + if (!context) { + throw new Error('useEmailWebSocketContext must be used within EmailWebSocketProvider'); + } + return context; +}; +``` + +## 🎨 CSS Styles + +```css +/* styles/email-notifications.css */ +.email-notifications { + position: relative; + display: inline-block; +} + +.connection-status { + position: absolute; + top: -8px; + right: -8px; + font-size: 10px; + padding: 2px 6px; + border-radius: 10px; + display: flex; + align-items: center; + gap: 4px; +} + +.connection-status.connected { + background: #d4edda; + color: #155724; +} + +.connection-status.disconnected { + background: #f8d7da; + color: #721c24; +} + +.status-dot { + width: 6px; + height: 6px; + border-radius: 50%; + background: currentColor; +} + +.notification-bell { + position: relative; + cursor: pointer; + font-size: 24px; + padding: 8px; + border-radius: 50%; + transition: background-color 0.2s; +} + +.notification-bell:hover { + background-color: #f5f5f5; +} + +.badge { + position: absolute; + top: 0; + right: 0; + background: #dc3545; + color: white; + border-radius: 50%; + min-width: 18px; + height: 18px; + font-size: 11px; + display: flex; + align-items: center; + justify-content: center; + font-weight: bold; +} + +.notifications-dropdown { + position: absolute; + top: 100%; + right: 0; + width: 350px; + max-height: 400px; + background: white; + border: 1px solid #ddd; + border-radius: 8px; + box-shadow: 0 4px 12px rgba(0, 0, 0, 0.15); + z-index: 1000; + overflow: hidden; +} + +.notifications-header { + padding: 12px 16px; + background: #f8f9fa; + border-bottom: 1px solid #dee2e6; + display: flex; + justify-content: space-between; + align-items: center; +} + +.notifications-list { + max-height: 300px; + overflow-y: auto; +} + +.notification-item { + padding: 12px 16px; + border-bottom: 1px solid #eee; + cursor: pointer; + transition: background-color 0.2s; +} + +.notification-item:hover { + background-color: #f8f9fa; +} + +.notification-item:last-child { + border-bottom: none; +} + +.sender { + display: flex; + align-items: center; + gap: 8px; + margin-bottom: 4px; +} + +.attachment-icon { + font-size: 12px; +} + +.subject { + font-weight: 500; + margin-bottom: 4px; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} + +.preview { + color: #666; + font-size: 14px; + margin-bottom: 4px; + display: -webkit-box; + -webkit-line-clamp: 2; + -webkit-box-orient: vertical; + overflow: hidden; +} + +.timestamp { + font-size: 12px; + color: #999; +} + +.no-notifications { + padding: 20px; + text-align: center; + color: #666; +} + +/* Dashboard Styles */ +.email-dashboard { + padding: 20px; +} + +.stats-grid { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); + gap: 16px; + margin-top: 20px; +} + +.stat-card { + background: white; + padding: 20px; + border-radius: 8px; + border: 1px solid #ddd; + text-align: center; +} + +.stat-card h3 { + margin: 0 0 8px 0; + font-size: 32px; + font-weight: bold; +} + +.stat-card h3.connected { + color: #28a745; +} + +.stat-card h3.disconnected { + color: #dc3545; +} + +.stat-card p { + margin: 0; + color: #666; + font-size: 14px; +} +``` + +## 🧪 Testing & Manual Triggers + +### Manual Email Check + +```typescript +// API call to manually trigger email check +const triggerEmailCheck = async () => { + try { + const response = await fetch("/api/email/gateway/check-emails", { + method: "POST", + headers: { + Authorization: `Bearer ${userToken}`, + "Content-Type": "application/json", + }, + }); + + const result = await response.json(); + console.log("✅ Manual email check triggered:", result); + } catch (error) { + console.error("❌ Failed to trigger email check:", error); + } +}; +``` + +### Send Test Notification + +```typescript +const sendTestNotification = async () => { + try { + const response = await fetch("/api/email/gateway/test-notification", { + method: "POST", + headers: { + Authorization: `Bearer ${userToken}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + message: "Test notification from React app!", + }), + }); + + const result = await response.json(); + console.log("✅ Test notification sent:", result); + } catch (error) { + console.error("❌ Failed to send test notification:", error); + } +}; +``` + +## 🔧 Complete App Integration + +```typescript +// App.tsx - Complete example +import React from 'react'; +import { EmailWebSocketProvider } from './context/EmailWebSocketContext'; +import { EmailNotifications } from './components/EmailNotifications'; +import { EmailDashboard } from './components/EmailDashboard'; + +function App() { + const userToken = localStorage.getItem('jwt-token') || ''; + + const handleEmailClick = (email: any) => { + console.log('Email clicked:', email); + // Navigate to email detail page + // router.push(`/emails/${email.messageId}`); + }; + + if (!userToken) { + return
Please login to access email notifications
; + } + + return ( + +
+
+

📧 Email App

+ +
+ +
+ +
+
+
+ ); +} + +export default App; +``` + +## 🚨 Error Handling & Edge Cases + +### Connection Recovery + +```typescript +const useConnectionRecovery = (socket: Socket | null) => { + useEffect(() => { + if (!socket) return; + + const handleReconnect = () => { + console.log("🔄 WebSocket reconnecting..."); + // Optionally refresh data or show reconnecting state + }; + + const handleReconnectError = () => { + console.error("❌ Failed to reconnect"); + // Show user notification or retry logic + }; + + socket.on("reconnect", handleReconnect); + socket.on("reconnect_error", handleReconnectError); + + return () => { + socket.off("reconnect", handleReconnect); + socket.off("reconnect_error", handleReconnectError); + }; + }, [socket]); +}; +``` + +### Token Refresh + +```typescript +const useTokenRefresh = (socket: Socket | null) => { + useEffect(() => { + const refreshToken = () => { + const newToken = localStorage.getItem("jwt-token"); + if (socket && newToken) { + socket.auth = { token: newToken }; + socket.disconnect(); + socket.connect(); + } + }; + + // Listen for token refresh events + window.addEventListener("token-refreshed", refreshToken); + + return () => { + window.removeEventListener("token-refreshed", refreshToken); + }; + }, [socket]); +}; +``` + +## 📝 Environment Configuration + +```typescript +// config/websocket.ts +export const WEBSOCKET_CONFIG = { + development: { + serverUrl: "ws://localhost:3000/email", + reconnectionAttempts: 5, + timeout: 20000, + }, + production: { + serverUrl: "wss://your-domain.com/email", + reconnectionAttempts: 10, + timeout: 30000, + }, +}; + +export const getWebSocketConfig = () => { + const env = process.env.NODE_ENV || "development"; + return WEBSOCKET_CONFIG[env as keyof typeof WEBSOCKET_CONFIG]; +}; +``` + +## 🎯 Best Practices + +1. **🔐 Security**: Always use HTTPS/WSS in production +2. **🔄 Reconnection**: Implement proper reconnection logic +3. **🎯 Performance**: Limit notification history (max 10-50 items) +4. **📱 UX**: Request notification permissions gracefully +5. **🔊 Audio**: Make notification sounds optional +6. **💾 Persistence**: Consider storing notifications in localStorage +7. **🎨 Theming**: Support light/dark mode for notifications +8. **♿ Accessibility**: Add proper ARIA labels and keyboard navigation + +## 🐛 Troubleshooting + +**Connection Issues:** + +- Check JWT token validity +- Verify server URL and port +- Check browser console for errors +- Ensure CORS is properly configured + +**Missing Notifications:** + +- Verify WebSocket event names match backend +- Check browser notification permissions +- Confirm user is authenticated properly +- Test with manual triggers + +**Performance Issues:** + +- Limit notification history +- Debounce rapid events +- Use React.memo for components +- Implement virtual scrolling for long lists + +--- + +## 🚀 Ready to Go! + +Your React app is now ready to receive real-time email notifications! The WebSocket connection will automatically: + +- ✅ Authenticate users with JWT tokens +- ✅ Receive new email notifications instantly +- ✅ Update unread counts in real-time +- ✅ Handle connection recovery automatically +- ✅ Show browser notifications (with permission) +- ✅ Provide a beautiful UI for email notifications + +Happy coding! 🎉 diff --git a/src/common/DTO/pagination.dto.ts b/src/common/DTO/pagination.dto.ts index a0a707b..097013f 100755 --- a/src/common/DTO/pagination.dto.ts +++ b/src/common/DTO/pagination.dto.ts @@ -15,7 +15,7 @@ export class PaginationDto { @IsNotEmpty() @IsNumber() @Min(1) - @Max(50) + @Max(100) @Type(() => Number) @ApiPropertyOptional({ type: "number", required: false }) limit?: number; diff --git a/src/common/enums/message.enum.ts b/src/common/enums/message.enum.ts index 02f0a7b..60ed64f 100755 --- a/src/common/enums/message.enum.ts +++ b/src/common/enums/message.enum.ts @@ -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 { diff --git a/src/configs/wildduck.config.ts b/src/configs/wildduck.config.ts index e39746c..0ae8b3a 100644 --- a/src/configs/wildduck.config.ts +++ b/src/configs/wildduck.config.ts @@ -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("WILDDUCK_BASE_URL"), - accessToken: configService.getOrThrow("WILDDUCK_ACCESS_TOKEN"), + baseURL: configService.getOrThrow("WILDDUCK_BASE_URL"), timeout: configService.getOrThrow("WILDDUCK_TIMEOUT"), retries: configService.getOrThrow("WILDDUCK_RETRIES"), + headers: { + Accept: "application/json", + "X-Access-Token": configService.getOrThrow("WILDDUCK_ACCESS_TOKEN"), + }, }), }); diff --git a/src/core/filters/ws-exception.filter.ts b/src/core/filters/ws-exception.filter.ts new file mode 100644 index 0000000..e689831 --- /dev/null +++ b/src/core/filters/ws-exception.filter.ts @@ -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(); + 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); + } +} diff --git a/src/core/interceptors/pagination.interceptor.ts b/src/core/interceptors/pagination.interceptor.ts index fccf21d..00771b6 100755 --- a/src/core/interceptors/pagination.interceptor.ts +++ b/src/core/interceptors/pagination.interceptor.ts @@ -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()}`; } } diff --git a/src/modules/email/DTO/send-email.dto.ts b/src/modules/email/DTO/send-email.dto.ts index 833b775..ec34722 100644 --- a/src/modules/email/DTO/send-email.dto.ts +++ b/src/modules/email/DTO/send-email.dto.ts @@ -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 }) diff --git a/src/modules/email/constants/email-events.constant.ts b/src/modules/email/constants/email-events.constant.ts index 2a63446..23bcf02 100644 --- a/src/modules/email/constants/email-events.constant.ts +++ b/src/modules/email/constants/email-events.constant.ts @@ -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", +}; diff --git a/src/modules/email/controllers/email-gateway.controller.ts b/src/modules/email/controllers/email-gateway.controller.ts new file mode 100644 index 0000000..d3d8ea5 --- /dev/null +++ b/src/modules/email/controllers/email-gateway.controller.ts @@ -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", +// }; +// } +// } diff --git a/src/modules/email/email.module.ts b/src/modules/email/email.module.ts index 2ece2ba..0cccc61 100644 --- a/src/modules/email/email.module.ts +++ b/src/modules/email/email.module.ts @@ -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 {} diff --git a/src/modules/email/gateways/email.gateway.ts b/src/modules/email/gateways/email.gateway.ts new file mode 100644 index 0000000..a874aec --- /dev/null +++ b/src/modules/email/gateways/email.gateway.ts @@ -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(); + private readonly userSessions = new Map>(); + + 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) { + 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 { + 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 { + 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, 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, + }); + } + } +} diff --git a/src/modules/email/guards/websocket-auth.guard.ts b/src/modules/email/guards/websocket-auth.guard.ts new file mode 100644 index 0000000..e01e266 --- /dev/null +++ b/src/modules/email/guards/websocket-auth.guard.ts @@ -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 { + const client: Socket = context.switchToWs().getClient(); + + 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; + } +} diff --git a/src/modules/email/interfaces/email-events.interface.ts b/src/modules/email/interfaces/email-events.interface.ts index 50ddf60..48da1d1 100644 --- a/src/modules/email/interfaces/email-events.interface.ts +++ b/src/modules/email/interfaces/email-events.interface.ts @@ -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; - 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; +} diff --git a/src/modules/email/interfaces/websocket.interface.ts b/src/modules/email/interfaces/websocket.interface.ts index 449b20c..58f2184 100644 --- a/src/modules/email/interfaces/websocket.interface.ts +++ b/src/modules/email/interfaces/websocket.interface.ts @@ -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; } /** - * 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 { + 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; -} - -/** - * 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; } diff --git a/src/modules/email/queue/email-monitoring.processor.ts b/src/modules/email/queue/email-monitoring.processor.ts new file mode 100644 index 0000000..91cedc5 --- /dev/null +++ b/src/modules/email/queue/email-monitoring.processor.ts @@ -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) { + 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; + } + } +} diff --git a/src/modules/email/services/email-monitoring-scheduler.service.ts b/src/modules/email/services/email-monitoring-scheduler.service.ts new file mode 100644 index 0000000..ddb3b13 --- /dev/null +++ b/src/modules/email/services/email-monitoring-scheduler.service.ts @@ -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(); + } +} diff --git a/src/modules/email/services/email-monitoring.service.ts b/src/modules/email/services/email-monitoring.service.ts new file mode 100644 index 0000000..6bec000 --- /dev/null +++ b/src/modules/email/services/email-monitoring.service.ts @@ -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 = { + 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) { + 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 { + 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"); + } +} diff --git a/src/modules/email/services/email-notification.service.ts b/src/modules/email/services/email-notification.service.ts new file mode 100644 index 0000000..2250d4c --- /dev/null +++ b/src/modules/email/services/email-notification.service.ts @@ -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) { + 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(), + }; + } + } +} diff --git a/src/modules/email/services/email.service.ts b/src/modules/email/services/email.service.ts index 854ec34..d2c80f6 100644 --- a/src/modules/email/services/email.service.ts +++ b/src/modules/email/services/email.service.ts @@ -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 { + const transformedQuery: Record = { ...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); diff --git a/src/modules/email/services/websocket-auth.service.ts b/src/modules/email/services/websocket-auth.service.ts new file mode 100644 index 0000000..98b8e31 --- /dev/null +++ b/src/modules/email/services/websocket-auth.service.ts @@ -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 { + 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(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; + } +} diff --git a/src/modules/mail-server/DTO/message.dto.ts b/src/modules/mail-server/DTO/message.dto.ts index cf07ff1..54bf0d0 100644 --- a/src/modules/mail-server/DTO/message.dto.ts +++ b/src/modules/mail-server/DTO/message.dto.ts @@ -32,6 +32,11 @@ export class ListMessagesQueryDto { @IsOptional() @IsString() previous?: string; + + @ApiPropertyOptional({ description: "Unseen messages only" }) + @IsOptional() + @IsBoolean() + unseen?: boolean; } export class UpdateMessageDto { diff --git a/src/modules/mail-server/mail-server.gateway.example.ts b/src/modules/mail-server/mail-server.gateway.example.ts deleted file mode 100644 index 447f5fc..0000000 --- a/src/modules/mail-server/mail-server.gateway.example.ts +++ /dev/null @@ -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: `

Hello ${recipientName}!

Welcome to our email service!

`, - }), - ); - - 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; - } - } -} diff --git a/src/modules/mail-server/services/mail-server.service.ts b/src/modules/mail-server/services/mail-server.service.ts index 37c4275..e93095b 100644 --- a/src/modules/mail-server/services/mail-server.service.ts +++ b/src/modules/mail-server/services/mail-server.service.ts @@ -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 { - 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 { - 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 { - 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); - }), - ); - } } diff --git a/src/modules/mail-server/services/wildduck-base.service.ts b/src/modules/mail-server/services/wildduck-base.service.ts index 21b956d..20f8be2 100644 --- a/src/modules/mail-server/services/wildduck-base.service.ts +++ b/src/modules/mail-server/services/wildduck-base.service.ts @@ -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("WILDDUCK_BASE_URL", "http://localhost:8080"); - this.accessToken = this.configService.get("WILDDUCK_ACCESS_TOKEN"); + this.baseURL = this.configService.getOrThrow("WILDDUCK_BASE_URL"); } - protected buildUrl(path: string, params?: Record): string { - const url = new URL(path, this.baseUrl); + protected buildUrl(path: string, params?: Record): 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(path: string, params?: Record): Observable { const url = this.buildUrl(path, params); + this.logger.debug(`WildDuck GET: ${url}`); + if (params) { + this.logger.debug(`WildDuck GET Params:`, params); + } return this.handleResponse(this.httpService.get(url)); } diff --git a/src/modules/utils/services/extract-token.utils copy.ts b/src/modules/utils/services/extract-token.utils copy.ts new file mode 100644 index 0000000..8467993 --- /dev/null +++ b/src/modules/utils/services/extract-token.utils copy.ts @@ -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; +}