Restored mail job (#306)

Co-authored-by: Rajat Saxena <[email protected]>

apps/queue/.env

@@ -3,4 +3,6 @@ EMAIL_USER=email_user
 EMAIL_PASS=email_pass
 EMAIL_HOST=email_host
 DB_CONNECTION_STRING=mongodb://db.string
-SEQUENCE_BOUNCE_LIMIT=3
+REDIS_HOST=localhost
+REDIS_PORT=6379
+SEQUENCE_BOUNCE_LIMIT=3

apps/queue/package.json

@@ -13,6 +13,8 @@
     },
     "dependencies": {
         "@courselit/common-models": "workspace:^",
+        "bullmq": "^4.14.0",
+        "express": "^4.18.2",
         "mongoose": "^8.0.0",
         "nodemailer": "^6.9.2",
         "pino": "^8.14.1",

apps/queue/src/domain/mail/handler.ts

@@ -0,0 +1,13 @@
+import type { MailJob } from "./model/mail-job";
+import mailQueue from "./queue";
+
+export async function addMailJob({ to, subject, body, from }: MailJob) {
+    for (let recipient of to) {
+        await mailQueue.add("mail", {
+            to: recipient,
+            subject,
+            body,
+            from,
+        });
+    }
+}

apps/queue/src/domain/mail/model/mail-job.ts

@@ -0,0 +1,10 @@
+import { z } from "zod";
+
+export const MailJob = z.object({
+    to: z.string().array(),
+    from: z.string(),
+    subject: z.string(),
+    body: z.string(),
+});
+
+export type MailJob = z.infer<typeof MailJob>;

apps/queue/src/domain/mail/process-ongoing-sequences.ts

@@ -17,6 +17,10 @@ const transporter = nodemailer.createTransport({
     },
 });
 
+const sequenceBounceLimit = process.env.SEQUENCE_BOUNCE_LIMIT
+    ? +process.env.SEQUENCE_BOUNCE_LIMIT
+    : 3;
+
 export async function processOngoingSequences(): Promise<void> {
     while (true) {
         const currentTime = new Date().getTime();
@@ -25,7 +29,7 @@ export async function processOngoingSequences(): Promise<void> {
         const dueOngoingSequences: OngoingSequence[] =
             await OngoingSequenceModel.find({
                 nextEmailScheduledTime: { $lt: currentTime },
-                retryCount: { $lt: +process.env.SEQUENCE_BOUNCE_LIMIT },
+                retryCount: { $lt: sequenceBounceLimit },
             });
 
         for (const ongoingSequence of dueOngoingSequences) {
@@ -134,7 +138,7 @@ async function attemptMailSending({
         });
     } catch (err: any) {
         ongoingSequence.retryCount++;
-        if (ongoingSequence.retryCount >= +process.env.SEQUENCE_BOUNCE_LIMIT) {
+        if (ongoingSequence.retryCount >= sequenceBounceLimit) {
             sequence.report.sequence.failed = [
                 ...sequence.report.sequence.failed,
                 ongoingSequence.userId,

apps/queue/src/domain/mail/queue.ts

@@ -0,0 +1,6 @@
+import { Queue } from "bullmq";
+import redis from "../../redis";
+
+const mailQueue = new Queue("mail", { connection: redis });
+
+export default mailQueue;

apps/queue/src/domain/mail/worker.ts

@@ -0,0 +1,34 @@
+import { Worker } from "bullmq";
+import nodemailer from "nodemailer";
+import redis from "../../redis";
+import { logger } from "../../logger";
+
+const transporter = nodemailer.createTransport({
+    host: process.env.EMAIL_HOST,
+    port: +(process.env.EMAIL_PORT || 587),
+    auth: {
+        user: process.env.EMAIL_USER,
+        pass: process.env.EMAIL_PASS,
+    },
+});
+
+const worker = new Worker(
+    "mail",
+    async (job) => {
+        const { to, from, subject, body } = job.data;
+
+        try {
+            await transporter.sendMail({
+                from,
+                to,
+                subject,
+                html: body,
+            });
+        } catch (err: any) {
+            logger.error(err);
+        }
+    },
+    { connection: redis },
+);
+
+export default worker;

apps/queue/src/index.ts

@@ -1,10 +1,18 @@
-import { connectToDatabase } from "./db";
-import { processOngoingSequences } from "./domain/mail/process-ongoing-sequences";
-import { processRules } from "./domain/mail/process-rules";
+import express from "express";
+import jobRoutes from "./job/routes";
 
-(async () => {
-    await connectToDatabase();
+// start workers
+import "./domain/mail/worker";
+import { startEmailAutomation } from "./start-email-automation";
 
-    processOngoingSequences();
-    processRules();
-})();
+const app = express();
+app.use(express.json());
+
+app.use("/job", jobRoutes);
+
+startEmailAutomation();
+
+const port = process.env.PORT || 80;
+app.listen(port, () => {
+    console.log(`Queue server running at ${port}`);
+});

apps/queue/src/job/routes.ts

@@ -0,0 +1,22 @@
+import express from "express";
+import { addMailJob } from "../domain/mail/handler";
+import { logger } from "../logger";
+import { MailJob } from "../domain/mail/model/mail-job";
+
+const router = express.Router();
+
+router.post("/mail", async (req: express.Request, res: express.Response) => {
+    try {
+        const { to, from, subject, body } = req.body;
+        MailJob.parse({ to, from, subject, body });
+
+        await addMailJob({ to, from, subject, body });
+
+        res.status(200).json({ message: "Success" });
+    } catch (err: any) {
+        logger.error(err);
+        res.status(500).json({ error: err.message });
+    }
+});
+
+export default router;

apps/queue/src/redis.ts

@@ -0,0 +1,4 @@
+export default {
+    host: process.env.REDIS_HOST || "localhost",
+    port: +(process.env.REDIS_PORT || 6379),
+};

apps/queue/src/start-email-automation.ts

@@ -0,0 +1,10 @@
+import { connectToDatabase } from "./db";
+import { processOngoingSequences } from "./domain/mail/process-ongoing-sequences";
+import { processRules } from "./domain/mail/process-rules";
+
+export async function startEmailAutomation() {
+    await connectToDatabase();
+
+    processOngoingSequences();
+    processRules();
+}