Listeners and transports

Libraries publish with emit and trace. Your app reacts with listeners: provider methods decorated with @OnDiagnostic. The transport you select decides where a listener runs and what happens when it throws.

ListenersCopy Anchor Link

Register DiagnosticsModule once at the app root, then decorate provider methods:

import { Injectable, Module } from '@nestjs/common';
import type { DiagnosticEvent } from '@dudousxd/nestjs-diagnostics';
import {
  type DiagnosticEventOf,
  DiagnosticsModule,
  OnDiagnostic,
} from '@dudousxd/nestjs-diagnostics/nestjs';

@Injectable()
export class TrackAgentRun {
  @OnDiagnostic('agent', 'run.finished')
  async handle(event: DiagnosticEventOf<'agent', 'run.finished'>) {
    // event.payload is typed from the ChannelRegistry
  }

  @OnDiagnostic('durable', /^run\./) // every durable event whose name matches
  async onRun(event: DiagnosticEvent) {}

  @OnDiagnostic('resilience') // every resilience event, current and future
  async onAnything(event: DiagnosticEvent) {}
}

@Module({
  imports: [DiagnosticsModule.forRoot()],
  providers: [TrackAgentRun],
})
export class AppModule {}

• event takes an exact name, a RegExp, or nothing (every event of the lib).

• For a channel declared in the ChannelRegistry, lib and event autocomplete and the handler's parameter is checked against the declared payload. Untyped channels accept any annotation. A library (or your app) declares its channels like this:

  Copy Text

  declare module '@dudousxd/nestjs-diagnostics' {
    interface ChannelRegistry {
      agent: { 'run.finished': { runId: string; tokens: number } };
    }
  }

• Every listener has a stable id, <ClassName>#<method>. Delivering transports route work by this id, so every process must register the same listeners, and you should not rename a listener class while deliveries for it may still be queued. Two listener classes with the same name fail the bootstrap.

• Each envelope carries a unique id (a UUID). It is the same in every process the event reaches; use it to make a listener idempotent.

Listeners only see events from emit. Span events from trace go to observers (raw diagnostics_channel subscribers, Telescope, OpenTelemetry), which keep working the same way whatever transport you choose.

TransportsCopy Anchor Link

Without a transport, listeners run in the process that emitted the event. To run them across processes, list transports and pick a default:

import { Module } from '@nestjs/common';
import { DiagnosticsModule, transports } from '@dudousxd/nestjs-diagnostics/nestjs';
import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL!);

@Module({
  imports: [
    DiagnosticsModule.forRoot({
      default: 'queue',
      transports: {
        redis: transports.redis({ pub: redis, sub: redis.duplicate() }),
        queue: transports.queue({ connection: { host: 'localhost', port: 6379 } }),
        durable: transports.durable({ retries: 5 }),
      },
    }),
  ],
})
export class AppModule {}

Only the default transport starts. Listing the others costs nothing: their optional dependency is imported only when selected.

Build the options from DI with forRootAsync:

DiagnosticsModule.forRootAsync({
  imports: [RedisModule],
  inject: [REDIS],
  useFactory: (redis: Redis) => ({
    default: 'redis',
    transports: { redis: transports.redis({ pub: redis, sub: redis.duplicate() }) },
  }),
});

GuaranteesCopy Anchor Link

With queue and durable, delivery happens once, from the process that called emit: copies of the event that reached other processes through a relay are not delivered again. Deliveries are handed to the transport after the current event loop turn, in emit order, never on the emitter's call stack. If the transport cannot store them (Redis down), the failure is logged and those deliveries are lost.

transports.redis()Copy Anchor Link

Broadcast over Redis pub/sub. Every process receives the event and runs its own listeners. Best-effort: a process that is down misses the event.

You own the Redis clients: diagnostics never opens or closes them. Payloads cross as JSON.

transports.queue()Copy Anchor Link

One BullMQ job per (listener, event). A handler that throws fails its job, and BullMQ retries it. Requires the optional peer bullmq:

pnpm add bullmq

The job id is diag.<event id>.<listener id>, so the same delivery is never queued twice. Every process that is not role: 'api' runs a worker for the queue: you don't add a BullMQ processor yourself.

transports.durable()Copy Anchor Link

Each (listener, event) becomes a durable run, retried with backoff and idempotent per event id. Requires @dudousxd/nestjs-durable-diagnostics and a configured DurableModule.

Split deployments: roleCopy Anchor Link

role says which half of a split deployment this process is:

DiagnosticsModule.forRoot({
  role: process.env.PROCESS_ROLE === 'worker' ? 'worker' : 'api',
  default: 'queue',
  transports: { queue: transports.queue({ connection }) },
});

Per transport:

'api' and 'worker' need a default transport: without one, nothing carries events between the halves, so the bootstrap fails. An unknown role, or a default that names no entry of transports, fails the bootstrap too.