(Feat) BullMQ Integration for Background Jobs

.env.example

@@ -21,6 +21,9 @@ REFRESH_TOKEN_EXPIRY=
 # Bcrypt Config
 BCRYPT_SALT_ROUNDS=
 
+# HMAC Config
+HMAC_SECRET=
+
 # Google OAuth Config
 GOOGLE_CLIENT_ID=
 GOOGLE_CLIENT_SECRET=

TODO.md

@@ -29,6 +29,7 @@
 - [x] OTP expiration must be less than 15m. Decide on the exact duration (e.g., 5m)
 - [ ] Add middleware to check if user is deleted
 - [ ] Centralize error messages
+- [x] Integrate with BullMQ for background job processing
 
 ## Improvements
 

deploy.sh

@@ -15,13 +15,21 @@ npx prisma generate --schema=src/database/prisma/schema.prisma
 echo "🏗 Building the app..."
 npm run build
 
-echo "🚀 Restarting app with PM2..."
+echo "🚀 Restarting apps with PM2..."
 if pm2 describe taskora-api > /dev/null 2>&1; then
-    echo "📍 Restarting existing process..."
+    echo "📍 Restarting taskora-api..."
     pm2 restart taskora-api
 else
-    echo "🆕 Starting new process with ecosystem config..."
-    pm2 start ecosystem.config.js
+    echo "🆕 Starting taskora-api from ecosystem config..."
+    pm2 start ecosystem.config.js --only taskora-api
+fi
+
+if pm2 describe taskora-email-worker > /dev/null 2>&1; then
+    echo "📍 Restarting taskora-email-worker..."
+    pm2 restart taskora-email-worker
+else
+    echo "🆕 Starting taskora-email-worker from ecosystem config..."
+    pm2 start ecosystem.config.js --only taskora-email-worker
 fi
 
 echo "💾 Saving PM2 process list..."

ecosystem.config.js

@@ -2,7 +2,7 @@ module.exports = {
   apps: [
     {
       name: 'taskora-api',
-      script: './dist/index.js',
+      script: 'dist/index.js',
       instances: 1,
       exec_mode: 'fork',
       node_args: '--expose-gc --max-old-space-size=1024',
@@ -15,5 +15,16 @@ module.exports = {
       kill_timeout: 5000,
       watch: false,
     },
+    {
+      name: 'taskora-email-worker',
+      script: 'dist/workers/email.worker.js',
+      instances: 1,
+      autorestart: true,
+      watch: false,
+      env: {
+        NODE_ENV: 'production',
+        FRONTEND_URL: 'https://taskora.live',
+      },
+    },
   ],
 };

package.json

@@ -8,6 +8,7 @@
     "post:build": "cpy src/templates/**/*.html dist/templates",
     "dev": "concurrently \"tsc-watch --onSuccess 'node ./dist/index.js'\" \"rm -rf dist\" \"npm run build\"",
     "start": "NODE_ENV=production node dist/index.js",
+    "start:worker": "node dist/workers/email.worker.js",
     "format": "prettier --write \"src/**/*.ts\"",
     "format:check": "prettier --check \"src/**/*.ts\"",
     "lint": "eslint \"src/**/*.ts\" --fix",

src/config/general.env.ts

@@ -14,3 +14,4 @@ export const mailPort = env('MAIL_PORT');
 export const mailAuthUser = env('MAIL_AUTH_USER');
 export const mailAuthPassword = env('MAIL_AUTH_PASSWORD');
 export const correctAnswer = env('CORRECT_ANSWER');
+export const hmacSecret = env('HMAC_SECRET');

src/jobs/email.job.ts

@@ -0,0 +1,76 @@
+import { JobsOptions } from 'bullmq';
+import { emailQueue } from '../queues';
+import { HashingService } from '../services';
+import { SendEmailVerificationJob, SendForgetPasswordJob } from '../types';
+import { logger, MAGIC_NUMBERS, WORKERS } from '../utils';
+
+export class EmailJob {
+  static async addVerificationEmailJob(
+    verificationJobDetails: SendEmailVerificationJob,
+  ) {
+    try {
+      const hashedEmail = this.hashEmail(verificationJobDetails.email);
+      const key = `verify-${hashedEmail}`;
+      const jobOptions = this.createJobOptions(key);
+      await emailQueue.add(
+        WORKERS.SEND_VERIFICATION_EMAIL,
+        {
+          email: verificationJobDetails.email,
+          name: verificationJobDetails.name,
+          token: verificationJobDetails.token,
+        },
+        jobOptions,
+      );
+
+      logger.info(
+        `Verification email job scheduled for ${verificationJobDetails.email} with key ${key}`,
+      );
+    } catch (error) {
+      logger.error(`Failed to add verification email job: ${error}`);
+      throw error;
+    }
+  }
+
+  static async addForgetPasswordEmailJob(
+    forgetPasswordDetails: SendForgetPasswordJob,
+  ) {
+    try {
+      const hashedEmail = this.hashEmail(forgetPasswordDetails.email);
+      const key = `forget-${hashedEmail}`;
+      const jobOptions = this.createJobOptions(key);
+      await emailQueue.add(
+        WORKERS.SEND_FORGET_PASSWORD_EMAIL,
+        {
+          email: forgetPasswordDetails.email,
+          name: forgetPasswordDetails.name,
+          otp: forgetPasswordDetails.otp,
+        },
+        jobOptions,
+      );
+
+      logger.info(
+        `Forget password email job scheduled for ${forgetPasswordDetails.email} with key ${key}`,
+      );
+    } catch (error) {
+      logger.error(`Failed to add forget password email job: ${error}`);
+      throw error;
+    }
+  }
+
+  private static createJobOptions(jobId: string): JobsOptions {
+    return {
+      jobId,
+      attempts: MAGIC_NUMBERS.MAX_NUMBER_OF_RETRIES,
+      backoff: {
+        type: 'exponential',
+        delay: MAGIC_NUMBERS.FIVE_SECONDS_IN_MILLISECONDS,
+      },
+      removeOnComplete: MAGIC_NUMBERS.MAX_COUNT_FOR_REMOVE_ON_COMPLETE,
+      removeOnFail: MAGIC_NUMBERS.MAX_COUNT_FOR_REMOVE_ON_FAILURE,
+    };
+  }
+
+  private static hashEmail(email: string): string {
+    return HashingService.generateHashWithHmac(email);
+  }
+}

src/jobs/index.ts

@@ -0,0 +1 @@
+export * from './email.job';

src/queues/email.queue.ts

@@ -0,0 +1,8 @@
+import { Queue } from 'bullmq';
+import { redisHost, redisPort } from '../config';
+import { EmailJobData } from '../types';
+import { QUEUES } from '../utils';
+
+export const emailQueue = new Queue<EmailJobData>(QUEUES.EMAIL_QUEUE, {
+  connection: { host: redisHost, port: Number(redisPort) },
+});

src/queues/index.ts

@@ -1 +1,2 @@
 export * from './competition.queue';
+export * from './email.queue';

src/services/auth.service.ts

@@ -1,6 +1,5 @@
 import { Provider } from '@prisma/client';
 import {
-  emailService,
   HashingService,
   JwtService,
   otpService,
@@ -9,6 +8,7 @@ import {
   userService,
 } from '.';
 import { IAuth, IResetPassword, IUser, IVerifyOtp } from '../interfaces';
+import { EmailJob } from '../jobs';
 import { IUserInfo } from '../types';
 import {
   ApiError,
@@ -82,11 +82,11 @@ export class AuthService extends BaseAuthService {
       MAGIC_NUMBERS.ONE_DAY_IN_SECONDS,
     );
 
-    emailService.sendVerificationEmail(
-      user.email,
-      user.name,
-      verificationToken,
-    );
+    await EmailJob.addVerificationEmailJob({
+      email: user.email,
+      name: user.name,
+      token: verificationToken,
+    });
   }
 
   async login(userLoginInfo: IAuth) {
@@ -272,7 +272,11 @@ export class AuthService extends BaseAuthService {
       ),
     ]);
 
-    emailService.sendForgetPasswordEmail(user.email, user.name, otp);
+    await EmailJob.addForgetPasswordEmailJob({
+      email: user.email,
+      name: user.name,
+      otp,
+    });
   }
 
   async verifyOTP(otpInfo: IVerifyOtp) {

src/services/email.service.ts

@@ -4,50 +4,76 @@ import {
   getOTPTemplate,
   getVerifyEmailTemplate,
   getWinnersTemplate,
+  logger,
 } from '../utils';
 import { smtpMailProvider } from './providers';
 
 export class EmailService {
   constructor(private readonly mailProvider: IMailProvider) {}
 
   async sendVerificationEmail(email: string, name: string, token: string) {
-    const verifyEmailTemplate = getVerifyEmailTemplate();
-    const verifyEmailUrl = `${frontendUrl}/verify-email?token=${token}`;
-
-    const html = verifyEmailTemplate
-      .replace(/{{verifyEmailUrl}}/g, verifyEmailUrl)
-      .replace(/{{name}}/g, name);
-
-    await this.mailProvider.sendEmail({
-      to: email,
-      subject: 'Verify your email',
-      text: 'Verify your email',
-      html,
-    });
+    try {
+      const verifyEmailTemplate = getVerifyEmailTemplate();
+      const verifyEmailUrl = `${frontendUrl}/verify-email?token=${token}`;
+
+      const html = verifyEmailTemplate
+        .replace(/{{verifyEmailUrl}}/g, verifyEmailUrl)
+        .replace(/{{name}}/g, name);
+
+      await this.mailProvider.sendEmail({
+        to: email,
+        subject: 'Verify your email',
+        text: 'Verify your email',
+        html,
+      });
+
+      logger.info(`Verification email sent to ${email}`);
+    } catch (error) {
+      logger.error(`Failed to send verification email to ${email}: ${error}`);
+      throw error;
+    }
   }
 
   async sendForgetPasswordEmail(email: string, name: string, otp: string) {
-    const html = getOTPTemplate()
-      .replace(/{{otp}}/g, otp)
-      .replace(/{{name}}/g, name);
-
-    await this.mailProvider.sendEmail({
-      to: email,
-      subject: 'Reset your password',
-      text: 'Reset your password',
-      html,
-    });
+    try {
+      const html = getOTPTemplate()
+        .replace(/{{otp}}/g, otp)
+        .replace(/{{name}}/g, name);
+
+      await this.mailProvider.sendEmail({
+        to: email,
+        subject: 'Reset your password',
+        text: 'Reset your password',
+        html,
+      });
+
+      logger.info(`Forget password email sent to ${email}`);
+    } catch (error) {
+      logger.error(
+        `Failed to send forget password email to ${email}: ${error}`,
+      );
+      throw error;
+    }
   }
 
   async notifyWinnerViaEmail(email: string, name: string) {
-    const html = getWinnersTemplate().replace(/{{name}}/g, name);
-
-    await this.mailProvider.sendEmail({
-      to: email,
-      subject: 'Congratulations! You are a winner',
-      text: 'Congratulations! You are a winner',
-      html,
-    });
+    try {
+      const html = getWinnersTemplate().replace(/{{name}}/g, name);
+
+      await this.mailProvider.sendEmail({
+        to: email,
+        subject: 'Congratulations! You are a winner',
+        text: 'Congratulations! You are a winner',
+        html,
+      });
+
+      logger.info(`Winner notification email sent to ${email}`);
+    } catch (error) {
+      logger.error(
+        `Failed to send winner notification email to ${email}: ${error}`,
+      );
+      throw error;
+    }
   }
 }
 

src/services/hashing.service.ts

@@ -1,5 +1,6 @@
 import { compare, hash } from 'bcryptjs';
-import { bcryptSaltRounds } from '../config';
+import crypto from 'crypto';
+import { bcryptSaltRounds, hmacSecret } from '../config';
 import { ApiError, BAD_REQUEST } from '../utils';
 
 export class HashingService {
@@ -9,10 +10,15 @@ export class HashingService {
     }
     return await hash(text, Number(bcryptSaltRounds));
   }
+
   static compare(text: string, hashedText: string): Promise<boolean> {
     if (!text || !hashedText) {
       throw new ApiError('Missing information to compare', BAD_REQUEST);
     }
     return compare(text, hashedText);
   }
+
+  static generateHashWithHmac(text: string) {
+    return crypto.createHmac('sha256', hmacSecret).update(text).digest('hex');
+  }
 }

src/types/index.ts

@@ -1,4 +1,5 @@
 export * from './jwt';
+export * from './queue';
 export * from './status';
 export * from './statusCodes';
 export * from './user';

src/types/queue.ts

@@ -0,0 +1,17 @@
+import { WORKERS } from '../utils';
+
+export type SendEmailVerificationJob = {
+  email: string;
+  name: string;
+  token: string;
+};
+
+export type SendForgetPasswordJob = {
+  email: string;
+  name: string;
+  otp: string;
+};
+
+export type EmailJobData = SendEmailVerificationJob | SendForgetPasswordJob;
+
+export type EmailJobName = (typeof WORKERS)[keyof typeof WORKERS];