# NatsBridge API

EventBridge implementation for NATS core messaging with optional JetStream.

---
Canonical: /handbook/api/classes/_purista_natsbridge.NatsBridge/
Source: natsbridge/src/NatsBridge.ts
Format: Markdown for agents
---

EventBridge implementation for NATS core messaging with optional JetStream.

Package: `@purista/natsbridge`

## Signature

```typescript
class NatsBridge
```

## Examples

```typescript
import { NatsBridge } from '@purista/natsbridge'

const eventBridge = new NatsBridge({
  servers: 'nats://localhost:4222',
  durableSubscriptionMode: 'strict',
})

await eventBridge.start()
```

## Members

### Constructors

- `new constructor(config?: { commandResponsePublishTwice: "always" | "eventOnly" | "eventAndError" | "never"; defaultCommandTimeout: number; defaultConsumerFailureHandling: NatsConsumerFailureHandlingDefaults; defaultMessageExpiryInterval: number; durableSubscriptionMode: "strict" | "best-effort"; emptyTopicPartString: string; ... })` — Creates a NATS bridge with defaults merged into NATS connection options.

### Properties

- `capabilities: EventBridgeCapabilities` — Runtime capability matrix used for strict startup validation.
- `commands: Map<string, JetStreamSubscription | Subscription>` — Registered command subscriptions keyed by PURISTA service address.
- `config: Complete<EventBridgeConfig<ConfigType>>`
- `connection: NatsConnection | undefined` — Active NATS connection, available after start.
- `defaultCommandTimeout: number` — The default time until when a command invocation automatically returns a time out error
- `inFlightExecutions: InFlightExecutionTracker`
- `instanceId: string` — Stable runtime instance id used to distinguish bridge processes.
- `isJetStreamEnabled: boolean` — Indicates whether the connected broker reports JetStream support.
- `js: JetStreamClient | undefined` — JetStream client used for durable publications and subscriptions.
- `jsm: JetStreamManager | undefined` — JetStream manager used for stream and consumer provisioning.
- `logger: Logger`
- `metricsRecorder: PuristaMetricsRecorder`
- `name: string` — Human-readable bridge name used in logs, traces, and metrics.
- `sc: Codec<unknown>` — JSON codec used for NATS message payload serialization.
- `subscriptions: Map<string, RegisteredSubscription>` — Registered event subscriptions keyed by PURISTA subscriber address.
- `traceProvider: NodeTracerProvider`

### Methods

- `destroy(): Promise<void>` — Waits for in-flight handlers, drains subscriptions, and closes the NATS
connection.
- `emitMessage<T>(message: Omit<EBMessage, "id" | "timestamp" | "correlationId">, contentType: string, contentEncoding: string): Promise<Readonly<EBMessage>>` — Publishes a PURISTA message as a NATS JSON payload.
- `getInFlightExecutionCount(): number` — Number of currently running handlers across all work kinds.
- `getInFlightExecutionCounts(): InFlightExecutionCounts` — Number of currently running handlers grouped by work kind.
- `getPausedSubscriptionConsumers(): object` — Returns currently paused subscription consumers keyed by registration key.
- `getTracer(): Tracer` — Returns open telemetry tracer of this service
- `invoke<T>(input: Omit<Command, "id" | "messageType" | "timestamp" | "correlationId">, commandTimeout: number): Promise<T>` — Invokes a command with NATS request/reply and waits for the response.
- `isHealthy(): Promise<boolean>` — Indicates whether the NATS connection is open and not draining.
- `isReady(): Promise<boolean>` — Indicates whether the NATS connection is open and not draining.
- `openStream<Chunk, Final>(_input: Omit<StreamOpenRequest, "id" | "messageType" | "timestamp" | "correlationId">, _ttl?: number): Promise<StreamHandle<Chunk, Final>>` — Open a stream invocation.
- `registerCommand(address: EBMessageAddress, cb: (message: { contentEncoding: string; contentType: string; correlationId: string; eventName: string; id: string; messageType: Command; ... }) => Promise<{ contentEncoding: "utf-8"; contentType: "application/json"; correlationId: string; eventName: string; id: string; isHandledError: boolean; ... } | { contentEncoding: string; contentType: string; correlationId: string; eventName: string; id: string; messageType: CommandSuccessResponse; ... }>, metadata: CommandDefinitionMetadataBase, eventBridgeConfig: DefinitionEventBridgeConfig): Promise<string>` — Registers a command handler.
- `registerStream(_address: EBMessageAddress, _cb: (message: StreamMessage) => Promise<void>, _metadata: StreamDefinitionMetadataBase, _eventBridgeConfig: DefinitionEventBridgeConfig): Promise<string>` — Register a service stream handler for a service target.
- `registerSubscription(subscription: Subscription, cb: (message: EBMessage) => Promise<Omit<{ contentEncoding: unknown; contentType: unknown; correlationId: unknown; eventName: unknown; id: unknown; messageType: unknown; ... }, unknown | unknown> | undefined>): Promise<string>` — Registers a subscription handler.
- `resumeSubscriptionConsumer(registrationKey: string): Promise<void>` — Resumes a subscription consumer paused by a `stop-consumer` control signal.
- `runInFlight<T>(fn: () => Promise<T>, kind?: "command" | "subscription" | "stream" | "generic"): Promise<T>`
- `start(): Promise<void>` — Connects to NATS and initializes JetStream clients when available.
- `startActiveSpan<F>(name: string, opts: SpanOptions, context: Context | undefined, fn: (span: Span) => Promise<F>): Promise<F>` — Start a child span for opentelemetry tracking
- `unregisterCommand(address: EBMessageAddress): Promise<void>` — Unregisters and drains/destroys a command handler subscription.
- `unregisterStream(_address: EBMessageAddress): Promise<void>` — Unregister a service stream
- `unregisterSubscription(address: EBMessageAddress): Promise<void>` — Unregisters and drains/destroys a subscription handler.
- `waitForInFlightDrain(timeoutMs?: number): Promise<boolean>`
- `wrapInSpan<F>(name: string, opts: SpanOptions, fn: (span: Span) => Promise<F>, context?: Context): Promise<F>` — Start span for opentelemetry tracking on same level.
The created span will not become the "active" span within opentelemetry!