Claude Code for NATS: High-Performance Messaging and Event Streaming

NATS is a lightweight, high-performance messaging system — a single 15MB binary handles millions of messages per second. Core NATS provides fire-and-forget pub/sub and request/reply. JetStream adds persistence, replay, and exactly-once delivery. NATS KV gives you a replicated key-value store. NATS Object Store handles large file distribution across a cluster. Claude Code writes NATS subject hierarchies, JetStream consumer definitions, and the Go and TypeScript client patterns that connect microservices cleanly.

CLAUDE.md for NATS Projects

## Messaging Stack
- NATS 2.x with JetStream enabled (persistence layer)
- Subject naming: domain.resource.action.version (orders.created.v1)
- Core NATS: ephemeral pub/sub for real-time events, request/reply for sync calls
- JetStream streams: durable, replayable event log; max-age TTL per stream
- JetStream consumers: push (server delivers) vs pull (client requests batches)
- Acks: AckExplicit for at-least-once; exactly-once via dedup window
- Error subjects: original.subject.error for dead letter pattern

Subject Design

# Hierarchical subject naming
orders.created.v1          # Order created event
orders.status.updated.v1   # Status change event
payments.processed.v1
payments.refunded.v1

# Request/reply (inbox pattern)
inventory.reserve           # Request subject
_INBOX.uniqueid             # Auto-generated reply-to

# Wildcards
orders.*          # Matches orders.created, orders.updated — one token
orders.>          # Matches everything under orders.*

# Per-tenant isolation (multi-tenant)
tenants.acme.orders.created.v1
tenants.globex.orders.created.v1

Core NATS: Pub/Sub and Request/Reply (Go)

// messaging/client.go
package messaging

import (
    "encoding/json"
    "time"
    nats "github.com/nats-io/nats.go"
)

type Client struct {
    nc *nats.Conn
}

func NewClient(url string) (*Client, error) {
    nc, err := nats.Connect(url,
        nats.MaxReconnects(-1),       // Reconnect forever
        nats.ReconnectWait(2*time.Second),
        nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
            log.Printf("NATS disconnected: %v", err)
        }),
        nats.ReconnectHandler(func(nc *nats.Conn) {
            log.Printf("NATS reconnected to %s", nc.ConnectedUrl())
        }),
    )
    if err != nil {
        return nil, err
    }
    return &Client{nc: nc}, nil
}

// Publish: fire-and-forget
func (c *Client) Publish(subject string, payload any) error {
    data, err := json.Marshal(payload)
    if err != nil {
        return err
    }
    return c.nc.Publish(subject, data)
}

// Subscribe: receive messages on subject
func (c *Client) Subscribe(subject string, handler func(msg *nats.Msg)) (*nats.Subscription, error) {
    return c.nc.Subscribe(subject, handler)
}

// Request/Reply: synchronous call-response pattern
func (c *Client) Request(subject string, payload any, timeout time.Duration) ([]byte, error) {
    data, _ := json.Marshal(payload)
    msg, err := c.nc.Request(subject, data, timeout)
    if err != nil {
        return nil, err
    }
    return msg.Data, nil
}

// Reply: handle incoming requests and respond
func (c *Client) SubscribeReply(subject string, handler func(req []byte) (any, error)) {
    c.nc.Subscribe(subject, func(msg *nats.Msg) {
        result, err := handler(msg.Data)
        if err != nil {
            msg.Respond([]byte(`{"error":"` + err.Error() + `"}`))
            return
        }
        data, _ := json.Marshal(result)
        msg.Respond(data)
    })
}

JetStream: Durable Streams and Consumers

// messaging/jetstream.go
package messaging

import (
    "context"
    nats "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

func SetupStreams(nc *nats.Conn) (jetstream.JetStream, error) {
    js, err := jetstream.New(nc)
    if err != nil {
        return nil, err
    }
    
    // Orders stream: persists all order events for 7 days
    _, err = js.CreateOrUpdateStream(context.Background(), jetstream.StreamConfig{
        Name:       "ORDERS",
        Subjects:   []string{"orders.>"},
        MaxAge:     7 * 24 * time.Hour,
        Storage:    jetstream.FileStorage,
        Replicas:   3,  // Replicated across 3 NATS servers
        Retention:  jetstream.LimitsPolicy,
        Discard:    jetstream.DiscardOld,
    })
    
    return js, err
}

// Durable consumer: resumes from last ack'd position after restart
func CreateOrderConsumer(js jetstream.JetStream, consumerName string) (jetstream.Consumer, error) {
    return js.CreateOrUpdateConsumer(context.Background(), "ORDERS", jetstream.ConsumerConfig{
        Name:           consumerName,
        Durable:        consumerName,
        FilterSubjects: []string{"orders.created.v1", "orders.status.updated.v1"},
        AckPolicy:      jetstream.AckExplicitPolicy,  // Must ack each message
        MaxDeliver:     5,                             // Retry up to 5 times
        AckWait:        30 * time.Second,
        DeliverPolicy:  jetstream.DeliverNewPolicy,
    })
}

// Pull consumer: batch processing
func ProcessOrderBatch(consumer jetstream.Consumer, batchSize int) error {
    msgs, err := consumer.Fetch(batchSize, jetstream.FetchMaxWait(5*time.Second))
    if err != nil {
        return err
    }
    
    for msg := range msgs.Messages() {
        var event OrderEvent
        if err := json.Unmarshal(msg.Data(), &event); err != nil {
            msg.Nak()  // Negative ack — redeliver
            continue
        }
        
        if err := processOrderEvent(event); err != nil {
            msg.NakWithDelay(30 * time.Second)  // Back off on error
            continue
        }
        
        msg.Ack()  // Explicit ack — message won't be redelivered
    }
    
    return msgs.Error()
}

TypeScript Client Patterns

// src/messaging/nats-client.ts
import { connect, StringCodec, JetStreamClient, RetentionPolicy } from 'nats';

const sc = StringCodec();
const nc = await connect({ servers: process.env.NATS_URL || 'nats://localhost:4222' });
const js = nc.jetstream();

// Publish to JetStream
export async function publishEvent(subject: string, event: object): Promise<void> {
  const ack = await js.publish(subject, sc.encode(JSON.stringify(event)));
  console.log(`Published to ${subject}, seq=${ack.seq}`);
}

// Key/Value store — replicated, observable config/state
const kv = await js.views.kv('feature-flags', {
  history: 5,  // Keep last 5 versions of each key
  ttl: 24 * 60 * 60 * 1000,  // 1 day
});

export async function setFlag(name: string, enabled: boolean): Promise<void> {
  await kv.put(name, sc.encode(JSON.stringify({ enabled, updatedAt: new Date() })));
}

export async function getFlag(name: string): Promise<boolean> {
  const entry = await kv.get(name);
  if (!entry) return false;
  const { enabled } = JSON.parse(sc.decode(entry.value));
  return enabled;
}

// Watch for real-time flag changes
export async function watchFlags(onChange: (name: string, enabled: boolean) => void) {
  const watcher = await kv.watch();
  for await (const entry of watcher) {
    const { enabled } = JSON.parse(sc.decode(entry.value));
    onChange(entry.key, enabled);
  }
}

For the Kafka comparison — when to use NATS JetStream vs Kafka Streams for event streaming at different scales, the Kafka Streams guide covers high-throughput stream processing patterns. For the event-driven architecture patterns that both NATS and Kafka enable, the event-driven architecture guide covers message contracts and consumer group design. The Claude Skills 360 bundle includes NATS skill sets covering JetStream streams/consumers, KV store patterns, and microservice request/reply. Start with the free tier to try NATS messaging pattern generation.