Custom Backend Driver

Agenda v6 introduces a pluggable backend system (AgendaBackend), allowing you to use any database and optionally provide real-time notifications.

Official Backend Packages

Before implementing a custom backend, check if an official package already exists:

Package	Backend	Notifications	Install
@agendajs/mongo-backend	MongoDB	Change Streams (optional)	npm install @agendajs/mongo-backend
@agendajs/postgres-backend	PostgreSQL	LISTEN/NOTIFY	npm install @agendajs/postgres-backend
@agendajs/redis-backend	Redis	Pub/Sub	npm install @agendajs/redis-backend

MongoDB:

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

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

PostgreSQL:

import { Agenda } from 'agenda';
import { PostgresBackend } from '@agendajs/postgres-backend';

const agenda = new Agenda({
  backend: new PostgresBackend({
    connectionString: 'postgresql://user:pass@localhost:5432/mydb'
  })
});

Redis:

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

const agenda = new Agenda({
  backend: new RedisBackend({
    connectionString: 'redis://localhost:6379'
  })
});

If you need a different database (SQLite, MySQL, etc.), continue reading to learn how to implement a custom backend.

Architecture

A backend provides:

• Storage (required): Via JobRepository interface
• Notifications (optional): Via NotificationChannel interface

Backend Capabilities

Backend	Storage	Notifications	Notes
MongoDB (MongoBackend)	✅	❌	Storage only by default.
MongoDB (MongoChangeStreamNotificationChannel)	❌	✅	Native Change Streams notifications. Requires replica set.
PostgreSQL (PostgresBackend)	✅	✅	Full backend. Uses LISTEN/NOTIFY for notifications.
Redis (RedisBackend)	✅	✅	Full backend. Uses Pub/Sub for notifications.
InMemoryNotificationChannel	❌	✅	Notifications only. For single-process/testing.

Common Configurations

1. MongoDB only (default): Storage with polling-based job processing
2. MongoDB + Change Streams: MongoDB for storage AND real-time notifications (requires replica set)
3. MongoDB + Redis notifications: MongoDB for storage, Redis Pub/Sub for real-time notifications
4. MongoDB + PostgreSQL notifications: MongoDB for storage, PostgreSQL LISTEN/NOTIFY for notifications
5. PostgreSQL unified: Single backend providing both storage AND notifications
6. Redis unified: Single backend providing both storage AND notifications

Mixing Storage and Notification Backends

You can use MongoDB for storage while using a different system for real-time notifications:

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

// MongoDB for storage + Redis for real-time notifications
const redisBackend = new RedisBackend({ connectionString: 'redis://localhost:6379' });
const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db }),
  notificationChannel: redisBackend.notificationChannel
});

This is useful when you want MongoDB's proven durability and flexible queries for job storage, but need faster real-time notifications across multiple processes.

MongoDB Backend

For MongoDB, use the @agendajs/mongo-backend package:

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

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

// Via existing MongoDB connection
import { MongoClient } from 'mongodb';
const client = await MongoClient.connect('mongodb://localhost');
const agenda = new Agenda({
  backend: new MongoBackend({ mongo: client.db('agenda') })
});

MongoDB with Change Streams (Real-Time)

If your MongoDB deployment is a replica set, you can use MongoChangeStreamNotificationChannel for native real-time notifications:

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

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

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

// Jobs are now processed immediately when created
await agenda.start();
await agenda.now('myJob'); // Triggers instant processing via change stream

How it works:

• Uses MongoDB Change Streams to watch the jobs collection for changes
• Automatically detects job inserts/updates and notifies the processor
• The publish() method is a no-op since changes are detected automatically
• Supports resume tokens for recovery after disconnections

Requirements:

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

Custom Backend

To use a different database, implement the AgendaBackend interface:

import { Agenda, AgendaBackend, JobRepository, NotificationChannel } from 'agenda';

class PostgresBackend implements AgendaBackend {
  readonly repository: JobRepository;
  readonly notificationChannel?: NotificationChannel;

  constructor(config: { connectionString: string }) {
    this.repository = new PostgresRepository(config);
    // Optional: provide LISTEN/NOTIFY based notifications
    this.notificationChannel = new PostgresNotificationChannel(config);
  }

  async connect(): Promise<void> {
    // Connect to PostgreSQL
  }

  async disconnect(): Promise<void> {
    // Disconnect from PostgreSQL
  }
}

const agenda = new Agenda({
  backend: new PostgresBackend({ connectionString: 'postgres://...' })
});

AgendaBackend Interface

interface AgendaBackend {
  /**
   * The job repository for storage operations.
   * Required - every backend must provide storage.
   */
  readonly repository: JobRepository;

  /**
   * Optional notification channel for real-time job notifications.
   * If provided, Agenda will use this for immediate job processing.
   * If not provided, Agenda falls back to periodic polling.
   */
  readonly notificationChannel?: NotificationChannel;

  /**
   * Connect to the backend.
   * Called when Agenda is initialized.
   */
  connect(): Promise<void>;

  /**
   * Disconnect from the backend.
   * Called when agenda.stop() is invoked.
   */
  disconnect(): Promise<void>;
}

JobRepository Interface

The JobRepository interface defines all database operations required by Agenda:

interface JobRepository {
  /**
   * Connect to the database
   */
  connect(): Promise<void>;

  /**
   * Query jobs with filtering, pagination, and state computation
   */
  queryJobs(options?: JobsQueryOptions): Promise<JobsResult>;

  /**
   * Get overview statistics for all job types
   */
  getJobsOverview(): Promise<JobsOverview[]>;

  /**
   * Get all distinct job names
   */
  getDistinctJobNames(): Promise<string[]>;

  /**
   * Get a single job by ID
   */
  getJobById(id: string): Promise<JobParameters | null>;

  /**
   * Get count of jobs ready to run (nextRunAt < now)
   */
  getQueueSize(): Promise<number>;

  /**
   * Remove jobs matching the given options
   * @returns Number of jobs removed
   */
  removeJobs(options: RemoveJobsOptions): Promise<number>;

  /**
   * Save a job (insert or update)
   */
  saveJob<DATA = unknown>(job: JobParameters<DATA>): Promise<JobParameters<DATA>>;

  /**
   * Update job state fields (lockedAt, lastRunAt, progress, etc.)
   */
  saveJobState(job: JobParameters): Promise<void>;

  /**
   * Attempt to lock a job for processing
   * @returns The locked job data, or undefined if lock failed
   */
  lockJob(job: JobParameters): Promise<JobParameters | undefined>;

  /**
   * Unlock a single job
   */
  unlockJob(job: JobParameters): Promise<void>;

  /**
   * Unlock multiple jobs by ID
   */
  unlockJobs(jobIds: (JobId | string)[]): Promise<void>;

  /**
   * Find and lock the next job to run for a given job type
   */
  getNextJobToRun(
    jobName: string,
    nextScanAt: Date,
    lockDeadline: Date,
    now?: Date
  ): Promise<JobParameters | undefined>;
}

NotificationChannel Interface

For real-time job notifications:

interface NotificationChannel {
  readonly state: NotificationChannelState;
  connect(): Promise<void>;
  disconnect(): Promise<void>;
  subscribe(handler: NotificationHandler): () => void;
  publish(notification: JobNotification): Promise<void>;
  on(event: 'stateChange' | 'error', listener: Function): this;
  off(event: 'stateChange' | 'error', listener: Function): this;
}

type NotificationChannelState = 'disconnected' | 'connecting' | 'connected' | 'reconnecting' | 'error';

interface JobNotification {
  jobId: JobId;
  jobName: string;
  nextRunAt: Date | null;
  priority: number;
  timestamp: Date;
  source?: string;
}

Supporting Types

JobId

A branded string type for job IDs:

type JobId = string & { readonly __brand: 'AgendaJobId' };

// Convert a string to JobId
import { toJobId } from 'agenda';
const id = toJobId('my-job-id');

RemoveJobsOptions

Options for removing jobs:

interface RemoveJobsOptions {
  /** Remove job by ID */
  id?: JobId | string;
  /** Remove jobs by IDs */
  ids?: (JobId | string)[];
  /** Remove jobs by name */
  name?: string;
  /** Remove jobs by names (include) */
  names?: string[];
  /** Remove jobs NOT matching these names (exclude) */
  notNames?: string[];
  /** Remove jobs matching data */
  data?: unknown;
}

JobsQueryOptions

Options for querying jobs:

interface JobsQueryOptions {
  /** Filter by job name */
  name?: string;
  /** Filter by job names (multiple) */
  names?: string[];
  /** Filter by computed state */
  state?: JobState;
  /** Filter by job ID */
  id?: string;
  /** Filter by job IDs (multiple) */
  ids?: string[];
  /** Text to search in job name */
  search?: string;
  /** Filter by job data */
  data?: unknown;
  /** Include disabled jobs (default: true) */
  includeDisabled?: boolean;
  /** Sort order */
  sort?: JobsSort;
  /** Number of jobs to skip (pagination) */
  skip?: number;
  /** Maximum number of jobs to return */
  limit?: number;
}

JobState

Computed job states:

type JobState = 'running' | 'scheduled' | 'queued' | 'completed' | 'failed' | 'repeating';

Implementation Notes

When implementing a custom backend:

1. Job IDs: Use JobId (a branded string) for all job identifiers. Convert from your database's native ID format at the repository boundary.

2. State Computation: Job states are computed from job fields, not stored. Use the computeJobState() helper:

   import { computeJobState } from 'agenda';
   const state = computeJobState(job);
   

3. Locking: Implement atomic lock operations to prevent duplicate job execution in distributed environments.

4. Upserts: The saveJob() method must handle both inserts (no _id) and updates (has _id), as well as unique constraints and type: 'single' jobs.

5. Notifications: If your database supports real-time notifications (like PostgreSQL LISTEN/NOTIFY), implement NotificationChannel and provide it via notificationChannel.

Example: SQLite Backend

Here's an example structure for implementing a SQLite backend:

import {
  AgendaBackend,
  JobRepository,
  BaseNotificationChannel,
  JobNotification
} from 'agenda';

class SQLiteRepository implements JobRepository {
  // Implement all repository methods using better-sqlite3 or similar...
}

// Optional: For multi-process setups, implement a notification channel
class SQLiteNotificationChannel extends BaseNotificationChannel {
  // Could use file-based notifications, IPC, or external pub/sub
  async connect(): Promise<void> {
    this.setState('connected');
  }

  async disconnect(): Promise<void> {
    this.setState('disconnected');
  }

  async publish(notification: JobNotification): Promise<void> {
    // Publish notification to other processes
  }
}

class SQLiteBackend implements AgendaBackend {
  readonly repository: JobRepository;
  readonly notificationChannel?: NotificationChannel;

  constructor(config: { path: string }) {
    this.repository = new SQLiteRepository(config);
    // Notification channel is optional for single-process apps
  }

  async connect(): Promise<void> {
    await this.repository.connect();
  }

  async disconnect(): Promise<void> {
    // Close database connection
  }
}

// Usage
const agenda = new Agenda({
  backend: new SQLiteBackend({ path: './jobs.db' })
});

Reference Implementations

For complete implementation examples, see the source code of the official backend packages:

• MongoDB: @agendajs/mongo-backend
• PostgreSQL: @agendajs/postgres-backend
• Redis: @agendajs/redis-backend