runWithCorrelationId

runWithCorrelationId<T>(correlationId, fn): T

Defined in: correlation.store.ts:74

Run a callback within a correlation context.

Use this in non-HTTP entry points (queue workers, message consumers, WebSocket handlers, scheduled jobs) to propagate an external correlation ID into everything the callback runs, wrapped operations included.

If correlationId is falsy (undefined or empty), the id is resolved via getCorrelationId (the current one, or a new one from correlationSource.create()), so fn always runs with a correlation id. A pipeline dispatched inside fn takes it as its context.correlationId.

Type Parameters

T

T

Parameters

correlationId

string | undefined

The correlation ID to propagate. If falsy, falls back to getCorrelationId.

fn

() => T

The callback to execute within the correlation context.

Returns

T

Example

// BullMQ worker
new Worker('emails', (job) =>
  runWithCorrelationId(job.data.correlationId, () => sendEmail(job.data)),
);

// RabbitMQ consumer (amqplib)
channel.consume('users', (message) =>
  runWithCorrelationId(message.properties.correlationId, () =>
    onUserCreated(JSON.parse(message.content.toString())),
  ),
);

// Scheduled job: no incoming id, so a new one is made
setInterval(() => runWithCorrelationId(undefined, () => hourlySync()), 3_600_000);