How to register lifecycle hooks

Use whenyou want the transport to feed your metrics, tracing or alerting.

You getevery event it emits, the signature of each, and where in the flow they fire.

Available events

Every event is a member of the TransportEvent enum:

Event	Signature	When it fires
Connect	(server: string) => void	NATS connection established
Disconnect	() => void	NATS connection lost
Reconnect	(server: string) => void	NATS connection re-established after a disconnect
Error	(error: Error, context?: string) => void	Any transport-level error
RpcTimeout	(subject: string, correlationId: string) => void	A deadline expires on either side of an RPC call
MessageRouted	(subject: string, kind: MessageKind) => void	A message is successfully routed to its handler
HandlerCompleted	(subject, kind: StreamKind, durationMs, status) => void	A handler returns or throws (success/error/retried/terminated).
Published	(subject, kind: StreamKind, durationMs, status) => void	Every client-side publish leg completes (event emit or RPC publish).
RpcCompleted	(subject, durationMs, status) => void	RPC round-trip settles on the caller side (success / error / timeout).
ConsumerRecovered	(label, attempts: number) => void	Self-healing recovers a consumer after one or more failed restarts
ShutdownStart	() => void	Graceful shutdown sequence begins
ShutdownComplete	() => void	Graceful shutdown sequence finishes
DeadLetter	(info: DeadLetterInfo) => void	A message exhausts all delivery attempts
ScheduleReconciled	(report: ScheduleReconcileReport) => void	Once per connection at boot, after the declared recurring schedules are written

• MessageKind on MessageRouted has two values, Event and Rpc.
• ScheduleReconcileReport lists handler patterns in three arrays: created, updated and unchanged.
• The built-in metrics subscribe to these events next to your hooks, so a hook of yours never replaces them.

RpcTimeout fires on the caller when a request never gets its reply, and on the responder when a handler runs past the deadline. Core-mode caller timeouts have no correlation ID to report and pass an empty string, so read subject first and treat correlationId as optional detail.

Registering hooks

Pass a hooks object in forRoot() or forRootAsync() with the events you care about.

src/app.module.ts

import { Module } from '@nestjs/common';
import { JetstreamModule, TransportEvent } from '@horizon-republic/nestjs-jetstream';

@Module({
  imports: [
    JetstreamModule.forRoot({
      name: 'orders',
      servers: ['nats://localhost:4222'],
      hooks: {
        [TransportEvent.Connect]: (server) => {
          console.log(`Connected to ${server}`);
        },
        [TransportEvent.Disconnect]: () => {
          console.warn('NATS connection lost');
        },
        [TransportEvent.Error]: (error, context) => {
          console.error(`Transport error [${context}]:`, error);
        },
      },
    }),
  ],
})
export class AppModule {}

Practical examples

Sentry error tracking

import * as Sentry from '@sentry/node';
import { JetstreamModule, TransportEvent } from '@horizon-republic/nestjs-jetstream';

JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  hooks: {
    [TransportEvent.Error]: (error, context) => {
      Sentry.captureException(error, {
        tags: { component: 'jetstream', context },
      });
    },
    [TransportEvent.DeadLetter]: (info) => {
      Sentry.captureMessage(`Dead letter: ${info.subject}`, {
        level: 'error',
        extra: {
          stream: info.stream,
          deliveryCount: info.deliveryCount,
          streamSequence: info.streamSequence,
        },
      });
    },
  },
})

Prometheus metrics

metrics: true gives you counters and histograms for handlers, publishes, RPC, dead letters and connection state without writing hooks. See Metrics.

Structured logging

import { JetstreamModule, TransportEvent } from '@horizon-republic/nestjs-jetstream';

const log = (event: string, data?: Record<string, unknown>) =>
  console.log(JSON.stringify({ event, ts: new Date().toISOString(), ...data }));

JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  hooks: {
    [TransportEvent.Connect]: (server) => log('nats.connect', { server }),
    [TransportEvent.Disconnect]: () => log('nats.disconnect'),
    [TransportEvent.Reconnect]: (server) => log('nats.reconnect', { server }),
    [TransportEvent.Error]: (error, context) =>
      log('nats.error', { message: error.message, context }),
    [TransportEvent.ShutdownStart]: () => log('nats.shutdown.start'),
    [TransportEvent.ShutdownComplete]: () => log('nats.shutdown.complete'),
  },
})

Hooks with multiple connections

Hooks are registered once, at the root, and fire for every connection. With more than one connection configured, each callback gets the originating connection name as a trailing argument:

JetstreamModule.forRoot({
  name: 'orders',
  defaultConnection: 'primary',
  hooks: {
    [TransportEvent.Error]: (error, context, connection) => {
      sentry.captureException(error, { tags: { connection } });
    },
    [TransportEvent.HandlerCompleted]: (subject, kind, durationMs, status, connection) => {
      metrics.histogram('handler_ms', durationMs, { subject, connection });
    },
  },
  connections: {
    primary: { servers: ['nats://primary:4222'] },
    analytics: { servers: ['nats://analytics:4222'] },
  },
})

connection is the last parameter of every hook and is undefined in single-connection applications.

No hook = silence

An event without a registered hook is dropped: no default logging, no warning, no overhead beyond one lookup. To log everything during development, register a hook for every event.

Async hook error handling

Hooks can be synchronous or return a Promise. A hook that throws, or a Promise that rejects, is caught and logged through the NestJS Logger, with no unhandledRejection. Hook errors never crash the application and never affect message processing.

hooks: {
  // Safe: if Sentry is down, the error is logged but the app keeps running
  [TransportEvent.Error]: async (error) => {
    await sentry.captureException(error); // rejection is caught by EventBus
  },
}

DeadLetter hook vs onDeadLetter callback

	hooks[TransportEvent.DeadLetter]	onDeadLetter callback
Type	Sync or async hook (fire-and-forget)	Async callback (awaited)
Fires	Always, before the callback	Only if configured
Affects message fate?	No	Yes: success = term(), failure = nak()
Use case	Metrics, logging, alerting	Persisting dead letters to a store
Error behavior	Caught and logged	Causes message to be nak'd for retry

They work together:

JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  hooks: {
    // Lightweight: increment a counter, always runs
    [TransportEvent.DeadLetter]: (info) => {
      metrics.increment('dead_letter_total', { stream: info.stream });
    },
  },
  // Heavyweight: persist to database, must succeed before term()
  onDeadLetter: async (info) => {
    await dlqRepository.save(info);
  },
})

If the callback fails and the message is naked, the hook fires again on the next delivery attempt.

What's next?

• Dead Letter Queue: full guide on dead letter handling and the onDeadLetter callback
• Health Checks: monitor connection health with RTT latency
• Graceful Shutdown: ShutdownStart and ShutdownComplete events in context
• Module Configuration: hooks option in the full options reference