Implemented HTTP(s) Request, added add and delete to and from job queue

Server/controllers/authController.js

@@ -24,6 +24,7 @@ const {
  * @returns {String}
  */
 const issueToken = (payload) => {
+  //TODO Add proper expiration date
   return jwt.sign(payload, process.env.JWT_SECRET, { expiresIn: "99d" });
 };
 

Server/controllers/monitorController.js

@@ -116,6 +116,8 @@ const createMonitor = async (req, res, next) => {
 
   try {
     const monitor = await req.db.createMonitor(req, res);
+    // Add monitor to job queue
+    req.jobQueue.addJob(monitor._id, monitor);
     return res
       .status(201)
       .json({ success: true, msg: "Monitor created", data: monitor });
@@ -147,6 +149,8 @@ const deleteMonitor = async (req, res, next) => {
 
   try {
     const monitor = await req.db.deleteMonitor(req, res, next);
+    req.jobQueue.deleteJob(monitor);
+
     /**
      * TODO
      * We should remove all checks and alerts associated with this monitor

Server/index.js

@@ -38,7 +38,6 @@ const startApp = async () => {
     ? DB_TYPE[process.env.DB_TYPE]()
     : require("./db/FakeDb");
 
-  const jobQueue = await JobQueue.createJobQueue();
   /**
    * NOTES
    * Email Service will be added
@@ -69,7 +68,7 @@ const startApp = async () => {
 
   //routes
   app.use("/api/v1/auth", authRouter);
-  app.use("/api/v1/monitors", monitorRouter);
+  app.use("/api/v1/monitors", verifyJWT, monitorRouter);
   app.use("/api/v1/checks", verifyJWT, checkRouter);
   app.use("/api/v1/alerts", verifyJWT, alertRouter);
   //Temporary route for testing, remove later
@@ -92,7 +91,8 @@ const startApp = async () => {
    */
   app.use(handleErrors);
 
-  connectDbAndRunServer(app, db);
+  await connectDbAndRunServer(app, db);
+  const jobQueue = await JobQueue.createJobQueue(db);
 };
 
 startApp().catch((error) => {

Server/package-lock.json

@@ -10,6 +10,7 @@
       "license": "ISC",
       "dependencies": {
         "@sendgrid/mail": "^8.1.3",
+        "axios": "^1.7.2",
         "bcrypt": "^5.1.1",
         "bullmq": "5.7.15",
         "cors": "^2.8.5",
@@ -373,7 +374,6 @@
       "version": "1.7.2",
       "resolved": "https://registry.npmjs.org/axios/-/axios-1.7.2.tgz",
       "integrity": "sha512-2A8QhOMrbomlDuiLeK9XibIBzuHeRcqqNOHp0Cyp5EoJ1IFDh+XZH3A6BkXtv0K4gFGCI0Y4BM7B1wOEi0Rmgw==",
-      "license": "MIT",
       "dependencies": {
         "follow-redirects": "^1.15.6",
         "form-data": "^4.0.0",

Server/package.json

@@ -12,6 +12,7 @@
   "license": "ISC",
   "dependencies": {
     "@sendgrid/mail": "^8.1.3",
+    "axios": "^1.7.2",
     "bcrypt": "^5.1.1",
     "bullmq": "5.7.15",
     "cors": "^2.8.5",

Server/service/jobQueue.js

@@ -6,6 +6,8 @@ const connection = {
 };
 const JOBS_PER_WORKER = 5;
 const logger = require("../utils/logger");
+const SERVICE_NAME = "JobQueue";
+const axios = require("axios");
 
 class JobQueue {
   /**
@@ -18,6 +20,7 @@ class JobQueue {
       connection,
     });
     this.workers = [];
+    this.db = null;
   }
 
   /**
@@ -27,9 +30,11 @@ class JobQueue {
    * @returns {Promise<JobQueue>} - Returns a new JobQueue
    *
    */
-  static async createJobQueue() {
+  static async createJobQueue(db) {
     const queue = new JobQueue();
     try {
+      queue.db = db;
+      const monitors = await db.getAllMonitors();
       const workerStats = await queue.getWorkerStats();
       await queue.scaleWorkers(workerStats);
       return queue;
@@ -47,8 +52,9 @@ class JobQueue {
     const worker = new Worker(
       QUEUE_NAME,
       async (job) => {
-        // TODO Ping a monitor
-        console.log(`${job.name} completed, workers: ${this.workers.length}`);
+        const response = await axios.get(job.data.url);
+        // TODO create a check object and save it to the db
+        console.log(response.status);
       },
       {
         connection,
@@ -147,7 +153,6 @@ class JobQueue {
   async getJobs() {
     try {
       const jobs = await this.queue.getRepeatableJobs();
-      console.log("jobs", jobs);
       return jobs;
     } catch (error) {
       throw error;
@@ -166,8 +171,7 @@ class JobQueue {
     try {
       await this.queue.add(jobName, payload, {
         repeat: {
-          every: 1000,
-          limit: 100,
+          every: payload.interval,
         },
       });
       const workerStats = await this.getWorkerStats();
@@ -177,6 +181,34 @@ class JobQueue {
     }
   }
 
+  /**
+   * Deletes a job from the queue.
+   *
+   * @async
+   * @param {Monitor} monitor - The monitor to remove.
+   * @throws {Error}
+   */
+  async deleteJob(monitor) {
+    try {
+      const deleted = await this.queue.removeRepeatable(monitor.id, {
+        every: monitor.interval,
+      });
+      if (deleted) {
+        logger.info("Job removed from queue", {
+          service: SERVICE_NAME,
+          jobId: monitor.id,
+        });
+      } else {
+        logger.error("Job not found in queue", {
+          service: SERVICE_NAME,
+          jobId: monitor.id,
+        });
+      }
+    } catch (error) {
+      throw error;
+    }
+  }
+
   /**
    * @async
    * @returns {Promise<boolean>} - Returns true if obliteration is successful