@agendajs/mongo-backend

MongoDB backend for Agenda job scheduler.

Installation

npm install agenda @agendajs/mongo-backend
# or
pnpm add agenda @agendajs/mongo-backend
# or
yarn add agenda @agendajs/mongo-backend

Requirements:

• Node.js 18+
• MongoDB 4+

Usage

import { Agenda } from 'agenda';
import { MongoBackend } from '@agendajs/mongo-backend';

// Via connection string
const agenda = new Agenda({
  backend: new MongoBackend({
    address: 'mongodb://localhost/agenda'
  })
});

// Or via existing MongoDB connection
import { MongoClient } from 'mongodb';

const client = await MongoClient.connect('mongodb://localhost');
const db = client.db('agenda');

const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db })
});

// Define jobs
agenda.define('send-email', async (job) => {
  const { to, subject } = job.attrs.data;
  await sendEmail(to, subject);
});

// Start processing
await agenda.start();

// Schedule jobs
await agenda.every('1 hour', 'send-email', { to: '[email protected]', subject: 'Hello' });

Configuration Options

interface IMongoBackendConfig {
  /** MongoDB connection string */
  address?: string;

  /** Existing MongoDB database instance */
  mongo?: Db;

  /** Collection name for jobs (default: 'agendaJobs') */
  collection?: string;

  /** MongoDB client options */
  options?: MongoClientOptions;

  /** Name to identify this Agenda instance (stored as lastModifiedBy) */
  name?: string;

  /** Whether to create indexes on connect (default: true) */
  ensureIndex?: boolean;

  /** Sort order for job queries */
  sort?: { [key: string]: SortDirection };
}

Full Example with Options

const agenda = new Agenda({
  backend: new MongoBackend({
    address: 'mongodb://localhost/agenda',
    collection: 'jobs',           // Custom collection name
    sort: { nextRunAt: 'asc', priority: 'desc' }
  }),
  processEvery: '30 seconds',     // Job polling interval
  maxConcurrency: 20,             // Max concurrent jobs
  defaultConcurrency: 5           // Default per job type
});

Real-Time Notifications

MongoBackend provides storage only and uses polling to check for new jobs. For real-time job processing across distributed systems, you have several options:

Option 1: MongoDB Change Streams (Native)

If your MongoDB deployment is a replica set (even a single-node replica set works), you can use MongoChangeStreamNotificationChannel for native real-time notifications without any external dependencies:

import { Agenda } from 'agenda';
import { MongoBackend, MongoChangeStreamNotificationChannel } from '@agendajs/mongo-backend';

const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db }),
  notificationChannel: new MongoChangeStreamNotificationChannel({ db })
});

// Jobs are processed immediately when created (no polling delay)
await agenda.start();
await agenda.now('myJob'); // Triggers instant processing

Requirements:

• MongoDB replica set (single-node replica sets work fine)
• WiredTiger storage engine (default since MongoDB 3.2)

Configuration options:

const channel = new MongoChangeStreamNotificationChannel({
  db: mongoDb,                    // Required: MongoDB database instance
  collection: 'agendaJobs',       // Optional: collection name (default: 'agendaJobs')
  resumeToken: savedToken,        // Optional: resume from specific point
  fullDocument: true              // Optional: include full document on updates (default: true)
});

Option 2: In-Memory (Single Process)

import { Agenda, InMemoryNotificationChannel } from 'agenda';
import { MongoBackend } from '@agendajs/mongo-backend';

// For single-process apps (testing/development)
const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db }),
  notificationChannel: new InMemoryNotificationChannel()
});

Option 3: Redis Pub/Sub (Multi-Process)

import { Agenda } from 'agenda';
import { MongoBackend } from '@agendajs/mongo-backend';
import { RedisBackend } from '@agendajs/redis-backend';

// For production: use Redis for notifications
const redisBackend = new RedisBackend({ connectionString: 'redis://localhost:6379' });
const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db }),
  notificationChannel: redisBackend.notificationChannel
});

Database Index

By default, MongoBackend automatically creates the findAndLockNextJobIndex index on connect for optimal job query performance. The index includes:

{
  "name": 1,
  "nextRunAt": 1,
  "priority": -1,
  "lockedAt": 1,
  "disabled": 1
}

To disable automatic index creation (e.g., if you manage indexes separately):

const backend = new MongoBackend({
  mongo: db,
  ensureIndex: false  // Skip automatic index creation
});

Additional Indexes

If you use Agendash, add this index for better dashboard performance:

db.agendaJobs.createIndex({
  "nextRunAt": -1,
  "lastRunAt": -1,
  "lastFinishedAt": -1
}, { name: "agendash" })

If you have job definitions with thousands of instances, this index can improve lock queries:

db.agendaJobs.createIndex({
  "name": 1,
  "disabled": 1,
  "lockedAt": 1
}, { name: "findAndLockDeadJobs" })

Job Logging

MongoBackend includes a built-in MongoJobLogger that stores structured job lifecycle events in a dedicated MongoDB collection (agenda_logs by default). The logger is lightweight and only creates its collection on first use.

Automatic Usage

When you enable logging in Agenda, the backend's built-in logger is used automatically:

const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db }),
  logging: true  // Uses MongoJobLogger automatically
});

Standalone Usage

You can also use MongoJobLogger independently — for example, to log to MongoDB while using a different backend for storage:

import { MongoJobLogger } from '@agendajs/mongo-backend';

const logger = new MongoJobLogger({
  mongo: db,                        // Required: MongoDB database instance
  logCollection: 'agenda_logs'      // Optional: collection name (default: 'agenda_logs')
});

// Use with any backend
import { RedisBackend } from '@agendajs/redis-backend';
const agenda = new Agenda({
  backend: new RedisBackend({ connectionString: 'redis://...' }),
  logging: logger
});

Querying Logs

const { entries, total } = await agenda.getLogs({
  jobName: 'send-email',
  level: 'error',
  limit: 50,
  sort: 'desc'
});

// Clear old logs
await agenda.clearLogs({ to: new Date(Date.now() - 30 * 86400000) });

See the Agenda README for full logging documentation.

Related Packages

• agenda - Core scheduler
• @agendajs/postgres-backend - PostgreSQL backend with LISTEN/NOTIFY
• @agendajs/redis-backend - Redis backend with Pub/Sub

License

MIT