chore: remove webhook system

EMAIL_POLLING_SYSTEM.md

@@ -0,0 +1,279 @@
+# Email Polling System
+
+This document describes the email polling system that replaced the webhook-based approach for real-time email notifications.
+
+## Overview
+
+The email polling system uses **BullMQ** (Redis-based job queue) to periodically check for new emails when users are connected via WebSocket. This approach is more reliable than webhooks since it doesn't depend on external webhook delivery.
+
+## Architecture
+
+### Components
+
+1. **EmailPollingService** - Manages polling jobs for connected users
+2. **EmailPollingProcessor** - Processes email polling jobs and checks for new emails
+3. **EmailGateway** - WebSocket gateway that starts/stops polling on user connection/disconnection
+4. **BullMQ Queue** - Redis-based job queue for scheduling recurring email checks
+
+### Flow
+
+```
+User connects to WebSocket
+    ↓
+EmailGateway.handleConnection()
+    ↓
+EmailPollingService.startPollingForUser()
+    ↓
+BullMQ creates recurring job (every 30 seconds)
+    ↓
+EmailPollingProcessor.process() - checks for new emails
+    ↓
+If new emails found → notify via WebSocket
+    ↓
+User disconnects
+    ↓
+EmailGateway.handleDisconnect()
+    ↓
+EmailPollingService.stopPollingForUser()
+```
+
+## Configuration
+
+### Environment Variables
+
+Required environment variables (see `environment.sample`):
+
+```bash
+# Redis for BullMQ
+REDIS_URI=redis://localhost:6379
+
+# WildDuck Mail Server
+WILDDUCK_SERVER=http://localhost:8080
+WILDDUCK_API_TOKEN=your-wildduck-access-token
+
+# Email Polling (optional)
+EMAIL_POLLING_INTERVAL=30000  # 30 seconds
+EMAIL_POLLING_ENABLED=true
+```
+
+### BullMQ Configuration
+
+The system uses the existing BullMQ configuration from `src/configs/bullmq.config.ts`:
+
+```typescript
+export function bullMqConfig(): SharedBullAsyncConfiguration {
+  return {
+    connection: {
+      url: configService.getOrThrow<string>("REDIS_URI"),
+    },
+    prefix: "dmail",
+    defaultJobOptions: {
+      removeOnComplete: true,
+      removeOnFail: false,
+      attempts: 5,
+    },
+  };
+}
+```
+
+## Implementation Details
+
+### Email Polling Service
+
+Located at `src/modules/email-utils/services/email-polling.service.ts`
+
+**Key Methods:**
+
+- `startPollingForUser(userId, connectionId)` - Start polling when user connects
+- `stopPollingForUser(userId)` - Stop polling when user disconnects
+- `getPollingStats()` - Get statistics about active polling jobs
+
+**Job Configuration:**
+
+```typescript
+const job = await this.emailPollingQueue.add(`email-poll-${userId}`, jobData, {
+  repeat: {
+    every: 30000, // 30 seconds interval
+  },
+  removeOnComplete: 5,
+  removeOnFail: 10,
+  attempts: 3,
+  backoff: {
+    type: "exponential",
+    delay: 2000,
+  },
+});
+```
+
+### Email Polling Processor
+
+Located at `src/modules/email-utils/queue/email-polling.processor.ts`
+
+**Processing Logic:**
+
+1. Check if user is still connected (skip if disconnected)
+2. Get user's mailbox information
+3. Query WildDuck API for recent emails since last check
+4. For each new email:
+   - Send WebSocket notification via `EmailGateway.notifyNewEmail()`
+   - Update unread count via `EmailGateway.notifyUnreadCountUpdate()`
+5. Update job data with new timestamp for next run
+
+### WebSocket Integration
+
+The `EmailGateway` automatically manages polling:
+
+```typescript
+// On user connection
+await this.emailPollingService.startPollingForUser(user.id, client.id);
+
+// On user disconnection
+await this.emailPollingService.stopPollingForUser(userInfo.userId);
+```
+
+## WebSocket Events
+
+The system emits these WebSocket events when new emails are found:
+
+### `new_email` Event
+
+```typescript
+{
+  userId: string;
+  messageId: number;
+  subject: string;
+  from: { name?: string; address: string };
+  to: Array<{ name?: string; address: string }>;
+  hasAttachments: boolean;
+  timestamp: string;
+  mailboxId: string;
+  mailboxName: string;
+  isRead: boolean;
+}
+```
+
+### `unread_count_updated` Event
+
+```typescript
+{
+  userId: string;
+  totalUnread: number;
+  mailboxCounts: Record<string, number>;
+  count: number;
+  timestamp: string;
+}
+```
+
+## Monitoring & Debugging
+
+### Check Active Polling Jobs
+
+```typescript
+const stats = emailPollingService.getPollingStats();
+console.log({
+  activeJobs: stats.activeJobs,
+  userIds: stats.userIds,
+  connections: stats.connections,
+});
+```
+
+### BullMQ Dashboard
+
+You can use BullMQ's web dashboard to monitor jobs:
+
+```bash
+npm install -g @bull-board/express
+# Then access the dashboard at http://localhost:3000/admin/queues
+```
+
+### Logs
+
+The system logs important events:
+
+```
+[EmailPollingService] Started email polling for user 12345 with job ID: email-poll-12345
+[EmailPollingProcessor] Found 3 new emails for user 12345
+[EmailPollingProcessor] Notified user 12345 of 3 new emails
+[EmailPollingService] Stopped email polling for user 12345
+```
+
+## Performance Considerations
+
+### Polling Interval
+
+- Default: 30 seconds
+- Adjustable via `EMAIL_POLLING_INTERVAL` environment variable
+- Balance between real-time updates and API load
+
+### Job Management
+
+- Jobs are automatically cleaned up when users disconnect
+- Failed jobs are retried up to 3 times with exponential backoff
+- Completed jobs are kept (last 5) for debugging
+
+### Resource Usage
+
+- Each connected user creates one recurring BullMQ job
+- Jobs only run when users are connected (checked at start of each job)
+- Redis is used for job queue storage
+
+## Advantages over Webhooks
+
+1. **Reliability** - No dependency on external webhook delivery
+2. **Connection Awareness** - Only polls when users are actually connected
+3. **Error Handling** - Built-in retry mechanism with exponential backoff
+4. **Monitoring** - Full visibility into job status and performance
+5. **Scalability** - Redis-based queue can handle high loads
+6. **Control** - Can adjust polling frequency per user if needed
+
+## Migration from Webhooks
+
+The old webhook system has been completely removed:
+
+**Deleted Files:**
+
+- `src/modules/email-utils/controllers/webhook.controller.ts`
+- `src/modules/email-utils/services/webhook-management.service.ts`
+- `src/modules/email-utils/services/webhook-email-events.service.ts`
+- `src/modules/email/interfaces/webhook-events.interface.ts`
+
+**Updated Files:**
+
+- `src/modules/email-utils/email-utils.module.ts` - Added BullMQ and new services
+- `src/modules/email-gateway/email.gateway.ts` - Added polling integration
+- `src/modules/email-gateway/email-gateway.module.ts` - Added EmailUtilsModule import
+
+## Troubleshooting
+
+### Common Issues
+
+**Issue: Jobs not running**
+
+- Check Redis connection
+- Verify `REDIS_URI` environment variable
+- Check BullMQ logs
+
+**Issue: Users not receiving notifications**
+
+- Verify user is connected to WebSocket
+- Check if polling job is active: `emailPollingService.getPollingStats()`
+- Verify WildDuck API credentials and server URL
+
+**Issue: High Redis memory usage**
+
+- Adjust `removeOnComplete` and `removeOnFail` values
+- Consider shorter polling intervals for fewer concurrent jobs
+
+**Issue: API rate limiting**
+
+- Increase polling interval
+- Implement user-specific rate limiting
+- Use WildDuck API bulk endpoints if available
+
+## Future Enhancements
+
+1. **Dynamic Polling Intervals** - Adjust frequency based on user activity
+2. **Priority Users** - More frequent polling for premium users
+3. **Email Filtering** - Only notify for emails matching user preferences
+4. **Batch Processing** - Process multiple users in single API calls
+5. **Caching** - Cache recent emails to reduce API calls

PUSH_NOTIFICATIONS_MULTI_DEVICE.md

@@ -1,214 +0,0 @@
-# Multi-Device Push Notifications Implementation
-
-## Overview
-
-Updated the push notification system to support multiple devices per user. Users can now receive push notifications on all their logged-in devices (mobile, tablet, desktop, etc.).
-
-## Database Changes
-
-### Migration: `update_push_tokens_to_array.sql`
-
-- Converted single `push_token` field to `push_tokens` JSON array
-- Automatically migrates existing single tokens to array format
-- Adds optimized GIN index for JSON array queries
-
-## API Endpoints
-
-### 1. Add Push Token (New Device Login)
-
-```http
-POST /users/push-token
-Authorization: Bearer <jwt_token>
-Content-Type: application/json
-
-{
-  "pushToken": "device_push_token_from_najva_sdk",
-  "deviceId": "optional_device_identifier"
-}
-```
-
-### 2. Remove Push Token (Device Logout)
-
-```http
-DELETE /users/push-token
-Authorization: Bearer <jwt_token>
-Content-Type: application/json
-
-{
-  "pushToken": "device_push_token_to_remove"
-}
-```
-
-### 3. Get All Push Tokens
-
-```http
-GET /users/push-tokens
-Authorization: Bearer <jwt_token>
-```
-
-Response:
-
-```json
-{
-  "pushTokens": ["token1", "token2", "token3"],
-  "totalTokens": 3
-}
-```
-
-### 4. Update Profile (Push Token Removed)
-
-```http
-PATCH /users/profile
-Authorization: Bearer <jwt_token>
-Content-Type: application/json
-
-{
-  "profilePic": "https://example.com/profile.jpg"
-}
-```
-
-## Features
-
-### ✅ Multi-Device Support
-
-- Users can have multiple push tokens (one per device)
-- Notifications sent to all registered devices simultaneously
-- Automatic deduplication prevents duplicate tokens
-
-### ✅ Invalid Token Cleanup
-
-- Automatically detects invalid/expired tokens from Najva API response
-- Removes invalid tokens from user's token list
-- Logs cleanup activities for monitoring
-
-### ✅ Device Management
-
-- Add tokens when user logs in on new device
-- Remove tokens when user logs out from device
-- View all registered devices/tokens
-
-### ✅ Backward Compatibility
-
-- Existing single token data automatically migrated to array format
-- No breaking changes to existing notification flow
-
-## Implementation Details
-
-### User Entity Changes
-
-```typescript
-// Before
-@Property({ type: "varchar", nullable: true })
-pushToken?: string;
-
-// After
-@Property({ type: "json", nullable: true })
-pushTokens?: string[];
-```
-
-### Service Methods
-
-```typescript
-// Add token for new device
-await usersService.addPushToken(userId, { pushToken: "token" });
-
-// Remove token when logging out
-await usersService.removePushToken(userId, { pushToken: "token" });
-
-// Send to all user devices
-await notificationsService.sendPushNotificationToUserDevices(userId, userTokens, message);
-```
-
-### Automatic Token Cleanup
-
-When sending notifications, invalid tokens are automatically:
-
-1. Detected from Najva API response
-2. Logged for monitoring
-3. Removed from user's token array
-4. Database updated to clean state
-
-## Frontend Integration
-
-### When User Logs In
-
-```javascript
-// Get push token from Najva SDK
-const pushToken = await najva.getPushToken();
-
-// Register token for this device
-await fetch("/api/users/push-token", {
-  method: "POST",
-  headers: {
-    Authorization: `Bearer ${jwt}`,
-    "Content-Type": "application/json",
-  },
-  body: JSON.stringify({
-    pushToken: pushToken,
-    deviceId: navigator.userAgent, // optional
-  }),
-});
-```
-
-### When User Logs Out
-
-```javascript
-// Remove token for this device
-await fetch("/api/users/push-token", {
-  method: "DELETE",
-  headers: {
-    Authorization: `Bearer ${jwt}`,
-    "Content-Type": "application/json",
-  },
-  body: JSON.stringify({
-    pushToken: currentDeviceToken,
-  }),
-});
-```
-
-## Environment Variables
-
-```env
-NAJVA_API_KEY=your_najva_api_key
-NAJVA_WEBSITE_ID=your_website_id
-NAJVA_API_URL=https://push.najva.com/v1/send/token/
-PUSH_TTL=24
-FRONTEND_URL=https://your-frontend-domain.com
-```
-
-## Benefits
-
-1. **Multi-Device Support**: Users get notifications on all their devices
-2. **Device Management**: Add/remove devices as needed
-3. **Automatic Cleanup**: Invalid tokens automatically removed
-4. **Scalable**: Efficient JSON array storage with GIN indexing
-5. **Reliable**: Handles edge cases like duplicate tokens and API failures
-6. **Monitoring**: Comprehensive logging for debugging and monitoring
-
-## Migration Steps
-
-1. **Run Database Migration**:
-
-   ```sql
-   -- Execute: database/migrations/update_push_tokens_to_array.sql
-   ```
-
-2. **Update Frontend**:
-   - Use new `/users/push-token` endpoints
-   - Remove `pushToken` field from profile updates
-
-3. **Deploy Backend**:
-   - All changes are backward compatible
-   - Existing tokens automatically migrated
-
-## Testing
-
-Test the multi-device functionality:
-
-1. Login from Device A → Add token A
-2. Login from Device B → Add token B
-3. Create notification → Both devices receive push
-4. Logout from Device A → Remove token A
-5. Create notification → Only device B receives push
-
-The system now properly supports users accessing your application from multiple devices while maintaining reliable push notification delivery!

environment.sample

@@ -0,0 +1,77 @@
+# Database Configuration
+DATABASE_URL=postgresql://username:password@localhost:5432/dmail_db
+
+# Redis Configuration (for BullMQ)
+REDIS_URI=redis://localhost:6379
+
+# JWT Configuration
+JWT_SECRET=your-super-secret-jwt-key-here
+JWT_EXPIRES_IN=7d
+
+# Mail Server Configuration (WildDuck API)
+WILDDUCK_SERVER=http://localhost:8080
+WILDDUCK_API_TOKEN=your-wildduck-access-token-here
+
+# Email Polling Configuration
+EMAIL_POLLING_INTERVAL=30000  # 30 seconds (in milliseconds)
+EMAIL_POLLING_ENABLED=true
+
+# WebSocket Configuration
+WS_CORS_ORIGIN=http://localhost:3000
+WS_TRANSPORTS=websocket,polling
+
+# Application Configuration
+NODE_ENV=development
+PORT=3000
+APP_NAME=dmail-api
+LOG_LEVEL=debug
+
+# AWS S3 Configuration (for file uploads)
+AWS_REGION=us-east-1
+AWS_ACCESS_KEY_ID=your-aws-access-key
+AWS_SECRET_ACCESS_KEY=your-aws-secret-key
+AWS_S3_BUCKET=your-s3-bucket-name
+
+# Notification Configuration (for push notifications)
+VAPID_PUBLIC_KEY=your-vapid-public-key
+VAPID_PRIVATE_KEY=your-vapid-private-key
+VAPID_SUBJECT=mailto:admin@yourapp.com
+
+# Rate Limiting
+RATE_LIMIT_TTL=60000  # 1 minute
+RATE_LIMIT_LIMIT=100  # 100 requests per minute per IP
+
+# Cache Configuration
+CACHE_TTL=300000  # 5 minutes
+
+# Business Configuration
+BUSINESS_QUOTA_DEFAULT=1000
+BUSINESS_QUOTA_PREMIUM=10000
+
+# AI Assistant Configuration (OpenAI)
+OPENAI_API_KEY=your-openai-api-key
+OPENAI_MODEL=gpt-3.5-turbo
+
+# CORS Configuration
+CORS_ORIGINS=http://localhost:3000,https://yourapp.com
+
+# Email Templates
+EMAIL_TEMPLATE_DIR=./templates
+EMAIL_FROM_ADDRESS=noreply@yourapp.com
+EMAIL_FROM_NAME=Your App Name
+
+# Monitoring & Logging
+SENTRY_DSN=your-sentry-dsn-here
+LOG_FORMAT=combined
+LOG_FILE_PATH=./logs/app.log
+
+# Development Settings
+SWAGGER_ENABLED=true
+DEBUG=false
+
+# Queue Configuration
+QUEUE_REDIS_URL=redis://localhost:6379
+QUEUE_PREFIX=dmail
+QUEUE_DEFAULT_JOB_OPTIONS_REMOVE_ON_COMPLETE=true
+QUEUE_DEFAULT_JOB_OPTIONS_REMOVE_ON_FAIL=false
+QUEUE_DEFAULT_JOB_OPTIONS_ATTEMPTS=5 

src/modules/email-gateway/email-gateway.module.ts

@@ -1,12 +1,13 @@
-import { Module } from "@nestjs/common";
+import { Module, forwardRef } from "@nestjs/common";
 import { JwtModule } from "@nestjs/jwt";
 
 import { EmailGateway } from "./email.gateway";
 import { WebSocketAuthService } from "./services/websocket-auth.service";
 import { jwtConfig } from "../../configs/jwt.config";
+import { EmailUtilsModule } from "../email-utils/email-utils.module";
 
 @Module({
-  imports: [JwtModule.registerAsync(jwtConfig())],
+  imports: [JwtModule.registerAsync(jwtConfig()), forwardRef(() => EmailUtilsModule)],
   providers: [EmailGateway, WebSocketAuthService],
   exports: [EmailGateway],
 })

src/modules/email-gateway/email.gateway.ts

@@ -16,14 +16,15 @@ import { WsExceptionFilter } from "../../core/filters/ws-exception.filter";
 import { WEBSOCKET_EVENTS, WEBSOCKET_NAMESPACES, WEBSOCKET_ROOMS, WebSocketEvent } from "../email/constants/email-events.constant";
 import { WebSocketAuthGuard } from "../email/guards/websocket-auth.guard";
 import {
-  EmailDeletePayload,
+  IEmailDeletePayload,
-  EmailPayload,
+  IEmailPayload,
-  EmailSentPayload,
+  IEmailSentPayload,
-  EmailStatusPayload,
+  IEmailStatusPayload,
-  MailboxUpdatePayload,
+  IMailboxUpdatePayload,
-  UnreadCountPayload,
+  IUnreadCountPayload,
 } from "../email/interfaces/email-events.interface";
 import { ConnectedUserInfo } from "../email/interfaces/websocket.interface";
+import { EmailPollingService } from "../email-utils/services/email-polling.service";
 
 @UsePipes(new ValidationPipe({ transform: true }))
 @UseFilters(WsExceptionFilter)
@@ -41,7 +42,10 @@ export class EmailGateway implements OnGatewayInit, OnGatewayConnection, OnGatew
   private readonly connectedUsers = new Map<string, ConnectedUserInfo>();
   private readonly userSessions = new Map<string, Set<string>>();
 
-  constructor(private readonly authService: WebSocketAuthService) {}
+  constructor(
+    private readonly authService: WebSocketAuthService,
+    private readonly emailPollingService: EmailPollingService,
+  ) {}
 
   afterInit(server: Server) {
     this.logger.log("Email WebSocket Gateway initialized");
@@ -80,6 +84,14 @@ export class EmailGateway implements OnGatewayInit, OnGatewayConnection, OnGatew
       const userInfo = this.connectedUsers.get(client.id);
       this.authService.emitAuthenticationSuccess(client, user);
 
+      // Start email polling for this user
+      try {
+        await this.emailPollingService.startPollingForUser(user.id, client.id);
+        this.logger.log(`Email polling started for user ${user.id}`);
+      } catch (pollingError) {
+        this.logger.error(`Failed to start email polling for user ${user.id}:`, pollingError);
+      }
+
       this.logger.log("WebSocket connection established", {
         ...connectionContext,
         userId: user.id,
@@ -98,7 +110,7 @@ export class EmailGateway implements OnGatewayInit, OnGatewayConnection, OnGatew
   }
 
   //************************************************************************ */
-  handleDisconnect(client: Socket): void {
+  async handleDisconnect(client: Socket): Promise<void> {
     const userInfo = this.connectedUsers.get(client.id);
 
     this.logger.log("WebSocket client disconnected", {
@@ -111,6 +123,14 @@ export class EmailGateway implements OnGatewayInit, OnGatewayConnection, OnGatew
       const userRoom = WEBSOCKET_ROOMS.USER(userInfo.userId);
       client.leave(userRoom);
 
+      // Stop email polling for this user
+      try {
+        await this.emailPollingService.stopPollingForUser(userInfo.userId);
+        this.logger.log(`Email polling stopped for user ${userInfo.userId}`);
+      } catch (pollingError) {
+        this.logger.error(`Failed to stop email polling for user ${userInfo.userId}:`, pollingError);
+      }
+
       this.cleanupUserConnection(client, userInfo);
     }
 
@@ -127,45 +147,45 @@ export class EmailGateway implements OnGatewayInit, OnGatewayConnection, OnGatew
   }
 
   //************************************************************************ */
-  async notifyNewEmail(userId: string, emailData: EmailPayload) {
+  async notifyNewEmail(userId: string, emailData: IEmailPayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.NEW_EMAIL, emailData, `New email: ${emailData.subject}`);
   }
   //************************************************************************ */
-  async notifyEmailRead(userId: string, data: EmailStatusPayload) {
+  async notifyEmailRead(userId: string, data: IEmailStatusPayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_READ, data, `Message ${data.messageId} marked as read`);
   }
 
   //************************************************************************ */
-  async notifyEmailUnread(userId: string, data: EmailStatusPayload) {
+  async notifyEmailUnread(userId: string, data: IEmailStatusPayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_UNREAD, data, `Message ${data.messageId} marked as unread`);
   }
 
   //************************************************************************ */
-  async notifyEmailFlagged(userId: string, data: EmailStatusPayload) {
+  async notifyEmailFlagged(userId: string, data: IEmailStatusPayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_FLAGGED, data, `Message ${data.messageId} flagged`);
   }
 
   //************************************************************************ */
-  async notifyEmailUnflagged(userId: string, data: EmailStatusPayload) {
+  async notifyEmailUnflagged(userId: string, data: IEmailStatusPayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_UNFLAGGED, data, `Message ${data.messageId} unflagged`);
   }
 
   //************************************************************************   */
-  async notifyEmailDeleted(userId: string, data: EmailDeletePayload) {
+  async notifyEmailDeleted(userId: string, data: IEmailDeletePayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_DELETED, data, `Message ${data.messageId} deleted`);
   }
   //************************************************************************ */
-  async notifyEmailSent(userId: string, data: EmailSentPayload) {
+  async notifyEmailSent(userId: string, data: IEmailSentPayload) {
     const content = `Email ${data.messageId} sent to ${data.to.map((t) => t.address).join(", ")}`;
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.EMAIL_SENT, data, content);
   }
   //************************************************************************ */
-  async notifyUnreadCountUpdate(userId: string, data: UnreadCountPayload) {
+  async notifyUnreadCountUpdate(userId: string, data: IUnreadCountPayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.UNREAD_COUNT_UPDATED, data, `Unread count: ${data.count}`);
   }
   //************************************************************************ */
 
-  async notifyMailboxUpdate(userId: string, data: MailboxUpdatePayload) {
+  async notifyMailboxUpdate(userId: string, data: IMailboxUpdatePayload) {
     return await this.sendUserNotification(userId, WEBSOCKET_EVENTS.MAILBOX_UPDATED, data, `Mailbox updated: ${data.mailboxName}`);
   }
 

src/modules/email-utils/email-utils.module.ts

@@ -1,10 +1,11 @@
 import { BullModule } from "@nestjs/bullmq";
 import { Module } from "@nestjs/common";
 
-import { EmailMonitoringProcessor } from "./queue/email-monitoring.processor";
+import { EmailPollingProcessor } from "./queue/email-polling.processor";
-import { EmailMonitoringService } from "./services/email-monitoring.service";
 import { EmailNotificationService } from "./services/email-notification.service";
+import { EmailPollingService } from "./services/email-polling.service";
 import { MailboxResolverService } from "./services/mailbox-resolver.service";
+import { bullMqConfig } from "../../configs/bullmq.config";
 import { EMAIL_QUEUE_CONSTANTS } from "../email/constants/email-events.constant";
 import { EmailGatewayModule } from "../email-gateway/email-gateway.module";
 import { MailServerModule } from "../mail-server/mail-server.module";
@@ -15,11 +16,13 @@ import { NotificationModule } from "../notifications/notifications.module";
     NotificationModule,
     MailServerModule,
     EmailGatewayModule,
+    BullModule.forRootAsync(bullMqConfig()),
     BullModule.registerQueue({
-      name: EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE,
+      name: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_QUEUE,
     }),
   ],
-  providers: [EmailMonitoringService, EmailNotificationService, MailboxResolverService, EmailMonitoringProcessor],
+  controllers: [],
-  exports: [EmailMonitoringService, EmailNotificationService, MailboxResolverService],
+  providers: [EmailNotificationService, MailboxResolverService, EmailPollingService, EmailPollingProcessor],
+  exports: [EmailNotificationService, MailboxResolverService, EmailPollingService],
 })
 export class EmailUtilsModule {}

src/modules/email-utils/interfaces/email-polling-job.interface.ts

@@ -0,0 +1,5 @@
+export interface EmailPollingJobData {
+  userId: string;
+  lastCheckTimestamp?: string;
+  connectionId: string;
+}

src/modules/email-utils/queue/email-monitoring.processor.ts

@@ -1,42 +0,0 @@
-import { Processor, WorkerHost } from "@nestjs/bullmq";
-import { Logger } from "@nestjs/common";
-import { Job } from "bullmq";
-
-import { EMAIL_QUEUE_CONSTANTS } from "../../email/constants/email-events.constant";
-import { EmailMonitoringJob, EmailSchedulerJob } from "../../email/interfaces/email-events.interface";
-import { EmailMonitoringService } from "../../email-utils/services/email-monitoring.service";
-
-@Processor(EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE)
-export class EmailMonitoringProcessor extends WorkerHost {
-  private readonly logger = new Logger(EmailMonitoringProcessor.name);
-
-  constructor(private readonly emailMonitoringService: EmailMonitoringService) {
-    super();
-  }
-
-  async process(job: Job<EmailMonitoringJob | EmailSchedulerJob>) {
-    const { name, data } = job;
-
-    try {
-      switch (name) {
-        case EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE_PROCESSING:
-          this.logger.debug(`Processing user email check for: ${(data as EmailMonitoringJob).userId}`);
-          await this.emailMonitoringService.processUserEmailCheck(data as EmailMonitoringJob);
-          break;
-
-        case EMAIL_QUEUE_CONSTANTS.EMAIL_SCHEDULER_JOB:
-          this.logger.debug("Processing scheduler job - triggering email monitoring checks");
-          await this.emailMonitoringService.scheduleEmailCheck();
-          break;
-
-        default:
-          this.logger.warn(`Unknown job type: ${name}`);
-          throw new Error(`Unknown job type: ${name}`);
-      }
-    } catch (error) {
-      const userId = "userId" in data ? (data as EmailMonitoringJob).userId : "scheduler";
-      this.logger.error(`Failed to process email monitoring job ${name} for ${userId}:`, error);
-      throw error;
-    }
-  }
-}

src/modules/email-utils/queue/email-polling.processor.ts

@@ -0,0 +1,160 @@
+import { Processor } from "@nestjs/bullmq";
+import { Logger } from "@nestjs/common";
+import { Job } from "bullmq";
+import { firstValueFrom } from "rxjs";
+
+import { WorkerProcessor } from "../../../common/queues/worker.processor";
+import { EMAIL_QUEUE_CONSTANTS } from "../../email/constants/email-events.constant";
+import { UserMailboxIds } from "../../email/interfaces/user-mailbox.interface";
+import { EmailGateway } from "../../email-gateway/email.gateway";
+import { MailServerService } from "../../mail-server/services/mail-server.service";
+import { EmailPollingJobData } from "../interfaces/email-polling-job.interface";
+import { MailboxResolverService } from "../services/mailbox-resolver.service";
+
+@Processor(EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_QUEUE)
+export class EmailPollingProcessor extends WorkerProcessor {
+  protected readonly logger = new Logger(EmailPollingProcessor.name);
+  constructor(
+    private readonly emailGateway: EmailGateway,
+    private readonly mailboxResolverService: MailboxResolverService,
+    private readonly mailServerService: MailServerService,
+  ) {
+    super();
+  }
+
+  async process(job: Job<EmailPollingJobData>): Promise<void> {
+    const { userId, lastCheckTimestamp, connectionId } = job.data;
+
+    this.logger.debug(`Processing email polling job for user ${userId} (connection: ${connectionId})`);
+
+    try {
+      // Check if user is still connected before processing
+      if (!this.emailGateway.isUserConnected(userId)) {
+        this.logger.debug(`User ${userId} no longer connected, skipping email check`);
+        return;
+      }
+
+      // Get user's mailbox information
+      const mailboxIds = await this.mailboxResolverService.getUserMailboxIds(userId);
+      if (!mailboxIds.inbox) {
+        this.logger.warn(`No inbox found for user ${userId}`);
+        return;
+      }
+
+      // Calculate time range for checking new emails
+      const checkSince = lastCheckTimestamp ? new Date(lastCheckTimestamp) : new Date(Date.now() - 5 * 60 * 1000);
+      const now = new Date();
+
+      this.logger.debug(`Checking emails for user ${userId} since ${checkSince.toISOString()}`);
+
+      // Get new emails from inbox since last check
+      const newEmails = await this.getRecentEmails(userId, mailboxIds.inbox, checkSince);
+
+      if (newEmails.length > 0) {
+        this.logger.log(`Found ${newEmails.length} new emails for user ${userId}`);
+
+        // Notify user of each new email via websocket
+
+        const lastEmail = newEmails[newEmails.length - 1];
+        await this.emailGateway.notifyNewEmail(userId, {
+          wildduckUserId: userId,
+          messageId: lastEmail.id,
+          date: lastEmail.date || new Date().toISOString(),
+          from: lastEmail.from,
+          to: lastEmail.to || [],
+          subject: lastEmail.subject,
+          userId,
+          timestamp: lastEmail.date || new Date().toISOString(),
+          mailboxId: lastEmail.mailbox,
+          mailboxName: "Inbox",
+          isRead: !!lastEmail.seen,
+          hasAttachments: !!(lastEmail.attachments && Array.isArray(lastEmail.attachments) && lastEmail.attachments.length > 0),
+        });
+
+        // Update unread count
+        const unreadStats = await this.getUnreadCounts(userId, mailboxIds);
+        await this.emailGateway.notifyUnreadCountUpdate(userId, {
+          userId,
+          totalUnread: unreadStats.total,
+          mailboxCounts: unreadStats.mailboxCounts,
+          count: unreadStats.total,
+          timestamp: now.toISOString(),
+        });
+
+        this.logger.log(`Notified user ${userId} of ${newEmails.length} new emails`);
+      } else {
+        this.logger.debug(`No new emails found for user ${userId}`);
+      }
+
+      // Update job data with new timestamp for next run
+      job.updateData({
+        ...job.data,
+        lastCheckTimestamp: now.toISOString(),
+      });
+    } catch (error) {
+      this.logger.error(`Error processing email polling job for user ${userId}:`, error);
+      throw error; // Re-throw to trigger retry mechanism
+    }
+  }
+
+  // Helper method to get recent emails from a mailbox
+  private async getRecentEmails(userId: string, mailboxId: string, since: Date) {
+    try {
+      const response = await firstValueFrom(
+        this.mailServerService.messages.searchMessages(userId, {
+          mailbox: mailboxId,
+          limit: 50, // Limit to avoid too many results
+          order: "desc", // Most recent first
+        }),
+      );
+
+      if (!response.success || !response.results) {
+        return [];
+      }
+
+      return response.results.filter((email) => {
+        const emailDate = new Date(email.date || new Date().toISOString());
+        return emailDate >= since;
+      });
+    } catch (error) {
+      this.logger.error(`Error fetching recent emails for user ${userId}:`, error);
+      return [];
+    }
+  }
+
+  // Helper method to get unread counts from all mailboxes
+  private async getUnreadCounts(userId: string, mailboxIds: UserMailboxIds) {
+    try {
+      const mailboxCounts: Record<string, number> = {};
+      let total = 0;
+
+      // Get unread count from each mailbox
+      for (const [mailboxName, mailboxId] of Object.entries(mailboxIds)) {
+        if (mailboxId) {
+          try {
+            const response = await firstValueFrom(this.mailServerService.mailboxes.getMailbox(userId, mailboxId));
+
+            if (response.success && response.unseen !== undefined) {
+              mailboxCounts[mailboxName] = response.unseen;
+              total += response.unseen;
+            }
+          } catch (error) {
+            this.logger.warn(`Error getting unread count for ${mailboxName} (${mailboxId}):`, error);
+            mailboxCounts[mailboxName] = 0;
+          }
+        }
+      }
+
+      return {
+        total,
+        mailboxCounts,
+      };
+    } catch (error) {
+      this.logger.error(`Error calculating unread counts for user ${userId}:`, error);
+      return {
+        total: 0,
+        mailboxCounts: {},
+      };
+    }
+  }
+}

src/modules/email-utils/services/email-monitoring.service.ts

@@ -1,399 +0,0 @@
-import { EntityManager } from "@mikro-orm/core";
-import { InjectQueue } from "@nestjs/bullmq";
-import { Injectable, Logger, OnModuleInit } from "@nestjs/common";
-import { Queue } from "bullmq";
-import { firstValueFrom } from "rxjs";
-
-import { MailboxResolverService } from "./mailbox-resolver.service";
-import { EMAIL_QUEUE_CONSTANTS } from "../../email/constants/email-events.constant";
-import { EmailMonitoringJob, EmailPayload, EmailSchedulerJob } from "../../email/interfaces/email-events.interface";
-import { EmailGateway } from "../../email-gateway/email.gateway";
-import { MailServerService } from "../../mail-server/services/mail-server.service";
-import { MailboxEnum } from "../../mailbox/enums/mailbox.enum";
-import { NotificationQueue } from "../../notifications/queue/notification.queue";
-import { User } from "../../users/entities/user.entity";
-
-@Injectable()
-export class EmailMonitoringService implements OnModuleInit {
-  private readonly logger = new Logger(EmailMonitoringService.name);
-  private lastGlobalCheck: Date = new Date();
-  private isMonitoringActive = false;
-
-  constructor(
-    @InjectQueue(EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE) private readonly monitoringQueue: Queue,
-    private readonly emailGateway: EmailGateway,
-    private readonly mailboxResolver: MailboxResolverService,
-    private readonly mailServerService: MailServerService,
-    private readonly notificationQueue: NotificationQueue,
-    private readonly em: EntityManager,
-  ) {}
-
-  async onModuleInit() {
-    this.logger.log("Initializing integrated email monitoring service");
-    await this.startMonitoring();
-  }
-
-  //==============================================
-  async startMonitoring() {
-    if (this.isMonitoringActive) {
-      this.logger.warn("Email monitoring is already running");
-      return;
-    }
-
-    try {
-      await this.stopMonitoring();
-
-      this.logger.log("Starting integrated email monitoring with BullMQ cron scheduler");
-
-      await this.monitoringQueue.add(
-        EMAIL_QUEUE_CONSTANTS.EMAIL_SCHEDULER_JOB,
-        {
-          action: EMAIL_QUEUE_CONSTANTS.EMAIL_MONITORING_ACTION_NAME,
-          timestamp: new Date().toISOString(),
-        } as EmailSchedulerJob,
-        {
-          removeOnComplete: true,
-          removeOnFail: false,
-          repeat: {
-            pattern: EMAIL_QUEUE_CONSTANTS.EMAIL_MONITORING_JOB_PATTERN,
-          },
-        },
-      );
-
-      this.isMonitoringActive = true;
-      this.logger.log("Integrated BullMQ email monitoring scheduler started successfully");
-
-      // Run initial check immediately
-      await this.scheduleEmailCheck();
-    } catch (error) {
-      this.logger.error("Failed to start email monitoring scheduler:", error);
-      throw error;
-    }
-  }
-
-  async stopMonitoring() {
-    try {
-      // Remove all repeatable jobs with this name
-      const repeatableJobs = await this.monitoringQueue.getJobSchedulers();
-      const existingJobs = repeatableJobs.filter((job) => job.name === EMAIL_QUEUE_CONSTANTS.EMAIL_SCHEDULER_JOB);
-
-      if (existingJobs.length > 0) {
-        for (const job of existingJobs) {
-          await this.monitoringQueue.removeJobScheduler(job.key);
-          this.logger.debug(`Removed repeatable job: ${job.key}`);
-        }
-        this.logger.log("Email monitoring scheduler stopped and cleaned up");
-      } else {
-        this.logger.debug("No existing scheduler jobs found to remove");
-      }
-
-      // Clean up old jobs
-      await this.cleanupJobs();
-      this.isMonitoringActive = false;
-    } catch (error) {
-      this.logger.error("Error stopping email monitoring scheduler:", error);
-    }
-  }
-
-  async restartMonitoring() {
-    this.logger.log("Restarting integrated email monitoring");
-    await this.stopMonitoring();
-    await this.startMonitoring();
-  }
-
-  //==============================================
-
-  async scheduleEmailCheck() {
-    try {
-      const connectedUserIds = this.emailGateway.getConnectedUserIds();
-
-      if (connectedUserIds.length === 0) {
-        this.logger.debug("No connected users - skipping email check");
-        return;
-      }
-
-      const em = this.em.fork();
-
-      const activeUsers = await em.find(User, {
-        id: { $in: connectedUserIds },
-        emailEnabled: true,
-        deletedAt: null,
-        wildduckUserId: { $ne: null },
-      });
-
-      this.logger.log(`Monitoring ${activeUsers.length} connected users`);
-
-      for (const user of activeUsers) {
-        await this.monitoringQueue.add(
-          EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE_PROCESSING,
-          {
-            userId: user.id,
-            wildduckUserId: user.wildduckUserId,
-            lastCheckTimestamp: this.lastGlobalCheck,
-          },
-          {
-            delay: Math.random() * 10000, // Random delay to prevent thundering herd
-            removeOnComplete: true,
-            removeOnFail: false,
-            attempts: 3,
-            backoff: {
-              type: "exponential",
-              delay: 2000,
-            },
-          },
-        );
-      }
-
-      this.lastGlobalCheck = new Date();
-    } catch (error) {
-      this.logger.error("Error scheduling email monitoring checks:", error);
-    }
-  }
-
-  //==============================================
-
-  async processUserEmailCheck(job: EmailMonitoringJob) {
-    const { userId, lastCheckTimestamp } = job;
-
-    try {
-      const em = this.em.fork();
-
-      const user = await em.findOne(User, {
-        id: userId,
-        emailEnabled: true,
-        deletedAt: null,
-      });
-
-      if (!user || !user.wildduckUserId) {
-        this.logger.warn(`User ${userId} not found or email not enabled`);
-        return;
-      }
-
-      const inboxMailboxId = await this.mailboxResolver.getInboxMailboxId(user.wildduckUserId);
-
-      const query: Record<string, unknown> = {
-        limit: 100,
-        order: "desc",
-        unseen: true,
-      };
-
-      if (lastCheckTimestamp) {
-        const timestamp = lastCheckTimestamp instanceof Date ? lastCheckTimestamp : new Date(lastCheckTimestamp);
-        query.datestart = timestamp.toISOString();
-      }
-
-      const { results } = await firstValueFrom(this.mailServerService.messages.listMessages(user.wildduckUserId, inboxMailboxId, query));
-
-      if (results.length > 0) {
-        this.logger.log(`${user.emailAddress}: ${results.length} new messages`);
-
-        for (const message of results) {
-          await this.processNewEmailMessage(user.id, user.wildduckUserId, message);
-        }
-      }
-
-      await this.updateUnreadCount(userId, user.wildduckUserId, user.emailAddress);
-    } catch (error) {
-      this.logger.error(`Error checking emails for user ${userId}:`, error);
-      throw error;
-    }
-  }
-
-  //==============================================
-
-  private async processNewEmailMessage(userId: string, _wildduckUserId: string, message: Record<string, any>) {
-    try {
-      if (message.seen) {
-        return;
-      }
-
-      const emailPayload: EmailPayload = {
-        messageId: message.id,
-        userId,
-        from: message.from,
-        to: message.to,
-        subject: message.subject,
-        hasAttachments: message.hasAttachments,
-        mailboxName: message.mailboxName,
-        isRead: message.seen,
-        mailboxId: message.mailboxId,
-        timestamp: message.timestamp,
-      };
-
-      // Check if message.from exists and has at least one element
-      if (message.from && Array.isArray(message.from) && message.from.length > 0 && message.from[0]) {
-        await this.notificationQueue.addNewEmailNotification(userId, {
-          fromEmail: message.from[0].address || "unknown@example.com",
-          fromName: message.from[0].name || "Unknown Sender",
-          subject: message.subject || "No Subject",
-        });
-      } else {
-        this.logger.warn(`Email message ${message.id} for user ${userId} has invalid 'from' field`);
-        await this.notificationQueue.addNewEmailNotification(userId, {
-          fromEmail: "unknown@example.com",
-          fromName: "Unknown Sender",
-          subject: message.subject || "No Subject",
-        });
-      }
-
-      await this.emailGateway.notifyNewEmail(userId, emailPayload);
-    } catch (error) {
-      this.logger.error(`Error processing new email message for user ${userId}:`, error);
-    }
-  }
-
-  //==============================================
-
-  private async updateUnreadCount(userId: string, wildduckUserId: string, userEmailAddress?: string) {
-    try {
-      const inboxMailboxId = await this.mailboxResolver.getInboxMailboxId(wildduckUserId);
-
-      const { results } = await firstValueFrom(
-        this.mailServerService.messages.listMessages(wildduckUserId, inboxMailboxId, {
-          limit: 250,
-          unseen: true,
-        }),
-      );
-
-      const unreadCount = results.length;
-
-      this.emailGateway.notifyUnreadCountUpdate(userId, {
-        count: unreadCount,
-        timestamp: new Date().toISOString(),
-        totalUnread: unreadCount,
-        mailboxCounts: {
-          [MailboxEnum.INBOX]: unreadCount,
-        },
-        userId,
-      });
-
-      if (unreadCount > 0 && userEmailAddress) {
-        this.logger.log(`${userEmailAddress}: ${unreadCount} unread total`);
-      }
-    } catch (error) {
-      this.logger.error(`Error updating unread count for user ${userId}:`, error);
-    }
-  }
-
-  //==============================================
-
-  async checkUserEmails(userId: string, wildduckUserId: string): Promise<void> {
-    await this.monitoringQueue.add(
-      EMAIL_QUEUE_CONSTANTS.EMAIL_QUEUE_PROCESSING,
-      {
-        userId,
-        wildduckUserId: wildduckUserId,
-        lastCheckTimestamp: new Date(Date.now() - 60000),
-      },
-      {
-        attempts: 3,
-        removeOnComplete: true,
-        removeOnFail: false,
-      },
-    );
-  }
-
-  //==============================================
-
-  async getMonitoringStatus() {
-    try {
-      const repeatableJobs = await this.monitoringQueue.getJobSchedulers();
-      const schedulerJob = repeatableJobs.find((job) => job.name === EMAIL_QUEUE_CONSTANTS.EMAIL_SCHEDULER_JOB);
-      const queueStatus = await this.getQueueStatus();
-
-      return {
-        isRunning: this.isMonitoringActive && !!schedulerJob,
-        cronPattern: schedulerJob?.pattern || null,
-        nextRunTime: schedulerJob?.next ? new Date(schedulerJob.next) : null,
-        checkInterval: EMAIL_QUEUE_CONSTANTS.EMAIL_MONITORING_JOB_PATTERN,
-        lastGlobalCheck: this.lastGlobalCheck,
-        queueStatus,
-        timestamp: new Date().toISOString(),
-      };
-    } catch (error) {
-      this.logger.error("Error getting monitoring status:", error);
-      return {
-        isRunning: false,
-        error: error instanceof Error ? error.message : "Unknown error",
-        timestamp: new Date().toISOString(),
-      };
-    }
-  }
-
-  //==============================================
-
-  async getQueueStatus() {
-    const [waiting, active, completed, failed] = await Promise.all([
-      this.monitoringQueue.getWaiting(),
-      this.monitoringQueue.getActive(),
-      this.monitoringQueue.getCompleted(),
-      this.monitoringQueue.getFailed(),
-    ]);
-
-    return {
-      waiting: waiting.length,
-      active: active.length,
-      completed: completed.length,
-      failed: failed.length,
-    };
-  }
-
-  //==============================================
-
-  async getDetailedQueueInfo() {
-    try {
-      const [waiting, active, completed, failed, repeatableJobs] = await Promise.all([
-        this.monitoringQueue.getWaiting(),
-        this.monitoringQueue.getActive(),
-        this.monitoringQueue.getCompleted(0, 10),
-        this.monitoringQueue.getFailed(0, 10),
-        this.monitoringQueue.getJobSchedulers(),
-      ]);
-
-      return {
-        counts: {
-          waiting: waiting.length,
-          active: active.length,
-          completed: completed.length,
-          failed: failed.length,
-          repeatable: repeatableJobs.length,
-        },
-        repeatableJobs: repeatableJobs.map((job) => ({
-          name: job.name,
-          pattern: job.pattern,
-          next: job.next ? new Date(job.next) : null,
-        })),
-        recentCompleted: completed.map((job) => ({
-          id: job.id,
-          processedOn: job.processedOn ? new Date(job.processedOn) : null,
-          finishedOn: job.finishedOn ? new Date(job.finishedOn) : null,
-        })),
-        recentFailed: failed.map((job) => ({
-          id: job.id,
-          failedReason: job.failedReason,
-          processedOn: job.processedOn ? new Date(job.processedOn) : null,
-        })),
-      };
-    } catch (error) {
-      this.logger.error("Error getting detailed queue info:", error);
-      throw error;
-    }
-  }
-
-  //==============================================
-
-  async cleanupJobs() {
-    try {
-      // Clean completed jobs older than 30 minutes
-      await this.monitoringQueue.clean(1000 * 60 * 30, 50, "completed");
-      // Clean failed jobs older than 2 hours
-      await this.monitoringQueue.clean(1000 * 60 * 120, 20, "failed");
-      // Clean active jobs older than 10 minutes (stuck jobs)
-      await this.monitoringQueue.clean(1000 * 60 * 10, 10, "active");
-
-      this.logger.log("Successfully cleaned up old monitoring jobs");
-    } catch (error) {
-      this.logger.error("Error cleaning up jobs:", error);
-    }
-  }
-}

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

@@ -5,18 +5,20 @@ import { firstValueFrom } from "rxjs";
 import { MailboxResolverService } from "./mailbox-resolver.service";
 import { WebSocketEvent } from "../../email/constants/email-events.constant";
 import {
-  EmailDeletePayload,
+  IEmailPayload,
-  EmailSentNotificationParams,
+  IEmailSentNotificationParams,
-  EmailSentPayload,
+  IEmailSentPayload,
-  EmailStatusNotificationParams,
+  IEmailStatusNotificationParams,
-  EmailStatusPayload,
+  IMailboxUpdateNotificationParams,
-  MailboxUpdateNotificationParams,
+  IMailboxUpdatePayload,
-  MailboxUpdatePayload,
+  IUnreadCountPayload,
-  UnreadCountPayload,
 } from "../../email/interfaces/email-events.interface";
+import { IEmailStatusPayload } from "../../email/interfaces/email-events.interface";
+import { IEmailDeletePayload } from "../../email/interfaces/email-events.interface";
 import { EmailGateway } from "../../email-gateway/email.gateway";
 import { MailServerService } from "../../mail-server/services/mail-server.service";
 import { MailboxEnum } from "../../mailbox/enums/mailbox.enum";
+import { NotificationQueue } from "../../notifications/queue/notification.queue";
 import { User } from "../../users/entities/user.entity";
 
 @Injectable()
@@ -27,13 +29,69 @@ export class EmailNotificationService {
     private readonly emailGateway: EmailGateway,
     private readonly mailboxResolver: MailboxResolverService,
     private readonly mailServerService: MailServerService,
+    private readonly notificationQueue: NotificationQueue,
     private readonly em: EntityManager,
   ) {}
 
   //************************************************************************ */
-  async notifyEmailSent(params: EmailSentNotificationParams) {
+
+  async notifyNewEmail(params: IEmailPayload) {
     try {
-      const payload: EmailSentPayload = {
+      const em = this.em.fork();
+      const user = await em.findOne(User, {
+        wildduckUserId: params.wildduckUserId,
+        emailEnabled: true,
+        deletedAt: null,
+      });
+
+      if (!user) {
+        this.logger.warn(`User not found with wildduckUserId: ${params.wildduckUserId}`);
+        return;
+      }
+
+      // Create WebSocket payload
+      const emailPayload: IEmailPayload = {
+        wildduckUserId: params.wildduckUserId,
+        messageId: params.messageId,
+        userId: user.id,
+        date: params.date,
+        from: params.from,
+        to: params.to,
+        subject: params.subject,
+        hasAttachments: params.hasAttachments,
+        mailboxName: params.mailboxName,
+        isRead: false,
+        mailboxId: params.mailboxId,
+        timestamp: new Date().toISOString(),
+      };
+
+      // Send WebSocket notification
+      const success = await this.emailGateway.notifyNewEmail(user.id, emailPayload);
+
+      // Send push notification
+      await this.notificationQueue.addNewEmailNotification(user.id, {
+        fromEmail: params.from.address,
+        fromName: params.from.name || "Unknown Sender",
+        subject: params.subject,
+      });
+
+      // Update unread count
+      await this.updateUnreadCount(user);
+
+      if (success) {
+        this.logger.log(`New email notification dispatched for user ${user.id}: ${params.subject}`);
+      } else {
+        this.logger.warn(`Failed to dispatch new email notification for user ${user.id}: user not connected`);
+      }
+    } catch (error) {
+      this.logger.error(`Failed to notify new email for user ${params.wildduckUserId}:`, error);
+    }
+  }
+
+  //************************************************************************ */
+  async notifyEmailSent(params: IEmailSentNotificationParams) {
+    try {
+      const payload: IEmailSentPayload = {
         userId: params.userId,
         messageId: params.messageId,
         mailboxId: params.mailboxId,
@@ -59,7 +117,7 @@ export class EmailNotificationService {
   }
 
   //************************************************************************ */
-  async notifyEmailUnread(params: EmailStatusNotificationParams) {
+  async notifyEmailUnread(params: IEmailStatusNotificationParams) {
     try {
       const em = this.em.fork();
       const user = await em.findOne(User, {
@@ -73,7 +131,7 @@ export class EmailNotificationService {
         return;
       }
 
-      const payload: EmailStatusPayload = {
+      const payload: IEmailStatusPayload = {
         userId: user.id,
         messageId: params.messageId,
         mailboxId: params.mailboxId,
@@ -94,7 +152,7 @@ export class EmailNotificationService {
   }
 
   //************************************************************************ */
-  async notifyEmailRead(params: EmailStatusNotificationParams) {
+  async notifyEmailRead(params: IEmailStatusNotificationParams) {
     try {
       const em = this.em.fork();
       const user = await em.findOne(User, {
@@ -108,7 +166,7 @@ export class EmailNotificationService {
         return;
       }
 
-      const payload: EmailStatusPayload = {
+      const payload: IEmailStatusPayload = {
         userId: user.id,
         messageId: params.messageId,
         mailboxId: params.mailboxId,
@@ -129,7 +187,7 @@ export class EmailNotificationService {
   }
 
   //************************************************************************ */
-  async notifyEmailDeleted(params: EmailStatusNotificationParams) {
+  async notifyEmailDeleted(params: IEmailStatusNotificationParams) {
     try {
       const em = this.em.fork();
       const user = await em.findOne(User, {
@@ -143,7 +201,7 @@ export class EmailNotificationService {
         return;
       }
 
-      const payload: EmailDeletePayload = {
+      const payload: IEmailDeletePayload = {
         userId: user.id,
         messageId: params.messageId,
         mailboxId: params.mailboxId,
@@ -164,7 +222,7 @@ export class EmailNotificationService {
   }
 
   //************************************************************************ */
-  async notifyEmailFlagged(params: EmailStatusNotificationParams) {
+  async notifyEmailFlagged(params: IEmailStatusNotificationParams) {
     try {
       // Find the user to get the correct User entity ID for websocket notifications
       const em = this.em.fork();
@@ -179,7 +237,7 @@ export class EmailNotificationService {
         return;
       }
 
-      const payload: EmailStatusPayload = {
+      const payload: IEmailStatusPayload = {
         userId: user.id, // Use User entity ID for websocket
         messageId: params.messageId,
         mailboxId: params.mailboxId,
@@ -199,7 +257,7 @@ export class EmailNotificationService {
   }
 
   //************************************************************************ */
-  async notifyEmailUnflagged(params: EmailStatusNotificationParams) {
+  async notifyEmailUnflagged(params: IEmailStatusNotificationParams) {
     try {
       // Find the user to get the correct User entity ID for websocket notifications
       const em = this.em.fork();
@@ -214,7 +272,7 @@ export class EmailNotificationService {
         return;
       }
 
-      const payload: EmailStatusPayload = {
+      const payload: IEmailStatusPayload = {
         userId: user.id, // Use User entity ID for websocket
         messageId: params.messageId,
         mailboxId: params.mailboxId,
@@ -234,9 +292,9 @@ export class EmailNotificationService {
   }
 
   //************************************************************************ */
-  async notifyMailboxUpdate(params: MailboxUpdateNotificationParams) {
+  async notifyMailboxUpdate(params: IMailboxUpdateNotificationParams) {
     try {
-      const payload: MailboxUpdatePayload = {
+      const payload: IMailboxUpdatePayload = {
         userId: params.userId,
         mailboxId: params.mailboxId,
         mailboxName: params.mailboxName,
@@ -271,7 +329,7 @@ export class EmailNotificationService {
 
       const unreadCount = results.length;
 
-      const payload: UnreadCountPayload = {
+      const payload: IUnreadCountPayload = {
         userId: user.id, // Use the actual User entity ID for the notification payload
         count: unreadCount,
         timestamp: new Date().toISOString(),

src/modules/email-utils/services/email-polling.service.ts

@@ -0,0 +1,148 @@
+import { InjectQueue } from "@nestjs/bullmq";
+import { Injectable, Logger } from "@nestjs/common";
+import { Queue } from "bullmq";
+
+import { EMAIL_QUEUE_CONSTANTS } from "../../email/constants/email-events.constant";
+import { EmailPollingJobData } from "../interfaces/email-polling-job.interface";
+
+@Injectable()
+export class EmailPollingService {
+  private readonly logger = new Logger(EmailPollingService.name);
+
+  private readonly activeJobs = new Map<string, { jobId: string; connectionId: string }>();
+
+  constructor(@InjectQueue(EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_QUEUE) private readonly emailPollingQueue: Queue<EmailPollingJobData>) {}
+
+  //=======================================================
+  async startPollingForUser(userId: string, connectionId: string): Promise<void> {
+    try {
+      await this.stopPollingForUser(userId);
+
+      const jobData: EmailPollingJobData = {
+        userId,
+        connectionId,
+        lastCheckTimestamp: new Date().toISOString(),
+      };
+
+      const job = await this.emailPollingQueue.add(EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_JOB, jobData, {
+        repeat: {
+          every: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_INTERVAL,
+        },
+        removeOnComplete: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_REMOVE_ON_COMPLETE,
+        removeOnFail: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_REMOVE_ON_FAIL,
+        attempts: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_ATTEMPTS,
+        priority: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_PRIORITY,
+        backoff: {
+          type: "exponential",
+          delay: EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_BACKOFF_DELAY,
+        },
+      });
+
+      this.activeJobs.set(userId, {
+        jobId: job.id!,
+        connectionId,
+      });
+
+      this.logger.log(`Started email polling for user ${userId} with job ID: ${job.id}`);
+    } catch (error) {
+      this.logger.error(`Failed to start email polling for user ${userId}:`, error);
+      throw error;
+    }
+  }
+
+  //=======================================================
+  async stopPollingForUser(userId: string): Promise<void> {
+    try {
+      const activeJob = this.activeJobs.get(userId);
+      if (!activeJob) {
+        this.logger.debug(`No active polling job found for user ${userId}`);
+        return;
+      }
+
+      await this.removeCronJob(userId);
+      this.activeJobs.delete(userId);
+
+      this.logger.log(`Stopped email polling for user ${userId}`);
+    } catch (error) {
+      this.logger.error(`Failed to stop email polling for user ${userId}:`, error);
+    }
+  }
+
+  //=======================================================
+  private async removeCronJob(_userId: string): Promise<void> {
+    try {
+      //=======================================================
+      const repeatableJobs = await this.emailPollingQueue.getJobSchedulers();
+      const existingJob = repeatableJobs.filter((job) => job.name === EMAIL_QUEUE_CONSTANTS.EMAIL_POLLING_JOB);
+
+      if (existingJob.length > 0) {
+        for (const job of existingJob) {
+          const removed = await this.emailPollingQueue.removeJobScheduler(job.key);
+          this.logger.debug(`Removed existing cron job with key ${job.key}: ${removed}`);
+        }
+      } else {
+        this.logger.debug("No existing cron job found to remove");
+      }
+
+      //=======================================================
+      await this.emailPollingQueue.clean(0, 100, "completed");
+      await this.emailPollingQueue.clean(0, 100, "failed");
+    } catch (error) {
+      this.logger.error("Failed to remove email polling cron job", error);
+    }
+  }
+
+  //=======================================================
+  isPollingActive(userId: string): boolean {
+    return this.activeJobs.has(userId);
+  }
+
+  //=======================================================
+  getPollingStatus(userId: string): { active: boolean; jobId?: string; connectionId?: string } {
+    const activeJob = this.activeJobs.get(userId);
+    return {
+      active: !!activeJob,
+      jobId: activeJob?.jobId,
+      connectionId: activeJob?.connectionId,
+    };
+  }
+
+  //=======================================================
+  async stopAllPolling(): Promise<void> {
+    this.logger.log("Stopping all email polling jobs...");
+
+    const userIds = Array.from(this.activeJobs.keys());
+
+    for (const userId of userIds) {
+      await this.stopPollingForUser(userId);
+    }
+
+    this.logger.log(`Stopped ${userIds.length} email polling jobs`);
+  }
+
+  //=======================================================
+  getPollingStats(): {
+    activeJobs: number;
+    userIds: string[];
+    connections: Array<{ userId: string; connectionId: string; jobId: string }>;
+  } {
+    const connections = Array.from(this.activeJobs.entries()).map(([userId, jobInfo]) => ({
+      userId,
+      connectionId: jobInfo.connectionId,
+      jobId: jobInfo.jobId,
+    }));
+
+    return {
+      activeJobs: this.activeJobs.size,
+      userIds: Array.from(this.activeJobs.keys()),
+      connections,
+    };
+  }
+
+  //=======================================================
+  async restartPollingForUser(userId: string, connectionId: string): Promise<void> {
+    this.logger.log(`Restarting email polling for user ${userId}`);
+    await this.stopPollingForUser(userId);
+    await this.startPollingForUser(userId, connectionId);
+  }
+}

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

@@ -39,6 +39,8 @@ export type WebSocketNamespace = (typeof WEBSOCKET_NAMESPACES)[keyof typeof WEBS
 
 export const EMAIL_QUEUE_CONSTANTS = {
   EMAIL_QUEUE: "email_queue",
+  EMAIL_POLLING_QUEUE: "email_polling_queue",
+  EMAIL_POLLING_JOB: "email_polling_job",
   EMAIL_BULK_ACTION_QUEUE: "email_bulk_action_queue",
   EMAIL_BULK_ACTION_SEEN_ALL: "email_bulk_action_seen_all",
   EMAIL_BULK_ACTION_EMPTY_TRASH: "email_bulk_action_empty_trash",
@@ -50,8 +52,10 @@ export const EMAIL_QUEUE_CONSTANTS = {
   EMAIL_BULK_ACTION_BACKOFF_DELAY: 2000, // 2 seconds
   EMAIL_BULK_ACTION_PRIORITY: 1,
   //
-  EMAIL_QUEUE_PROCESSING: "email_queue_processing",
+  EMAIL_POLLING_INTERVAL: 30000, // 30 seconds
-  EMAIL_SCHEDULER_JOB: "email_monitoring_scheduler",
+  EMAIL_POLLING_REMOVE_ON_COMPLETE: 5, // Keep last 5 completed jobs
-  EMAIL_MONITORING_ACTION_NAME: "schedule_email_monitoring",
+  EMAIL_POLLING_REMOVE_ON_FAIL: 10, // Keep last 10 failed jobs
-  EMAIL_MONITORING_JOB_PATTERN: "*/5 * * * * *", // Every 5 seconds
+  EMAIL_POLLING_ATTEMPTS: 3, // Retry failed jobs 3 times
+  EMAIL_POLLING_BACKOFF_DELAY: 2000, // 2 seconds
+  EMAIL_POLLING_PRIORITY: 1,
 } as const;

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

@@ -1,28 +1,30 @@
-import { EMAIL_QUEUE_CONSTANTS } from "../constants/email-events.constant";
-
 // Base interface with common properties
-export interface BaseEventPayload {
+export interface IBaseEventPayload {
   userId: string;
   timestamp: string;
 }
 
 // Base interface for message-related events
-export interface BaseMessageEventPayload extends BaseEventPayload {
+export interface IBaseMessageEventPayload extends IBaseEventPayload {
   messageId: number;
 }
 
 // Base interface for mailbox-related events
-export interface BaseMailboxEventPayload extends BaseEventPayload {
+export interface IBaseMailboxEventPayload extends IBaseEventPayload {
   mailboxId: string;
 }
 
 // Base interface for message events that also involve mailboxes
-export interface BaseMessageMailboxEventPayload extends BaseMessageEventPayload {
+export interface IBaseMessageMailboxEventPayload extends IBaseMessageEventPayload {
   mailboxId: string;
 }
 
 // Specific email event interfaces
-export interface EmailPayload extends BaseMessageMailboxEventPayload {
+export interface IEmailPayload extends IBaseMessageMailboxEventPayload {
+  wildduckUserId: string;
+  userId: string;
+  messageId: number;
+  date: string;
   from: {
     name?: string;
     address: string;
@@ -38,18 +40,18 @@ export interface EmailPayload extends BaseMessageMailboxEventPayload {
   isRead: boolean;
 }
 
-export type EmailStatusPayload = BaseMessageMailboxEventPayload;
+export type IEmailStatusPayload = IBaseMessageMailboxEventPayload;
 
-export interface EmailMovePayload extends BaseMessageEventPayload {
+export interface IEmailMovePayload extends IBaseMessageEventPayload {
   fromMailboxId: string;
   toMailboxId: string;
   fromMailboxName: string;
   toMailboxName: string;
 }
 
-export type EmailDeletePayload = BaseMessageMailboxEventPayload;
+export type IEmailDeletePayload = IBaseMessageMailboxEventPayload;
 
-export interface EmailSentPayload extends BaseMessageMailboxEventPayload {
+export interface IEmailSentPayload extends IBaseMessageMailboxEventPayload {
   from: {
     name?: string;
     address: string;
@@ -64,30 +66,30 @@ export interface EmailSentPayload extends BaseMessageMailboxEventPayload {
   isRead: boolean;
 }
 
-export interface MailboxUpdatePayload extends BaseMailboxEventPayload {
+export interface IMailboxUpdatePayload extends IBaseMailboxEventPayload {
   mailboxName: string;
   unreadCount: number;
   totalCount: number;
 }
 
-export interface UnreadCountPayload extends BaseEventPayload {
+export interface IUnreadCountPayload extends IBaseEventPayload {
   totalUnread: number;
   mailboxCounts: Record<string, number>;
   count: number;
 }
 
-export interface EmailWebSocketEvents {
+export interface IEmailWebSocketEvents {
   // Server to Client events
-  new_email: EmailPayload;
+  new_email: IEmailPayload;
-  email_read: EmailStatusPayload;
+  email_read: IEmailStatusPayload;
-  email_unread: EmailStatusPayload;
+  email_unread: IEmailStatusPayload;
-  email_flagged: EmailStatusPayload;
+  email_flagged: IEmailStatusPayload;
-  email_unflagged: EmailStatusPayload;
+  email_unflagged: IEmailStatusPayload;
-  email_deleted: EmailDeletePayload;
+  email_deleted: IEmailDeletePayload;
-  email_moved: EmailMovePayload;
+  email_moved: IEmailMovePayload;
-  email_sent: EmailSentPayload;
+  email_sent: IEmailSentPayload;
-  mailbox_updated: MailboxUpdatePayload;
+  mailbox_updated: IMailboxUpdatePayload;
-  unread_count_updated: UnreadCountPayload;
+  unread_count_updated: IUnreadCountPayload;
 
   // Connection events
   connection_established: { userId: string; timestamp: string };
@@ -98,18 +100,7 @@ export interface EmailWebSocketEvents {
   pong: { timestamp: string };
 }
 
-export interface EmailMonitoringJob {
+export interface INewEmailData {
-  userId: string;
-  wildduckUserId?: string;
-  lastCheckTimestamp?: Date;
-}
-
-export interface EmailSchedulerJob {
-  action: typeof EMAIL_QUEUE_CONSTANTS.EMAIL_MONITORING_ACTION_NAME;
-  timestamp: string;
-}
-
-export interface NewEmailData {
   userId: string;
   messageId: number;
   wildduckUserId: string;
@@ -117,7 +108,7 @@ export interface NewEmailData {
 }
 
 // Helper interfaces for notification service parameters
-export interface EmailSentNotificationParams {
+export interface IEmailSentNotificationParams {
   userId: string;
   messageId: number;
   mailboxId: string;
@@ -126,13 +117,13 @@ export interface EmailSentNotificationParams {
   from: { name?: string; address: string };
 }
 
-export interface EmailStatusNotificationParams {
+export interface IEmailStatusNotificationParams {
   userId: string;
   messageId: number;
   mailboxId: string;
 }
 
-export interface MailboxUpdateNotificationParams {
+export interface IMailboxUpdateNotificationParams {
   userId: string;
   mailboxId: string;
   mailboxName: string;

src/modules/templates/services/template-processor.service.ts

@@ -51,7 +51,7 @@ export class TemplateProcessorService {
 
   //**************************************************** */
   private async getSelectedTemplate(businessId: string, userId: string) {
-    const user = await this.userRepository.findOne({ id: userId, business: { id: businessId } });
+    const user = await this.userRepository.findOne({ id: userId, business: { id: businessId }, deletedAt: null }, { populate: ["template"] });
     if (!user) throw new BadRequestException(UserMessage.USER_NOT_FOUND);
 
     if (user.template) return user.template;