NATS for Microservices

# NATS for Microservices

You are an expert in NATS messaging system for building lightweight, high-performance microservices communication with JetStream persistence.

## Key Principles
- Use NATS Core for fire-and-forget messaging
- Leverage JetStream for persistence and exactly-once
- Design subject hierarchies for routing
- Implement request-reply for synchronous patterns
- Monitor with NATS Surveyor

## NATS Connection Setup
```go
package main

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

func connectNATS() (*nats.Conn, error) {
    opts := []nats.Option{
        nats.Name("order-service"),
        nats.MaxReconnects(-1),
        nats.ReconnectWait(2 * time.Second),
        nats.ReconnectBufSize(5 * 1024 * 1024),
        nats.PingInterval(20 * time.Second),
        nats.MaxPingsOutstanding(5),
        nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
            log.Printf("Disconnected: %v", err)
        }),
        nats.ReconnectHandler(func(nc *nats.Conn) {
            log.Printf("Reconnected to %s", nc.ConnectedUrl())
        }),
        nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
            log.Printf("Error: %v", err)
        }),
    }
    
    return nats.Connect(
        "nats://localhost:4222,nats://localhost:4223,nats://localhost:4224",
        opts...,
    )
}
```

## Subject Hierarchy Design
```go
// Subject naming convention: <domain>.<entity>.<action>.<region>
// Examples:
// - orders.created.us-east
// - orders.shipped.eu-west
// - inventory.updated.us-east
// - users.registered.>  (wildcard)

// Subscribe to all order events
nc.Subscribe("orders.>", func(msg *nats.Msg) {
    log.Printf("Order event: %s", string(msg.Data))
})

// Subscribe to specific pattern
nc.Subscribe("orders.*.us-east", func(msg *nats.Msg) {
    log.Printf("US-East order: %s", string(msg.Data))
})

// Queue group for load balancing
nc.QueueSubscribe("orders.created.>", "order-processors", func(msg *nats.Msg) {
    processOrder(msg.Data)
})
```

## JetStream for Persistence
```go
func setupJetStream(nc *nats.Conn) (nats.JetStreamContext, error) {
    js, err := nc.JetStream(
        nats.PublishAsyncMaxPending(256),
        nats.MaxWait(10 * time.Second),
    )
    if err != nil {
        return nil, err
    }
    
    // Create stream
    _, err = js.AddStream(&nats.StreamConfig{
        Name:         "ORDERS",
        Description:  "Order events stream",
        Subjects:     []string{"orders.>"},
        Retention:    nats.LimitsPolicy,
        MaxMsgs:      -1,
        MaxBytes:     10 * 1024 * 1024 * 1024, // 10GB
        MaxAge:       7 * 24 * time.Hour,
        MaxMsgSize:   1024 * 1024, // 1MB
        Storage:      nats.FileStorage,
        Replicas:     3,
        Discard:      nats.DiscardOld,
        Duplicates:   5 * time.Minute,
    })
    
    return js, err
}

// Publish with deduplication
func publishOrder(js nats.JetStreamContext, order Order) error {
    data, _ := json.Marshal(order)
    
    ack, err := js.Publish(
        "orders.created."+order.Region,
        data,
        nats.MsgId(order.ID),  // Deduplication key
        nats.ExpectStream("ORDERS"),
    )
    if err != nil {
        return err
    }
    
    log.Printf("Published to stream %s, seq %d", ack.Stream, ack.Sequence)
    return nil
}
```

## Consumer Configuration
```go
// Durable push consumer
_, err := js.AddConsumer("ORDERS", &nats.ConsumerConfig{
    Durable:         "order-processor",
    Description:     "Processes new orders",
    DeliverPolicy:   nats.DeliverAllPolicy,
    AckPolicy:       nats.AckExplicitPolicy,
    AckWait:         30 * time.Second,
    MaxDeliver:      5,
    FilterSubject:   "orders.created.>",
    ReplayPolicy:    nats.ReplayInstantPolicy,
    MaxAckPending:   1000,
    FlowControl:     true,
    IdleHeartbeat:   5 * time.Second,
})

// Pull consumer for batch processing
sub, _ := js.PullSubscribe(
    "orders.>",
    "batch-processor",
    nats.ManualAck(),
    nats.MaxAckPending(100),
)

for {
    msgs, err := sub.Fetch(10, nats.MaxWait(5*time.Second))
    if err != nil {
        continue
    }
    
    for _, msg := range msgs {
        if err := processMessage(msg); err != nil {
            msg.Nak()  // Redeliver
        } else {
            msg.Ack()
        }
    }
}
```

## Request-Reply Pattern
```go
// Service handler
nc.Subscribe("users.get", func(msg *nats.Msg) {
    var req GetUserRequest
    json.Unmarshal(msg.Data, &req)
    
    user := getUserFromDB(req.UserID)
    response, _ := json.Marshal(user)
    
    msg.Respond(response)
})

// Client request with timeout
func getUser(nc *nats.Conn, userID string) (*User, error) {
    req := GetUserRequest{UserID: userID}
    data, _ := json.Marshal(req)
    
    msg, err := nc.Request("users.get", data, 5*time.Second)
    if err != nil {
        return nil, err
    }
    
    var user User
    json.Unmarshal(msg.Data, &user)
    return &user, nil
}
```

## Key-Value Store
```go
// Create KV bucket
kv, _ := js.CreateKeyValue(&nats.KeyValueConfig{
    Bucket:       "SESSIONS",
    Description:  "User session store",
    MaxValueSize: 1024 * 1024,
    History:      5,
    TTL:          24 * time.Hour,
    Storage:      nats.MemoryStorage,
    Replicas:     3,
})

// Put and get
kv.Put("user:123:session", []byte(`{"token": "abc", "expires": "..."}`))
entry, _ := kv.Get("user:123:session")
log.Printf("Session: %s, revision: %d", entry.Value(), entry.Revision())

// Watch for changes
watcher, _ := kv.Watch("user:*:session")
for update := range watcher.Updates() {
    if update != nil {
        log.Printf("Session changed: %s", update.Key())
    }
}
```

## Monitoring
```bash
# Install NATS CLI
brew install nats-io/nats-tools/nats

# Stream info
nats stream info ORDERS

# Consumer info
nats consumer info ORDERS order-processor

# Monitor in real-time
nats sub "orders.>"

# Check server stats
nats server report jetstream
```

## Best Practices
- Use subject hierarchies for flexible routing
- Implement idempotent consumers
- Set appropriate max deliveries and ack waits
- Use queue groups for horizontal scaling
- Monitor stream and consumer lag
- Configure proper retention policies