@nage-api/queue

Typed background jobs (PLAN.md §8, §25 P2).

The Job contract is retained from the legacy framework so migrating is a driver swap rather than a rewrite of every consumer (§26). What changes is that both ends now share a type:

import type { QueueService } from '@nage-api/queue';

// A `type`, not an `interface`: an interface has to say `extends QueueJobMap` to
// satisfy the constraint, and inheriting that index signature widens `keyof` to
// `string` — which silently stops the job *name* from being checked at all.
type Jobs = {
  'email.welcome': { userId: string };
  'report.build': { month: string };
};

declare const queue: QueueService<Jobs>;
declare const userId: string;
declare function sendWelcomeEmail(id: string, attempt: number): Promise<void>;

await queue.enqueue('email.welcome', { userId }); // name and payload both checked

queue.process('email.welcome', async ({ job, attempt }) => {
  await sendWelcomeEmail(job.payload.userId, attempt);
});

The legacy bus passed any in both directions, so a renamed field was a runtime failure in a worker nobody was watching.

Wiring it up

The feature is off unless queue.enabled is true, and registering the module publishes a publisher, not a worker: nothing consumes until something calls start(), and the driver forRoot builds has no poll timer of its own. A process that is meant to run jobs supplies a driver with an interval and starts it.

import { Injectable, Module, type OnApplicationBootstrap } from '@nestjs/common';
import {
  MemoryJobLogStore,
  MemoryQueueDriver,
  NageQueueModule,
  QueueService,
} from '@nage-api/queue';
import type { NodeEnvironment } from '@nage-api/contracts';

type Jobs = {
  'email.welcome': { userId: string };
};

declare const env: { NODE_ENV: NodeEnvironment };
declare function sendWelcomeEmail(userId: string): Promise<void>;

// One store, given to both: `forRoot` publishes the log store for readers, but a
// driver you construct yourself is never handed it, so nothing would write.
const jobLogs = new MemoryJobLogStore();

@Injectable()
export class EmailWorker implements OnApplicationBootstrap {
  // The provider is the bare class, so the job map is a compile-time view of it.
  constructor(private readonly queue: QueueService<Jobs>) {}

  async onApplicationBootstrap(): Promise<void> {
    // Register before starting: a job whose name has no handler is dead on
    // arrival, not queued until one appears.
    this.queue.process('email.welcome', async ({ job }) => {
      await sendWelcomeEmail(job.payload.userId);
    });

    await this.queue.start();
  }
}

@Module({
  imports: [
    NageQueueModule.forRoot({
      queue: { enabled: true, jobLogs: true },
      // Passed so the module can refuse a driver that loses jobs on restart.
      environment: env.NODE_ENV,
      logs: jobLogs,
      driver: new MemoryQueueDriver({ concurrency: 4, pollIntervalMs: 250, logs: jobLogs }),
    }),
  ],
  providers: [EmailWorker],
})
export class WorkerModule {}

onApplicationShutdown drains whatever the module started, so SIGTERM does not abandon a job that had already been taken.

What the driver guarantees

Retries with jittered exponential backoff. The jitter is the point: a batch of jobs that failed together at the same instant will otherwise retry together, and the dependency they were waiting on goes down again.

A dead-letter state. A job that exhausts its attempts becomes dead and fires onDead, rather than vanishing or retrying forever. A job with no registered handler is dead immediately — retrying would burn the attempts, and dropping it silently would hide a deployment mistake.

De-duplication by key, while the first is still pending. That is what makes an at-least-once queue tolerable for "send the welcome email".

The correlation id travels with the job. A job is usually the tail of a request, and losing the id at the queue boundary is where a trace stops being useful.

Graceful drain on shutdown, so a worker does not exit mid-job.

Job logs

Opt-in (queue.jobLogs). Each transition records state, attempt, duration and the failure message — but never the payload, which may carry personal data and would outlive the job by months.

Drivers

MemoryQueueDriver is a real implementation, not a stub: retries, delays, de-duplication, concurrency and drain all behave the way the BullMQ adapter must, which makes it the executable specification of the port. It is driven by an explicit tick()/drain() as well as a timer, so tests advance it deterministically instead of sleeping.

It is not durable, so the module refuses to use it in production — the failure otherwise presents as "some emails were never sent", weeks later, with nothing in the logs.

Not yet implemented

• The BullMQ driver itself (§27.5). QueueDriver is the seam; the application constructs BullMQ and passes it in, so ioredis stays out of the install for deployments that run no workers.
• Repeatable/cron jobs, priorities, and JobContext.progress reporting — the hook exists and the memory driver ignores it.
• A queue-depth gauge wired to @nage-api/observability.