### forRoot() Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/module-configuration

Example of using `forRoot()` for global setup in `AppModule`, configuring NATS servers, stream, consumer, and RPC settings.

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

@Module({
  imports: [
    JetstreamModule.forRoot({
      name: 'user-events',
      servers: ['nats://localhost:4222'],
      events: {
        stream: {
          max_age: toNanos(30, 'days'),
          max_bytes: 10 * 1024 * 1024 * 1024, // 10 GB
          num_replicas: 3,
        },
        consumer: {
          max_ack_pending: 500,
          ack_wait: toNanos(30, 'seconds'),
        },
        consume: { idle_heartbeat: 10_000 },
        concurrency: 200,
        ackExtension: true,
      },
      rpc: { mode: 'core', timeout: 10_000 },
      shutdownTimeout: 15_000,
      hooks: {
        [TransportEvent.Error]: (err, ctx) => console.error(`[${ctx}]`, err),
        [TransportEvent.Connect]: (server) => console.log(`Connected to ${server}`),
      },
    }),
  ],
})
export class AppModule {}
```

--------------------------------

### Integration Test Setup with NATS Container

Source: https://nestjs-jetstream.horizon-republic.dev/docs/development/testing

Demonstrates the standard pattern for integration tests, including starting a NATS container, establishing a connection, and cleaning up resources after the tests.

```typescript
import type { StartedTestContainer } from 'testcontainers';

import { startNatsContainer } from './nats-container';

describe('My Feature', () => {
  let container: StartedTestContainer;
  let port: number;
  let nc: NatsConnection;
  
  beforeAll(async () => {
    ({ container, port } = await startNatsContainer());
    nc = await createNatsConnection(port);
  });
  
  afterAll(async () => {
    try {
      await nc?.drain();
    } finally {
      await container?.stop();
    }
  });
  
  it('should do something', async () => {
    const { app } = await createTestApp({ name: 'my-service', port }, [MyController]);
    
    // ... test against real NATS ...
    
    await app.close();
  });

});

```

--------------------------------

### Main Application Setup

Source: https://nestjs-jetstream.horizon-republic.dev/docs/getting-started/quick-start

Connect the JetstreamStrategy as a microservice transport in your main application file.

```typescript
import { NestFactory } from '@nestjs/core';
import { JetstreamStrategy } from '@horizon-republic/nestjs-jetstream';
import { AppModule } from './app.module';

const bootstrap = async () => {
  const app = await NestFactory.create(AppModule);

  // Retrieve the strategy instance from the DI container
  app.connectMicroservice(
    { strategy: app.get(JetstreamStrategy) },
    { inheritAppConfig: true },
  );

  // Required so the JetStream transport drains in-flight handlers on SIGTERM.
  // Skip this and messages in flight when the pod dies will be redelivered
  // only after `ack_wait` expires — slower recovery, noisier metrics.
  app.enableShutdownHooks();

  await app.startAllMicroservices();
  await app.listen(3000);
};

void bootstrap();
```

--------------------------------

### Clone Repository and Install Dependencies

Source: https://nestjs-jetstream.horizon-republic.dev/docs/development/contributing

Clone your fork of the repository and install the necessary project dependencies using pnpm.

```bash
git clone https://github.com/your-username/nestjs-jetstream.git  

cd nestjs-jetstream  

pnpm install  

```

--------------------------------

### Install NestJS Jetstream with yarn

Source: https://nestjs-jetstream.horizon-republic.dev/docs/getting-started/installation

Use yarn to install the package. Ensure you have Node.js and yarn installed.

```bash
yarn add @horizon-republic/nestjs-jetstream  
```

--------------------------------

### Install msgpackr Peer Dependency

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/classes/MsgpackCodec

Install the optional msgpackr peer dependency using npm or pnpm.

```bash
npm install msgpackr

# or: pnpm add msgpackr
```

--------------------------------

### Gateway Controller Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/getting-started/quick-start

An example NestJS controller demonstrating how to inject the ClientProxy and use emit() for fire-and-forget and broadcast events, and send() for RPC commands.

```typescript
import { Controller, Get, Inject, Param, ParseIntPipe } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { Observable } from 'rxjs';

@Controller('orders')
export class GatewayController {
  constructor(
    @Inject('orders') private readonly client: ClientProxy,
  ) {}

  /** Emit a workqueue event (fire-and-forget, at-least-once delivery). */
  @Get('create')
  createOrder(): Observable<void> {
    return this.client.emit('order.created', { orderId: 42, total: 99.99 });
  }

  /** Emit a broadcast event (all service instances receive it). */
  @Get('broadcast')
  broadcastConfig(): Observable<void> {
    return this.client.emit('broadcast:config.updated', {
      key: 'maintenance',
      value: 'true',
    });
  }

  /** Send an RPC command and wait for a response. */
  @Get(':id')
  getOrder(@Param('id', ParseIntPipe) id: number): Observable<{ id: number; status: string }> {
    return this.client.send<{ id: number; status: string }>('order.get', { id });
  }
}
```

--------------------------------

### Install NestJS Jetstream with npm

Source: https://nestjs-jetstream.horizon-republic.dev/docs/getting-started/installation

Use npm to install the package. Ensure you have Node.js and npm installed.

```bash
npm install @horizon-republic/nestjs-jetstream  
```

--------------------------------

### Install the library

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/migration

Install the necessary packages for migrating to JetStream.

```bash
npm install @horizon-republic/nestjs-jetstream @nats-io/transport-node @nats-io/jetstream  
```

--------------------------------

### Install NestJS Jetstream with pnpm

Source: https://nestjs-jetstream.horizon-republic.dev/docs/getting-started/installation

Use pnpm to install the package. Ensure you have Node.js and pnpm installed.

```bash
pnpm add @horizon-republic/nestjs-jetstream  
```

--------------------------------

### JetstreamRecord Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/classes/JetstreamRecord

Demonstrates how to create and use a JetstreamRecord using the JetstreamRecordBuilder. This example shows setting a message payload, adding a custom header, setting a timeout, and then sending the record using `client.send()`.

```APIDOC
## Example
```typescript
const record = new JetstreamRecordBuilder({ id: 1 })
  .setHeader('x-tenant', 'acme')
  .setTimeout(5000)
  .build();

client.send('get.user', record);
```
```

--------------------------------

### Testing the Application

Source: https://nestjs-jetstream.horizon-republic.dev/docs/getting-started/quick-start

Command-line instructions to start the NestJS application and send various types of messages using curl.

```bash
# Start the app
npm run start:dev

# Send a workqueue event
curl http://localhost:3000/orders/create

# Send a broadcast event
curl http://localhost:3000/orders/broadcast

# Send an RPC command
curl http://localhost:3000/orders/42
```

--------------------------------

### MsgpackCodec Installation

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/custom-codec

msgpackr is an optional peer dependency and should be installed only if you opt into this codec.

```bash
npm install msgpackr
# or:
pnpm add msgpackr
# or:
yarn add msgpackr
```

--------------------------------

### RPC Configuration Examples

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/module-configuration

Examples demonstrating how to configure RPC modes ('core' and 'jetstream') with different timeouts and stream/consumer overrides.

```typescript
// Core mode (default) -- NATS native request/reply  
rpc: { mode: 'core', timeout: 10_000 }  

  
// JetStream mode -- commands persisted in a stream  
rpc: {  
  mode: 'jetstream',  
  timeout: 60_000,  
  stream: { max_age: toNanos(1, 'minutes') },     // stream overrides  
  consumer: { max_deliver: 3 },            // consumer overrides  
}
```

--------------------------------

### Publisher-only Mode - Main Application Setup

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/module-configuration

Example of setting up the main application in publisher-only mode, omitting microservice connection.

```typescript
const bootstrap = async () => {
  const app = await NestFactory.create(AppModule);
  
  // No microservice connection needed in publisher-only mode
  await app.listen(3000);
};

void bootstrap();
```

--------------------------------

### Basic NodeSDK Setup

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability/tracing

This snippet shows the basic setup for an OpenTelemetry Node.js SDK with an OTLP trace exporter. It should be loaded before your AppModule.

```typescript
// tracing.ts — load this BEFORE your AppModule  

import { NodeSDK } from '@opentelemetry/sdk-node';  

import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';  

  

new NodeSDK({  

  serviceName: 'orders-service',  

  traceExporter: new OTLPTraceExporter({ url: 'http://collector:4318/v1/traces' }),  

}).start();  

```

--------------------------------

### All - Full replay on every start (default)

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/ordered-events

Example configuration for the 'All' deliver policy, showing how to set up a Jetstream module with ordered events and a stream max age.

```typescript
JetstreamModule.forRoot({
  name: 'projections',
  servers: ['nats://localhost:4222'],
  ordered: {
    // DeliverPolicy.All is the default -- you can omit this entirely
    stream: {
      max_age: toNanos(7, 'days'), // 7 days
    },
  },
})
```

--------------------------------

### Install prom-client

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability/metrics

Install the optional peer dependency `prom-client` to enable metrics.

```bash
pnpm add prom-client
```

--------------------------------

### Protobuf Codec Implementation

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/custom-codec

A conceptual example of a Protobuf codec implementation for NestJS Jetstream, demonstrating how to wrap generated message classes for strongly-typed schemas and cross-language compatibility. This example is adaptable to various Protobuf libraries like protobufjs, ts-proto, or google-protobuf.

```typescript
import type { Codec } from '@horizon-republic/nestjs-jetstream';  

  

/**  

 * Conceptual example — adapt to your Protobuf library  

 * (protobufjs, ts-proto, google-protobuf, etc.)  

 */  

export class ProtobufCodec implements Codec {  

  constructor(  

    private readonly messageType: {  

      encode(data: unknown): { finish(): Uint8Array };  

      decode(data: Uint8Array): unknown;  

    },  

  ) {}  

  

  encode(data: unknown): Uint8Array {  

    return this.messageType.encode(data).finish();  

  }  

  

  decode(data: Uint8Array): unknown {  

    return this.messageType.decode(data);  

  }  

}
```

--------------------------------

### Conventional Commit Message Examples

Source: https://nestjs-jetstream.horizon-republic.dev/docs/development/contributing

Examples of commit messages following the Conventional Commits specification. Use scopes for better changelog organization.

```git
feat: add support for custom stream configuration  

fix(strategy): remove private logger that shadows Server base class  

docs: update module configuration examples  

```

--------------------------------

### Switching to in-memory streams (Before)

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/stream-migration

Example configuration for using File storage (default) before migration.

```typescript
// Before — File storage (default)  

JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
});  

```

--------------------------------

### Async Hook Error Handling Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/lifecycle-hooks

This example shows how to handle errors in asynchronous hooks safely, ensuring the application continues to run even if an external service like Sentry is unavailable.

```typescript
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  
  },  
}

```

--------------------------------

### Capture Body Configuration Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability/tracing

Example demonstrating how to configure message body capture with a maximum byte limit and subject allowlist.

```typescript
otel: {
  captureBody: { maxBytes: 8192, subjectAllowlist: ['orders.*'] },
}
```

--------------------------------

### Configuring JetStream Mode

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/rpc

Example of configuring the JetStream mode for RPC in NestJS microservices.

```typescript
JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  rpc: { mode: 'jetstream' },
});  


```

--------------------------------

### Decode Error Logging Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/custom-codec

Example log output demonstrating how NestJS Jetstream handles and logs decode errors when a consumer receives a message encoded with an incompatible codec. This helps in diagnosing serialization issues between services.

```text
[Jetstream:EventRouter] Decode error for orders__microservice.ev.order.created: Error: ...  

[Jetstream:RpcRouter]   Decode error for RPC orders__microservice.cmd.get.order: Error: ...  
```

--------------------------------

### Keep your handlers

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/migration

Example of message and event handlers, which remain unchanged.

```typescript
import { Controller } from '@nestjs/common';  
import { EventPattern, MessagePattern, Payload } from '@nestjs/microservices';  
  
@Controller()
export class OrdersController {
  @EventPattern('order.created')  
  handleOrderCreated(@Payload() data: { orderId: string }) {
    // Same code as before. Throws here will trigger JetStream retries.  
  }
  
  @MessagePattern('order.get')  
  getOrder(@Payload() data: { id: string }) {
    return { id: data.id, status: 'shipped' };  
  }

}

```

--------------------------------

### Ordered Consumer Configuration Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/ordered-events

Demonstrates how to configure an ordered consumer, highlighting the workaround for DeliverPolicy.All.

```typescript
ordered: {
  deliverPolicy: DeliverPolicy.All,
}

// This also works (All is the default):
ordered: {}
```

--------------------------------

### Example DLQ service

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/dead-letter-queue

A sample DLQ service that demonstrates how to persist dead letter information, including logging and saving to a repository.

```typescript
import {
  Injectable,
  Logger,
} from '@nestjs/common';
import {
  DeadLetterInfo,
} from '@horizon-republic/nestjs-jetstream';
import {
  DlqRepository,
} from './dlq.repository';

@Injectable()
export class DlqService {
  private readonly logger = new Logger(DlqService.name);

  constructor(private readonly repository: DlqRepository) {}

  async persist(info: DeadLetterInfo): Promise<void> {
    this.logger.error(
      `Dead letter on ${info.subject} (stream: ${info.stream}, seq: ${info.streamSequence})`,
      info.error,
    );

    // Store in your database for later investigation or replay
    await this.repository.save({
      subject: info.subject,
      payload: JSON.stringify(info.data),
      error: info.error instanceof Error ? info.error.message : String(info.error),
      deliveryCount: info.deliveryCount,
      stream: info.stream,
      streamSequence: info.streamSequence,
      occurredAt: info.timestamp,
    });
  }
}
```

--------------------------------

### Sentry Integration

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability/tracing

Example of initializing Sentry with automatic OpenTelemetry setup and trace sampling rate.

```typescript
// Sentry — automatic OTel setup with tracesSampleRate  

import * as Sentry from '@sentry/node';  
Sentry.init({ dsn: process.env.SENTRY_DSN, tracesSampleRate: 1.0 });  

```

--------------------------------

### Run Linting and Tests

Source: https://nestjs-jetstream.horizon-republic.dev/docs/development/contributing

Verify your changes by running linting checks and all tests. Ensure all checks pass before pushing.

```bash
pnpm lint  

pnpm test  

```

--------------------------------

### New - Start from "now", skip backlog

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/ordered-events

Example configuration for the 'New' deliver policy, setting the deliver policy to DeliverPolicy.New for a dashboard consumer.

```typescript
import { DeliverPolicy } from '@nats-io/jetstream';
  
JetstreamModule.forRoot({
  name: 'dashboard',
  servers: ['nats://localhost:4222'],
  ordered: {
    deliverPolicy: DeliverPolicy.New,
  },
})
```

--------------------------------

### Structured Logging Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/lifecycle-hooks

This snippet demonstrates how to emit structured JSON logs for all lifecycle events using `JetstreamModule.forRoot` and custom logging logic.

```typescript
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'),  
  },  
})  


```

--------------------------------

### Metrics example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/interfaces/JetstreamModuleOptions

Example of enabling built-in Prometheus metrics.

```typescript
JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  metrics: true,
})
```

--------------------------------

### Create and Send JetstreamRecord

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/classes/JetstreamRecord

Demonstrates how to create a JetstreamRecord using the builder pattern, set headers and timeouts, and then send it using the client. This is useful for sending messages with custom metadata and configurations.

```typescript
const record = new JetstreamRecordBuilder({ id: 1 })  
  .setHeader('x-tenant', 'acme')  
  .setTimeout(5000)  
  .build();  
  
client.send('get.user', record);
```

--------------------------------

### onDeadLetter Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/interfaces/JetstreamModuleOptions

Example of how to use the onDeadLetter callback to persist dead-lettered messages.

```typescript
JetstreamModule.forRootAsync({
  name: 'my-service',
  imports: [DlqModule],
  inject: [DlqService, JETSTREAM_CONNECTION],
  useFactory: (dlqService, connection) => ({
    servers: ['nats://localhost:4222'],
    onDeadLetter: async (info) => {
      await dlqService.persist(info);
    },
  }),
})
```

--------------------------------

### DLQ stream example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/interfaces/JetstreamModuleOptions

Example of configuring a Dead-Letter Queue (DLQ) stream.

```typescript
JetstreamModule.forRootAsync({
  name: 'my-service',
  servers: ['nats://localhost:4222'],
  dlq: {
    stream: {
      max_age: toNanos(30, 'days'),
    },
  },
})
```

--------------------------------

### MsgpackCodec Constructor

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/classes/MsgpackCodec

Initializes a new MsgpackCodec instance with a provided msgpackr Packr instance.

```APIDOC
## new MsgpackCodec(packr)

### Description
Initializes a new MsgpackCodec instance.

### Parameters
#### packr
- **packr** (PackrLike) - The msgpackr Packr instance to use for encoding and decoding.

### Returns
- `MsgpackCodec` - A new instance of MsgpackCodec.
```

--------------------------------

### Terminate Message Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/handler-context

Example of using ctx.terminate() to permanently reject a message that is no longer relevant.

```typescript
@EventPattern('order.process')
async handle(@Payload() data: OrderDto, @Ctx() ctx: RpcContext) {
  const order = await this.orderService.find(data.orderId);

  if (order.status === 'cancelled') {
    ctx.terminate('Order already cancelled');
    return;
  }

  await this.process(order);
}
```

--------------------------------

### Custom Consumer Configuration Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/events

Shows how to override default consumer settings such as max_deliver, ack_wait, and max_ack_pending in the JetstreamModule configuration.

```typescript
JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  events: {
    consumer: {
      max_deliver: 5,              // 5 retries instead of 3
      ack_wait: toNanos(30, 'seconds'),  // 30s ack timeout instead of 10s
      max_ack_pending: 50,         // Limit in-flight messages to 50
    },
  },
})
```

--------------------------------

### forRootAsync() useClass Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/module-configuration

Example of using `forRootAsync()` with `useClass` to provide a configuration class for JetStream module.

```typescript
JetstreamModule.forRootAsync({
  name: 'orders',
  useClass: NatsConfigService,
})
```

--------------------------------

### Business Retry Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/handler-context

Example of using ctx.retry() for a business-level retry when conditions are not met for immediate processing.

```typescript
@EventPattern('order.fulfill')
async handle(@Payload() data: FulfillDto, @Ctx() ctx: RpcContext) {
  if (!await this.inventoryService.isAvailable()) {
    ctx.retry({ delayMs: 30_000 }); // try again in 30 seconds
    return;
  }

  await this.fulfillOrder(data);
  // auto-ack — no flags set
}
```

--------------------------------

### JsonCodec Usage Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/classes/JsonCodec

Instantiate JsonCodec and use its encode and decode methods to serialize and deserialize data. Ensure data is JSON-serializable for encoding and be aware that decoding returns unknown.

```typescript
const codec = new JsonCodec();

const bytes = codec.encode({ hello: 'world' });

const data = codec.decode(bytes); // { hello: 'world' }
```

--------------------------------

### Meta structure examples

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/handler-metadata

Examples of different meta object structures for various use cases.

```json
// HTTP routing
{ meta: { http: { method: 'POST', path: '/orders' } } }

// Auth requirements
{ meta: { http: { method: 'GET', path: '/orders/:id' }, auth: 'bearer' } }

// Feature flags
{ meta: { feature: 'orders-v2', canary: true } }

// Documentation hints
{ meta: { description: 'Creates a new order', tags: ['orders'] } }
```

--------------------------------

### Publishing Headers from Python with OpenTelemetry

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/header-contract

Injects W3C Trace Context headers into a NATS message for cross-language tracing. Requires OpenTelemetry to be set up.

```python
from opentelemetry import propagate
from nats.aio.msg import Msg

headers = {}
propagate.inject(headers)

await js.publish(
    subject="orders__microservice.ev.orders.created",
    payload=payload,
    headers=headers,
)
```

--------------------------------

### OtelOptions Interface

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api

Options for OpenTelemetry integration.

```APIDOC
## Interface: `OtelOptions`

Options for OpenTelemetry integration.

### Properties

*   `tracerName?: string`
```

--------------------------------

### forRootAsync() useExisting Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/module-configuration

Example of using `forRootAsync()` with `useExisting` to reuse an already registered provider for JetStream configuration.

```typescript
JetstreamModule.forRootAsync({
  name: 'orders',
  imports: [NatsConfigModule],
  useExisting: NatsConfigService,
})
```

--------------------------------

### Writing events in CQRS example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/ordered-events

Example of publishing events to an ordered stream within a CQRS writer service.

```typescript
import { lastValueFrom } from 'rxjs';

@Injectable()
export class WriterService {
  constructor(@Inject('projections') private readonly client: ClientProxy) {}

  async recordEvent(event: DomainEvent) {
    // Publish to the ordered stream
    await lastValueFrom(
      this.client.emit(`ordered:${event.type}`, event.payload),
    );
  }
}
```

--------------------------------

### Run NATS Server with JetStream using Docker

Source: https://nestjs-jetstream.horizon-republic.dev/docs/getting-started/installation

Start a NATS server with JetStream enabled locally using Docker. This command maps the NATS port and enables JetStream.

```bash
docker run -d --name nats -p 4222:4222 nats:2.12 -js  
```

--------------------------------

### Custom Error Classifier Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability/tracing

Example of how to implement a custom error classifier for OpenTelemetry spans in NestJS Jetstream.

```typescript
otel: {
  errorClassifier: (err) => {
    if (err instanceof MyDomainError) return 'expected';
    if (typeof err === 'object' && err !== null && 'code' in err && /^BIZ_/.test((err as { code: string }).code)) {
      return 'expected';
    }
    return 'unexpected';
  },
}
```

--------------------------------

### Publishing Headers from Go with OpenTelemetry

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/header-contract

Injects W3C Trace Context headers into a NATS message for end-to-end tracing. Ensure OpenTelemetry is configured.

```go
import (
  "go.opentelemetry.io/otel"
  "go.opentelemetry.io/otel/propagation"
  "github.com/nats-io/nats.go"
)

ctx, span := tracer.Start(ctx, "create-order")
deferspan.End()

headers := nats.Header{}
otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(headers))

js.PublishMsg(&nats.Msg{
  Subject: "orders__microservice.ev.orders.created",
  Data:    payload,
  Header:  headers,
})
```

--------------------------------

### check() - plain status object example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/health-checks

Example of using the check() method in a NestJS controller to expose a health endpoint.

```typescript
import {
  Controller,
  Get,
} from '@nestjs/common';
import { JetstreamHealthIndicator } from '@horizon-republic/nestjs-jetstream';

@Controller('health')
export class HealthController {
  constructor(private readonly jetstream: JetstreamHealthIndicator) {}

  @Get()
  async check() {
    const status = await this.jetstream.check();

    return {
      status: status.connected ? 'ok' : 'error',
      nats: {
        connected: status.connected,
        server: status.server,
        latency: status.latency,
      },
    };
  }
}
```

--------------------------------

### Switching to in-memory streams (After migration enabled)

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/stream-migration

Example configuration for using Memory storage with migration enabled.

```typescript
// After — Memory storage with migration enabled  

JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  allowDestructiveMigration: true,
  events: { stream: { storage: StorageType.Memory } },
  broadcast: { stream: { storage: StorageType.Memory } },
  ordered: { stream: { storage: StorageType.Memory } },
});  

```

--------------------------------

### Implement MsgPackCodec with Codec Interface

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/interfaces/Codec

Provides an example implementation of the Codec interface using MsgPack for serialization and deserialization. This class handles converting data to and from Uint8Array for NATS transmission.

```typescript
import { encode, decode } from '@msgpack/msgpack';  

class MsgPackCodec implements Codec {  
  encode(data: unknown): Uint8Array {  
    return encode(data);  
  }  
  decode(data: Uint8Array): unknown {  
    return decode(data);  
  }  
}
```

--------------------------------

### System Under Test (sut) Initialization

Source: https://nestjs-jetstream.horizon-republic.dev/docs/development/testing

Initialize the system under test (sut) before each test. The 'sut' variable should always refer to the primary object being tested.

```typescript
let sut: StreamProvider;

beforeEach(() => {
  sut = new StreamProvider(mockOptions, mockConnection);
});
```

--------------------------------

### forFeature() Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/module-configuration

Example of using `forFeature()` in a feature module (`OrdersModule`) to register lightweight `JetstreamClient` proxies for specific services.

```typescript
import {
  Module
} from '@nestjs/common';
import {
  JetstreamModule
} from '@horizon-republic/nestjs-jetstream';
import { OrdersService } from './orders.service';

@Module({
  imports: [
    JetstreamModule.forFeature({ name: 'users' }),
    JetstreamModule.forFeature({ name: 'payments' }),
  ],
  providers: [OrdersService],
  exports: [OrdersService],
})
export class OrdersModule {}
```

--------------------------------

### Custom Stream Configuration Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/events

Illustrates how to override default stream settings like max_age, max_bytes, and duplicate_window in the JetstreamModule configuration.

```typescript
JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  events: {
    stream: {
      max_age: toNanos(14, 'days'), // 14 days instead of 7
      max_bytes: 10 * 1024 * 1024 * 1024,        // 10 GB instead of 5 GB
      duplicate_window: toNanos(5, 'minutes'),     // 5 min dedup window instead of 2 min
    },
  },
})
```

--------------------------------

### Handling the RPC command (server side)

Source: https://nestjs-jetstream.horizon-republic.dev/docs/patterns/rpc

Example of how a server-side controller handles an incoming RPC command using NestJS microservices.

```typescript
import { Controller } from '@nestjs/common';  

import { MessagePattern, Payload } from '@nestjs/microservices';  

  

@Controller()  

export class OrdersController {  

  @MessagePattern('get.order')  

  handleGetOrder(@Payload() data: { id: string }): Order {  

    return this.ordersService.findById(data.id);  

  }  

}  


```

--------------------------------

### Built-in DLQ stream configuration

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/dead-letter-queue

Example of configuring the built-in DLQ stream using `JetstreamModule.forRoot()` with custom stream options.

```typescript
JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  dlq: {
    stream: {
      max_age: toNanos(30, 'days'), // how long dead letters are retained
    },
  },
})
```

--------------------------------

### forRootAsync() useFactory Example

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/module-configuration

Example of using `forRootAsync()` with `useFactory` to dynamically configure JetStream module using environment variables via `ConfigModule`.

```typescript
import {
  Module
} from '@nestjs/common';
import {
  ConfigModule,
  ConfigService
} from '@nestjs/config';
import {
  JetstreamModule
} from '@horizon-republic/nestjs-jetstream';

@Module({
  imports: [
    ConfigModule.forRoot(),
    JetstreamModule.forRootAsync({
      name: 'orders',
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: (config: ConfigService) => {
        const mode = config.get<'core' | 'jetstream'>('RPC_MODE', 'core');

        return {
          servers: [config.getOrThrow('NATS_URL')],
          rpc: mode === 'jetstream' ? { mode, timeout: 60_000 } : { mode },
          shutdownTimeout: config.get('SHUTDOWN_TIMEOUT', 10_000),
        };
      },
    }),
  ],
})
export class AppModule {}
```

--------------------------------

### OtelOptions Methods

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/interfaces/OtelOptions

Customizable hooks for OpenTelemetry spans within the Jetstream module.

```APIDOC
## Methods

### publishHook()? 

> `optional` **publishHook**(`span`, `ctx`): `void`

Invoked after a publish span has been started and before the actual publish call executes. Use to enrich the span with custom attributes. Must be synchronous — thrown errors are caught and logged at debug level without affecting the publish path.
#### Parameters​
##### span
`Span`
##### ctx
`JetstreamPublishContext`
#### Returns
`void`

* * *

### consumeHook()? 

> `optional` **consumeHook**(`span`, `ctx`): `void`

Invoked after a consume span has been started and before the handler is dispatched. Use to enrich the span with custom attributes derived from the incoming message or handler metadata.
#### Parameters​
##### span
`Span`
##### ctx
`JetstreamConsumeContext`
#### Returns
`void`

* * *

### responseHook()? 

> `optional` **responseHook**(`span`, `ctx`): `void`

Invoked once an operation's outcome is known (success, error, or timeout) and the span status has been set, just before the span ends. Fires for publish, consume, and RPC client spans.

* * *

### errorClassifier()? 

> `optional` **errorClassifier**(`err`): `ErrorClassification`

Classify a thrown error as `'expected'` (business error, part of the RPC contract) or `'unexpected'` (infrastructure failure or bug). Drives OpenTelemetry span status and attributes only.
  * `'expected'` → span status `OK` with `jetstream.rpc.reply.has_error` and `jetstream.rpc.reply.error.code` attributes
  * `'unexpected'` → span status `ERROR` with `span.recordException(err)`

Reply envelopes delivered to RPC clients are identical in both cases. This classification affects only the observability artifact.
The default recognizes NestJS-idiomatic primitives for business errors: `RpcException` and `HttpException`. Teams with custom error hierarchies override to recognize their own types.
#### Parameters​
##### err
`unknown`
#### Returns
`ErrorClassification`
#### Example
```
errorClassifier: (err) => {
    if (err instanceof MyDomainError) return 'expected';
    if (err?.code?.startsWith('BIZ_')) return 'expected';
    return 'unexpected';
  }
```
```

--------------------------------

### Example of enabling metrics and custom hooks

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability/metrics

This example demonstrates how to enable metrics and define custom hooks for dead-letter events within the JetstreamModule configuration.

```typescript
JetstreamModule.forRoot({
  name: 'orders',
  servers: ['nats://localhost:4222'],
  metrics: true,
  hooks: {
    [TransportEvent.DeadLetter]: async (info) => {
      await sentry.captureMessage(`Dead letter: ${info.subject}`, { extra: info });
    },
  },
});
```

--------------------------------

### Constants

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api

Default configuration values and constants used in the library.

```APIDOC
## Constants

### `DEFAULT_BROADCAST_CONSUMER_CONFIG`

Default configuration for broadcast consumers.

### `DEFAULT_BROADCAST_STREAM_CONFIG`

Default configuration for broadcast streams.

### `DEFAULT_COMMAND_CONSUMER_CONFIG`

Default configuration for command consumers.

### `DEFAULT_COMMAND_STREAM_CONFIG`

Default configuration for command streams.

### `DEFAULT_DLQ_STREAM_CONFIG`

Default configuration for Dead Letter Queue streams.

### `DEFAULT_EVENT_CONSUMER_CONFIG`

Default configuration for event consumers.

### `DEFAULT_EVENT_STREAM_CONFIG`

Default configuration for event streams.

### `DEFAULT_JETSTREAM_RPC_TIMEOUT`

Default RPC timeout for JetStream operations.

### `DEFAULT_METADATA_BUCKET`

Default name for the metadata bucket.

### `DEFAULT_METADATA_HISTORY`

Default history size for metadata.

### `DEFAULT_METADATA_REPLICAS`

Default number of replicas for metadata storage.

### `DEFAULT_METADATA_TTL`

Default Time-To-Live for metadata.

### `DEFAULT_ORDERED_STREAM_CONFIG`

Default configuration for ordered streams.

### `DEFAULT_RPC_TIMEOUT`

Default RPC timeout.

### `DEFAULT_SHUTDOWN_TIMEOUT`

Default shutdown timeout.

### `DEFAULT_TRACES`

Default trace configuration.

### `JETSTREAM_CODEC`

Injection token for the JetStream codec.

### `JETSTREAM_CONNECTION`

Injection token for the JetStream connection.

### `JETSTREAM_OPTIONS`

Injection token for JetStream module options.

### `MIN_METADATA_TTL`

Minimum Time-To-Live for metadata.

### `RESERVED_HEADERS`

Reserved headers in JetStream messages.

### `TRACER_NAME`

Name of the tracer used for JetStream operations.
```

--------------------------------

### Example of using a dedicated Registry for multi-registry deployments

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability/metrics

This example shows how to configure the JetstreamModule to use a specific `prom-client` Registry instance, useful for applications exposing multiple metrics endpoints.

```typescript
import { Registry } from 'prom-client';

const tenantRegister = new Registry();

JetstreamModule.forRoot({
  // ...
  metrics: { register: tenantRegister },
});
```

--------------------------------

### build()

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/classes/JetstreamRecordBuilder

Builds and returns the immutable JetstreamRecord instance, ready for sending or emitting.

```APIDOC
## build()

> **build**(): `JetstreamRecord`<`TData`>

Build the immutable JetstreamRecord.

#### Returns​
`JetstreamRecord`<`TData`>

A frozen record ready to pass to `client.send()` or `client.emit()`.
```

--------------------------------

### Quick start: Tracing and Metrics

Source: https://nestjs-jetstream.horizon-republic.dev/docs/observability

This code snippet demonstrates how to enable both tracing with OpenTelemetry and Prometheus metrics in a NestJS application using `@willsoto/nestjs-prometheus` and `@horizon-republic/nestjs-jetstream`.

```typescript
// 1. Tracing — register any OpenTelemetry SDK before your AppModule loads.  
// Sentry, Datadog, NewRelic, or @opentelemetry/sdk-node — pick one.  
import { NodeSDK } from '@opentelemetry/sdk-node';  
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';  
new NodeSDK({  
  serviceName: 'orders-service',  
  traceExporter: new OTLPTraceExporter({ url: 'http://collector:4318/v1/traces' }),  
}).start();  
  
// 2. Metrics — enable in forRoot.  
import { PrometheusModule } from '@willsoto/nestjs-prometheus';  
import { JetstreamModule } from '@horizon-republic/nestjs-jetstream';  
  
@Module({  
  imports: [  
    PrometheusModule.register(),  
    JetstreamModule.forRoot({  
      name: 'orders',  
      servers: ['nats://localhost:4222'],  
      metrics: true,  
    }),  
  ],  
})  
export class AppModule {}  
```

--------------------------------

### JetstreamRecordBuilder Constructor

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/api/classes/JetstreamRecordBuilder

Initializes a new instance of JetstreamRecordBuilder. It can optionally take initial data for the record.

```APIDOC
## Constructor

> **new JetstreamRecordBuilder** <`TData`>(`data?`): `JetstreamRecordBuilder`<`TData`>

#### Parameters​
##### data?​
`TData`

#### Returns​
`JetstreamRecordBuilder`<`TData`>
```

--------------------------------

### Module Configuration

Source: https://nestjs-jetstream.horizon-republic.dev/

Example of configuring the JetstreamModule in your NestJS application.

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

@Module({
  imports: [
    JetstreamModule.forRoot({
      servers: ['nats://localhost:4222'],
    }),
  ],
})
export class AppModule {}
```

--------------------------------

### Verify NATS Server and Connectivity

Source: https://nestjs-jetstream.horizon-republic.dev/docs/guides/troubleshooting

Use these commands to check if the NATS server is running and test basic connectivity to the server.

```bash
# Verify NATS is running  
nats-server --version  
docker ps | grep nats  
  
# Test connectivity  
nats server check connection --server nats://localhost:4222  
```

--------------------------------

### Broadcast max_age override

Source: https://nestjs-jetstream.horizon-republic.dev/docs/reference/release-notes

Example of how to override the default broadcast max_age.

```yaml
broadcast: { stream: { max_age: toNanos(1, 'days') } }
```