✨ Update README with configuration details, add default test configuration, and implement TelegramMapper tests. Enhance message processing to set timestamps and improve error handling in the processor. Refactor NATS consumer configuration and ensure proper handling of message attributes.

README.md

@@ -70,6 +70,9 @@ psql -U postgres -f internal/repository/telegrams.ddl
 
 Configuration is loaded from TOML files and environment variables. The configuration file should be located at `configs/config.{env}.toml` where `{env}` is determined by the `GO_ENV` environment variable (defaults to `dev`).
 
+- `nats.url` and `nats.stream` are required (the latter defaults to `TELEGRAM` when omitted).
+- `subscription.topic` is optional; when not provided the application subscribes to `telegram.>`.
+
 ### Configuration Structure
 
 ```toml
@@ -92,6 +95,7 @@ ack_wait = "30s"
 max_ack_pending = 1024
 
 [subscription]
+# Optional. Defaults to "telegram.>" when omitted.
 topic = "telegram.serial"
 
 [publisher]
@@ -110,6 +114,10 @@ monitor_interval = "30s"
 [log]
 level = "info"
 format = "json"
+
+### Timeouts and Ack Wait
+
+`[timeouts]` is optional, but if you plan to tune JetStream redelivery you should set `timeouts.ack_wait` and/or `[nats.consumer].ack_wait`. When neither is specified the application defaults both values to `30s`, ensuring predictable redelivery timing.
 ```
 
 ### Environment Variables
@@ -472,6 +480,8 @@ Logs include contextual information:
 
 - **Batch Processing**: Messages are processed in configurable batches (default: 50)
 - **Database Inserts**: Uses PostgreSQL `COPY FROM` for efficient batch inserts
+  via `Repository.InsertBatch`. The default processor issues single inserts,
+  but you can switch to buffered batches in high-throughput deployments.
 - **Connection Pooling**: Configurable PostgreSQL connection pool
 - **JetStream**: Reliable message delivery with automatic retries
 

configs/config.testdefaults.toml

@@ -0,0 +1,21 @@
+[nats]
+url = "nats://localhost:4222"
+stream = "TELEGRAM"
+
+[publisher]
+topic = "telegram.json"
+
+[postgres]
+url = "postgres://user:password@localhost:5432/aviation?sslmode=disable"
+max_conns = 2
+min_conns = 1
+
+[app]
+batch_size = 1
+batch_timeout = "1s"
+monitor_interval = "1s"
+
+[log]
+level = "info"
+format = "json"
+

internal/adapter/mapper/telegram.go

@@ -4,6 +4,8 @@ import (
 	"caatsm/internal/domain"
 	"encoding/json"
 	"fmt"
+	"time"
+
 	"github.com/google/uuid"
 )
 
@@ -42,28 +44,87 @@ func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, erro
 	secondaryAddresses := msg.SecondaryAddresses
 
 	return []interface{}{
-		msgUUID,                    // uuid
+		msgUUID,                // uuid
-		msg.MessageID,              // message_id
+		msg.MessageID,          // message_id
-		msg.DateTime,               // date_time
+		msg.DateTime,           // date_time
-		msg.PriorityIndicator,      // priority_indicator
+		msg.PriorityIndicator,  // priority_indicator
-		msg.PrimaryAddress,         // primary_address
+		msg.PrimaryAddress,     // primary_address
-		secondaryAddresses,         // secondary_addresses (TEXT)
+		secondaryAddresses,     // secondary_addresses (TEXT)
-		msg.Originator,             // originator
+		msg.Originator,         // originator
-		msg.OriginatorDateTime,     // originator_date_time
+		msg.OriginatorDateTime, // originator_date_time
-		msg.Category,               // category
+		msg.Category,           // category
-		msg.Content,                // content (TEXT, original message)
+		msg.Content,            // content (TEXT, original message)
-		bodyDataJSON,               // body_data (JSONB)
+		bodyDataJSON,           // body_data (JSONB)
-		msg.ReceivedAt,             // received_at
+		msg.ReceivedAt,         // received_at
-		msg.ParsedAt,               // parsed_at
+		msg.ParsedAt,           // parsed_at
-		msg.DispatchedAt,           // dispatched_at
+		msg.DispatchedAt,       // dispatched_at
-		msg.NeedDispatch,           // need_dispatch
+		msg.NeedDispatch,       // need_dispatch
 	}, nil
 }
 
 // FromDBRow converts a database row to a domain.ParsedMessage
 func (m *TelegramMapper) FromDBRow(row []interface{}) (*domain.ParsedMessage, error) {
-	// This is a placeholder - will be implemented if needed for queries
+	const expectedColumns = 15
-	// For now, we only need ToDBRow for inserts
+	if len(row) < expectedColumns {
-	return nil, fmt.Errorf("FromDBRow not implemented")
+		return nil, fmt.Errorf("expected %d columns, got %d", expectedColumns, len(row))
-}
+	}
 
+	msgUUID, ok := row[0].(uuid.UUID)
+	if !ok {
+		return nil, fmt.Errorf("column 0 must be uuid.UUID")
+	}
+
+	toString := func(v interface{}) string {
+		if v == nil {
+			return ""
+		}
+		if s, ok := v.(string); ok {
+			return s
+		}
+		return fmt.Sprint(v)
+	}
+
+	parseTime := func(v interface{}) time.Time {
+		if v == nil {
+			return time.Time{}
+		}
+		if t, ok := v.(time.Time); ok {
+			return t
+		}
+		return time.Time{}
+	}
+
+	var bodyData interface{}
+	if raw := row[10]; raw != nil {
+		switch val := raw.(type) {
+		case []byte:
+			if len(val) > 0 {
+				if err := json.Unmarshal(val, &bodyData); err != nil {
+					return nil, fmt.Errorf("failed to unmarshal body data: %w", err)
+				}
+			}
+		default:
+			bodyData = val
+		}
+	}
+
+	needDispatch, _ := row[14].(bool)
+
+	return &domain.ParsedMessage{
+		Uuid:               msgUUID.String(),
+		MessageID:          toString(row[1]),
+		DateTime:           toString(row[2]),
+		PriorityIndicator:  toString(row[3]),
+		PrimaryAddress:     toString(row[4]),
+		SecondaryAddresses: toString(row[5]),
+		Originator:         toString(row[6]),
+		OriginatorDateTime: toString(row[7]),
+		Category:           toString(row[8]),
+		Content:            toString(row[9]),
+		BodyData:           bodyData,
+		ReceivedAt:         parseTime(row[11]),
+		ParsedAt:           parseTime(row[12]),
+		DispatchedAt:       parseTime(row[13]),
+		NeedDispatch:       needDispatch,
+	}, nil
+}

internal/adapter/mapper/telegram_test.go

@@ -0,0 +1,71 @@
+package mapper
+
+import (
+	"testing"
+	"time"
+
+	"caatsm/internal/domain"
+
+	"github.com/google/uuid"
+)
+
+func TestTelegramMapper_ToDBRow_GeneratesUUIDWhenEmpty(t *testing.T) {
+	mapper := NewTelegramMapper()
+	msg := &domain.ParsedMessage{}
+
+	row, err := mapper.ToDBRow(msg)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	value, ok := row[0].(uuid.UUID)
+	if !ok {
+		t.Fatalf("expected first column to be uuid.UUID, got %T", row[0])
+	}
+	if value == uuid.Nil {
+		t.Fatalf("expected generated uuid to be non-nil")
+	}
+}
+
+func TestTelegramMapper_FromDBRow_RoundTrip(t *testing.T) {
+	mapper := NewTelegramMapper()
+	now := time.Now().UTC()
+
+	original := &domain.ParsedMessage{
+		Uuid:               uuid.NewString(),
+		MessageID:          "TMQ1324",
+		DateTime:           "150631",
+		PriorityIndicator:  "FF",
+		PrimaryAddress:     "ZBTJZPZX",
+		SecondaryAddresses: "150630 ZBACZQZX",
+		Originator:         "ORIGIN",
+		OriginatorDateTime: "150630",
+		Category:           "FPL",
+		Content:            "raw telegram",
+		BodyData:           map[string]string{"key": "value"},
+		ReceivedAt:         now,
+		ParsedAt:           now,
+		DispatchedAt:       now,
+		NeedDispatch:       true,
+	}
+
+	row, err := mapper.ToDBRow(original)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	roundTrip, err := mapper.FromDBRow(row)
+	if err != nil {
+		t.Fatalf("unexpected error reading row: %v", err)
+	}
+
+	if roundTrip.Uuid != original.Uuid {
+		t.Fatalf("expected uuid %s, got %s", original.Uuid, roundTrip.Uuid)
+	}
+	if roundTrip.MessageID != original.MessageID {
+		t.Fatalf("expected message_id %s, got %s", original.MessageID, roundTrip.MessageID)
+	}
+	if roundTrip.NeedDispatch != original.NeedDispatch {
+		t.Fatalf("expected need_dispatch %v, got %v", original.NeedDispatch, roundTrip.NeedDispatch)
+	}
+}

internal/adapter/parser/parser.go

@@ -6,5 +6,6 @@ import "caatsm/internal/domain"
 type Parser interface {
 	// Parse parses a raw message string and returns a ParsedMessage
 	Parse(rawText string) *domain.ParsedMessage
-}
 
+	// TODO: consider returning (*domain.ParsedMessage, error) to surface parse failures explicitly.
+}

internal/app/processor.go

@@ -5,6 +5,8 @@ import (
 	"caatsm/internal/adapter/parser"
 	"context"
 	"fmt"
+	"strings"
+	"time"
 
 	"go.uber.org/zap"
 )
@@ -38,26 +40,42 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string)
 		return Permanent(fmt.Errorf("empty message"))
 	}
 
-	// Parse the message
+	receivedAt := time.Now()
+
 	parsed := p.parser.Parse(string(raw))
 	if parsed == nil {
 		return Permanent(fmt.Errorf("parser returned nil"))
 	}
 
-	// Set the message ID from NATS
+	if msgID != "" {
-	parsed.Uuid = msgID
+		if parsed.Comments == "" {
+			parsed.Comments = fmt.Sprintf("nats_msg_id=%s", msgID)
+		} else if !strings.Contains(parsed.Comments, "nats_msg_id=") {
+			parsed.Comments = fmt.Sprintf("%s; nats_msg_id=%s", parsed.Comments, msgID)
+		}
+	}
+	if parsed.ReceivedAt.IsZero() {
+		parsed.ReceivedAt = receivedAt
+	}
+	if parsed.ParsedAt.IsZero() {
+		parsed.ParsedAt = time.Now()
+	}
 
 	// Log parsing result
 	if !parsed.Parsed {
-		p.logger.Info("Message not parsed",
+		p.logger.Warn("Message not parsed",
 			zap.String("msg_id", msgID),
-			zap.String("content_preview", truncateContent(parsed.Content, 200)),
+			zap.String("message_id", parsed.MessageID),
+			zap.String("category", parsed.Category),
+			zap.String("content_preview", truncateContent(parsed.Content, 256)),
 		)
 	} else {
 		p.logger.Info("Message parsed successfully",
 			zap.String("msg_id", msgID),
 			zap.String("message_id", parsed.MessageID),
 			zap.String("category", parsed.Category),
+			zap.Time("received_at", parsed.ReceivedAt),
+			zap.Time("parsed_at", parsed.ParsedAt),
 		)
 	}
 

internal/app/processor_test.go

@@ -1,16 +1,182 @@
 package app
 
 import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
 	"caatsm/internal/adapter"
 	"caatsm/internal/adapter/parser"
 	"caatsm/internal/domain"
-	"context"
-	"errors"
-	"testing"
 
+	"github.com/google/uuid"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 )
 
+func TestHandleEmptyMessageIsPermanent(t *testing.T) {
+	proc := newTestProcessor(&stubParser{}, &stubRepository{}, &stubPublisher{})
+	err := proc.Handle(context.Background(), nil, "id-1")
+	if err == nil || !IsPermanent(err) {
+		t.Fatalf("expected permanent error for empty message, got %v", err)
+	}
+}
+
+func TestHandleNilParserResultIsPermanent(t *testing.T) {
+	proc := newTestProcessor(&stubParser{value: nil}, &stubRepository{}, &stubPublisher{})
+	err := proc.Handle(context.Background(), []byte("payload"), "id-2")
+	if err == nil || !IsPermanent(err) {
+		t.Fatalf("expected permanent error for nil parser result, got %v", err)
+	}
+}
+
+func TestHandleSuccessDoesNotOverwriteUuid(t *testing.T) {
+	originalUUID := uuid.NewString()
+	parsed := &domain.ParsedMessage{Uuid: originalUUID, Parsed: true}
+
+	repo := &stubRepository{}
+	pub := &stubPublisher{}
+	proc := newTestProcessor(&stubParser{value: parsed}, repo, pub)
+
+	const msgID = "msg-123"
+	if err := proc.Handle(context.Background(), []byte("payload"), msgID); err != nil {
+		t.Fatalf("expected success, got %v", err)
+	}
+
+	if repo.last() == nil {
+		t.Fatalf("expected message to be inserted")
+	}
+	if repo.last().Uuid != originalUUID {
+		t.Fatalf("expected uuid to remain %s, got %s", originalUUID, repo.last().Uuid)
+	}
+	if !strings.Contains(repo.last().Comments, "nats_msg_id=msg-123") {
+		t.Fatalf("expected comments to contain msg id, got %q", repo.last().Comments)
+	}
+	if pub.last == nil {
+		t.Fatalf("expected publisher to receive message")
+	}
+}
+
+func TestHandlePublisherErrorIsPermanent(t *testing.T) {
+	parsed := &domain.ParsedMessage{Parsed: true}
+
+	repo := &stubRepository{}
+	pub := &stubPublisher{err: errors.New("publish failed")}
+	proc := newTestProcessor(&stubParser{value: parsed}, repo, pub)
+
+	err := proc.Handle(context.Background(), []byte("payload"), "id-3")
+	if err == nil {
+		t.Fatalf("expected error when publisher fails")
+	}
+	if !IsPermanent(err) {
+		t.Fatalf("publisher failure should be permanent")
+	}
+	if repo.last() == nil {
+		t.Fatalf("expected message to insert before publish failure")
+	}
+}
+
+func TestMessageProcessor_Handle_SetsReceivedAndParsedAtWhenZero(t *testing.T) {
+	parsed := &domain.ParsedMessage{
+		Uuid:   uuid.NewString(),
+		Parsed: true,
+	}
+	repo := &stubRepository{}
+	publisher := &stubPublisher{}
+	logger := zap.NewNop()
+	processor := NewMessageProcessor(&stubParser{value: parsed}, repo, publisher, logger)
+
+	start := time.Now()
+	if err := processor.Handle(context.Background(), []byte("raw"), "msg-4"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	saved := repo.last()
+	if saved == nil {
+		t.Fatal("expected repository to receive a message")
+	}
+	if saved.ReceivedAt.IsZero() || saved.ParsedAt.IsZero() {
+		t.Fatalf("expected timestamps to be set, got received=%v parsed=%v", saved.ReceivedAt, saved.ParsedAt)
+	}
+	if saved.ReceivedAt.Before(start.Add(-time.Second)) || saved.ParsedAt.Before(start.Add(-time.Second)) {
+		t.Fatalf("timestamps look stale: received=%v parsed=%v", saved.ReceivedAt, saved.ParsedAt)
+	}
+}
+
+func TestMessageProcessor_Handle_DoesNotOverrideExistingTimestamps(t *testing.T) {
+	received := time.Now().Add(-2 * time.Minute)
+	parsedAt := time.Now().Add(-time.Minute)
+
+	parsed := &domain.ParsedMessage{
+		Uuid:       uuid.NewString(),
+		Parsed:     true,
+		ReceivedAt: received,
+		ParsedAt:   parsedAt,
+	}
+	repo := &stubRepository{}
+	publisher := &stubPublisher{}
+	logger := zap.NewNop()
+	processor := NewMessageProcessor(&stubParser{value: parsed}, repo, publisher, logger)
+
+	if err := processor.Handle(context.Background(), []byte("raw"), "msg-5"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	saved := repo.last()
+	if saved.ReceivedAt != received {
+		t.Fatalf("expected received_at to remain %v, got %v", received, saved.ReceivedAt)
+	}
+	if saved.ParsedAt != parsedAt {
+		t.Fatalf("expected parsed_at to remain %v, got %v", parsedAt, saved.ParsedAt)
+	}
+}
+
+func TestContentPreview_TruncatesLongContent(t *testing.T) {
+	longContent := strings.Repeat("a", 1024)
+	preview := truncateContent(longContent, 256)
+	if len(preview) != 256 {
+		t.Fatalf("expected preview length 256, got %d", len(preview))
+	}
+	if !strings.HasSuffix(preview, "...") {
+		t.Fatalf("expected preview to end with ellipsis, got %q", preview[len(preview)-10:])
+	}
+}
+
+func TestMessageProcessor_Handle_NotParsedLogsPreviewOnly(t *testing.T) {
+	core, logs := observer.New(zap.WarnLevel)
+	logger := zap.New(core)
+
+	parser := &stubParser{
+		value: &domain.ParsedMessage{
+			Content: strings.Repeat("x", 1024),
+			Parsed:  false,
+		},
+	}
+	repo := &stubRepository{}
+	publisher := &stubPublisher{}
+	processor := NewMessageProcessor(parser, repo, publisher, logger)
+
+	if err := processor.Handle(context.Background(), []byte("raw"), "msg-6"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	entries := logs.FilterMessage("Message not parsed").All()
+	if len(entries) == 0 {
+		t.Fatal("expected a warning log for unparsed message")
+	}
+	preview, ok := entries[0].ContextMap()["content_preview"].(string)
+	if !ok {
+		t.Fatal("expected content_preview field in log")
+	}
+	if len(preview) > 256 {
+		t.Fatalf("expected preview <= 256 chars, got %d", len(preview))
+	}
+}
+
+func newTestProcessor(p parser.Parser, repo adapter.Repository, pub adapter.Publisher) *MessageProcessor {
+	return NewMessageProcessor(p, repo, pub, zap.NewNop())
+}
+
 type stubParser struct {
 	value *domain.ParsedMessage
 }
@@ -36,6 +202,13 @@ func (s *stubRepository) InsertBatch(ctx context.Context, msgs []*domain.ParsedM
 	return errors.New("not implemented")
 }
 
+func (s *stubRepository) last() *domain.ParsedMessage {
+	if len(s.inserted) == 0 {
+		return nil
+	}
+	return s.inserted[len(s.inserted)-1]
+}
+
 type stubPublisher struct {
 	last interface{}
 	err  error
@@ -45,71 +218,3 @@ func (s *stubPublisher) Publish(message interface{}) error {
 	s.last = message
 	return s.err
 }
-
-func newTestProcessor(p parser.Parser, repo adapter.Repository, pub adapter.Publisher) *MessageProcessor {
-	return NewMessageProcessor(p, repo, pub, zap.NewNop())
-}
-
-func TestHandleEmptyMessageIsPermanent(t *testing.T) {
-	proc := newTestProcessor(&stubParser{}, &stubRepository{}, &stubPublisher{})
-	err := proc.Handle(context.Background(), nil, "id-1")
-	if err == nil || !IsPermanent(err) {
-		t.Fatalf("expected permanent error for empty message, got %v", err)
-	}
-}
-
-func TestHandleNilParserResultIsPermanent(t *testing.T) {
-	proc := newTestProcessor(&stubParser{value: nil}, &stubRepository{}, &stubPublisher{})
-	err := proc.Handle(context.Background(), []byte("payload"), "id-2")
-	if err == nil || !IsPermanent(err) {
-		t.Fatalf("expected permanent error for nil parser result, got %v", err)
-	}
-}
-
-func TestHandleSuccessSetsUuidAndPublishes(t *testing.T) {
-	parsed := domain.NewParsedMessage()
-	parsed.Parsed = true
-
-	repo := &stubRepository{}
-	pub := &stubPublisher{}
-	proc := newTestProcessor(&stubParser{value: parsed}, repo, pub)
-
-	const msgID = "uuid-123"
-	err := proc.Handle(context.Background(), []byte("payload"), msgID)
-	if err != nil {
-		t.Fatalf("expected success, got %v", err)
-	}
-
-	if len(repo.inserted) != 1 {
-		t.Fatalf("expected one inserted message, got %d", len(repo.inserted))
-	}
-	if repo.inserted[0].Uuid != msgID {
-		t.Fatalf("expected message uuid to be %s, got %s", msgID, repo.inserted[0].Uuid)
-	}
-	if pub.last == nil {
-		t.Fatalf("expected publisher to receive message")
-	}
-	if pub.last != repo.inserted[0] {
-		t.Fatalf("publisher received unexpected message pointer")
-	}
-}
-
-func TestHandlePublisherErrorIsPermanent(t *testing.T) {
-	parsed := domain.NewParsedMessage()
-	parsed.Parsed = true
-
-	repo := &stubRepository{}
-	pub := &stubPublisher{err: errors.New("publish failed")}
-	proc := newTestProcessor(&stubParser{value: parsed}, repo, pub)
-
-	err := proc.Handle(context.Background(), []byte("payload"), "id-3")
-	if err == nil {
-		t.Fatalf("expected error when publisher fails")
-	}
-	if !IsPermanent(err) {
-		t.Fatalf("publisher failure should be permanent")
-	}
-	if len(repo.inserted) != 1 {
-		t.Fatalf("expected message to insert before publish failure")
-	}
-}

internal/domain/aviation.go

@@ -34,7 +34,7 @@ MessageID: "TMQ1324".
 DateTime: "150631".
 PriorityIndicator: "FF".
 PrimaryAddress: "ZBTJZPZX".
-SecondaryAddresses: ["150630", "ZBACZQZX"].
+SecondaryAddresses: "150630 ZBACZQZX".
 Originator: "".
 OriginatorDateTime: "".
 Category: "".
@@ -72,7 +72,7 @@ MessageID: "XMP4567".
 DateTime: "120915".
 PriorityIndicator: "DD".
 PrimaryAddress: "KLAXZPZX".
-SecondaryAddresses: ["120914", "KSFOZQZX"].
+SecondaryAddresses: "120914 KSFOZQZX".
 Originator: "".
 OriginatorDateTime: "".
 Category: "".
@@ -92,7 +92,7 @@ type ParsedMessage struct {
 	DateTime           string      `json:"dateTime"`                     // 日期时间: The date and time of the message (e.g., '150631').
 	PriorityIndicator  string      `json:"priorityIndicator"`            // 优先级标识: The priority level of the message (e.g., 'FF').
 	PrimaryAddress     string      `json:"primaryAddress"`               // 主要地址: The primary recipient address (e.g., 'ZBTJZPZX').
-	SecondaryAddresses string      `json:"secondaryAddresses,omitempty"` // 次要地址: Additional recipient addresses (e.g., ['150630', 'ZBACZQZX']).
+	SecondaryAddresses string      `json:"secondaryAddresses,omitempty"` // 次要地址: Additional recipient addresses (space-separated string such as "150630 ZBACZQZX").
 	Originator         string      `json:"originator,omitempty"`         // 发件人: The sender of the message.
 	OriginatorDateTime string      `json:"originatorDateTime,omitempty"` // 发件日期时间: The date and time when the originator sent the message.
 	Category           string      `json:"category,omitempty"`           // 类别: The category of the message.

internal/infra/config/config.go

@@ -174,12 +174,11 @@ func LoadConfig() (*Config, error) {
 	if cfg.NATS.ConsumerRules.MaxDeliver == 0 {
 		cfg.NATS.ConsumerRules.MaxDeliver = 5
 	}
+	if cfg.Timeouts.AckWait == 0 {
+		cfg.Timeouts.AckWait = 30 * time.Second
+	}
 	if cfg.NATS.ConsumerRules.AckWait == 0 {
-		if cfg.Timeouts.AckWait != 0 {
+		cfg.NATS.ConsumerRules.AckWait = cfg.Timeouts.AckWait
-			cfg.NATS.ConsumerRules.AckWait = cfg.Timeouts.AckWait
-		} else {
-			cfg.NATS.ConsumerRules.AckWait = 30 * time.Second
-		}
 	}
 	if cfg.NATS.ConsumerRules.MaxAckPending == 0 {
 		cfg.NATS.ConsumerRules.MaxAckPending = 1024
@@ -198,8 +197,8 @@ func (c *Config) Validate() error {
 	if c.NATS.URL == "" {
 		return fmt.Errorf("nats.url is required")
 	}
-	if c.Subscription.Topic == "" && c.NATS.Stream == "" {
+	if c.NATS.Stream == "" {
-		return fmt.Errorf("subscription.topic or nats.stream is required")
+		return fmt.Errorf("nats.stream is required")
 	}
 	if c.Publisher.Topic == "" {
 		return fmt.Errorf("publisher.topic is required")

internal/infra/config/config_test.go

@@ -0,0 +1,37 @@
+package config
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestLoadConfig_DefaultAckWait(t *testing.T) {
+	t.Setenv("GO_ENV", "testdefaults")
+
+	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)
+	})
+
+	cfg, err := LoadConfig()
+	if err != nil {
+		t.Fatalf("failed to load config: %v", err)
+	}
+
+	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)
+	}
+}

internal/infra/nats/consumer.go

@@ -6,9 +6,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
+
 	"github.com/nats-io/nats.go"
 	"go.uber.org/zap"
-	"time"
 )
 
 // Consumer handles NATS JetStream message consumption
@@ -69,6 +70,7 @@ func (c *Consumer) ensureConsumer() error {
 	if ackWait == 0 {
 		ackWait = 30 * time.Second
 	}
+	c.cfg.NATS.ConsumerRules.AckWait = ackWait
 
 	consumerConfig := &nats.ConsumerConfig{
 		Durable:       c.consumerName,
@@ -90,6 +92,7 @@ func (c *Consumer) ensureConsumer() error {
 			zap.String("consumer", c.consumerName),
 			zap.String("stream", streamName),
 			zap.String("subject", c.subject),
+			zap.Duration("ack_wait", ackWait),
 		)
 	}
 
@@ -157,6 +160,7 @@ func (c *Consumer) Start(ctx context.Context) error {
 		}
 
 		// Process each message
+		// TODO: consider buffering messages to take advantage of Repository.InsertBatch for higher throughput.
 		for _, msg := range msgs {
 			if err := c.processMessage(ctx, msg); err != nil {
 				isPermanent := app.IsPermanent(err)