# API (/integrations/nestjs/reference/api)



This page is the API lookup for `@kubemq/nestjs-transport` — the `KubeMQModule` static methods,
the five handler decorators and their options, the per-pattern contexts, the codec exports, and
the module constants and enums. For connection options and configuration interfaces see
[Configuration](/integrations/nestjs/reference/configuration); for the error catalog see [error codes](/integrations/nestjs/reference/error-codes). For
task-oriented walkthroughs, start with [Getting Started](/integrations/nestjs/tutorials/getting-started); this page is the
lookup table you return to.

## Module API [#module-api]

`KubeMQModule` is a dynamic module exposing seven static factory methods. `forRoot` / `forRootAsync` establish the shared connection; `register` / `registerAsync` create named client proxies for injection; `forFeature` / `forFeatureAsync` create channel-prefixed scoped clients on top of the shared connection; `forTest` swaps in mocks.

| Method                                  | Purpose                                                                                                      |
| --------------------------------------- | ------------------------------------------------------------------------------------------------------------ |
| `KubeMQModule.forRoot(options)`         | Register the global, shared KubeMQ connection from a static `KubeMQModuleOptions` (global by default)        |
| `KubeMQModule.forRootAsync(options)`    | Same, but resolve options asynchronously via `useFactory` / `useClass` / `useExisting`                       |
| `KubeMQModule.register(options)`        | Create a named `KubeMQClientProxy` for `@Inject`-ed message sending                                          |
| `KubeMQModule.registerAsync(options)`   | Same, with async option resolution                                                                           |
| `KubeMQModule.forFeature(options)`      | Create a `ScopedKubeMQClientProxy` that prepends `channelPrefix` to every channel, reusing the shared client |
| `KubeMQModule.forFeatureAsync(options)` | Same, with async resolution of the feature options                                                           |
| `KubeMQModule.forTest(options?)`        | Provide a `MockKubeMQClient` / `MockKubeMQServer` for unit tests without a broker                            |

```typescript title="app.module.ts"
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { KubeMQModule } from '@kubemq/nestjs-transport';

@Module({
  imports: [
    // Shared connection (global by default)
    KubeMQModule.forRoot({
      address: 'localhost:50000',
      clientId: 'orders-app',
      isGlobal: true,
    }),

    // Async shared connection from ConfigService
    KubeMQModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (config: ConfigService) => ({
        address: config.getOrThrow('KUBEMQ_ADDRESS'),
        clientId: config.get('KUBEMQ_CLIENT_ID'),
      }),
      inject: [ConfigService],
    }),

    // Named client proxy for injection
    KubeMQModule.register({
      name: 'ORDER_SERVICE',
      address: 'localhost:50000',
      clientId: 'order-client',
    }),

    // Channel-scoped client: every channel is prefixed with "orders"
    KubeMQModule.forFeature({ name: 'ORDER_BUS', channelPrefix: 'orders' }),
  ],
})
export class AppModule {}
```

<Callout type="info">
  `forRoot` and `forRootAsync` are global by default — `isGlobal` defaults to `true` (it is treated as global unless explicitly set to `false`). `register`, `registerAsync`, `forFeature`, and `forFeatureAsync` are scoped to the importing module. `forTest` is global unless `isGlobal: false` is passed.
</Callout>

The option interfaces accepted by these methods (`KubeMQModuleOptions`, `KubeMQClientOptions`, and the async variants) are documented in [Configuration](/integrations/nestjs/reference/configuration).

## Decorators [#decorators]

Five custom decorators replace the manual `@MessagePattern` / `@EventPattern` metadata you would otherwise hand-configure for each pattern. Each accepts a single channel string **or an array of channels**, plus an optional options object. They are exported from the package root (`src/decorators/index.ts`).

| Decorator                            | Pattern type  | Semantics                   | Accepted option keys                                                                        |
| ------------------------------------ | ------------- | --------------------------- | ------------------------------------------------------------------------------------------- |
| `@CommandHandler(channel, opts?)`    | `command`     | Request-reply               | base + `validate`                                                                           |
| `@QueryHandler(channel, opts?)`      | `query`       | Request-reply               | base + `validate`                                                                           |
| `@EventHandler(channel, opts?)`      | `event`       | Fire-and-forget             | base + `validate`                                                                           |
| `@EventStoreHandler(channel, opts?)` | `event_store` | Fire-and-forget, persistent | base + `startFrom`, `startValue`, `validate`                                                |
| `@QueueHandler(channel, opts?)`      | `queue`       | At-least-once delivery      | base + `manualAck`, `maxMessages`, `waitTimeoutSeconds`, `batch`, `idempotency`, `validate` |

```typescript title="handlers.ts"
import {
  CommandHandler,
  QueryHandler,
  EventHandler,
  EventStoreHandler,
  QueueHandler,
} from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

@CommandHandler('orders.create', { group: 'order-writers', maxConcurrent: 5 })
handleCreate(@Payload() data: CreateOrderDto) { /* ... */ }

@QueryHandler('orders.get')
handleGet(@Payload() data: { id: string }) { /* ... */ }

// A single handler can subscribe to multiple channels
@EventHandler(['orders.updated', 'orders.created'])
onChange(@Payload() data: OrderEvent) { /* ... */ }

@EventStoreHandler('orders.history', { startFrom: 'first' })
replay(@Payload() data: unknown) { /* ... */ }

@QueueHandler('orders.process', { manualAck: true, maxMessages: 5 })
process(@Payload() data: unknown, @Ctx() ctx) { /* ... */ }
```

## Decorator Options [#decorator-options]

All decorator option interfaces extend `KubeMQHandlerBaseOptions`. The base keys apply to every handler; each pattern then adds its own extras. Defaults below come from `src/interfaces/decorator-options.interface.ts` and `src/interfaces/handler-metadata.interface.ts`.

### Base options (all handlers) [#base-options-all-handlers]

<TypeTable
  type="{
  group: {
    type: 'string',
    description: 'Consumer group — handlers sharing a group load-balance message delivery.',
  },
  maxConcurrent: {
    type: 'number',
    description: 'Max concurrent handler executions. Excess messages queue in an internal FIFO.',
  },
  concurrency: {
    type: 'number',
    description: 'Alias for maxConcurrent. When both are set, concurrency takes precedence.',
  },
  maxQueueDepth: {
    type: 'number',
    default: '1000',
    description: 'Max depth of the internal FIFO backpressure queue when concurrency is limited.',
  },
  deadLetterChannel: {
    type: 'string',
    description: 'Channel that failed messages are routed to after retries are exhausted.',
  },
  maxRetries: {
    type: 'number',
    description: 'Max retry attempts before DLQ routing. Defaults to 0 when deadLetterChannel is unset; 3 when it is set and maxRetries is omitted.',
  },
  validate: {
    type: 'ClassConstructor',
    description: 'DTO class used for class-validator validation of the incoming payload.',
  },
}"
/>

### Events Store extras (`@EventStoreHandler`) [#events-store-extras-eventstorehandler]

<TypeTable
  type="{
  startFrom: {
    type: &#x22;EventStoreStartFrom&#x22;,
    description: &#x22;Replay start position: 'new' | 'first' | 'last' | 'sequence' | 'time' | 'timeDelta', or the numeric equivalents 1-6.&#x22;,
  },
  startValue: {
    type: 'number',
    description: 'Sequence number or time value used when startFrom is sequence/time/timeDelta.',
  },
}"
/>

### Queue extras (`@QueueHandler`) [#queue-extras-queuehandler]

<TypeTable
  type="{
  manualAck: {
    type: 'boolean',
    default: 'false',
    description: 'When true, the context exposes ack()/nack()/reQueue() and the message is not auto-acked.',
  },
  maxMessages: {
    type: 'number',
    description: 'Max messages received per poll.',
  },
  waitTimeoutSeconds: {
    type: 'number',
    description: 'Poll wait timeout in seconds.',
  },
  batch: {
    type: 'boolean',
    default: 'false',
    description: 'When true, the handler receives the full batch of messages as a KubeMQQueueBatchContext.',
  },
  idempotency: {
    type: '{ ttlSeconds?: number; maxCacheSize?: number }',
    description: 'Deduplication window. ttlSeconds defaults to 300; maxCacheSize defaults to 10000. (QueueHandler only in v1.0.)',
  },
}"
/>

```typescript title="decorator-options.ts"
// Replay an events-store stream from sequence 100
@EventStoreHandler('orders.history', { startFrom: 'sequence', startValue: 100 })

// Manual-ack queue with bounded concurrency and a DLQ after 3 retries
@QueueHandler('orders.process', {
  manualAck: true,
  maxMessages: 5,
  waitTimeoutSeconds: 60,
  concurrency: 4,
  deadLetterChannel: 'orders.dlq',
  maxRetries: 3,
})

// Batch processing with idempotent dedup
@QueueHandler('orders.batch', {
  batch: true,
  maxMessages: 10,
  idempotency: { ttlSeconds: 600, maxCacheSize: 50_000 },
})
```

## Contexts [#contexts]

The object bound by `@Ctx()` is a typed context that extends `BaseRpcContext`. `KubeMQContext` is the base; per-pattern contexts add fields. Pick the context type that matches the handler decorator.

| Context                   | Adds         | Notable members                                                                                           |
| ------------------------- | ------------ | --------------------------------------------------------------------------------------------------------- |
| `KubeMQContext`           | base         | `channel`, `id`, `timestamp`, `tags`, `metadata`, `patternType`, `getCorrelationId()`, `getCausationId()` |
| `KubeMQCommandContext`    | command      | `fromClientId`, `replyChannel` (re-exported from the merged request context)                              |
| `KubeMQQueryContext`      | query        | `fromClientId`, `replyChannel` (re-exported from the merged request context)                              |
| `KubeMQRequestContext`    | union        | Shared request context backing both command and query                                                     |
| `KubeMQEventStoreContext` | events store | `sequence`                                                                                                |
| `KubeMQQueueContext`      | queue        | `sequence`, `receiveCount`, `isReRouted`, `reRouteFromQueue`, `ack()`, `nack()`, `reQueue(channel)`       |
| `KubeMQQueueBatchContext` | batch queue  | `size`, `getContexts()`, `getContext(index)`, `ackAll()`, `nackAll()`                                     |

```typescript title="contexts.ts"
import {
  EventStoreHandler,
  QueueHandler,
  KubeMQEventStoreContext,
  KubeMQQueueContext,
} from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

@EventStoreHandler('orders.history', { startFrom: 'first' })
onHistory(@Payload() data: unknown, @Ctx() ctx: KubeMQEventStoreContext) {
  console.log(`Event #${ctx.sequence} on ${ctx.channel}`);
}

@QueueHandler('orders.process', { manualAck: true })
async onProcess(@Payload() data: unknown, @Ctx() ctx: KubeMQQueueContext) {
  console.log(`Delivery #${ctx.receiveCount}, re-routed: ${ctx.isReRouted}`);
  try {
    await handle(data);
    ctx.ack();
  } catch {
    ctx.reQueue('orders.dlq'); // or ctx.nack()
  }
}
```

<Callout type="warn">
  `ack()`, `nack()`, and `reQueue()` (and the batch `ackAll()` / `nackAll()`) are only available when the handler is declared with `{ manualAck: true }`. Calling them in auto-ack mode throws an `Error`.
</Callout>

## Codecs [#codecs]

The default codec is JSON. Swap in MessagePack or Protobuf by passing a serializer/deserializer pair to the server, client, or CQRS options. Custom codecs implement the `KubeMQSerializer` / `KubeMQDeserializer` interfaces.

| Export                                              | Description                                     |
| --------------------------------------------------- | ----------------------------------------------- |
| `KubeMQSerializer`                                  | Interface: `serialize(value): Uint8Array`       |
| `KubeMQDeserializer`                                | Interface: `deserialize(data, tags?): any`      |
| `JsonSerializer` / `JsonDeserializer`               | Default JSON codec pair                         |
| `MessagePackSerializer` / `MessagePackDeserializer` | MessagePack codec (requires `@msgpack/msgpack`) |
| `ProtobufSerializer` / `ProtobufDeserializer`       | Protobuf codec (requires `protobufjs`)          |

```typescript title="codecs.ts"
import { KubeMQServer, MessagePackSerializer, MessagePackDeserializer } from '@kubemq/nestjs-transport';

new KubeMQServer({
  address: 'localhost:50000',
  serializer: new MessagePackSerializer(),
  deserializer: new MessagePackDeserializer(),
});
```

<Callout type="warn">
  The sender and receiver must use the same codec. If a `MessagePackSerializer` produces bytes that a `JsonDeserializer` tries to parse, a `SerializationError` is thrown on the receiving side.
</Callout>

## Constants & Enums [#constants--enums]

Exported tokens, metadata keys, message tags, and enums. The injection token and metadata-key constants come from `src/constants.ts`; `KubeMQStatus` is from `src/events/kubemq.events.ts`; `EventStoreStartFrom` is the start-position type.

| Export                    | Type     | Value / shape                                                                                                    |
| ------------------------- | -------- | ---------------------------------------------------------------------------------------------------------------- |
| `KUBEMQ_TRANSPORT`        | `string` | `'kubemq'` — transport identifier                                                                                |
| `KUBEMQ_MODULE_OPTIONS`   | `string` | `'KUBEMQ_MODULE_OPTIONS'` — injection token for module options                                                   |
| `KUBEMQ_HANDLER_METADATA` | `string` | `'kubemq:handler'` — metadata key for decorator options                                                          |
| `TAG_PATTERN`             | `string` | `'nestjs:pattern'`                                                                                               |
| `TAG_ID`                  | `string` | `'nestjs:id'`                                                                                                    |
| `TAG_TYPE`                | `string` | `'nestjs:type'`                                                                                                  |
| `TAG_CONTENT_TYPE`        | `string` | `'nestjs:content-type'`                                                                                          |
| `TAG_CORRELATION_ID`      | `string` | `'x-correlation-id'`                                                                                             |
| `TAG_CAUSATION_ID`        | `string` | `'x-causation-id'`                                                                                               |
| `TAG_IDEMPOTENCY_KEY`     | `string` | `'x-idempotency-key'`                                                                                            |
| `TAG_TRACEPARENT`         | `string` | `'traceparent'`                                                                                                  |
| `TAG_TRACESTATE`          | `string` | `'tracestate'`                                                                                                   |
| `KubeMQPatternType`       | type     | `'command' \| 'query' \| 'event' \| 'event_store' \| 'queue'`                                                    |
| `KubeMQStatus`            | enum     | `DISCONNECTED` `'disconnected'`, `RECONNECTING` `'reconnecting'`, `CONNECTED` `'connected'`, `CLOSED` `'closed'` |
| `EventStoreStartFrom`     | type     | `'new' \| 'first' \| 'last' \| 'sequence' \| 'time' \| 'timeDelta' \| 1 \| 2 \| 3 \| 4 \| 5 \| 6`                |

<Callout type="info">
  There is no exported `KUBEMQ_CLIENT_TOKEN` constant — choose your own string or symbol token for `register` / `forTest`. When `forTest` is called without a `name`, it defaults the injection token to the string `'KUBEMQ_SERVICE'`.
</Callout>

## See Also [#see-also]

<Cards>
  <Card title="Configuration" href="/integrations/nestjs/reference/configuration" description="Package facts, connection options, and every configuration interface." />

  <Card title="Error codes" href="/integrations/nestjs/reference/error-codes" description="The exported error types and the troubleshooting / FAQ table." />

  <Card title="Usage" href="/integrations/nestjs/how-to/usage" description="The decorators and KubeMQRecord builder in task-oriented examples." />
</Cards>