Ts.ED Integration

@monque/tsed registers job handlers through decorators, injects their dependencies, and starts and stops the scheduler with your Ts.ED application.

Quick Reference

Method / Type	Description
MonqueModule	Main Ts.ED module
@JobController()	Register a job controller
@Job()	Register a job handler
@Cron()	Register a scheduled job
MonqueService	Injectable job management service

Installation

bun

bun add @monque/tsed @monque/core mongodb

npm

npm install @monque/tsed @monque/core mongodb

pnpm

pnpm add @monque/tsed @monque/core mongodb

yarn

yarn add @monque/tsed @monque/core mongodb

Dependencies

Install these peer dependencies alongside the integration:

Package	Version	Required
@monque/core	^1.18.0	Yes
mongodb	^7.7.0	Yes
@tsed/core	^8.41.2	Yes
@tsed/di	^8.41.2	Yes
@tsed/mongoose	^8.41.2	Only for Mongoose integration

Configuration

Import MonqueModule in your Server.ts and configure the connection via the @Configuration decorator. Choose an existing database connection, an asynchronous factory, or a Ts.ED Mongoose connection.

Direct Strategy (Db instance)

Use this if you already have a connected native MongoDB Db instance available synchronously.

import { Configuration } from "@tsed/di";
import { MonqueModule } from "@monque/tsed";

@Configuration({
  imports: [MonqueModule],
  monque: {
    enabled: true,
    db: myExistingDbInstance,
  },
})
export class Server {}

Factory Strategy (Async)

Use an asynchronous factory to connect to MongoDB during application startup.

// src/Server.ts
import { Configuration } from "@tsed/di";
import { MonqueModule } from "@monque/tsed";
import "@tsed/platform-express";
import { MongoClient } from "mongodb";

@Configuration({
  imports: [MonqueModule],
  monque: {
    enabled: true,
    dbFactory: async () => {
      const client = await MongoClient.connect("mongodb://localhost:27017/my-app");
      return client.db("my-app");
    },
    collectionName: "custom_jobs", // Optional: defaults to "monque_jobs"
  },
})
export class Server {}

Close clients you create in dbFactory during application shutdown, after Monque stops. Reuse an application-managed connection when you already have one.

Mongoose Strategy

If your application uses @tsed/mongoose, set mongooseConnectionId to reuse its native MongoDB connection and pool. You do not need a separate connection string for Monque.

import { Configuration } from "@tsed/di";
import { MongooseModule, MongooseService } from "@tsed/mongoose";
import { MonqueModule } from "@monque/tsed";

@Configuration({
  imports: [
    MongooseModule, // Establish Mongoose connections
    MonqueModule, // Hooks into existing Mongoose connections
  ],
  mongoose: [
    {
      id: "default",
      url: "mongodb://localhost:27017/my-app", // Mongoose manages this connection
    },
  ],
  monque: {
    enabled: true,
    dbToken: MongooseService, // Reference the Mongoose service
    mongooseConnectionId: "default", // Identify which connection to reuse
  },
})
export class Server {}

Usage

1. Define a Job Controller

Group related handlers in a class decorated with @JobController. Decorate each handler with @Job to register it.

Type Safety

Define payload types with TypeScript interfaces and pass them to the Job<T> type:

import { JobController, Job } from "@monque/tsed";
import type { Job as MonqueJob } from "@monque/core";

interface SendEmailPayload {
  to: string;
  subject: string;
  templateId: string;
}

@JobController("email") // Optional namespace prefix: "email."
export class EmailJobs {
  @Job("send", { concurrency: 5, maxRetries: 5 }) // Full job name: "email.send"
  async sendEmail(job: MonqueJob<SendEmailPayload>) {
    // job.data is fully typed
    const { to, subject } = job.data;
    console.log(`Sending email to ${to}...`);
  }
}

@Job() also accepts baseRetryInterval and maxBackoffDelay. Omitted retry settings inherit your application’s monque configuration. See per-worker retry settings.

For runtime payload validation, pass any Standard Schema compatible validator:

import { z } from "zod";

const emailSchema = z.object({ email: z.email() });

@JobController("email")
export class ValidatedEmailJobs {
  @Job("deliver", { schema: emailSchema })
  async deliver(job: MonqueJob<z.output<typeof emailSchema>>) {
    await sendEmail(job.data.email);
  }
}

The handler receives validated output; invalid payloads fail before the method or its request-scoped dependencies are invoked. See payload validation for transformation, retry, and stored-input behavior.

Use MonqueService.pause(), resume(), and isPaused() for local processing controls. Pass the fully qualified name to control one worker, for example monqueService.pause('email.send'). Omitting the name controls the local instance. Running jobs continue; other instances are unaffected. See pausing local processing.

2. Schedule Jobs (Cron)

Use the @Cron decorator to schedule recurring tasks. Cron jobs are automatically registered and scheduled with Monque during the application’s $onInit lifecycle phase.

import { JobController, Cron } from "@monque/tsed";

@JobController()
export class ReportJobs {
  @Cron("0 9 * * *", { name: "daily-report", timezone: "Europe/Berlin", priority: 10 })
  async generateDailyReport() {
    console.log("Generating report...");
  }
}

Set priority, timezone, and uniqueKey with either @Cron or MonqueService.schedule():

await monqueService.schedule(
  "0 9 * * *",
  "daily-report",
  {},
  {
    timezone: "Europe/Berlin",
    priority: 10,
    uniqueKey: "daily-report",
  },
);

Use a version of @monque/core with timezone support on every worker before enabling the option. Ts.ED uses the application’s installed core peer dependency; upgrading Ts.ED alone does not upgrade core. Without timezone, existing server-local scheduling remains unchanged. See timezone scheduling for daylight saving behavior and rollout guidance.

Priority belongs to each Job, not its Worker. MonqueService.now(name, data, { priority, session }) accepts options for immediate Jobs; existing two-argument calls still work. The default priority is 0, and supplied values must be signed safe integers. Recurring runs and retries retain priority. Submitting an active Unique Key again keeps the original Job. Use @monque/core 1.18.0 or newer on every producer and Scheduler Instance for consistent priority ordering. To change a pending Job’s priority, call MonqueService.setJobPriority(id, priority).

3. Enqueue Jobs

Inject MonqueService into any Ts.ED Service or Controller to dispatch jobs.

import { Service, Inject } from "@tsed/di";
import { MonqueService } from "@monque/tsed";

@Service()
export class AuthService {
  @Inject()
  declare private monque: MonqueService;

  async registerUser(user: { email: string }) {
    // Dispatch background job
    await this.monque.enqueue("email.send", {
      to: user.email,
      subject: "Welcome!",
      templateId: "welcome",
    });
  }
}

Non-retryable failures

Import NonRetryableError from @monque/core and throw it from an @Job or @Cron handler when another attempt cannot help:

import { NonRetryableError } from "@monque/core";
import { Job, JobController } from "@monque/tsed";

@JobController("reports")
export class ReportJobs {
  @Job("deliver")
  async deliver() {
    throw new NonRetryableError("Destination account no longer exists");
  }
}

Ts.ED preserves the error when passing it to the scheduler. Requires @monque/core 1.14.0 or later in the application; upgrading Ts.ED alone does not upgrade its core peer. A recurring job stops too. Manual retry through MonqueService, Management, or Dashboard remains available after fixing the cause. See Retry & Backoff.

Job Isolation

Each execution gets its own DIContext, so services with ProviderScope.REQUEST are created separately for each job. You can use Ts.ED context APIs such as PlatformContext inside the handler.

Producer-only Mode

Set disableJobProcessing on API servers that enqueue jobs but leave processing to separate worker servers:

// API Server - Only enqueues jobs, doesn't process them
@Configuration({
  imports: [MonqueModule],
  monque: {
    dbFactory: async () => client.db("myapp"),
    disableJobProcessing: true, // Producer-only mode
  },
})
export class ApiServer {}

// Worker Server - Processes jobs
@Configuration({
  imports: [MonqueModule],
  monque: {
    dbFactory: async () => client.db("myapp"),
    // disableJobProcessing defaults to false
  },
})
export class WorkerServer {}

Note

When disableJobProcessing: true: - The database connection is still established - MonqueService can enqueue jobs via enqueue(), now(), etc. - Jobs are not registered (even if @Job decorators exist) - The polling loop is not started - isHealthy() returns false (since the scheduler isn’t running)

API Reference

Services

MonqueService

Inject MonqueService to call methods on the application’s Monque instance.

MonqueService.getQueueViewSummaries() returns queue counts and current local worker activity, pause state, retry settings, and schema presence. You can pass the injected service directly as the Management API’s monque option, keeping its reads and actions attached to the same scheduler.

Commonly used methods:

• enqueue() - Add a job to the queue.
• enqueueMany() - Submit a batch with per-job priorities, scheduling, and unique keys, including in producer-only applications. Returns inserted and deduplicated counts; see batch submission for partial-failure behavior.
• schedule() - Schedule a recurring job.
• now() - Run a job immediately.

For transactional submission, pass { session } to enqueue(), now(), enqueueMany(), or schedule(). Create the session from the same MongoClient used by your configured database connection. Your application owns the transaction and session lifetime.

Tip

See the Ts.ED API Reference for queries, bulk actions, and other methods.

Dashboard and Management API

@monque/tsed configures the scheduler and workers. To serve job inspection and actions over HTTP, use the Management API. Express applications can also mount the dashboard.

In producer-only mode, an unhealthy scheduler status is expected because processing is not started. It does not indicate that workers in another process have stopped.

Next Steps

• Core Concepts: job data and lifecycle.
• Scheduling: cron expressions and recurring jobs.
• Core API Reference: scheduler methods and options.