✨ Update dependencies, enhance NATS consumer configuration, and improve error handling in message processing. Introduce telemetry support with OpenTelemetry for tracing and metrics. Refactor README to include new configuration options and update tests for improved coverage of error scenarios.

internal/infra/config/config.go

@@ -19,6 +19,7 @@ type Config struct {
 	App       AppConfig       `koanf:"app"`
 	Log       LogConfig       `koanf:"log"`
 	Publisher PublisherConfig `koanf:"publisher"`
+	Telemetry TelemetryConfig `koanf:"telemetry"`
 	// Legacy fields for backward compatibility during migration
 	Subscription SubscriptionConfig `koanf:"subscription"`
 	Timeouts     TimeoutsConfig     `koanf:"timeouts"`
@@ -51,6 +52,11 @@ type ConsumerRulesConfig struct {
 	MaxDeliver    int           `koanf:"max_deliver"`
 	AckWait       time.Duration `koanf:"ack_wait"`
 	MaxAckPending int           `koanf:"max_ack_pending"`
+	DeliverPolicy string        `koanf:"deliver_policy"`
+	ReplayPolicy  string        `koanf:"replay_policy"`
+	Backoff       []time.Duration `koanf:"backoff"`
+	StartSequence uint64        `koanf:"start_sequence"`
+	StartTime     string        `koanf:"start_time"`
 }
 
 // PostgresConfig holds PostgreSQL configuration
@@ -78,6 +84,13 @@ type PublisherConfig struct {
 	Topic string `koanf:"topic"`
 }
 
+// TelemetryConfig controls tracing/metrics exporters.
+type TelemetryConfig struct {
+	Enabled  bool   `koanf:"enabled"`
+	Endpoint string `koanf:"endpoint"`
+	Insecure bool   `koanf:"insecure"`
+}
+
 // SubscriptionConfig holds subscription configuration (legacy)
 type SubscriptionConfig struct {
 	Topic      string `koanf:"topic"`
@@ -183,6 +196,15 @@ func LoadConfig() (*Config, error) {
 	if cfg.NATS.ConsumerRules.MaxAckPending == 0 {
 		cfg.NATS.ConsumerRules.MaxAckPending = 1024
 	}
+	if cfg.NATS.ConsumerRules.DeliverPolicy == "" {
+		cfg.NATS.ConsumerRules.DeliverPolicy = "all"
+	}
+	if cfg.NATS.ConsumerRules.ReplayPolicy == "" {
+		cfg.NATS.ConsumerRules.ReplayPolicy = "instant"
+	}
+	if cfg.Telemetry.Endpoint == "" {
+		cfg.Telemetry.Endpoint = ""
+	}
 
 	// Validate configuration
 	if err := cfg.Validate(); err != nil {
@@ -236,6 +258,29 @@ func (c *Config) Validate() error {
 	if c.NATS.ConsumerRules.MaxAckPending < 0 {
 		return fmt.Errorf("nats.consumer.max_ack_pending must be >= 0")
 	}
+	switch strings.ToLower(c.NATS.ConsumerRules.DeliverPolicy) {
+	case "", "all", "new", "last", "last_per_subject", "sequence", "time":
+	default:
+		return fmt.Errorf("nats.consumer.deliver_policy must be one of all,new,last,last_per_subject,sequence,time")
+	}
+	switch strings.ToLower(c.NATS.ConsumerRules.ReplayPolicy) {
+	case "", "instant", "original":
+	default:
+		return fmt.Errorf("nats.consumer.replay_policy must be instant or original")
+	}
+	if c.NATS.ConsumerRules.StartTime != "" {
+		if _, err := time.Parse(time.RFC3339, c.NATS.ConsumerRules.StartTime); err != nil {
+			return fmt.Errorf("nats.consumer.start_time must be RFC3339: %w", err)
+		}
+	}
+	for _, d := range c.NATS.ConsumerRules.Backoff {
+		if d < 0 {
+			return fmt.Errorf("nats.consumer.backoff durations must be >= 0")
+		}
+	}
+	if c.Telemetry.Endpoint == "" && c.Telemetry.Enabled {
+		return fmt.Errorf("telemetry.endpoint is required when telemetry.enabled=true")
+	}
 	return nil
 }
 

internal/infra/config/config_test.go

@@ -3,35 +3,38 @@ package config
 import (
 	"os"
 	"path/filepath"
-	"testing"
 	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
 )
 
-func TestLoadConfig_DefaultAckWait(t *testing.T) {
-	t.Setenv("GO_ENV", "testdefaults")
+var _ = Describe("LoadConfig", func() {
+	var (
+		originalWD string
+	)
 
-	wd, err := os.Getwd()
-	if err != nil {
-		t.Fatalf("failed to get working dir: %v", err)
-	}
-	repoRoot := filepath.Clean(filepath.Join(wd, "..", "..", ".."))
-	if err := os.Chdir(repoRoot); err != nil {
-		t.Fatalf("failed to chdir to repo root: %v", err)
-	}
-	t.Cleanup(func() {
-		_ = os.Chdir(wd)
+	BeforeEach(func() {
+		Expect(os.Setenv("GO_ENV", "testdefaults")).To(Succeed())
+
+		var err error
+		originalWD, err = os.Getwd()
+		Expect(err).NotTo(HaveOccurred())
+
+		repoRoot := filepath.Clean(filepath.Join(originalWD, "..", "..", ".."))
+		Expect(os.Chdir(repoRoot)).To(Succeed())
 	})
 
-	cfg, err := LoadConfig()
-	if err != nil {
-		t.Fatalf("failed to load config: %v", err)
-	}
+	AfterEach(func() {
+		Expect(os.Chdir(originalWD)).To(Succeed())
+	})
 
-	want := 30 * time.Second
-	if cfg.Timeouts.AckWait != want {
-		t.Fatalf("expected timeouts.ack_wait to default to %v, got %v", want, cfg.Timeouts.AckWait)
-	}
-	if cfg.NATS.ConsumerRules.AckWait != want {
-		t.Fatalf("expected consumer ack_wait to default to %v, got %v", want, cfg.NATS.ConsumerRules.AckWait)
-	}
-}
+	It("defaults ack waits when not provided", func() {
+		cfg, err := LoadConfig()
+		Expect(err).NotTo(HaveOccurred())
+
+		want := 30 * time.Second
+		Expect(cfg.Timeouts.AckWait).To(Equal(want))
+		Expect(cfg.NATS.ConsumerRules.AckWait).To(Equal(want))
+	})
+})

internal/infra/config/suite_test.go

@@ -0,0 +1,13 @@
+package config
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestInfraConfig(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Infra Config Suite")
+}

internal/infra/nats/consumer.go

@@ -6,9 +6,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"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"
 )
 
@@ -21,6 +26,11 @@ type Consumer struct {
 	logger       *zap.Logger
 	subject      string
 	consumerName string
+	meter        metric.Meter
+	ackPending   metric.Int64Histogram
+	redelivered  metric.Int64Histogram
+	pending      metric.Int64Histogram
+	delivered    metric.Int64Histogram
 }
 
 // ProvideConsumer creates a NATS consumer
@@ -47,6 +57,7 @@ func ProvideConsumer(
 		subject:      subject,
 		consumerName: consumerName,
 	}
+	consumer.initMetrics()
 
 	// Create consumer if it doesn't exist
 	if err := consumer.ensureConsumer(); err != nil {
@@ -74,12 +85,28 @@ func (c *Consumer) ensureConsumer() error {
 
 	consumerConfig := &nats.ConsumerConfig{
 		Durable:       c.consumerName,
-		DeliverPolicy: nats.DeliverAllPolicy,
+		DeliverPolicy: mapDeliverPolicy(c.cfg.NATS.ConsumerRules.DeliverPolicy),
 		AckPolicy:     nats.AckExplicitPolicy,
 		AckWait:       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
+		}
 	}
 
 	_, err := c.js.AddConsumer(streamName, consumerConfig)
@@ -93,6 +120,8 @@ func (c *Consumer) ensureConsumer() error {
 			zap.String("stream", streamName),
 			zap.String("subject", c.subject),
 			zap.Duration("ack_wait", ackWait),
+			zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy),
+			zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy),
 		)
 	}
 
@@ -133,6 +162,9 @@ func (c *Consumer) Start(ctx context.Context) error {
 		zap.Duration("batch_timeout", batchTimeout),
 		zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver),
 		zap.Duration("ack_wait", c.cfg.NATS.ConsumerRules.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)
@@ -177,8 +209,8 @@ func (c *Consumer) Start(ctx context.Context) error {
 					continue
 				}
 
-				// Transient error: request redelivery
-				if nakErr := msg.Nak(); nakErr != nil {
+				// Transient error: request redelivery with optional delay
+				if nakErr := c.nakWithStrategy(msg); nakErr != nil {
 					c.logger.Error("Failed to NAK message", zap.Error(nakErr))
 				}
 				continue
@@ -221,6 +253,7 @@ func (c *Consumer) emitConsumerStats(ctx context.Context, streamName string) {
 				zap.Uint64("delivered_consumer_seq", uint64(info.Delivered.Consumer)),
 				zap.Uint64("delivered_stream_seq", uint64(info.Delivered.Stream)),
 			)
+			c.recordConsumerMetrics(ctx, info)
 		}
 	}
 }
@@ -254,10 +287,80 @@ func (c *Consumer) Shutdown(ctx context.Context) error {
 	}
 }
 
+func (c *Consumer) initMetrics() {
+	meter := otel.Meter("caatsm/nats")
+	c.meter = meter
+
+	if hist, err := meter.Int64Histogram("nats.consumer.ack_pending"); err == nil {
+		c.ackPending = hist
+	}
+	if hist, err := meter.Int64Histogram("nats.consumer.redelivered"); err == nil {
+		c.redelivered = hist
+	}
+	if hist, err := meter.Int64Histogram("nats.consumer.pending"); err == nil {
+		c.pending = hist
+	}
+	if hist, err := meter.Int64Histogram("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))
+	}
+}
+
+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" {
@@ -276,9 +379,12 @@ func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error {
 
 	// 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
 }
 
@@ -294,3 +400,29 @@ func (c *Consumer) resolveMsgID(msg *nats.Msg) (string, string, error) {
 
 	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
+	}
+}

internal/infra/postgres/repository.go

@@ -5,10 +5,16 @@ import (
 	"caatsm/internal/adapter/mapper"
 	"caatsm/internal/domain"
 	"context"
+	"encoding/json"
 	"fmt"
+	"time"
 
+	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
 	"go.uber.org/zap"
 )
 
@@ -30,8 +36,14 @@ func ProvideRepository(pool *pgxpool.Pool, logger *zap.Logger) (adapter.Reposito
 
 // InsertOne inserts a single telegram message
 func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) error {
+	ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertOne")
+	defer span.End()
+	span.SetAttributes(attribute.String("db.table", "aviation.telegrams"))
+
 	row, err := r.mapper.ToDBRow(msg)
 	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return fmt.Errorf("failed to map message to DB row: %w", err)
 	}
 
@@ -39,18 +51,21 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e
 		INSERT INTO aviation.telegrams (
 			uuid, message_id, date_time, priority_indicator, primary_address,
 			secondary_addresses, originator, originator_date_time, category,
-			content, body_data, received_at, parsed_at, dispatched_at, need_dispatch
+			content, body_data, status, error_reason,
+			received_at, parsed_at, dispatched_at, need_dispatch
 		) VALUES (
-			$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15
+			$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17
 		)
 		ON CONFLICT (uuid) DO NOTHING
 	`
 
 	tag, err := r.pool.Exec(ctx, query,
 		row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[8],
-		row[9], row[10], row[11], row[12], row[13], row[14],
+		row[9], row[10], row[11], row[12], row[13], row[14], row[15], row[16],
 	)
 	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return fmt.Errorf("failed to insert message: %w", err)
 	}
 
@@ -72,6 +87,10 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e
 
 // InsertBatch inserts multiple telegram messages in a batch using CopyFrom
 func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error {
+	ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertBatch")
+	defer span.End()
+	span.SetAttributes(attribute.String("db.table", "aviation.telegrams"))
+
 	if len(msgs) == 0 {
 		return nil
 	}
@@ -93,14 +112,18 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa
 		[]string{
 			"uuid", "message_id", "date_time", "priority_indicator", "primary_address",
 			"secondary_addresses", "originator", "originator_date_time", "category",
-			"content", "body_data", "received_at", "parsed_at", "dispatched_at", "need_dispatch",
+			"content", "body_data", "status", "error_reason",
+			"received_at", "parsed_at", "dispatched_at", "need_dispatch",
 		},
 		pgx.CopyFromRows(rows),
 	)
 	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 		return fmt.Errorf("failed to batch insert messages: %w", err)
 	}
 
+	span.SetAttributes(attribute.Int64("db.inserted", copyCount))
 	r.logger.Info("Batch inserted messages",
 		zap.Int("count", int(copyCount)),
 		zap.Int("attempted", len(msgs)),
@@ -108,3 +131,69 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa
 
 	return nil
 }
+
+// InsertRaw inserts a failed telegram into aviation.telegrams_raw for post-processing.
+func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) error {
+	ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertRaw")
+	defer span.End()
+	span.SetAttributes(attribute.String("db.table", "aviation.telegrams_raw"))
+
+	if msg == nil {
+		err := fmt.Errorf("message is nil")
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return fmt.Errorf("message is nil")
+	}
+	if msg.Uuid == "" {
+		msg.Uuid = uuid.NewString()
+	}
+	if msg.ReceivedAt.IsZero() {
+		msg.ReceivedAt = time.Now()
+	}
+
+	metadata := map[string]interface{}{
+		"message_id": msg.MessageID,
+		"category":   msg.Category,
+		"comments":   msg.Comments,
+	}
+	metadataJSON, err := json.Marshal(metadata)
+	if err != nil {
+		return fmt.Errorf("failed to marshal metadata: %w", err)
+	}
+
+	query := `
+		INSERT INTO aviation.telegrams_raw (
+			uuid, status, error_reason, content, received_at, metadata
+		) VALUES (
+			$1, $2, $3, $4, $5, $6
+		)
+		ON CONFLICT (uuid) DO UPDATE
+		SET status = EXCLUDED.status,
+		    error_reason = EXCLUDED.error_reason,
+			content = EXCLUDED.content,
+			received_at = EXCLUDED.received_at,
+			metadata = EXCLUDED.metadata
+	`
+
+	_, err = r.pool.Exec(ctx, query,
+		msg.Uuid,
+		string(msg.Status),
+		msg.ErrorReason,
+		msg.Content,
+		msg.ReceivedAt,
+		metadataJSON,
+	)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return fmt.Errorf("failed to insert raw telegram: %w", err)
+	}
+
+	span.SetAttributes(attribute.String("telegram.uuid", msg.Uuid), attribute.String("telegram.status", string(msg.Status)))
+	r.logger.Debug("Persisted raw telegram",
+		zap.String("uuid", msg.Uuid),
+		zap.String("status", string(msg.Status)),
+	)
+
+	return nil
+}