Jobs Wiring

Recipes are working wiring examples with the sharp edges annotated. This page and the jobs-wiring recipe served to coding agents by the MCP server share one source (@modularityjs/manifest), so they never drift apart.

Three packages compose:

@modularityjs/queue — the transport contract: QueueService, @Consume, retry policies, dead-lettering. Jobs ride on top of it.
@modularityjs/queue-jobs — typed background jobs: abstract Job<TPayload> base class, @JobDefinition({ name, retryPolicy?, concurrency? }) class decorator, JobDispatcher.dispatch(JobClass, payload, options?), and the jobPoolEntries(...) registration helper. QueueJobsModule has no driver of its own.
@modularityjs/queue-{memory,redis,rabbitmq} — the queue driver actually moving messages. Jobs work identically over all three; retry/concurrency/delay enforcement is the driver's.

A job is a class: subclass Job<TPayload>, implement handle(payload) (constructor-inject services like any provider), and decorate with @JobDefinition. The decorator registers the class's queue consumer on topic jobs:<name> — application code never touches topic strings. Dispatch takes the class, so the payload type is checked at the call site.

typescript

import { Inject, Injectable } from '@modularityjs/di';
import { inversify } from '@modularityjs/di-inversify';
import { createApp, Module, ModularityModule } from '@modularityjs/modularity';
import { QueueModule } from '@modularityjs/queue';
import {
  Job,
  JobDefinition,
  JobDispatcher,
  jobPoolEntries,
  QueueJobsModule,
} from '@modularityjs/queue-jobs';
import { QueueMemoryModule } from '@modularityjs/queue-memory';

interface InvoicePayload {
  invoiceId: string;
}

@JobDefinition({
  name: 'send-invoice',
  retryPolicy: { maxAttempts: 5, strategy: 'exponential', delayMs: 2000 },
  concurrency: 2,
})
class SendInvoiceJob extends Job<InvoicePayload> {
  async handle(payload: InvoicePayload): Promise<void> {
    // render + email the invoice; a throw here triggers the retry policy
  }
}

@Injectable()
class BillingService {
  constructor(@Inject(JobDispatcher) private readonly jobs: JobDispatcher) {}

  async finalize(invoiceId: string): Promise<void> {
    await this.jobs.dispatch(SendInvoiceJob, { invoiceId });
    // or delayed: await this.jobs.dispatch(SendInvoiceJob, { invoiceId }, { delayMs: 60_000 });
  }
}

@Module({
  name: 'billing',
  imports: [QueueModule, QueueJobsModule],
  providers: [SendInvoiceJob, BillingService],
  pools: [...jobPoolEntries(SendInvoiceJob)],
})
class BillingModule {}

const app = await createApp({
  di: inversify,
  modules: [
    ModularityModule,
    QueueModule,
    QueueMemoryModule,
    QueueJobsModule,
    BillingModule,
  ],
});

Register with jobPoolEntries, and keep the class in providers too. The helper contributes each job class to both JobsPool (so the dispatcher can resolve it) and QueueConsumersPool (so the driver consumes its topic) — but pool useClass entries are DI-instantiated, so the class must also appear in the module's providers. Dispatching a job that was never wired throws a StateException naming the fix rather than publishing to a topic nobody consumes.

Driver switch is one line. Replace QueueMemoryModule with RedisModule.forRoot({...}), QueueRedisModule or QueueRabbitmqModule.forRoot({ url: 'amqp://...' }) — jobs, retry policies, and delayed dispatch are unchanged. The memory driver dispatches in-process (dev/tests); Redis and RabbitMQ survive restarts and scale out consumers.

Retry and concurrency are per job. retryPolicy and concurrency on @JobDefinition are ordinary @Consume options (defaults: 3 attempts, exponential backoff from 1s capped at 30s, concurrency 1). A job that exhausts retries dead-letters like any queue message — inspect with QueueService.getDeadLetters('jobs:<name>') or the queue:dead-letter:list CLI.

Jobs Wiring ​

Jobs Wiring