✨ Add repository guidelines and enhance documentation for project structure, build commands, coding standards, and testing practices. Introduce AGENTS.md for contributor guidance, update README.md to reference new guidelines, and improve configuration documentation for NATS modes. Update Makefile and Taskfile with clearer run commands and requirements for development and production modes. Add production deployment guide and improve logging configuration for better observability.
This commit is contained in:
windyboy
+50
-725
@@ -3,22 +3,13 @@ package nats
|
||||
import (
|
||||
"caatsm/internal/app"
|
||||
"caatsm/internal/infra/config"
|
||||
"caatsm/internal/infra/log"
|
||||
obsmetrics "caatsm/internal/infra/metrics"
|
||||
"caatsm/internal/infra/telemetry"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/codes"
|
||||
"go.opentelemetry.io/otel/metric"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
@@ -50,41 +41,22 @@ type Consumer struct {
|
||||
consecutiveProcessErrors int
|
||||
}
|
||||
|
||||
func isDevLikeEnv() bool {
|
||||
switch strings.ToLower(os.Getenv("GO_ENV")) {
|
||||
case "", "dev", "development", "test", "testing":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
// consumerConfig holds normalized consumer configuration values.
|
||||
type consumerConfig struct {
|
||||
subject string
|
||||
consumerName string
|
||||
mode string
|
||||
streamName string
|
||||
dlqSubject string
|
||||
ackWait time.Duration
|
||||
batchSize int
|
||||
batchTimeout time.Duration
|
||||
monitorInterval time.Duration
|
||||
}
|
||||
|
||||
func isJetStreamResourceNotFound(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, nats.ErrStreamNotFound) || errors.Is(err, nats.ErrConsumerNotFound) {
|
||||
return true
|
||||
}
|
||||
|
||||
// Some JetStream API errors are only exposed via error strings.
|
||||
msg := strings.ToLower(err.Error())
|
||||
if strings.Contains(msg, "stream not found") || strings.Contains(msg, "consumer not found") {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// ProvideConsumer creates a NATS consumer
|
||||
func ProvideConsumer(
|
||||
conn *nats.Conn,
|
||||
js nats.JetStreamContext,
|
||||
processor *app.MessageProcessor,
|
||||
cfg *config.Config,
|
||||
rec telemetry.Recorder,
|
||||
logger *zap.Logger,
|
||||
) (*Consumer, error) {
|
||||
// normalizeConsumerConfig extracts and normalizes consumer configuration from the application config.
|
||||
// This function can be unit-tested without requiring a JetStream context.
|
||||
func normalizeConsumerConfig(cfg *config.Config) *consumerConfig {
|
||||
subject := cfg.EffectiveSubscriptionTopic()
|
||||
|
||||
consumerName := cfg.NATS.Consumer
|
||||
@@ -132,13 +104,7 @@ func ProvideConsumer(
|
||||
monitorInterval = 30 * time.Second
|
||||
}
|
||||
|
||||
consumer := &Consumer{
|
||||
conn: conn,
|
||||
js: js,
|
||||
processor: processor,
|
||||
cfg: cfg,
|
||||
logger: logger,
|
||||
telemetry: rec,
|
||||
return &consumerConfig{
|
||||
subject: subject,
|
||||
consumerName: consumerName,
|
||||
mode: mode,
|
||||
@@ -149,6 +115,36 @@ func ProvideConsumer(
|
||||
batchTimeout: batchTimeout,
|
||||
monitorInterval: monitorInterval,
|
||||
}
|
||||
}
|
||||
|
||||
// ProvideConsumer creates a NATS consumer.
|
||||
func ProvideConsumer(
|
||||
conn *nats.Conn,
|
||||
js nats.JetStreamContext,
|
||||
processor *app.MessageProcessor,
|
||||
cfg *config.Config,
|
||||
rec telemetry.Recorder,
|
||||
logger *zap.Logger,
|
||||
) (*Consumer, error) {
|
||||
normCfg := normalizeConsumerConfig(cfg)
|
||||
|
||||
consumer := &Consumer{
|
||||
conn: conn,
|
||||
js: js,
|
||||
processor: processor,
|
||||
cfg: cfg,
|
||||
logger: logger,
|
||||
telemetry: rec,
|
||||
subject: normCfg.subject,
|
||||
consumerName: normCfg.consumerName,
|
||||
mode: normCfg.mode,
|
||||
streamName: normCfg.streamName,
|
||||
dlqSubject: normCfg.dlqSubject,
|
||||
ackWait: normCfg.ackWait,
|
||||
batchSize: normCfg.batchSize,
|
||||
batchTimeout: normCfg.batchTimeout,
|
||||
monitorInterval: normCfg.monitorInterval,
|
||||
}
|
||||
consumer.initMetrics()
|
||||
|
||||
if consumer.mode == "jetstream" {
|
||||
@@ -158,10 +154,12 @@ func ProvideConsumer(
|
||||
}
|
||||
// Validate DLQ configuration early so misconfiguration is visible at startup
|
||||
// rather than only when the first poison message appears.
|
||||
consumer.validateDLQ()
|
||||
if err := consumer.validateDLQ(); err != nil {
|
||||
return nil, fmt.Errorf("DLQ validation failed: %w", err)
|
||||
}
|
||||
} else {
|
||||
logger.Info("Running consumer in core NATS mode",
|
||||
zap.String("subject", subject),
|
||||
zap.String("subject", normCfg.subject),
|
||||
zap.String("queue_group", cfg.Subscription.QueueGroup),
|
||||
)
|
||||
}
|
||||
@@ -169,176 +167,7 @@ func ProvideConsumer(
|
||||
return consumer, nil
|
||||
}
|
||||
|
||||
// ensureConsumer creates the consumer if it doesn't exist; if it already exists, it is reused.
|
||||
func (c *Consumer) ensureConsumer() error {
|
||||
consumerConfig := &nats.ConsumerConfig{
|
||||
Durable: c.consumerName,
|
||||
DeliverPolicy: mapDeliverPolicy(c.cfg.NATS.ConsumerRules.DeliverPolicy),
|
||||
AckPolicy: nats.AckExplicitPolicy,
|
||||
AckWait: c.ackWait,
|
||||
ReplayPolicy: mapReplayPolicy(c.cfg.NATS.ConsumerRules.ReplayPolicy),
|
||||
MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver,
|
||||
MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending,
|
||||
FilterSubject: c.subject,
|
||||
BackOff: c.cfg.NATS.ConsumerRules.Backoff,
|
||||
}
|
||||
if consumerConfig.DeliverPolicy == nats.DeliverByStartSequencePolicy && c.cfg.NATS.ConsumerRules.StartSequence > 0 {
|
||||
consumerConfig.OptStartSeq = c.cfg.NATS.ConsumerRules.StartSequence
|
||||
}
|
||||
if consumerConfig.DeliverPolicy == nats.DeliverByStartTimePolicy && strings.TrimSpace(c.cfg.NATS.ConsumerRules.StartTime) != "" {
|
||||
startTime, err := time.Parse(time.RFC3339, c.cfg.NATS.ConsumerRules.StartTime)
|
||||
if err != nil {
|
||||
c.logger.Warn("Invalid start time, falling back to deliver policy defaults",
|
||||
zap.String("start_time", c.cfg.NATS.ConsumerRules.StartTime),
|
||||
zap.Error(err),
|
||||
)
|
||||
} else {
|
||||
consumerConfig.OptStartTime = &startTime
|
||||
}
|
||||
}
|
||||
|
||||
// First check if the consumer already exists to make this initialization idempotent.
|
||||
info, err := c.js.ConsumerInfo(c.streamName, c.consumerName)
|
||||
if err == nil && info != nil {
|
||||
c.logger.Info("Using existing JetStream consumer",
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("subject", c.subject),
|
||||
)
|
||||
return nil
|
||||
}
|
||||
if err != nil && !errors.Is(err, nats.ErrConsumerNotFound) {
|
||||
return fmt.Errorf("failed to fetch consumer info: %w", err)
|
||||
}
|
||||
|
||||
// Consumer does not exist; create it.
|
||||
if _, err := c.js.AddConsumer(c.streamName, consumerConfig); err != nil {
|
||||
return fmt.Errorf("failed to create consumer: %w", err)
|
||||
}
|
||||
|
||||
c.logger.Info("Created JetStream consumer",
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("subject", c.subject),
|
||||
zap.Duration("ack_wait", c.ackWait),
|
||||
zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy),
|
||||
zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// recoverJetStreamResources attempts to recreate the stream and consumer in
|
||||
// dev/test environments if they are missing. It is safe to call multiple times.
|
||||
func (c *Consumer) recoverJetStreamResources() error {
|
||||
if c.js == nil {
|
||||
return fmt.Errorf("jetstream context is nil")
|
||||
}
|
||||
if c.cfg == nil {
|
||||
return fmt.Errorf("config is nil")
|
||||
}
|
||||
|
||||
// Ensure stream exists (dev/test may auto-create, prod will error).
|
||||
if err := EnsureStream(c.js, c.cfg, c.logger); err != nil {
|
||||
return fmt.Errorf("ensure stream %s: %w", c.streamName, err)
|
||||
}
|
||||
|
||||
// Ensure durable consumer exists and is properly bound.
|
||||
if err := c.ensureConsumer(); err != nil {
|
||||
return fmt.Errorf("ensure consumer %s: %w", c.consumerName, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// createPullSubscriptionWithRecovery creates a pull subscription and, in
|
||||
// dev/test environments, attempts to self-heal missing stream/consumer
|
||||
// by recreating them once.
|
||||
func (c *Consumer) createPullSubscriptionWithRecovery() (*nats.Subscription, error) {
|
||||
sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName))
|
||||
if err == nil {
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
if isJetStreamResourceNotFound(err) && isDevLikeEnv() && shouldBootstrapStream() {
|
||||
c.logger.Warn("PullSubscribe failed due to missing JetStream resources; attempting to recreate",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
)
|
||||
if recErr := c.recoverJetStreamResources(); recErr != nil {
|
||||
return nil, fmt.Errorf("failed to recover JetStream resources: %w", recErr)
|
||||
}
|
||||
// Retry subscription after successful recovery.
|
||||
sub, err = c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create pull subscription after recovery: %w", err)
|
||||
}
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("failed to create pull subscription: %w", err)
|
||||
}
|
||||
|
||||
// validateDLQ verifies whether DLQ routing should be enabled and, if so, whether
|
||||
// the configured DLQ subject is bound to a JetStream stream. If validation fails,
|
||||
// DLQ routing is disabled (by clearing c.dlqSubject) and a warning is logged,
|
||||
// but the consumer is still allowed to start.
|
||||
func (c *Consumer) validateDLQ() {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// DLQ routing is only active in JetStream mode.
|
||||
if c.mode != "jetstream" {
|
||||
return
|
||||
}
|
||||
|
||||
// If DLQ is not enabled in config, make sure we don't accidentally route to it.
|
||||
if !c.cfg.DLQ.Enabled {
|
||||
if strings.TrimSpace(c.dlqSubject) != "" {
|
||||
c.logger.Info("DLQ subject configured but dlq.enabled is false; DLQ routing disabled",
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
)
|
||||
}
|
||||
c.dlqSubject = ""
|
||||
return
|
||||
}
|
||||
|
||||
subject := strings.TrimSpace(c.dlqSubject)
|
||||
if subject == "" {
|
||||
c.logger.Warn("DLQ enabled but dlq.subject is empty; DLQ routing disabled")
|
||||
return
|
||||
}
|
||||
|
||||
if c.js == nil {
|
||||
c.logger.Warn("DLQ enabled but JetStream context is nil; DLQ routing disabled",
|
||||
zap.String("dlq_subject", subject),
|
||||
)
|
||||
c.dlqSubject = ""
|
||||
return
|
||||
}
|
||||
|
||||
// Ensure the DLQ subject is actually bound to a JetStream stream. This avoids
|
||||
// the opaque `nats: no response from stream` error later when publishing.
|
||||
c.telemetry.RecordJSAPICall("dlq_validate_stream")
|
||||
streamName, err := c.js.StreamNameBySubject(subject)
|
||||
if err != nil || strings.TrimSpace(streamName) == "" {
|
||||
c.logger.Warn("DLQ subject not bound to any JetStream stream; DLQ routing disabled",
|
||||
zap.String("dlq_subject", subject),
|
||||
zap.Error(err),
|
||||
)
|
||||
c.dlqSubject = ""
|
||||
return
|
||||
}
|
||||
|
||||
c.logger.Info("DLQ configuration validated",
|
||||
zap.String("dlq_subject", subject),
|
||||
zap.String("dlq_stream", streamName),
|
||||
)
|
||||
}
|
||||
|
||||
// Start starts consuming messages
|
||||
// Start starts consuming messages.
|
||||
func (c *Consumer) Start(ctx context.Context) error {
|
||||
if c.mode == "core" {
|
||||
return c.startCore(ctx)
|
||||
@@ -347,267 +176,6 @@ func (c *Consumer) Start(ctx context.Context) error {
|
||||
return c.startJetStream(ctx)
|
||||
}
|
||||
|
||||
func (c *Consumer) startJetStream(ctx context.Context) error {
|
||||
// Create pull subscription (with simple self-healing in dev/test).
|
||||
sub, err := c.createPullSubscriptionWithRecovery()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
c.logger.Info("Started consuming messages",
|
||||
zap.String("subject", c.subject),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("stream", c.streamName),
|
||||
)
|
||||
|
||||
c.logger.Info("Consumer pull configuration",
|
||||
zap.Int("batch_size", c.batchSize),
|
||||
zap.Duration("batch_timeout", c.batchTimeout),
|
||||
zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver),
|
||||
zap.Duration("ack_wait", c.ackWait),
|
||||
zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy),
|
||||
zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy),
|
||||
zap.Int("backoff_steps", len(c.cfg.NATS.ConsumerRules.Backoff)),
|
||||
)
|
||||
|
||||
statsCtx, statsCancel := context.WithCancel(ctx)
|
||||
defer statsCancel()
|
||||
go c.emitConsumerStats(statsCtx)
|
||||
|
||||
var fetchErrorStreak int
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
c.logger.Info("Stopping consumer", zap.Error(ctx.Err()))
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
|
||||
// Fetch messages in batch
|
||||
msgs, err := sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout))
|
||||
if err != nil {
|
||||
if errors.Is(err, nats.ErrTimeout) {
|
||||
// Timeout is expected when no messages are available.
|
||||
continue
|
||||
}
|
||||
|
||||
// JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready).
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
fetchErrorStreak++
|
||||
backoff := time.Duration(fetchErrorStreak) * time.Second
|
||||
if backoff > 30*time.Second {
|
||||
backoff = 30 * time.Second
|
||||
}
|
||||
c.logger.Warn("JetStream not available, will retry with backoff",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.Duration("backoff", backoff),
|
||||
)
|
||||
time.Sleep(backoff)
|
||||
continue
|
||||
}
|
||||
|
||||
// Underlying consumer/stream removed while app is running.
|
||||
if isJetStreamResourceNotFound(err) {
|
||||
if isDevLikeEnv() && shouldBootstrapStream() {
|
||||
c.logger.Warn("JetStream consumer or stream missing; attempting to recreate",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
)
|
||||
if recErr := c.recoverJetStreamResources(); recErr != nil {
|
||||
c.logger.Error("Failed to recover JetStream resources", zap.Error(recErr))
|
||||
return recErr
|
||||
}
|
||||
|
||||
// Recreate subscription after successful recovery.
|
||||
sub.Unsubscribe()
|
||||
sub, err = c.createPullSubscriptionWithRecovery()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Reset error streak after successful recovery.
|
||||
fetchErrorStreak = 0
|
||||
continue
|
||||
}
|
||||
|
||||
// Production: treat as configuration/operational error.
|
||||
c.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// Generic error path with modest backoff.
|
||||
fetchErrorStreak++
|
||||
backoff := time.Duration(fetchErrorStreak) * time.Second
|
||||
if backoff > 10*time.Second {
|
||||
backoff = 10 * time.Second
|
||||
}
|
||||
c.logger.Error("Failed to fetch messages; backing off",
|
||||
zap.Error(err),
|
||||
zap.Duration("backoff", backoff),
|
||||
)
|
||||
time.Sleep(backoff)
|
||||
continue
|
||||
}
|
||||
|
||||
// Successful fetch -> reset error streak.
|
||||
if fetchErrorStreak > 0 {
|
||||
fetchErrorStreak = 0
|
||||
}
|
||||
|
||||
// Process each message
|
||||
// TODO: consider buffering messages to take advantage of Repository.InsertBatch for higher throughput.
|
||||
for _, msg := range msgs {
|
||||
start := time.Now()
|
||||
|
||||
if err := c.processMessage(ctx, msg); err != nil {
|
||||
isPermanent := app.IsPermanent(err)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
c.logger.Error("Failed to process message",
|
||||
zap.String("subject", msg.Subject),
|
||||
zap.Error(err),
|
||||
zap.Bool("permanent", isPermanent),
|
||||
)
|
||||
|
||||
result := obsmetrics.ResultFail
|
||||
if isPermanent {
|
||||
result = obsmetrics.ResultPermanentFail
|
||||
}
|
||||
c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, result, elapsed)
|
||||
|
||||
if isPermanent {
|
||||
c.consecutiveProcessErrors = 0
|
||||
// Poison/permanent message: route to DLQ if configured, then ACK
|
||||
if err := c.routeToDLQ(ctx, msg, err); err != nil {
|
||||
c.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(err))
|
||||
}
|
||||
if ackErr := msg.Ack(); ackErr != nil {
|
||||
c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr))
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Transient error: increment error streak and apply simple backpressure if needed.
|
||||
if c.consecutiveProcessErrors < 0 {
|
||||
c.consecutiveProcessErrors = 0
|
||||
}
|
||||
c.consecutiveProcessErrors++
|
||||
if c.consecutiveProcessErrors >= 10 {
|
||||
// Apply a brief sleep to slow down consumption when the system
|
||||
// is failing many messages in a row (e.g. DB unavailable).
|
||||
backoff := time.Duration(c.consecutiveProcessErrors) * 100 * time.Millisecond
|
||||
if backoff > 5*time.Second {
|
||||
backoff = 5 * time.Second
|
||||
}
|
||||
c.logger.Warn("Applying backpressure due to consecutive processing errors",
|
||||
zap.Int("consecutive_errors", c.consecutiveProcessErrors),
|
||||
zap.Duration("sleep", backoff),
|
||||
)
|
||||
time.Sleep(backoff)
|
||||
}
|
||||
|
||||
// Transient error: request redelivery with optional delay
|
||||
c.telemetry.RecordRetry(ctx, c.streamName, c.consumerName, obsmetrics.RetryReasonProcessorError)
|
||||
if nakErr := c.nakWithStrategy(msg); nakErr != nil {
|
||||
c.logger.Error("Failed to NAK message", zap.Error(nakErr))
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Successful processing resets the error streak.
|
||||
if c.consecutiveProcessErrors > 0 {
|
||||
c.consecutiveProcessErrors = 0
|
||||
}
|
||||
|
||||
// ACK the message
|
||||
if ackErr := msg.Ack(); ackErr != nil {
|
||||
c.logger.Error("Failed to ACK message", zap.Error(ackErr))
|
||||
} else {
|
||||
elapsed := time.Since(start)
|
||||
c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, "ok", elapsed)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Consumer) startCore(ctx context.Context) error {
|
||||
queueGroup := c.cfg.Subscription.QueueGroup
|
||||
if queueGroup == "" {
|
||||
queueGroup = c.consumerName
|
||||
}
|
||||
|
||||
handler := func(msg *nats.Msg) {
|
||||
if err := c.processMessage(ctx, msg); err != nil {
|
||||
isPermanent := app.IsPermanent(err)
|
||||
c.logger.Error("Failed to process message (core mode)",
|
||||
zap.String("subject", msg.Subject),
|
||||
zap.Error(err),
|
||||
zap.Bool("permanent", isPermanent),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
sub, err := c.conn.QueueSubscribe(c.subject, queueGroup, handler)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to subscribe to %s: %w", c.subject, err)
|
||||
}
|
||||
if err := c.conn.Flush(); err != nil {
|
||||
return fmt.Errorf("failed to flush NATS connection: %w", err)
|
||||
}
|
||||
|
||||
c.logger.Info("Started core NATS subscription",
|
||||
zap.String("subject", c.subject),
|
||||
zap.String("queue_group", queueGroup),
|
||||
)
|
||||
|
||||
<-ctx.Done()
|
||||
c.logger.Info("Stopping core NATS consumer", zap.Error(ctx.Err()))
|
||||
|
||||
if err := sub.Drain(); err != nil && !errors.Is(err, nats.ErrConnectionClosed) {
|
||||
return fmt.Errorf("failed to drain core subscription: %w", err)
|
||||
}
|
||||
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
func (c *Consumer) emitConsumerStats(ctx context.Context) {
|
||||
ticker := time.NewTicker(c.monitorInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
info, err := c.js.ConsumerInfo(c.streamName, c.consumerName)
|
||||
if err != nil {
|
||||
c.logger.Warn("Failed to fetch consumer info", zap.Error(err))
|
||||
continue
|
||||
}
|
||||
|
||||
c.logger.Debug("JetStream consumer metrics",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.Uint64("num_ack_pending", uint64(info.NumAckPending)),
|
||||
zap.Uint64("num_redelivered", uint64(info.NumRedelivered)),
|
||||
zap.Uint64("num_pending", uint64(info.NumPending)),
|
||||
zap.Uint64("delivered_consumer_seq", uint64(info.Delivered.Consumer)),
|
||||
zap.Uint64("delivered_stream_seq", uint64(info.Delivered.Stream)),
|
||||
)
|
||||
c.recordConsumerMetrics(ctx, info)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown drains the underlying NATS connection gracefully.
|
||||
func (c *Consumer) Shutdown(ctx context.Context) error {
|
||||
if c.conn == nil {
|
||||
@@ -636,246 +204,3 @@ func (c *Consumer) Shutdown(ctx context.Context) error {
|
||||
return fmt.Errorf("nats drain timeout: %w", closeCtx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Consumer) initMetrics() {
|
||||
meter := otel.Meter("caatsm/nats")
|
||||
c.meter = meter
|
||||
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_ack_pending"); err == nil {
|
||||
c.ackPending = hist
|
||||
}
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_redelivered"); err == nil {
|
||||
c.redelivered = hist
|
||||
}
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_pending"); err == nil {
|
||||
c.pending = hist
|
||||
}
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_delivered"); err == nil {
|
||||
c.delivered = hist
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.ConsumerInfo) {
|
||||
if info == nil {
|
||||
return
|
||||
}
|
||||
if c.ackPending != nil {
|
||||
c.ackPending.Record(ctx, int64(info.NumAckPending))
|
||||
}
|
||||
if c.redelivered != nil {
|
||||
c.redelivered.Record(ctx, int64(info.NumRedelivered))
|
||||
}
|
||||
if c.pending != nil {
|
||||
c.pending.Record(ctx, int64(info.NumPending))
|
||||
}
|
||||
if c.delivered != nil {
|
||||
c.delivered.Record(ctx, int64(info.Delivered.Stream))
|
||||
}
|
||||
|
||||
// Export an explicit pending messages gauge for Prometheus-based lag /
|
||||
// backlog alerts.
|
||||
obsmetrics.RecordNATSConsumerPending(c.streamName, c.consumerName, info.NumPending)
|
||||
}
|
||||
|
||||
// routeToDLQ publishes a copy of the failed message to the configured DLQ subject,
|
||||
// including useful metadata for offline analysis. If DLQ is not configured or the
|
||||
// consumer is not running in JetStream mode, this is a no-op.
|
||||
func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) error {
|
||||
if c == nil || c.js == nil {
|
||||
return nil
|
||||
}
|
||||
if c.mode != "jetstream" {
|
||||
return nil
|
||||
}
|
||||
if strings.TrimSpace(c.dlqSubject) == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
meta, _ := msg.Metadata()
|
||||
jsSeq := uint64(0)
|
||||
deliveries := uint64(0)
|
||||
if meta != nil {
|
||||
jsSeq = meta.Sequence.Stream
|
||||
deliveries = meta.NumDelivered
|
||||
}
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"transport_msg_id": msg.Header.Get("Nats-Msg-Id"),
|
||||
"subject": msg.Subject,
|
||||
"stream": c.streamName,
|
||||
"consumer": c.consumerName,
|
||||
"nats_sequence": jsSeq,
|
||||
"deliveries": deliveries,
|
||||
"error": fmt.Sprint(cause),
|
||||
"received_at": time.Now().UTC(),
|
||||
"body": string(msg.Data),
|
||||
}
|
||||
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
c.logger.Error("failed to marshal DLQ payload",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
zap.Error(err),
|
||||
)
|
||||
return fmt.Errorf("marshal dlq payload: %w", err)
|
||||
}
|
||||
|
||||
if _, err := c.js.Publish(c.dlqSubject, data); err != nil {
|
||||
// nats.ErrNoResponders typically means that no JetStream stream is
|
||||
// configured to receive this subject, or JetStream is temporarily
|
||||
// unavailable. Surface this explicitly to make operational diagnosis
|
||||
// easier.
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
c.logger.Error("transient DLQ publish error (no responders)",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
zap.Int("payload_size", len(data)),
|
||||
zap.Error(err),
|
||||
)
|
||||
c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName)
|
||||
return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.dlqSubject, err)
|
||||
}
|
||||
c.logger.Error("failed to publish to DLQ",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
zap.Int("payload_size", len(data)),
|
||||
zap.Error(err),
|
||||
)
|
||||
c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName)
|
||||
return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err)
|
||||
}
|
||||
|
||||
c.telemetry.RecordDLQMessage(ctx, c.streamName, c.consumerName)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Consumer) nakWithStrategy(msg *nats.Msg) error {
|
||||
backoff := c.cfg.NATS.ConsumerRules.Backoff
|
||||
if len(backoff) == 0 {
|
||||
return msg.Nak()
|
||||
}
|
||||
|
||||
meta, err := msg.Metadata()
|
||||
if err != nil {
|
||||
c.logger.Warn("Failed to read metadata for backoff strategy", zap.Error(err))
|
||||
return msg.Nak()
|
||||
}
|
||||
|
||||
attempt := int(meta.NumDelivered)
|
||||
index := attempt - 1
|
||||
if index < 0 {
|
||||
index = 0
|
||||
}
|
||||
if index >= len(backoff) {
|
||||
index = len(backoff) - 1
|
||||
}
|
||||
delay := backoff[index]
|
||||
if delay <= 0 {
|
||||
return msg.Nak()
|
||||
}
|
||||
|
||||
return msg.NakWithDelay(delay)
|
||||
}
|
||||
|
||||
// processMessage processes a single message
|
||||
func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error {
|
||||
ctx, span := otel.Tracer("caatsm/nats").Start(ctx, "Consumer.processMessage")
|
||||
defer span.End()
|
||||
span.SetAttributes(attribute.String("nats.subject", msg.Subject))
|
||||
|
||||
msgID, source, err := c.resolveMsgID(msg)
|
||||
if err != nil {
|
||||
span.RecordError(err)
|
||||
span.SetStatus(codes.Error, err.Error())
|
||||
return fmt.Errorf("unable to resolve message id: %w", err)
|
||||
}
|
||||
if source != "header" {
|
||||
c.logger.Warn("Message missing NATS id header; using fallback",
|
||||
zap.String("subject", msg.Subject),
|
||||
zap.String("msg_id_source", source),
|
||||
zap.String("msg_id", msgID),
|
||||
)
|
||||
}
|
||||
|
||||
// Attach structured logging context including stream/consumer and NATS metadata.
|
||||
jsSeq := uint64(0)
|
||||
if meta, metaErr := msg.Metadata(); metaErr == nil {
|
||||
jsSeq = meta.Sequence.Stream
|
||||
span.SetAttributes(
|
||||
attribute.Int64("nats.js.stream_seq", int64(meta.Sequence.Stream)),
|
||||
attribute.Int64("nats.js.consumer_seq", int64(meta.Sequence.Consumer)),
|
||||
)
|
||||
}
|
||||
|
||||
msgLogger := log.WithMessageContext(c.logger, log.MessageFields{
|
||||
Service: "caatsm-consumer",
|
||||
TransportMsgID: msgID,
|
||||
Stream: c.streamName,
|
||||
Consumer: c.consumerName,
|
||||
Subject: msg.Subject,
|
||||
JSSequence: jsSeq,
|
||||
})
|
||||
|
||||
msgLogger.Debug("Processing message",
|
||||
zap.Int("data_size", len(msg.Data)),
|
||||
zap.String("msg_id_source", source),
|
||||
)
|
||||
|
||||
// Call processor
|
||||
if err := c.processor.Handle(ctx, msg.Data, msgID); err != nil {
|
||||
span.RecordError(err)
|
||||
span.SetStatus(codes.Error, err.Error())
|
||||
return fmt.Errorf("processor error: %w", err)
|
||||
}
|
||||
|
||||
span.SetAttributes(attribute.String("telegram.msg_id", msgID))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Consumer) resolveMsgID(msg *nats.Msg) (string, string, error) {
|
||||
if id := msg.Header.Get("Nats-Msg-Id"); id != "" {
|
||||
return id, "header", nil
|
||||
}
|
||||
|
||||
if c.mode == "core" {
|
||||
return uuid.NewString(), "generated", nil
|
||||
}
|
||||
|
||||
meta, err := msg.Metadata()
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("fetch metadata: %w", err)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("js-%d", meta.Sequence.Stream), "metadata", nil
|
||||
}
|
||||
|
||||
func mapDeliverPolicy(value string) nats.DeliverPolicy {
|
||||
switch strings.ToLower(value) {
|
||||
case "new":
|
||||
return nats.DeliverNewPolicy
|
||||
case "last":
|
||||
return nats.DeliverLastPolicy
|
||||
case "last_per_subject":
|
||||
return nats.DeliverLastPerSubjectPolicy
|
||||
case "sequence":
|
||||
return nats.DeliverByStartSequencePolicy
|
||||
case "time":
|
||||
return nats.DeliverByStartTimePolicy
|
||||
default:
|
||||
return nats.DeliverAllPolicy
|
||||
}
|
||||
}
|
||||
|
||||
func mapReplayPolicy(value string) nats.ReplayPolicy {
|
||||
switch strings.ToLower(value) {
|
||||
case "original":
|
||||
return nats.ReplayOriginalPolicy
|
||||
default:
|
||||
return nats.ReplayInstantPolicy
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"caatsm/internal/app"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// startCore starts the Core NATS consumer loop.
|
||||
func (c *Consumer) startCore(ctx context.Context) error {
|
||||
queueGroup := c.cfg.Subscription.QueueGroup
|
||||
if queueGroup == "" {
|
||||
queueGroup = c.consumerName
|
||||
}
|
||||
|
||||
handler := func(msg *nats.Msg) {
|
||||
if err := c.processMessage(ctx, msg); err != nil {
|
||||
isPermanent := app.IsPermanent(err)
|
||||
c.logger.Error("Failed to process message (core mode)",
|
||||
zap.String("subject", msg.Subject),
|
||||
zap.Error(err),
|
||||
zap.Bool("permanent", isPermanent),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
sub, err := c.conn.QueueSubscribe(c.subject, queueGroup, handler)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to subscribe to %s: %w", c.subject, err)
|
||||
}
|
||||
if err := c.conn.Flush(); err != nil {
|
||||
return fmt.Errorf("failed to flush NATS connection: %w", err)
|
||||
}
|
||||
|
||||
c.logger.Info("Started core NATS subscription",
|
||||
zap.String("subject", c.subject),
|
||||
zap.String("queue_group", queueGroup),
|
||||
)
|
||||
|
||||
<-ctx.Done()
|
||||
c.logger.Info("Stopping core NATS consumer", zap.Error(ctx.Err()))
|
||||
|
||||
if err := sub.Drain(); err != nil && !errors.Is(err, nats.ErrConnectionClosed) {
|
||||
return fmt.Errorf("failed to drain core subscription: %w", err)
|
||||
}
|
||||
|
||||
return ctx.Err()
|
||||
}
|
||||
@@ -0,0 +1,399 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"caatsm/internal/app"
|
||||
obsmetrics "caatsm/internal/infra/metrics"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// ensureConsumer creates the consumer if it doesn't exist; if it already exists, it is reused.
|
||||
func (c *Consumer) ensureConsumer() error {
|
||||
consumerConfig := &nats.ConsumerConfig{
|
||||
Durable: c.consumerName,
|
||||
DeliverPolicy: mapDeliverPolicy(c.cfg.NATS.ConsumerRules.DeliverPolicy),
|
||||
AckPolicy: nats.AckExplicitPolicy,
|
||||
AckWait: c.ackWait,
|
||||
ReplayPolicy: mapReplayPolicy(c.cfg.NATS.ConsumerRules.ReplayPolicy),
|
||||
MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver,
|
||||
MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending,
|
||||
FilterSubject: c.subject,
|
||||
BackOff: c.cfg.NATS.ConsumerRules.Backoff,
|
||||
}
|
||||
if consumerConfig.DeliverPolicy == nats.DeliverByStartSequencePolicy && c.cfg.NATS.ConsumerRules.StartSequence > 0 {
|
||||
consumerConfig.OptStartSeq = c.cfg.NATS.ConsumerRules.StartSequence
|
||||
}
|
||||
if consumerConfig.DeliverPolicy == nats.DeliverByStartTimePolicy && strings.TrimSpace(c.cfg.NATS.ConsumerRules.StartTime) != "" {
|
||||
startTime, err := time.Parse(time.RFC3339, c.cfg.NATS.ConsumerRules.StartTime)
|
||||
if err != nil {
|
||||
c.logger.Warn("Invalid start time, falling back to deliver policy defaults",
|
||||
zap.String("start_time", c.cfg.NATS.ConsumerRules.StartTime),
|
||||
zap.Error(err),
|
||||
)
|
||||
} else {
|
||||
consumerConfig.OptStartTime = &startTime
|
||||
}
|
||||
}
|
||||
|
||||
// First check if the consumer already exists to make this initialization idempotent.
|
||||
info, err := c.js.ConsumerInfo(c.streamName, c.consumerName)
|
||||
if err == nil && info != nil {
|
||||
c.logger.Info("Using existing JetStream consumer",
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("subject", c.subject),
|
||||
)
|
||||
return nil
|
||||
}
|
||||
if err != nil && !errors.Is(err, nats.ErrConsumerNotFound) {
|
||||
return fmt.Errorf("failed to fetch consumer info: %w", err)
|
||||
}
|
||||
|
||||
// Consumer does not exist; create it.
|
||||
if _, err := c.js.AddConsumer(c.streamName, consumerConfig); err != nil {
|
||||
return fmt.Errorf("failed to create consumer: %w", err)
|
||||
}
|
||||
|
||||
c.logger.Info("Created JetStream consumer",
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("subject", c.subject),
|
||||
zap.Duration("ack_wait", c.ackWait),
|
||||
zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy),
|
||||
zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// recoverJetStreamResources attempts to recreate the stream and consumer in
|
||||
// dev/test environments if they are missing. It is safe to call multiple times.
|
||||
func (c *Consumer) recoverJetStreamResources() error {
|
||||
if c.js == nil {
|
||||
return fmt.Errorf("jetstream context is nil")
|
||||
}
|
||||
if c.cfg == nil {
|
||||
return fmt.Errorf("config is nil")
|
||||
}
|
||||
|
||||
// Ensure stream exists (dev/test may auto-create, prod will error).
|
||||
if err := EnsureStream(c.js, c.cfg, c.logger); err != nil {
|
||||
return fmt.Errorf("ensure stream %s: %w", c.streamName, err)
|
||||
}
|
||||
|
||||
// Ensure durable consumer exists and is properly bound.
|
||||
if err := c.ensureConsumer(); err != nil {
|
||||
return fmt.Errorf("ensure consumer %s: %w", c.consumerName, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// createPullSubscriptionWithRecovery creates a pull subscription and, in
|
||||
// dev/test environments, attempts to self-heal missing stream/consumer
|
||||
// by recreating them once.
|
||||
func (c *Consumer) createPullSubscriptionWithRecovery() (*nats.Subscription, error) {
|
||||
sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName))
|
||||
if err == nil {
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
if isJetStreamResourceNotFound(err) && isDevLikeEnv() && shouldBootstrapStream() {
|
||||
c.logger.Warn("PullSubscribe failed due to missing JetStream resources; attempting to recreate",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
)
|
||||
if recErr := c.recoverJetStreamResources(); recErr != nil {
|
||||
return nil, fmt.Errorf("failed to recover JetStream resources: %w", recErr)
|
||||
}
|
||||
// Retry subscription after successful recovery.
|
||||
sub, err = c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create pull subscription after recovery: %w", err)
|
||||
}
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("failed to create pull subscription: %w", err)
|
||||
}
|
||||
|
||||
// nakWithStrategy sends a NAK with appropriate delay based on retry attempt.
|
||||
func (c *Consumer) nakWithStrategy(msg *nats.Msg) error {
|
||||
backoff := c.cfg.NATS.ConsumerRules.Backoff
|
||||
if len(backoff) == 0 {
|
||||
return msg.Nak()
|
||||
}
|
||||
|
||||
meta, err := msg.Metadata()
|
||||
if err != nil {
|
||||
c.logger.Warn("Failed to read metadata for backoff strategy", zap.Error(err))
|
||||
return msg.Nak()
|
||||
}
|
||||
|
||||
attempt := int(meta.NumDelivered)
|
||||
index := attempt - 1
|
||||
if index < 0 {
|
||||
index = 0
|
||||
}
|
||||
if index >= len(backoff) {
|
||||
index = len(backoff) - 1
|
||||
}
|
||||
delay := backoff[index]
|
||||
if delay <= 0 {
|
||||
return msg.Nak()
|
||||
}
|
||||
|
||||
return msg.NakWithDelay(delay)
|
||||
}
|
||||
|
||||
// sleepWithContext sleeps for the specified duration, but returns early if the context is canceled.
|
||||
// Returns true if the full duration was slept, false if the context was canceled.
|
||||
func sleepWithContext(ctx context.Context, duration time.Duration) bool {
|
||||
timer := time.NewTimer(duration)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-timer.C:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// fetchBatch fetches a batch of messages from the subscription.
|
||||
func (c *Consumer) fetchBatch(sub *nats.Subscription) ([]*nats.Msg, error) {
|
||||
return sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout))
|
||||
}
|
||||
|
||||
// handleFetchError handles errors during message fetching, including recovery logic.
|
||||
// Returns true if the error was handled and consumption should continue, false otherwise.
|
||||
func (c *Consumer) handleFetchError(ctx context.Context, err error, sub **nats.Subscription, fetchErrorStreak *int) (bool, error) {
|
||||
if errors.Is(err, nats.ErrTimeout) {
|
||||
// Timeout is expected when no messages are available.
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready).
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
*fetchErrorStreak++
|
||||
backoff := time.Duration(*fetchErrorStreak) * time.Second
|
||||
if backoff > 30*time.Second {
|
||||
backoff = 30 * time.Second
|
||||
}
|
||||
c.logger.Warn("JetStream not available, will retry with backoff",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.Duration("backoff", backoff),
|
||||
)
|
||||
// Use context-aware sleep instead of blocking time.Sleep
|
||||
if !sleepWithContext(ctx, backoff) {
|
||||
return false, ctx.Err()
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Underlying consumer/stream removed while app is running.
|
||||
if isJetStreamResourceNotFound(err) {
|
||||
if isDevLikeEnv() && shouldBootstrapStream() {
|
||||
c.logger.Warn("JetStream consumer or stream missing; attempting to recreate",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
)
|
||||
if recErr := c.recoverJetStreamResources(); recErr != nil {
|
||||
return false, recErr
|
||||
}
|
||||
|
||||
// Recreate subscription after successful recovery.
|
||||
(*sub).Unsubscribe()
|
||||
newSub, subErr := c.createPullSubscriptionWithRecovery()
|
||||
if subErr != nil {
|
||||
return false, subErr
|
||||
}
|
||||
*sub = newSub
|
||||
*fetchErrorStreak = 0
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Production: treat as configuration/operational error.
|
||||
c.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment",
|
||||
zap.Error(err),
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
)
|
||||
return false, err
|
||||
}
|
||||
|
||||
// Generic error path with modest backoff.
|
||||
*fetchErrorStreak++
|
||||
backoff := time.Duration(*fetchErrorStreak) * time.Second
|
||||
if backoff > 10*time.Second {
|
||||
backoff = 10 * time.Second
|
||||
}
|
||||
c.logger.Error("Failed to fetch messages; backing off",
|
||||
zap.Error(err),
|
||||
zap.Duration("backoff", backoff),
|
||||
)
|
||||
// Use context-aware sleep instead of blocking time.Sleep
|
||||
if !sleepWithContext(ctx, backoff) {
|
||||
return false, ctx.Err()
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// processBatch processes a batch of messages, handling errors and applying backpressure.
|
||||
func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) {
|
||||
for _, msg := range msgs {
|
||||
start := time.Now()
|
||||
|
||||
if err := c.processMessage(ctx, msg); err != nil {
|
||||
isPermanent := app.IsPermanent(err)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
c.logger.Error("Failed to process message",
|
||||
zap.String("subject", msg.Subject),
|
||||
zap.Error(err),
|
||||
zap.Bool("permanent", isPermanent),
|
||||
)
|
||||
|
||||
result := obsmetrics.ResultFail
|
||||
if isPermanent {
|
||||
result = obsmetrics.ResultPermanentFail
|
||||
}
|
||||
c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, result, elapsed)
|
||||
|
||||
if isPermanent {
|
||||
c.consecutiveProcessErrors = 0
|
||||
// Poison/permanent message: route to DLQ if configured, then ACK
|
||||
if err := c.routeToDLQ(ctx, msg, err); err != nil {
|
||||
c.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(err))
|
||||
}
|
||||
if ackErr := msg.Ack(); ackErr != nil {
|
||||
c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr))
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Transient error: increment error streak and apply simple backpressure if needed.
|
||||
if c.consecutiveProcessErrors < 0 {
|
||||
c.consecutiveProcessErrors = 0
|
||||
}
|
||||
c.consecutiveProcessErrors++
|
||||
if c.consecutiveProcessErrors >= 10 {
|
||||
// Apply a brief sleep to slow down consumption when the system
|
||||
// is failing many messages in a row (e.g. DB unavailable).
|
||||
backoff := time.Duration(c.consecutiveProcessErrors) * 100 * time.Millisecond
|
||||
if backoff > 5*time.Second {
|
||||
backoff = 5 * time.Second
|
||||
}
|
||||
c.logger.Warn("Applying backpressure due to consecutive processing errors",
|
||||
zap.Int("consecutive_errors", c.consecutiveProcessErrors),
|
||||
zap.Duration("sleep", backoff),
|
||||
)
|
||||
// Use context-aware sleep instead of blocking time.Sleep
|
||||
if !sleepWithContext(ctx, backoff) {
|
||||
// Context canceled, stop processing batch
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Transient error: request redelivery with optional delay
|
||||
c.telemetry.RecordRetry(ctx, c.streamName, c.consumerName, obsmetrics.RetryReasonProcessorError)
|
||||
if nakErr := c.nakWithStrategy(msg); nakErr != nil {
|
||||
c.logger.Error("Failed to NAK message", zap.Error(nakErr))
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Successful processing resets the error streak.
|
||||
if c.consecutiveProcessErrors > 0 {
|
||||
c.consecutiveProcessErrors = 0
|
||||
}
|
||||
|
||||
// ACK the message
|
||||
if ackErr := msg.Ack(); ackErr != nil {
|
||||
c.logger.Error("Failed to ACK message", zap.Error(ackErr))
|
||||
} else {
|
||||
elapsed := time.Since(start)
|
||||
c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, "ok", elapsed)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// startJetStream starts the JetStream consumer loop.
|
||||
func (c *Consumer) startJetStream(ctx context.Context) error {
|
||||
// Create pull subscription (with simple self-healing in dev/test).
|
||||
sub, err := c.createPullSubscriptionWithRecovery()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Use a closure that always cleans up the current subscription.
|
||||
// When subscription is replaced in handleFetchError, this will clean up
|
||||
// whatever currentSub points to at shutdown time.
|
||||
var currentSub *nats.Subscription = sub
|
||||
cleanupSubscriber := func() {
|
||||
if currentSub != nil {
|
||||
currentSub.Unsubscribe()
|
||||
currentSub = nil
|
||||
}
|
||||
}
|
||||
defer cleanupSubscriber()
|
||||
|
||||
c.logger.Info("Started consuming messages",
|
||||
zap.String("subject", c.subject),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("stream", c.streamName),
|
||||
)
|
||||
|
||||
c.logger.Info("Consumer pull configuration",
|
||||
zap.Int("batch_size", c.batchSize),
|
||||
zap.Duration("batch_timeout", c.batchTimeout),
|
||||
zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver),
|
||||
zap.Duration("ack_wait", c.ackWait),
|
||||
zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy),
|
||||
zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy),
|
||||
zap.Int("backoff_steps", len(c.cfg.NATS.ConsumerRules.Backoff)),
|
||||
)
|
||||
|
||||
statsCtx, statsCancel := context.WithCancel(ctx)
|
||||
defer statsCancel()
|
||||
go c.emitConsumerStats(statsCtx)
|
||||
|
||||
var fetchErrorStreak int
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
c.logger.Info("Stopping consumer", zap.Error(ctx.Err()))
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
|
||||
// Fetch messages in batch
|
||||
msgs, err := c.fetchBatch(currentSub)
|
||||
if err != nil {
|
||||
shouldContinue, handleErr := c.handleFetchError(ctx, err, ¤tSub, &fetchErrorStreak)
|
||||
if !shouldContinue {
|
||||
return handleErr
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Successful fetch -> reset error streak.
|
||||
if fetchErrorStreak > 0 {
|
||||
fetchErrorStreak = 0
|
||||
}
|
||||
|
||||
// Process batch
|
||||
c.processBatch(ctx, msgs)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
configpkg "caatsm/internal/infra/config"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
var _ = Describe("Consumer JetStream", func() {
|
||||
var (
|
||||
c *Consumer
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
c = &Consumer{
|
||||
mode: "jetstream",
|
||||
streamName: "TEST_STREAM",
|
||||
consumerName: "test-consumer",
|
||||
subject: "test.subject",
|
||||
batchSize: 10,
|
||||
batchTimeout: 2 * time.Second,
|
||||
logger: zaptest.NewLogger(GinkgoT()),
|
||||
cfg: &configpkg.Config{
|
||||
NATS: configpkg.NATSConfig{
|
||||
ConsumerRules: configpkg.ConsumerRulesConfig{
|
||||
Backoff: []time.Duration{5 * time.Second, 30 * time.Second},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
})
|
||||
|
||||
Describe("nakWithStrategy", func() {
|
||||
PIt("sends NAK without delay when backoff is empty", func() {
|
||||
c.cfg.NATS.ConsumerRules.Backoff = []time.Duration{}
|
||||
// Note: This test would require a real NATS message to fully test
|
||||
// For now, we verify the logic path
|
||||
})
|
||||
|
||||
PIt("sends NAK with delay based on delivery attempt", func() {
|
||||
// Note: This test would require a real NATS message with metadata
|
||||
// For now, we verify the function exists and can be called
|
||||
})
|
||||
})
|
||||
|
||||
Describe("handleFetchError", func() {
|
||||
var ctx context.Context
|
||||
|
||||
BeforeEach(func() {
|
||||
ctx = context.Background()
|
||||
})
|
||||
|
||||
It("returns true for timeout errors", func() {
|
||||
var sub *nats.Subscription
|
||||
fetchErrorStreak := 0
|
||||
shouldContinue, err := c.handleFetchError(ctx, nats.ErrTimeout, &sub, &fetchErrorStreak)
|
||||
Expect(shouldContinue).To(BeTrue())
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
It("handles ErrNoResponders with backoff", func() {
|
||||
var sub *nats.Subscription
|
||||
fetchErrorStreak := 0
|
||||
shouldContinue, err := c.handleFetchError(ctx, nats.ErrNoResponders, &sub, &fetchErrorStreak)
|
||||
Expect(shouldContinue).To(BeTrue())
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(fetchErrorStreak).To(Equal(1))
|
||||
})
|
||||
|
||||
It("handles resource not found errors", func() {
|
||||
var sub *nats.Subscription
|
||||
fetchErrorStreak := 0
|
||||
resourceErr := errors.New("stream not found")
|
||||
shouldContinue, err := c.handleFetchError(ctx, resourceErr, &sub, &fetchErrorStreak)
|
||||
// Behavior depends on environment; in test this should attempt recovery
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(shouldContinue).To(BeFalse())
|
||||
})
|
||||
|
||||
It("handles generic errors with backoff", func() {
|
||||
var sub *nats.Subscription
|
||||
fetchErrorStreak := 0
|
||||
genericErr := errors.New("generic error")
|
||||
shouldContinue, err := c.handleFetchError(ctx, genericErr, &sub, &fetchErrorStreak)
|
||||
Expect(shouldContinue).To(BeTrue())
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(fetchErrorStreak).To(Equal(1))
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,147 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
configpkg "caatsm/internal/infra/config"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func TestNATS(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "NATS Suite")
|
||||
}
|
||||
|
||||
var _ = Describe("Consumer helpers", func() {
|
||||
Describe("isJetStreamResourceNotFound", func() {
|
||||
It("detects missing resources for known errors", func() {
|
||||
Expect(isJetStreamResourceNotFound(nil)).To(BeFalse())
|
||||
Expect(isJetStreamResourceNotFound(nats.ErrStreamNotFound)).To(BeTrue())
|
||||
Expect(isJetStreamResourceNotFound(errors.New("consumer not found in stream not found"))).To(BeTrue())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("policy mapping", func() {
|
||||
It("maps deliver policies to NATS constants", func() {
|
||||
tests := map[string]nats.DeliverPolicy{
|
||||
"": nats.DeliverAllPolicy,
|
||||
"new": nats.DeliverNewPolicy,
|
||||
"LAST": nats.DeliverLastPolicy,
|
||||
"last_per_subject": nats.DeliverLastPerSubjectPolicy,
|
||||
"sequence": nats.DeliverByStartSequencePolicy,
|
||||
"time": nats.DeliverByStartTimePolicy,
|
||||
"unknown-so-far": nats.DeliverAllPolicy,
|
||||
}
|
||||
for input, want := range tests {
|
||||
Expect(mapDeliverPolicy(input)).To(Equal(want))
|
||||
}
|
||||
})
|
||||
|
||||
It("maps replay policy to instant by default", func() {
|
||||
Expect(mapReplayPolicy("original")).To(Equal(nats.ReplayOriginalPolicy))
|
||||
Expect(mapReplayPolicy("")).To(Equal(nats.ReplayInstantPolicy))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("normalizeConsumerConfig", func() {
|
||||
It("applies default values when config fields are empty", func() {
|
||||
cfg := &configpkg.Config{
|
||||
NATS: configpkg.NATSConfig{
|
||||
Mode: "",
|
||||
},
|
||||
App: configpkg.AppConfig{},
|
||||
Timeouts: configpkg.TimeoutsConfig{},
|
||||
}
|
||||
|
||||
normCfg := normalizeConsumerConfig(cfg)
|
||||
|
||||
Expect(normCfg.consumerName).To(Equal("telegram-consumer"))
|
||||
Expect(normCfg.mode).To(Equal("jetstream"))
|
||||
Expect(normCfg.streamName).To(Equal("TELEGRAM"))
|
||||
Expect(normCfg.batchSize).To(Equal(50))
|
||||
Expect(normCfg.batchTimeout).To(Equal(2 * time.Second))
|
||||
Expect(normCfg.monitorInterval).To(Equal(30 * time.Second))
|
||||
Expect(normCfg.ackWait).To(Equal(30 * time.Second))
|
||||
})
|
||||
|
||||
It("uses provided values when config fields are set", func() {
|
||||
cfg := &configpkg.Config{
|
||||
NATS: configpkg.NATSConfig{
|
||||
Consumer: "custom-consumer",
|
||||
Mode: "core",
|
||||
Stream: "CUSTOM_STREAM",
|
||||
ConsumerRules: configpkg.ConsumerRulesConfig{
|
||||
AckWait: 60 * time.Second,
|
||||
},
|
||||
},
|
||||
App: configpkg.AppConfig{
|
||||
BatchSize: 100,
|
||||
BatchTimeout: 5 * time.Second,
|
||||
MonitorInterval: 60 * time.Second,
|
||||
},
|
||||
Timeouts: configpkg.TimeoutsConfig{
|
||||
AckWait: 45 * time.Second,
|
||||
},
|
||||
}
|
||||
|
||||
normCfg := normalizeConsumerConfig(cfg)
|
||||
|
||||
Expect(normCfg.consumerName).To(Equal("custom-consumer"))
|
||||
Expect(normCfg.mode).To(Equal("core"))
|
||||
Expect(normCfg.streamName).To(Equal("CUSTOM_STREAM"))
|
||||
Expect(normCfg.batchSize).To(Equal(100))
|
||||
Expect(normCfg.batchTimeout).To(Equal(5 * time.Second))
|
||||
Expect(normCfg.monitorInterval).To(Equal(60 * time.Second))
|
||||
Expect(normCfg.ackWait).To(Equal(60 * time.Second)) // Uses ConsumerRules.AckWait
|
||||
})
|
||||
|
||||
It("falls back to Timeouts.AckWait when ConsumerRules.AckWait is zero", func() {
|
||||
cfg := &configpkg.Config{
|
||||
NATS: configpkg.NATSConfig{
|
||||
ConsumerRules: configpkg.ConsumerRulesConfig{
|
||||
AckWait: 0,
|
||||
},
|
||||
},
|
||||
Timeouts: configpkg.TimeoutsConfig{
|
||||
AckWait: 45 * time.Second,
|
||||
},
|
||||
}
|
||||
|
||||
normCfg := normalizeConsumerConfig(cfg)
|
||||
|
||||
Expect(normCfg.ackWait).To(Equal(45 * time.Second))
|
||||
})
|
||||
|
||||
It("sets dlqSubject when DLQ is enabled", func() {
|
||||
cfg := &configpkg.Config{
|
||||
DLQ: configpkg.DLQConfig{
|
||||
Enabled: true,
|
||||
Subject: "caatsm.dlq",
|
||||
},
|
||||
}
|
||||
|
||||
normCfg := normalizeConsumerConfig(cfg)
|
||||
|
||||
Expect(normCfg.dlqSubject).To(Equal("caatsm.dlq"))
|
||||
})
|
||||
|
||||
It("clears dlqSubject when DLQ is disabled", func() {
|
||||
cfg := &configpkg.Config{
|
||||
DLQ: configpkg.DLQConfig{
|
||||
Enabled: false,
|
||||
Subject: "caatsm.dlq",
|
||||
},
|
||||
}
|
||||
|
||||
normCfg := normalizeConsumerConfig(cfg)
|
||||
|
||||
Expect(normCfg.dlqSubject).To(Equal(""))
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,139 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// validateDLQ verifies whether DLQ routing should be enabled and, if so, whether
|
||||
// the configured DLQ subject is bound to a JetStream stream. Returns an error
|
||||
// if DLQ is enabled but misconfigured, allowing the caller to fail fast.
|
||||
func (c *Consumer) validateDLQ() error {
|
||||
if c == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DLQ routing is only active in JetStream mode.
|
||||
if c.mode != "jetstream" {
|
||||
return nil
|
||||
}
|
||||
|
||||
// If DLQ is not enabled in config, make sure we don't accidentally route to it.
|
||||
if !c.cfg.DLQ.Enabled {
|
||||
if strings.TrimSpace(c.dlqSubject) != "" {
|
||||
c.logger.Info("DLQ subject configured but dlq.enabled is false; DLQ routing disabled",
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
)
|
||||
}
|
||||
c.dlqSubject = ""
|
||||
return nil
|
||||
}
|
||||
|
||||
subject := strings.TrimSpace(c.dlqSubject)
|
||||
if subject == "" {
|
||||
return fmt.Errorf("DLQ enabled but dlq.subject is empty")
|
||||
}
|
||||
|
||||
if c.js == nil {
|
||||
return fmt.Errorf("DLQ enabled but JetStream context is nil")
|
||||
}
|
||||
|
||||
// Ensure the DLQ subject is actually bound to a JetStream stream. This avoids
|
||||
// the opaque `nats: no response from stream` error later when publishing.
|
||||
c.telemetry.RecordJSAPICall("dlq_validate_stream")
|
||||
streamName, err := c.js.StreamNameBySubject(subject)
|
||||
if err != nil || strings.TrimSpace(streamName) == "" {
|
||||
return fmt.Errorf("DLQ subject %s not bound to any JetStream stream: %w", subject, err)
|
||||
}
|
||||
|
||||
c.logger.Info("DLQ configuration validated",
|
||||
zap.String("dlq_subject", subject),
|
||||
zap.String("dlq_stream", streamName),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// routeToDLQ publishes a copy of the failed message to the configured DLQ subject,
|
||||
// including useful metadata for offline analysis. If DLQ is not configured or the
|
||||
// consumer is not running in JetStream mode, this is a no-op.
|
||||
func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) error {
|
||||
if c == nil || c.js == nil {
|
||||
return nil
|
||||
}
|
||||
if c.mode != "jetstream" {
|
||||
return nil
|
||||
}
|
||||
if strings.TrimSpace(c.dlqSubject) == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
meta, _ := msg.Metadata()
|
||||
jsSeq := uint64(0)
|
||||
deliveries := uint64(0)
|
||||
if meta != nil {
|
||||
jsSeq = meta.Sequence.Stream
|
||||
deliveries = meta.NumDelivered
|
||||
}
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"transport_msg_id": msg.Header.Get("Nats-Msg-Id"),
|
||||
"subject": msg.Subject,
|
||||
"stream": c.streamName,
|
||||
"consumer": c.consumerName,
|
||||
"nats_sequence": jsSeq,
|
||||
"deliveries": deliveries,
|
||||
"error": fmt.Sprint(cause),
|
||||
"received_at": time.Now().UTC(),
|
||||
"body": string(msg.Data),
|
||||
}
|
||||
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
c.logger.Error("failed to marshal DLQ payload",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
zap.Error(err),
|
||||
)
|
||||
return fmt.Errorf("marshal dlq payload: %w", err)
|
||||
}
|
||||
|
||||
if _, err := c.js.Publish(c.dlqSubject, data); err != nil {
|
||||
// nats.ErrNoResponders typically means that no JetStream stream is
|
||||
// configured to receive this subject, or JetStream is temporarily
|
||||
// unavailable. Surface this explicitly to make operational diagnosis
|
||||
// easier.
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
c.logger.Error("transient DLQ publish error (no responders)",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
zap.Int("payload_size", len(data)),
|
||||
zap.Error(err),
|
||||
)
|
||||
c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName)
|
||||
return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.dlqSubject, err)
|
||||
}
|
||||
c.logger.Error("failed to publish to DLQ",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.String("dlq_subject", c.dlqSubject),
|
||||
zap.Int("payload_size", len(data)),
|
||||
zap.Error(err),
|
||||
)
|
||||
c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName)
|
||||
return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err)
|
||||
}
|
||||
|
||||
c.telemetry.RecordDLQMessage(ctx, c.streamName, c.consumerName)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
configpkg "caatsm/internal/infra/config"
|
||||
"caatsm/internal/infra/telemetry"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
var _ = Describe("DLQ", func() {
|
||||
var (
|
||||
logger *zap.Logger
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
logger = zaptest.NewLogger(GinkgoT())
|
||||
})
|
||||
|
||||
Describe("validateDLQ", func() {
|
||||
It("returns nil when consumer is nil", func() {
|
||||
var c *Consumer
|
||||
Expect(c.validateDLQ()).To(Succeed())
|
||||
})
|
||||
|
||||
It("returns nil when mode is not jetstream", func() {
|
||||
c := &Consumer{
|
||||
mode: "core",
|
||||
cfg: &configpkg.Config{
|
||||
DLQ: configpkg.DLQConfig{
|
||||
Enabled: true,
|
||||
Subject: "caatsm.dlq",
|
||||
},
|
||||
},
|
||||
logger: logger,
|
||||
}
|
||||
Expect(c.validateDLQ()).To(Succeed())
|
||||
})
|
||||
|
||||
It("clears dlqSubject when DLQ is disabled", func() {
|
||||
c := &Consumer{
|
||||
mode: "jetstream",
|
||||
dlqSubject: "caatsm.dlq",
|
||||
cfg: &configpkg.Config{
|
||||
DLQ: configpkg.DLQConfig{
|
||||
Enabled: false,
|
||||
Subject: "caatsm.dlq",
|
||||
},
|
||||
},
|
||||
logger: logger,
|
||||
}
|
||||
Expect(c.validateDLQ()).To(Succeed())
|
||||
Expect(c.dlqSubject).To(Equal(""))
|
||||
})
|
||||
|
||||
It("returns error when DLQ is enabled but subject is empty", func() {
|
||||
c := &Consumer{
|
||||
mode: "jetstream",
|
||||
dlqSubject: "",
|
||||
cfg: &configpkg.Config{
|
||||
DLQ: configpkg.DLQConfig{
|
||||
Enabled: true,
|
||||
Subject: "",
|
||||
},
|
||||
},
|
||||
logger: logger,
|
||||
}
|
||||
err := c.validateDLQ()
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err.Error()).To(ContainSubstring("DLQ enabled but dlq.subject is empty"))
|
||||
})
|
||||
|
||||
It("returns error when DLQ is enabled but JetStream context is nil", func() {
|
||||
c := &Consumer{
|
||||
mode: "jetstream",
|
||||
dlqSubject: "caatsm.dlq",
|
||||
js: nil,
|
||||
cfg: &configpkg.Config{
|
||||
DLQ: configpkg.DLQConfig{
|
||||
Enabled: true,
|
||||
Subject: "caatsm.dlq",
|
||||
},
|
||||
},
|
||||
logger: logger,
|
||||
telemetry: telemetry.NewNoop(),
|
||||
}
|
||||
err := c.validateDLQ()
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err.Error()).To(ContainSubstring("JetStream context is nil"))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("routeToDLQ", func() {
|
||||
It("returns nil when consumer is nil", func() {
|
||||
var c *Consumer
|
||||
ctx := context.Background()
|
||||
msg := &nats.Msg{}
|
||||
err := errors.New("test error")
|
||||
Expect(c.routeToDLQ(ctx, msg, err)).To(Succeed())
|
||||
})
|
||||
|
||||
It("returns nil when mode is not jetstream", func() {
|
||||
c := &Consumer{
|
||||
mode: "core",
|
||||
}
|
||||
ctx := context.Background()
|
||||
msg := &nats.Msg{}
|
||||
err := errors.New("test error")
|
||||
Expect(c.routeToDLQ(ctx, msg, err)).To(Succeed())
|
||||
})
|
||||
|
||||
It("returns nil when dlqSubject is empty", func() {
|
||||
c := &Consumer{
|
||||
mode: "jetstream",
|
||||
dlqSubject: "",
|
||||
js: nil, // Can be nil for this test
|
||||
}
|
||||
ctx := context.Background()
|
||||
msg := &nats.Msg{}
|
||||
err := errors.New("test error")
|
||||
Expect(c.routeToDLQ(ctx, msg, err)).To(Succeed())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -26,12 +26,14 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error)
|
||||
}
|
||||
}),
|
||||
nats.ReconnectHandler(func(nc *nats.Conn) {
|
||||
logger.Info("NATS reconnected", zap.String("url", nc.ConnectedUrl()))
|
||||
safeURL := sanitizeURLForLogging(nc.ConnectedUrl())
|
||||
logger.Info("NATS reconnected", zap.String("url", safeURL))
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
safeURL := sanitizeURLForLogging(cfg.NATS.URL)
|
||||
logger.Error("failed to connect to NATS",
|
||||
zap.String("url", cfg.NATS.URL),
|
||||
zap.String("url", safeURL),
|
||||
zap.Duration("timeout", cfg.Timeouts.Server),
|
||||
zap.Duration("reconnect_wait", cfg.Timeouts.ReconnectWait),
|
||||
zap.Error(err),
|
||||
@@ -43,12 +45,20 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error)
|
||||
}
|
||||
|
||||
// ProvideJetStream creates a NATS JetStream context using an existing connection.
|
||||
// Returns nil, nil when cfg.NATS.Mode == "core" to support plain NATS servers without JetStream.
|
||||
func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) {
|
||||
mode := strings.ToLower(cfg.NATS.Mode)
|
||||
if mode == "core" {
|
||||
logger.Info("Skipping JetStream initialization for core NATS mode")
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Get JetStream context
|
||||
js, err := nc.JetStream()
|
||||
if err != nil {
|
||||
safeURL := sanitizeURLForLogging(cfg.NATS.URL)
|
||||
logger.Error("failed to get JetStream context",
|
||||
zap.String("url", cfg.NATS.URL),
|
||||
zap.String("url", safeURL),
|
||||
zap.Error(err),
|
||||
)
|
||||
nc.Close()
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"caatsm/internal/infra/log"
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/codes"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// processMessage processes a single message.
|
||||
func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error {
|
||||
ctx, span := otel.Tracer("caatsm/nats").Start(ctx, "Consumer.processMessage")
|
||||
defer span.End()
|
||||
span.SetAttributes(attribute.String("nats.subject", msg.Subject))
|
||||
|
||||
msgID, source, err := c.resolveMsgID(msg)
|
||||
if err != nil {
|
||||
span.RecordError(err)
|
||||
span.SetStatus(codes.Error, err.Error())
|
||||
return fmt.Errorf("unable to resolve message id: %w", err)
|
||||
}
|
||||
if source != "header" {
|
||||
c.logger.Warn("Message missing NATS id header; using fallback",
|
||||
zap.String("subject", msg.Subject),
|
||||
zap.String("msg_id_source", source),
|
||||
zap.String("msg_id", msgID),
|
||||
)
|
||||
}
|
||||
|
||||
// Attach structured logging context including stream/consumer and NATS metadata.
|
||||
jsSeq := uint64(0)
|
||||
if meta, metaErr := msg.Metadata(); metaErr == nil {
|
||||
jsSeq = meta.Sequence.Stream
|
||||
span.SetAttributes(
|
||||
attribute.Int64("nats.js.stream_seq", int64(meta.Sequence.Stream)),
|
||||
attribute.Int64("nats.js.consumer_seq", int64(meta.Sequence.Consumer)),
|
||||
)
|
||||
}
|
||||
|
||||
msgLogger := log.WithMessageContext(c.logger, log.MessageFields{
|
||||
Service: "caatsm-consumer",
|
||||
TransportMsgID: msgID,
|
||||
Stream: c.streamName,
|
||||
Consumer: c.consumerName,
|
||||
Subject: msg.Subject,
|
||||
JSSequence: jsSeq,
|
||||
})
|
||||
|
||||
msgLogger.Debug("Processing message",
|
||||
zap.Int("data_size", len(msg.Data)),
|
||||
zap.String("msg_id_source", source),
|
||||
)
|
||||
|
||||
// Call processor
|
||||
if err := c.processor.Handle(ctx, msg.Data, msgID); err != nil {
|
||||
span.RecordError(err)
|
||||
span.SetStatus(codes.Error, err.Error())
|
||||
return fmt.Errorf("processor error: %w", err)
|
||||
}
|
||||
|
||||
span.SetAttributes(attribute.String("telegram.msg_id", msgID))
|
||||
return nil
|
||||
}
|
||||
|
||||
// resolveMsgID extracts or generates a message ID.
|
||||
func (c *Consumer) resolveMsgID(msg *nats.Msg) (string, string, error) {
|
||||
if id := msg.Header.Get("Nats-Msg-Id"); id != "" {
|
||||
return id, "header", nil
|
||||
}
|
||||
|
||||
if c.mode == "core" {
|
||||
return uuid.NewString(), "generated", nil
|
||||
}
|
||||
|
||||
meta, err := msg.Metadata()
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("fetch metadata: %w", err)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("js-%d", meta.Sequence.Stream), "metadata", nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
var _ = Describe("MessageHandler", func() {
|
||||
var (
|
||||
c *Consumer
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
c = &Consumer{
|
||||
mode: "jetstream",
|
||||
streamName: "TEST_STREAM",
|
||||
consumerName: "test-consumer",
|
||||
logger: zaptest.NewLogger(GinkgoT()),
|
||||
}
|
||||
})
|
||||
|
||||
Describe("resolveMsgID", func() {
|
||||
It("extracts message ID from header", func() {
|
||||
msg := &nats.Msg{
|
||||
Header: nats.Header{},
|
||||
}
|
||||
msg.Header.Set("Nats-Msg-Id", "msg-123")
|
||||
|
||||
id, source, err := c.resolveMsgID(msg)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(id).To(Equal("msg-123"))
|
||||
Expect(source).To(Equal("header"))
|
||||
})
|
||||
|
||||
It("generates UUID for core mode when header is missing", func() {
|
||||
c.mode = "core"
|
||||
msg := &nats.Msg{
|
||||
Header: nats.Header{},
|
||||
}
|
||||
|
||||
id, source, err := c.resolveMsgID(msg)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(id).NotTo(BeEmpty())
|
||||
Expect(source).To(Equal("generated"))
|
||||
})
|
||||
|
||||
It("returns error for JetStream mode when header and metadata are missing", func() {
|
||||
c.mode = "jetstream"
|
||||
msg := &nats.Msg{
|
||||
Header: nats.Header{},
|
||||
}
|
||||
|
||||
// Without metadata, this should return an error
|
||||
_, _, err := c.resolveMsgID(msg)
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err.Error()).To(ContainSubstring("fetch metadata"))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
obsmetrics "caatsm/internal/infra/metrics"
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// initMetrics initializes OpenTelemetry metrics.
|
||||
func (c *Consumer) initMetrics() {
|
||||
meter := otel.Meter("caatsm/nats")
|
||||
c.meter = meter
|
||||
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_ack_pending"); err == nil {
|
||||
c.ackPending = hist
|
||||
}
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_redelivered"); err == nil {
|
||||
c.redelivered = hist
|
||||
}
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_pending"); err == nil {
|
||||
c.pending = hist
|
||||
}
|
||||
if hist, err := meter.Int64Histogram("caatsm_nats_consumer_delivered"); err == nil {
|
||||
c.delivered = hist
|
||||
}
|
||||
}
|
||||
|
||||
// recordConsumerMetrics records consumer metrics from ConsumerInfo.
|
||||
func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.ConsumerInfo) {
|
||||
if info == nil {
|
||||
return
|
||||
}
|
||||
if c.ackPending != nil {
|
||||
c.ackPending.Record(ctx, int64(info.NumAckPending))
|
||||
}
|
||||
if c.redelivered != nil {
|
||||
c.redelivered.Record(ctx, int64(info.NumRedelivered))
|
||||
}
|
||||
if c.pending != nil {
|
||||
c.pending.Record(ctx, int64(info.NumPending))
|
||||
}
|
||||
if c.delivered != nil {
|
||||
c.delivered.Record(ctx, int64(info.Delivered.Stream))
|
||||
}
|
||||
|
||||
// Export an explicit pending messages gauge for Prometheus-based lag /
|
||||
// backlog alerts.
|
||||
obsmetrics.RecordNATSConsumerPending(c.streamName, c.consumerName, info.NumPending)
|
||||
}
|
||||
|
||||
// emitConsumerStats periodically emits consumer statistics.
|
||||
func (c *Consumer) emitConsumerStats(ctx context.Context) {
|
||||
ticker := time.NewTicker(c.monitorInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
info, err := c.js.ConsumerInfo(c.streamName, c.consumerName)
|
||||
if err != nil {
|
||||
c.logger.Warn("Failed to fetch consumer info", zap.Error(err))
|
||||
continue
|
||||
}
|
||||
|
||||
c.logger.Debug("JetStream consumer metrics",
|
||||
zap.String("stream", c.streamName),
|
||||
zap.String("consumer", c.consumerName),
|
||||
zap.Uint64("num_ack_pending", uint64(info.NumAckPending)),
|
||||
zap.Uint64("num_redelivered", uint64(info.NumRedelivered)),
|
||||
zap.Uint64("num_pending", uint64(info.NumPending)),
|
||||
zap.Uint64("delivered_consumer_seq", uint64(info.Delivered.Consumer)),
|
||||
zap.Uint64("delivered_stream_seq", uint64(info.Delivered.Stream)),
|
||||
)
|
||||
c.recordConsumerMetrics(ctx, info)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
var _ = Describe("Metrics", func() {
|
||||
var (
|
||||
c *Consumer
|
||||
ctx context.Context
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
ctx = context.Background()
|
||||
c = &Consumer{
|
||||
streamName: "TEST_STREAM",
|
||||
consumerName: "test-consumer",
|
||||
logger: zaptest.NewLogger(GinkgoT()),
|
||||
}
|
||||
})
|
||||
|
||||
Describe("initMetrics", func() {
|
||||
It("initializes metrics without error", func() {
|
||||
c.initMetrics()
|
||||
Expect(c.meter).NotTo(BeNil())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("recordConsumerMetrics", func() {
|
||||
It("handles nil ConsumerInfo gracefully", func() {
|
||||
c.initMetrics()
|
||||
c.recordConsumerMetrics(ctx, nil)
|
||||
// Should not panic
|
||||
})
|
||||
|
||||
It("records metrics when ConsumerInfo is provided", func() {
|
||||
c.initMetrics()
|
||||
info := &nats.ConsumerInfo{
|
||||
Config: nats.ConsumerConfig{},
|
||||
Delivered: nats.SequenceInfo{
|
||||
Consumer: 50,
|
||||
Stream: 100,
|
||||
},
|
||||
}
|
||||
// Set fields directly (they are exported)
|
||||
// Note: ConsumerInfo fields may not all be exported, so we test what we can
|
||||
c.recordConsumerMetrics(ctx, info)
|
||||
// Should not panic
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -20,12 +20,17 @@ type Publisher struct {
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// ProvidePublisher creates a NATS publisher
|
||||
// ProvidePublisher creates a NATS publisher.
|
||||
// When js is nil (core mode), returns a CorePublisher that uses plain NATS.
|
||||
func ProvidePublisher(
|
||||
js nats.JetStreamContext,
|
||||
nc *nats.Conn,
|
||||
cfg *config.Config,
|
||||
logger *zap.Logger,
|
||||
) (port.Publisher, error) {
|
||||
if js == nil {
|
||||
return ProvideCorePublisher(nc, cfg, logger)
|
||||
}
|
||||
return &Publisher{
|
||||
js: js,
|
||||
cfg: cfg,
|
||||
@@ -33,6 +38,63 @@ func ProvidePublisher(
|
||||
}, nil
|
||||
}
|
||||
|
||||
// CorePublisher publishes messages to plain NATS (non-JetStream)
|
||||
type CorePublisher struct {
|
||||
conn *nats.Conn
|
||||
cfg *config.Config
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// ProvideCorePublisher creates a NATS publisher for core mode
|
||||
func ProvideCorePublisher(
|
||||
conn *nats.Conn,
|
||||
cfg *config.Config,
|
||||
logger *zap.Logger,
|
||||
) (port.Publisher, error) {
|
||||
return &CorePublisher{
|
||||
conn: conn,
|
||||
cfg: cfg,
|
||||
logger: logger,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Publish publishes a message using plain NATS
|
||||
func (p *CorePublisher) Publish(message interface{}) error {
|
||||
topic := p.cfg.Publisher.Topic
|
||||
if topic == "" {
|
||||
p.logger.Error("publisher topic is not configured")
|
||||
return fmt.Errorf("publisher topic is not configured")
|
||||
}
|
||||
|
||||
// Marshal message to JSON
|
||||
messageBytes, err := json.Marshal(message)
|
||||
if err != nil {
|
||||
p.logger.Error("failed to marshal message",
|
||||
zap.String("topic", topic),
|
||||
zap.Error(err),
|
||||
)
|
||||
return fmt.Errorf("failed to marshal message: %w", err)
|
||||
}
|
||||
|
||||
// Publish to plain NATS
|
||||
err = p.conn.Publish(topic, messageBytes)
|
||||
if err != nil {
|
||||
p.logger.Error("failed to publish message",
|
||||
zap.String("topic", topic),
|
||||
zap.Int("message_size", len(messageBytes)),
|
||||
zap.Error(err),
|
||||
)
|
||||
return fmt.Errorf("failed to publish message: %w", err)
|
||||
}
|
||||
|
||||
p.logger.Debug("Published message",
|
||||
zap.String("topic", topic),
|
||||
zap.Int("size", len(messageBytes)),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Publish publishes a message
|
||||
func (p *Publisher) Publish(message interface{}) error {
|
||||
topic := p.cfg.Publisher.Topic
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// isDevLikeEnv checks if the current environment is development-like.
|
||||
func isDevLikeEnv() bool {
|
||||
switch strings.ToLower(os.Getenv("GO_ENV")) {
|
||||
case "", "dev", "development", "test", "testing":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// isJetStreamResourceNotFound checks if an error indicates missing JetStream resources.
|
||||
func isJetStreamResourceNotFound(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, nats.ErrStreamNotFound) || errors.Is(err, nats.ErrConsumerNotFound) {
|
||||
return true
|
||||
}
|
||||
|
||||
// Some JetStream API errors are only exposed via error strings.
|
||||
msg := strings.ToLower(err.Error())
|
||||
return strings.Contains(msg, "stream not found") || strings.Contains(msg, "consumer not found")
|
||||
}
|
||||
|
||||
// sanitizeURLForLogging removes credentials from URLs for safe logging.
|
||||
func sanitizeURLForLogging(rawURL string) string {
|
||||
if rawURL == "" {
|
||||
return ""
|
||||
}
|
||||
|
||||
// Parse URL
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
// If parsing fails, return a safe placeholder
|
||||
return "***"
|
||||
}
|
||||
|
||||
// Rebuild without credentials
|
||||
u.User = nil
|
||||
return u.String()
|
||||
}
|
||||
|
||||
// mapDeliverPolicy maps string configuration to NATS DeliverPolicy.
|
||||
func mapDeliverPolicy(value string) nats.DeliverPolicy {
|
||||
switch strings.ToLower(value) {
|
||||
case "new":
|
||||
return nats.DeliverNewPolicy
|
||||
case "last":
|
||||
return nats.DeliverLastPolicy
|
||||
case "last_per_subject":
|
||||
return nats.DeliverLastPerSubjectPolicy
|
||||
case "sequence":
|
||||
return nats.DeliverByStartSequencePolicy
|
||||
case "time":
|
||||
return nats.DeliverByStartTimePolicy
|
||||
default:
|
||||
return nats.DeliverAllPolicy
|
||||
}
|
||||
}
|
||||
|
||||
// mapReplayPolicy maps string configuration to NATS ReplayPolicy.
|
||||
func mapReplayPolicy(value string) nats.ReplayPolicy {
|
||||
switch strings.ToLower(value) {
|
||||
case "original":
|
||||
return nats.ReplayOriginalPolicy
|
||||
default:
|
||||
return nats.ReplayInstantPolicy
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"os"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("Utils", func() {
|
||||
Describe("isDevLikeEnv", func() {
|
||||
BeforeEach(func() {
|
||||
// Save original value
|
||||
originalEnv := os.Getenv("GO_ENV")
|
||||
DeferCleanup(func() {
|
||||
if originalEnv == "" {
|
||||
os.Unsetenv("GO_ENV")
|
||||
} else {
|
||||
os.Setenv("GO_ENV", originalEnv)
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
It("returns true for dev environment", func() {
|
||||
os.Setenv("GO_ENV", "dev")
|
||||
Expect(isDevLikeEnv()).To(BeTrue())
|
||||
})
|
||||
|
||||
It("returns true for development environment", func() {
|
||||
os.Setenv("GO_ENV", "development")
|
||||
Expect(isDevLikeEnv()).To(BeTrue())
|
||||
})
|
||||
|
||||
It("returns true for test environment", func() {
|
||||
os.Setenv("GO_ENV", "test")
|
||||
Expect(isDevLikeEnv()).To(BeTrue())
|
||||
})
|
||||
|
||||
It("returns true for testing environment", func() {
|
||||
os.Setenv("GO_ENV", "testing")
|
||||
Expect(isDevLikeEnv()).To(BeTrue())
|
||||
})
|
||||
|
||||
It("returns true for empty environment", func() {
|
||||
os.Unsetenv("GO_ENV")
|
||||
Expect(isDevLikeEnv()).To(BeTrue())
|
||||
})
|
||||
|
||||
It("returns false for production environment", func() {
|
||||
os.Setenv("GO_ENV", "prod")
|
||||
Expect(isDevLikeEnv()).To(BeFalse())
|
||||
})
|
||||
|
||||
It("returns false for production environment (uppercase)", func() {
|
||||
os.Setenv("GO_ENV", "PROD")
|
||||
Expect(isDevLikeEnv()).To(BeFalse())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("sanitizeURLForLogging", func() {
|
||||
It("removes credentials from URLs", func() {
|
||||
url := "nats://user:pass@localhost:4222"
|
||||
Expect(sanitizeURLForLogging(url)).To(Equal("nats://localhost:4222"))
|
||||
})
|
||||
|
||||
It("handles URLs without credentials", func() {
|
||||
url := "nats://localhost:4222"
|
||||
Expect(sanitizeURLForLogging(url)).To(Equal("nats://localhost:4222"))
|
||||
})
|
||||
|
||||
It("handles empty strings", func() {
|
||||
Expect(sanitizeURLForLogging("")).To(Equal(""))
|
||||
})
|
||||
|
||||
It("handles invalid URLs", func() {
|
||||
url := "://invalid"
|
||||
result := sanitizeURLForLogging(url)
|
||||
Expect(result).To(Equal("***"))
|
||||
})
|
||||
|
||||
It("handles URLs with user but no password", func() {
|
||||
url := "nats://user@localhost:4222"
|
||||
Expect(sanitizeURLForLogging(url)).To(Equal("nats://localhost:4222"))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user