A MongoDB job queue for TypeScript. Register workers, enqueue jobs, and schedule recurring work.

Features

• Atomic claims let multiple schedulers process jobs from one collection.
• Failed jobs retry with exponential backoff. You control the failure limit and delays.
• Cron expressions schedule recurring jobs.
• TypeScript generics describe job payloads and worker handlers.
• Job events let you record completion, failure, and processing time.
• Monque uses the native MongoDB driver and can share your application's connection.
• Shutdown waits for running jobs up to the configured timeout. Jobs left processing can be recovered on a later startup.

Installation

Using Bun:

bun add @monque/core mongodb

Or using npm/yarn/pnpm:

npm install @monque/core mongodb
yarn add @monque/core mongodb
pnpm add @monque/core mongodb

Usage

import { Monque } from '@monque/core';
import { MongoClient } from 'mongodb';

const client = new MongoClient('mongodb://localhost:27017');
await client.connect();

const monque = new Monque(client.db('myapp'), {
  collectionName: 'jobs',
  pollInterval: 1000,
  maxRetries: 10,
  defaultConcurrency: 5,
});

await monque.initialize();

// Register workers
monque.register('send-email', async (job) => {
  await sendEmail(job.data.to, job.data.subject);
});

// Start processing
monque.start();

// Enqueue jobs
await monque.enqueue('send-email', { to: '[email protected]', subject: 'Hello' });

// Schedule recurring jobs
await monque.schedule('0 9 * * *', 'daily-report', { type: 'summary' });

// Management
await monque.cancelJob('job-id');
const stats = await monque.getQueueStats();
const queueViews = await monque.getQueueViewSummaries();

// Graceful shutdown
await monque.stop();

API

new Monque(db, options?)

Pass a connected MongoDB database and any scheduler options you want to override.

Options:

• collectionName - MongoDB collection name (default: 'monque_jobs')
• pollInterval - Polling interval in ms (default: 1000)
• maxRetries - Max retry attempts (default: 10)
• baseRetryInterval - Base backoff interval in ms (default: 1000)
• shutdownTimeout - Graceful shutdown timeout in ms (default: 30000)
• defaultConcurrency - Jobs per worker (default: 5)
• lockTimeout - Stale job threshold in ms (default: 1800000)
• recoverStaleJobs - Recover stale jobs on startup (default: true)

Methods

• initialize() - Set up collection and indexes
• enqueue(name, data, options?) - Enqueue a job
• now(name, data) - Enqueue for immediate processing
• schedule(cron, name, data) - Schedule recurring job
• register(name, handler, options?) - Register a worker
• start() - Start processing jobs
• stop() - Graceful shutdown
• isHealthy() - Check scheduler health

Management:

• getJob(id) - Get job details
• getJobs(filter) - List jobs
• getJobsWithCursor(options) - Paginated list
• getQueueStats(filter?) - Queue statistics
• getQueueViewSummaries({ name }?) - Job counts and worker activity, optionally scoped to one name
• cancelJob(id) - Cancel a job
• retryJob(id) - Retry a job
• rescheduleJob(id, date) - Reschedule a job
• deleteJob(id) - Delete a job
• cancelJobs(filter) - Bulk cancel
• retryJobs(filter) - Bulk retry
• deleteJobs(filter) - Bulk delete

Events

monque.on('job:start', (job) => { /* job started */ });
monque.on('job:complete', ({ job, duration }) => { /* job completed */ });
monque.on('job:fail', ({ job, error, willRetry }) => { /* job failed */ });
monque.on('job:error', ({ error, job? }) => { /* unexpected error */ });
monque.on('job:cancelled', ({ job }) => { /* job cancelled */ });
monque.on('job:retried', ({ job, previousStatus }) => { /* job retried */ });
monque.on('job:deleted', ({ jobId }) => { /* job deleted */ });
monque.on('stale:recovered', ({ count }) => { /* stale jobs recovered */ });

Development

Running Tests

# Run tests once (fresh container each time)
bun run test

# Run tests in watch mode with container reuse (faster iteration)
bun run test:dev

# Or enable reuse globally in your shell profile
export TESTCONTAINERS_REUSE_ENABLE=true
bun run test:watch

With TESTCONTAINERS_REUSE_ENABLE=true, tests reuse the MongoDB container across runs. Ryuk, the Testcontainers cleanup daemon, still removes orphaned containers.

To manually clean up reusable containers:

docker ps -q --filter label=org.testcontainers=true | while read -r id; do docker stop "$id"; done

License

ISC