How to schedule functions in Node.js/Express and MongoDB

If you have ever wanted to schedule functions in your Node.js application and couldn't find a straightforward way to accomplish this, then you will find this article very useful.

Source code: https://github.com/MiguelCedric/agenda-server-starter

I came across Agenda.js while working on a personal project (a rideshare app). I was looking to update rides status automatically, send reminders to passengers a few hours before the departure time, prompt them to leave a review after the ride, and much more.

I didn't want to use cron jobs, but instead, have everything in nodejs and be able to update (or delete) these jobs like updating a simple database document.

Scheduled jobs are save as documents in your "jobs" collection. You can edit the scheduled time of a job or delete it so that it doesn't run at all.

Every job document has nextRunAt property, that is the date and time the job function will execute. You can perform any MongoDB CRUD operations on the document.

Here I walk you through setting up the agenda-starter-server.

Files:

controllers/
    rideController.js
jobs/
    - agenda.js
    jobs_list/
        - archiveRideJob.js
routes/
    - routes.js
- server.js
- package.json
- .env

package.json

{
  "name": "agenda-server-starter",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "author": "",
  "license": "ISC",
  "dependencies": {
    "agenda": "^3.0.0",
    "dotenv": "^8.2.0",
    "express": "^4.17.1",
    "express-mongo-db": "^2.0.4",
    "mongodb": "^3.5.3"
  },
  "devDependencies": {
    "nodemon": "^2.0.2"
  }
}

Since this is not a tutorial about creating an express.js server, I am going to jump straight into how to set up agenda.js to schedule jobs/functions.

To start, you need to create an agenda configuration file that connects to your database and run your defined jobs.

jobs/agenda.js

let Agenda = require("agenda");

// set the collection where the jobs will be save
// the collection can be name anything
let url = process.env.DB_URL  // DB_URL="mongodb://127.0.0.1:27017/test-db"
let agenda = new Agenda({ db: {address: process.env.DB_URL, collection: 'jobs'}}); 
 
// list the different jobs availale throughout your app
let jobTypes = ["archiveRideJob"];

// loop through the job_list folder and pass in the agenda instance to
// each job so that it has access to its API.
jobTypes.forEach( type => {
    // the type name should match the file name in the jobs_list folder
    require('./jobs_list/'+ type)(agenda);
})

if(jobTypes.length) {
    // if there are jobs in the jobsTypes array set up 
    agenda.on('ready', async () => await agenda.start());
}

let graceful = () => {
    agenda.stop(() => process.exit(0));
}

process.on("SIGTERM", graceful);
process.on("SIGINT", graceful);

module.exports = agenda;

.env

DB_URL="mongodb://127.0.0.1:27017/test"
MONGO_URL="mongodb://127.0.0.1:27017"
DB_NAME="test"

Now, let's define our jobs.

Although you can define your jobs inside only one file, splitting each job in its file makes it easy to scale, or make more sense to me.

In agenda, you define jobs, and then you can schedule or process them how you want throughout your app.

Let's look at the archiveRideJob job file.

jobs/jobs/archiveRideJob.js

let MongoClient = require("mongodb").MongoClient;

module.exports = (agenda) => {
    agenda.define("archive ride", (job, done) => {
        console.log("Archiving ride ...")

        // to be able to that you need to grab the ride id which is on the job document in the jobs collection in the database.
        // first let's connect to mongo db server
        //MONGO_URL="mongodb://127.0.0.1:27017"
        let client = new MongoClient(process.env.MONGO_URL) 
        
        client.connect((err) => {
            if(!err) {
                console.log('connection to db successful');
                
                // connect to the database
                // DB_NAME="test"
                let db = client.db(process.env.DB_NAME);
                
                // get the data passed when scheduling the job in the controller
                let rideId = job.attrs.data.rideId; 
                
                db.collection('rides').findOneAndUpdate({_id: rideId}, { $set: { status: 'expired' }}, (error) => {
                    if(!error) {
                        console.log('Successfully archived ride. rideId: ', rideId);
                        client.close();
                        done();
                    } else {
                        console.log("Error archiving ride ID: ", rideId);
                        console.log(error);
                        client.close();
                        done();
                    }
                })
            } else {
                console.log(err);
                client.close();
                done();
            }
        })
    })
}

Here, I use the native Node.js Driver for MongoDB (not Mongoose the ORM). If you are using Mongoose, then you need to import your models in this file and do your CRUD operations.

So first, we define our task 'archive ride'.

agenda.define('archive ride', function(job, done) {
    // your code goes here
    })

Next, I create a connection to my database.

let client = new MongoClient(process.env.MONGO_URL)
...
...
let db = client.db(process.env.DB_NAME);

Again, if you are using an ORM (object relational mapper) like Mongoose, then you don't need to create this connection.

db.collection('rides').findOneAndUpdate({_id: rideId},
                    { $set: { status: 'expired' }}, function (error) {
                    if(!error) {
                    ....
                    ....
                        done();
                    }
                    ...
                    ...
                })

That's it for defining the job.

Now we need to schedule our job when a user creates a ride. We do this in our controller.

controllers/rideController.js

let agenda = require('../jobs/agenda');

module.exports = {
    postRide: (req, res) => {
            // you ride data when someone posts a ride
            let myRide = {
                departureDate: new Date(Date.now()), // aslo the archive date
                rideData: "Some data"
            }
            
            // get the database object from the request object see server.js
            let db = req.db;

            // create a new ride document the ride data
            db.collection('rides').insertOne({...myRide}, (ride, err) => {
                if(!err) {
                    // if the ride is successfully saved in the database, schedule a job with a unique id in the job collection
                    agenda.schedule(
                        myRide.departureDate, // date the function will execute
                        "archive ride", 
                        { rideId: ride.insertedId } // add additional information to be accessed by the job function: job.attrs.data.rideId;
                    );
                    res.send("Hello World!")
                } else {
                    console.log(err)
                    res.send("Hello Error!")
                }
            })
    }
}

There are many ways to schedule and execute jobs with agenda.js. Here I use agenda.schedule(); to schedule a job to be executed in a future date I specified. Alternatively, you can use agenda.now() to execute the job immediately. Go to the documentation for [other methods].(https://github.com/agenda/agenda#creating-jobs).

If you are curious about the database connection middleware, I added this npm package express-mongo-db. (Very Optional)

Here is how the routes and server are set up:

routes/routes.js

const router = require("express").Router();

// import the controllers/functions to run when requesting a specific end point
let rideController = require('../controllers/rideController');


// run the function postRide in rideController when requestion "/" endpoint
router.get('/', rideController.postRide);


module.exports = router; // to be imported in server.js and used with route middleware

server.js

let express = require('express')
require('dotenv').config(); // make environment variables available throughout the app
let expressMongoDB = require('express-mongo-db') // attach the mongodb instance to the request object

// importing routes
let indexRoutes = require('./routes/routes');

let app = express();
let port = 3000

// connects to the database and attach a db object to the request object
app.use(expressMongoDB(process.env.DB_URL)) // e.g: DB_URL="mongodb://127.0.0.1:27017/test"

app.use('/', indexRoutes)

app.listen(port, () => {
    console.log(`Server is listenung on port ${port}...`)
})

That's it. If you have questions or suggestions, leave a comment below.