Carry the id through a job

To run a queue message’s work under the id the message carried, open a correlation context with CorrelationService.run(id, callback). CorrelationIdMiddleware opens none outside a request, so there getCorrelationId() returns undefined.

Open a correlation context in a queue consumer

RestockJob is a queue consumer in Baize’s orders service that opens its own correlation context with CorrelationService.run() from @evanion/nestjs-correlation-id. For each restock message it asks stock for the game’s current level. The code that queued the message, such as a request handler in orders, copied getCorrelationId() into the message’s correlationId field, so the message carries that request’s id when there was one.

CorrelationConfig is the type of the configuration forRoot() builds from its argument and the defaults, with the fields header, generator and validate. CORRELATION_CONFIG_TOKEN is the injection token forRoot() provides that configuration under, and RestockJob reads validate from it to check the id a message carried. RestockJob injects CorrelationService, the configuration, and StockClient, the client Forward the id builds:

import { Inject, Injectable } from '@nestjs/common';
import {
  CORRELATION_CONFIG_TOKEN,
  CorrelationService,
  type CorrelationConfig,
} from '@evanion/nestjs-correlation-id';
import { StockClient } from './stock.client.js';
 
export interface RestockMessage {
  game: string;
  correlationId?: string;
}
 
@Injectable()
export class RestockJob {
  constructor(
    private readonly correlation: CorrelationService,
    @Inject(CORRELATION_CONFIG_TOKEN)
    private readonly config: CorrelationConfig,
    private readonly stock: StockClient,
  ) {}
 
  handle(message: RestockMessage) {
    const carried = message.correlationId;
    const id =
      carried && this.config.validate?.(carried)
        ? carried
        : this.correlation.generate();
 
    return this.correlation.run(id, () => this.stock.level(message.game));
  }
}

• run(id, callback) runs callback inside a correlation context and returns what callback returns, so the promise from stock.level() comes straight back out of handle().
• Everything the callback calls or awaits reads id. StockModule registers StockClient’s HttpService with withCorrelation(), so its call to stock carries the id.
• generate() returns a new id from the configured generator, randomUUID unless forRoot() was given another. A cron job has no id to carry, so it runs its work inside run(generate(), …).

Declare RestockJob in a module that imports StockModule, and call handle() from your queue library’s consumer.

Why RestockJob validates the carried id

An id read off a queue message reaches the same log lines and outgoing headers as one read off a request, so RestockJob runs it through the same check @evanion/nestjs-correlation-id applies to a request. validate on the injected configuration is the function CorrelationIdMiddleware applies to an incoming header, DEFAULT_CORRELATION_ID_VALIDATOR unless forRoot() was given another. A carried id that fails the check is dropped, and the job generates its own.

See the id each restock message runs under

The code below builds RestockJob by hand on a CorrelationService from @evanion/nestjs-correlation-id, with a stand-in for StockClient that records the id each call to stock runs under. It hands the job three messages: one carrying storefront-4f1c9a, one carrying no id, and one carrying storefront 4f1c9a, whose space fails DEFAULT_CORRELATION_ID_VALIDATOR:

import {
  CorrelationService,
  DEFAULT_CORRELATION_ID_VALIDATOR,
} from '@evanion/nestjs-correlation-id';
import { RestockJob } from '../examples/restock.job.js';
 
const config = {
  header: 'X-Correlation-Id',
  generator: () => 'orders-7c41d2',
  validate: DEFAULT_CORRELATION_ID_VALIDATOR,
};
const correlation = new CorrelationService(config);
 
// Stands in for StockClient and records the id each call runs under, the id
// StockModule's HttpService sends to stock. `as never` lets it stand in for the
// class, which also holds an HttpService.
const sent: (string | undefined)[] = [];
const stock = {
  level: async (game: string) => {
    sent.push(correlation.getCorrelationId());
    return { game, available: 3 };
  },
};
const job = new RestockJob(correlation, config, stock as never);
 
await job.handle({ game: 'urn:game:azul', correlationId: 'storefront-4f1c9a' });
await job.handle({ game: 'urn:game:azul' });
await job.handle({ game: 'urn:game:azul', correlationId: 'storefront 4f1c9a' });
 
sent; // -> ['storefront-4f1c9a', 'orders-7c41d2', 'orders-7c41d2']

The first message ran under its own id. The other two ran under the id the generator returned.

Replace the id inside a correlation context

CorrelationService.setCorrelationId(id), from @evanion/nestjs-correlation-id, replaces the id for the rest of the context it is called in. It serves a consumer that opens its context before it has read the message: a wrapper calls run(generate(), …) around every handler, so each handler starts under a generated id. A handler that then reads an id off its message passes that id to setCorrelationId. Outside any context setCorrelationId throws.

The code below builds a queue worker, the process that runs such handlers, as a Nest application context, which starts no HTTP server. It calls generate, run and setCorrelationId:

import { NestFactory } from '@nestjs/core';
import {
  CorrelationModule,
  CorrelationService,
} from '@evanion/nestjs-correlation-id';
 
const worker = await NestFactory.createApplicationContext(
  CorrelationModule.forRoot({ generator: () => 'orders-7c41d2' }),
  { logger: false },
);
const correlation = worker.get(CorrelationService);
 
const generated = correlation.run(correlation.generate(), () =>
  correlation.getCorrelationId(),
);
 
let replaced: string | undefined;
correlation.run('orders-0f3a2b', () => {
  correlation.setCorrelationId('storefront-4f1c9a');
  replaced = correlation.getCorrelationId();
});
 
let refused = '';
try {
  correlation.setCorrelationId('storefront-4f1c9a');
} catch (error) {
  refused = (error as Error).message;
}
await worker.close();
 
generated; // -> 'orders-7c41d2'
replaced; // -> 'storefront-4f1c9a'
refused; // -> 'setCorrelationId() was called outside a correlation context. Apply CorrelationIdMiddleware, or wrap the work in CorrelationService.run().'

The generated id came from the generator passed to forRoot(). Inside run('orders-0f3a2b', …), setCorrelationId changed what getCorrelationId() returns. Outside, the call threw, and the error message names the two ways to open a context: CorrelationIdMiddleware and CorrelationService.run().

Inside a request, setCorrelationId leaves the response header alone. The header keeps the id the middleware wrote when the request arrived.

Change the header, the generator and the check

A job now carries the id its message brought, and @evanion/nestjs-correlation-id generates a fresh one when the message brought none. Configuration changes the header, the generator and the check @evanion/nestjs-correlation-id applies to an incoming id, through the fields forRoot() takes.