From 88d5922f904ee26203a1c11c5526f86ca471ef4b Mon Sep 17 00:00:00 2001 From: windyboy Date: Fri, 14 Nov 2025 23:18:10 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Update=20README=20with=20configurat?= =?UTF-8?q?ion=20details,=20add=20default=20test=20configuration,=20and=20?= =?UTF-8?q?implement=20TelegramMapper=20tests.=20Enhance=20message=20proce?= =?UTF-8?q?ssing=20to=20set=20timestamps=20and=20improve=20error=20handlin?= =?UTF-8?q?g=20in=20the=20processor.=20Refactor=20NATS=20consumer=20config?= =?UTF-8?q?uration=20and=20ensure=20proper=20handling=20of=20message=20att?= =?UTF-8?q?ributes.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 10 + configs/config.testdefaults.toml | 21 ++ internal/adapter/mapper/telegram.go | 99 +++++++-- internal/adapter/mapper/telegram_test.go | 71 +++++++ internal/adapter/parser/parser.go | 3 +- internal/app/processor.go | 28 ++- internal/app/processor_test.go | 247 ++++++++++++++++------- internal/domain/aviation.go | 6 +- internal/infra/config/config.go | 13 +- internal/infra/config/config_test.go | 37 ++++ internal/infra/nats/consumer.go | 6 +- 11 files changed, 434 insertions(+), 107 deletions(-) create mode 100644 configs/config.testdefaults.toml create mode 100644 internal/adapter/mapper/telegram_test.go create mode 100644 internal/infra/config/config_test.go diff --git a/README.md b/README.md index 7188e19..7cde5bb 100644 --- a/README.md +++ b/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 diff --git a/configs/config.testdefaults.toml b/configs/config.testdefaults.toml new file mode 100644 index 0000000..8ce2e73 --- /dev/null +++ b/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" + diff --git a/internal/adapter/mapper/telegram.go b/internal/adapter/mapper/telegram.go index 4b2b32f..132c376 100644 --- a/internal/adapter/mapper/telegram.go +++ b/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 - msg.MessageID, // message_id - msg.DateTime, // date_time - msg.PriorityIndicator, // priority_indicator - msg.PrimaryAddress, // primary_address - secondaryAddresses, // secondary_addresses (TEXT) - msg.Originator, // originator - msg.OriginatorDateTime, // originator_date_time - msg.Category, // category - msg.Content, // content (TEXT, original message) - bodyDataJSON, // body_data (JSONB) - msg.ReceivedAt, // received_at - msg.ParsedAt, // parsed_at - msg.DispatchedAt, // dispatched_at - msg.NeedDispatch, // need_dispatch + msgUUID, // uuid + msg.MessageID, // message_id + msg.DateTime, // date_time + msg.PriorityIndicator, // priority_indicator + msg.PrimaryAddress, // primary_address + secondaryAddresses, // secondary_addresses (TEXT) + msg.Originator, // originator + msg.OriginatorDateTime, // originator_date_time + msg.Category, // category + msg.Content, // content (TEXT, original message) + bodyDataJSON, // body_data (JSONB) + msg.ReceivedAt, // received_at + msg.ParsedAt, // parsed_at + msg.DispatchedAt, // dispatched_at + 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 - // For now, we only need ToDBRow for inserts - return nil, fmt.Errorf("FromDBRow not implemented") -} + const expectedColumns = 15 + if len(row) < expectedColumns { + 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 +} diff --git a/internal/adapter/mapper/telegram_test.go b/internal/adapter/mapper/telegram_test.go new file mode 100644 index 0000000..b32bd95 --- /dev/null +++ b/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) + } +} diff --git a/internal/adapter/parser/parser.go b/internal/adapter/parser/parser.go index b7d072c..b091ce0 100644 --- a/internal/adapter/parser/parser.go +++ b/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. +} diff --git a/internal/app/processor.go b/internal/app/processor.go index a4ed3df..ec79e34 100644 --- a/internal/app/processor.go +++ b/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 - parsed.Uuid = msgID + if 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), ) } diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go index 36aa8ae..40b29ad 100644 --- a/internal/app/processor_test.go +++ b/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") - } -} diff --git a/internal/domain/aviation.go b/internal/domain/aviation.go index eecc3b4..d77d2d6 100644 --- a/internal/domain/aviation.go +++ b/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. diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index c2ab16d..79def3d 100644 --- a/internal/infra/config/config.go +++ b/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 - } else { - cfg.NATS.ConsumerRules.AckWait = 30 * time.Second - } + cfg.NATS.ConsumerRules.AckWait = cfg.Timeouts.AckWait } 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 == "" { - return fmt.Errorf("subscription.topic or nats.stream is required") + if c.NATS.Stream == "" { + return fmt.Errorf("nats.stream is required") } if c.Publisher.Topic == "" { return fmt.Errorf("publisher.topic is required") diff --git a/internal/infra/config/config_test.go b/internal/infra/config/config_test.go new file mode 100644 index 0000000..8ee7c26 --- /dev/null +++ b/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) + } +} diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index cc4a345..556575e 100644 --- a/internal/infra/nats/consumer.go +++ b/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)