fix: Fix duplicate handler attachment for new cron jobs

devadathanmb · Feb 23, 2024 · efed838 · efed838
1 parent a379426
commit efed838
Show file tree

Hide file tree

Showing 5 changed files with 116 additions and 87 deletions.
diff --git a/src/app.ts b/src/app.ts
@@ -2,9 +2,11 @@ import createBot from "./createBot";
 import notifyUserCron from "./cron/notifyUserCron";
 import Bull = require("bull");
 const notifyUserQueue = new Bull("notify-user-queue");
+import createJobQueue from "./cron/queue";
 
 // Create the bot and initialize the database
 const { bot, db } = createBot();
+const queue = createJobQueue(bot);
 
 // Launch the bot
 const launchBot = async () => {
@@ -16,7 +18,7 @@ const launchBot = async () => {
         console.log(
           `Bot started in polling mode. Available at https://t.me/${res.username}`
         );
-        notifyUserCron(db, bot);
+        notifyUserCron(db, queue);
       });
   }
   // Launch in webhook mode if in production
@@ -32,7 +34,7 @@ const launchBot = async () => {
         console.log(
           `Bot started in webhook mode. Available at https://t.me/${res.username}`
         );
-        notifyUserCron(db, bot);
+        notifyUserCron(db, queue);
       });
   }
 };

diff --git a/src/cron/notifyUserCron.ts b/src/cron/notifyUserCron.ts
@@ -1,22 +1,14 @@
 import * as cron from "node-cron";
 import { Firestore } from "firebase-admin/firestore";
-import { Telegraf, TelegramError } from "telegraf";
-import { CustomContext } from "../types/customContext.type";
 import fetchAnnouncements from "../services/fetchAnnouncements";
 import { readFile, writeFile } from "fs";
 import fetchAttachment from "../services/fetchAttachment";
 import { Announcement, Attachment } from "../types/types";
 import findFilters from "../utils/findFilters";
+import getCaptionMsg from "../utils/getCaptionMsg";
 import Bull = require("bull");
 
-// Create queue
-const queue = new Bull("notify-user-queue", {
-  redis: {
-    host: "redis-queue-db",
-  },
-});
-
-async function notifyUserCron(db: Firestore, bot: Telegraf<CustomContext>) {
+async function notifyUserCron(db: Firestore, queue: Bull.Queue) {
   console.log("Cron job initialized");
   cron.schedule("*/10 * * * *", async () => {
     const startTime = new Date().toString();
@@ -89,14 +81,8 @@ async function notifyUserCron(db: Firestore, bot: Telegraf<CustomContext>) {
                 continue;
               }
 
-              const captionMsg = `
-
-<b>Subject:</b> ${announcement.subject ? announcement.subject : "N/A"}
-
-<b>Date:</b> ${announcement.date ? announcement.date : "N/A"}
-
-<b>Message:</b> ${announcement.message ? announcement.message : "N/A"}
-`;
+              // Get the caption message
+              const captionMsg = getCaptionMsg(announcement);
 
               // Get the data to fetch the attachments
               const attachments = announcement.attachments.map(
@@ -106,84 +92,40 @@ async function notifyUserCron(db: Firestore, bot: Telegraf<CustomContext>) {
                 })
               );
 
-              // Add all the chatIds to the queue
-              for (let i = 0; i < chatIds.length; i++) {
-                await queue.add({
-                  chatId: chatIds[i],
-                });
-              }
-
-              // Consumer
+              // Add stuff to the queue
+              // Pass fileBuffer as null since there are no attachments
               if (attachments.length === 0) {
-                queue.process(async (job) => {
-                  const { chatId } = job.data;
+                for (let i = 0; i < chatIds.length; i++) {
                   try {
-                    await bot.telegram.sendMessage(chatId, captionMsg, {
-                      parse_mode: "HTML",
+                    await queue.add({
+                      chatId: chatIds[i],
+                      fileBuffer: null,
+                      captionMsg: captionMsg,
+                      fileName: null,
                     });
-                  } catch (error: any) {
-                    if (error instanceof TelegramError) {
-                      if (error.code === 429) {
-                        const retryAfter = error.parameters?.retry_after!;
-                        await new Promise((resolve) =>
-                          setTimeout(resolve, retryAfter * 1000 + 2000)
-                        );
-                        await job.retry();
-                      } else if (error.code === 403) {
-                        try {
-                          await usersRef.doc(chatId.toString()).delete();
-                          await job.remove();
-                        } catch (error) {
-                          console.log(error);
-                        }
-                      }
-                    }
+                  } catch (error) {
+                    console.log(error);
                   }
-                });
+                }
               } else {
-                // fetch attachments
-                // send each attachment to each chatId
+                // Loop through each attachment and add to the queue
                 for (let i = 0; i < attachments.length; i++) {
                   const file = await fetchAttachment(attachments[i].encryptId);
-                  const fileBuffer = Buffer.from(file, "base64");
-                  queue.process(async (job) => {
-                    const chatId = job.data.chatId;
+
+                  for (let j = 0; j < chatIds.length; j++) {
                     try {
-                      await bot.telegram.sendDocument(
-                        chatId,
-                        {
-                          source: fileBuffer,
-                          filename: attachments[i].name,
-                        },
-                        { caption: captionMsg, parse_mode: "HTML" }
-                      );
-                    } catch (error: any) {
-                      if (error instanceof TelegramError) {
-                        if (error.code === 429) {
-                          const retryAfter = error.parameters?.retry_after!;
-                          await new Promise((resolve) =>
-                            setTimeout(resolve, retryAfter * 1000 + 2000)
-                          );
-                          await job.retry();
-                        } else if (error.code === 403) {
-                          try {
-                            await usersRef.doc(chatId.toString()).delete();
-                            await job.remove();
-                          } catch (error) {
-                            console.log(error);
-                          }
-                        }
-                      }
+                      await queue.add({
+                        chatId: chatIds[i],
+                        file: file,
+                        captionMsg: captionMsg,
+                        fileName: attachments[i].name,
+                      });
+                    } catch (error) {
+                      console.log(error);
                     }
-                  });
+                  }
                 }
               }
-
-              // Job completed event
-              queue.on("completed", async (job) => {
-                console.log(`✅ Message sent to ${job.data.chatId}`);
-                await job.remove();
-              });
             }
           }
         } catch (error) {

diff --git a/src/cron/queue.ts b/src/cron/queue.ts
@@ -0,0 +1,60 @@
+import { Telegraf, TelegramError } from "telegraf";
+import { CustomContext } from "../types/customContext.type";
+import Bull = require("bull");
+import { JobData } from "../types/types";
+
+function createJobQueue(bot: Telegraf<CustomContext>) {
+  // Create queue
+  const queue = new Bull<JobData>("notify-user-queue");
+
+  // Job completed event
+  queue.on("completed", async (job) => {
+    console.log(`✅ Message sent to ${job.data.chatId}`);
+    await job.remove();
+  });
+
+  // Consumer
+  queue.process(async (job) => {
+    const { chatId, file, captionMsg, fileName } = job.data;
+
+    try {
+      if (!file || !fileName) {
+        await bot.telegram.sendMessage(chatId, captionMsg, {
+          parse_mode: "HTML",
+        });
+      } else {
+        const fileBuffer = Buffer.from(file as string, "base64");
+        await bot.telegram.sendDocument(
+          chatId,
+          {
+            source: fileBuffer,
+            filename: fileName,
+          },
+          { caption: captionMsg, parse_mode: "HTML" }
+        );
+      }
+    } catch (error: any) {
+      console.log(error);
+      if (error instanceof TelegramError) {
+        if (error.code === 429) {
+          const retryAfter = error.parameters?.retry_after!;
+          await new Promise((resolve) =>
+            setTimeout(resolve, retryAfter * 1000 + 2000)
+          );
+          await job.retry();
+        } else if (error.code === 403) {
+          try {
+            /* await usersRef.doc(chatId.toString()).delete(); */
+            await job.remove();
+          } catch (error) {
+            console.log(error);
+          }
+        }
+      }
+    }
+  });
+
+  return queue;
+}
+
+export default createJobQueue;
diff --git a/src/types/types.ts b/src/types/types.ts
@@ -63,6 +63,13 @@ interface Command {
   description: string;
 }
 
+interface JobData {
+  chatId: number;
+  file: string | null;
+  captionMsg: string;
+  fileName: string | null;
+}
+
 export {
   ResultDetails,
   ResultSummary,
@@ -73,4 +80,5 @@ export {
   AcademicCalendar,
   Timetable,
   Command,
+  JobData,
 };
diff --git a/src/utils/getCaptionMsg.ts b/src/utils/getCaptionMsg.ts
@@ -0,0 +1,17 @@
+// Utility function to get the caption message from the message object
+import { Announcement } from "../types/types";
+
+function getCaptionMsg(announcement: Announcement) {
+  const captionMsg = `
+
+<b>Subject:</b> ${announcement.subject ? announcement.subject : "N/A"}
+
+<b>Date:</b> ${announcement.date ? announcement.date : "N/A"}
+
+<b>Message:</b> ${announcement.message ? announcement.message : "N/A"}
+`;
+
+  return captionMsg;
+}
+
+export default getCaptionMsg;