5. JetStream. Persistence и stream processing

js, err := nc.JetStream()
if err != nil {
    log.Fatal(err)
}

// Создаем stream с персистентной конфигурацией
stream, err := js.AddStream(&nats.StreamConfig{
    Name:      "ORDERS",
    Subjects:  []string{"orders.*"},        // Subject'ы, которые будут сохраняться в лог
    Retention: nats.WorkQueuePolicy,       // Удалять сообщение сразу после подтверждения обработки
    MaxAge:    7 * 24 * time.Hour,         // Хранить сообщения максимум 7 дней
    MaxBytes:  100 * 1024 * 1024,         // Ограничение размера лога: максимум 100MB
    Storage:   nats.FileStorage,           // File-based storage (дисковое хранилище)
    Replicas:  3,                          // 3 реплики для отказоустойчивости (Raft кворум)
})
if err != nil {
    log.Fatal(err)
}

type Msg struct {
    Subject      string    // Subject сообщения (например, orders.new)
    Reply        string    // Reply subject для подтверждений и ответов
    Data         []byte    // Полезная нагрузка (Payload)
    Header       Header    // Заголовки (key-value, включая Nats-Msg-Id)
    Sequence     uint64    // Монотонный Sequence number в рамках конкретного stream
    Time         time.Time // Время поступления на брокер (timestamp)
    Stream       string    // Имя stream'a (например, ORDERS)
    Duplicates   bool      // Флаг дедупликации (признак дубликата)
}

// Push consumer (сервер отправляет сообщения)
pushCons, err := js.Subscribe("orders.*", func(msg *nats.Msg) {
    fmt.Printf("Processing order: %s\n", string(msg.Data))
    msg.Ack() // Подтверждение получения
}, nats.Durable("order_processor")) // Durable consumer
if err != nil {
    log.Fatal(err)
}

// Pull consumer (клиент сам забирает сообщения)
pullCons, err := js.PullSubscribe("orders.*", "batch_processor")
if err != nil {
    log.Fatal(err)
}

// Батчевая обработка с контролем времени ожидания
msgs, err := pullCons.Fetch(10, nats.MaxWait(5*time.Second))
if err != nil {
    log.Printf("Fetch error: %v", err)
    return
}

for _, msg := range msgs {
    processOrder(msg)
    msg.Ack()
}

// Async push consumer
_, err = js.Subscribe("orders.new", func(msg *nats.Msg) {
    // Обработка в фоне
    go processOrderAsync(msg)
    msg.Ack()
}, nats.Durable("async_order_processor"))

// Pull consumer с контролем скорости (Rate Limiting & Backpressure)
cons, err := js.PullSubscribe("orders.*", "batch_processor")
if err != nil {
    log.Fatal(err)
}

for {
    // Забираем до 10 сообщений, ожидая не более 10 секунд
    msgs, err := cons.Fetch(10, nats.MaxWait(10*time.Second))
    if err != nil {
        if err == nats.ErrTimeout {
            continue // Нет новых сообщений, повторяем цикл
        }
        log.Printf("Fetch error: %v", err)
        break
    }

    // Обрабатываем batch
    for _, msg := range msgs {
        processOrder(msg)
        msg.Ack()
    }
}

stream, err := js.AddStream(&nats.StreamConfig{
    Name:      "LOGS",
    Subjects:  []string{"logs.*"},
    Retention: nats.LimitsPolicy,
    MaxMsgs:   100000,          // Максимум 100K сообщений в логе
    MaxBytes:  10 * 1024 * 1024, // 10MB
    MaxAge:    24 * time.Hour,  // 24 часа
})

stream, err := js.AddStream(&nats.StreamConfig{
    Name:      "TEMP_EVENTS",
    Subjects:  []string{"temp.*"},
    Retention: nats.InterestPolicy, // Удалять сообщение после того, как ВСЕ консьюмеры прислали Ack
})

stream, err := js.AddStream(&nats.StreamConfig{
    Name:      "TASK_QUEUE",
    Subjects:  []string{"tasks.*"},
    Retention: nats.WorkQueuePolicy, // Удалять сообщение из лога сразу после ПЕРВОГО успешного Ack
})

// Stream с включенной дедупликацией
stream, err := js.AddStream(&nats.StreamConfig{
    Name:            "ORDERS",
    Subjects:        []string{"orders.*"},
    DuplicateWindow: 2 * time.Minute, // Окно дедупликации 2 минуты
})
if err != nil {
    log.Fatal(err)
}

// Публикация с deduplication key
msg := nats.NewMsg("orders.new")
msg.Data = []byte("order data #42")
msg.Header.Set(nats.MsgIdHdr, "ORDER-12345") // Deduplication key (Nats-Msg-Id)

ack, err := js.PublishMsg(msg)
if err != nil {
    log.Printf("Publish error: %v", err)
} else {
    fmt.Printf("Message published with sequence: %d (duplicate: %v)\n", ack.Sequence, ack.Duplicate)
}

// Durable consumer (offset сохраняется)
_, err = js.Subscribe("orders.*", func(msg *nats.Msg) {
    processOrder(msg)
    msg.Ack()
}, nats.Durable("order_processor"), // Имя durable consumer'а
   nats.ManualAck())                // Ручное подтверждение (отключаем авто-ack)

// Ephemeral consumer (offset теряется при перезапуске)
_, err = js.Subscribe("orders.*", func(msg *nats.Msg) {
    processOrder(msg)
    msg.Ack()
}, nats.ManualAck())

// Consumer только для определенных subject'ов
cons, err := js.Subscribe("orders.new", func(msg *nats.Msg) {
    // Обрабатываем только создание новых заказов
    processOrder(msg)
    msg.Ack()
}, nats.Durable("new_orders_only"))

// Consumer для финансового аудита
js.Subscribe("orders.*", auditHandler, nats.Durable("audit_consumer"))

// Consumer для мгновенных push-уведомлений
js.Subscribe("orders.new", notificationHandler, nats.Durable("notification_consumer"))

// Consumer для бизнес-обработки платежей
js.Subscribe("orders.*", processingHandler, nats.Durable("processing_consumer"))

// Несколько экземпляров одного consumer'а разделяют поток между собой
for i := 0; i < 3; i++ {
    go func(instanceID int) {
        _, err := js.Subscribe("orders.*", func(msg *nats.Msg) {
            fmt.Printf("Instance %d processing: %s\n", instanceID, msg.Subject)
            processOrder(msg)
            msg.Ack()
        }, nats.Durable("order_workers")) // Одинаковое имя = queue group
        if err != nil {
            log.Printf("Worker %d error: %v", instanceID, err)
        }
    }(i)
}

// Memory storage (выше производительность, задержки единицы микросекунд, но данные сбрасываются при рестарте кластера)
stream, err := js.AddStream(&nats.StreamConfig{
    Name:    "HIGH_PERF_TEMP",
    Storage: nats.MemoryStorage, // Только в памяти
})

// File storage (полная durability, сохранение на диск через WAL)
stream, err := js.AddStream(&nats.StreamConfig{
    Name:    "PERSISTENT_LOGS", 
    Storage: nats.FileStorage, // На диск
})

stream, err := js.AddStream(&nats.StreamConfig{
    Name:        "COMPRESSED_STREAM",
    Subjects:    []string{"logs.*"},
    Compression: nats.S2Compression, // Использует S2 compression для сжатия блоков на диске
})

// Получение подробной информации о stream
info, err := js.StreamInfo("ORDERS")
if err != nil {
    log.Printf("Stream info error: %v", err)
    return
}

fmt.Printf("Messages: %d, Bytes: %d, First: %v, Last: %v\n",
    info.State.Messages,
    info.State.Bytes,
    info.State.FirstTime,
    info.State.LastTime,
)

// Получение информации о consumer
consInfo, err := js.ConsumerInfo("ORDERS", "order_processor")
if err != nil {
    log.Printf("Consumer info error: %v", err)
    return
}

fmt.Printf("Pending: %d, Ack Floor: %d\n",
    consInfo.NumPending,
    consInfo.AckFloor.Consumer,
)

// Service registration через JetStream
js.Publish("service.register", []byte(serviceInfoJSON))

// Регулярные Health checks
js.Publish("service.heartbeat", []byte(healthStatusJSON))

type Event struct {
    Type    string `json:"type"`
    Data    string `json:"data"`
    Version int    `json:"version"`
}

event := Event{
    Type:    "UserCreated",
    Data:    userData,
    Version: 1,
}

js.Publish("user.events", mustEncode(event))

// Command stream (поток команд)
js.Publish("commands.create.user", commandData)

// Event stream (поток фактов)  
js.Publish("events.user.created", eventData)