From 9cce4610b6ae237841eb6e00223d2290e06ada23 Mon Sep 17 00:00:00 2001 From: windyboy Date: Fri, 14 Nov 2025 21:42:04 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Enhance=20NATS=20configuration=20an?= =?UTF-8?q?d=20error=20handling.=20Introduce=20stream=20limits=20and=20con?= =?UTF-8?q?sumer=20rules=20in=20configuration=20files.=20Refactor=20messag?= =?UTF-8?q?e=20processing=20to=20handle=20permanent=20errors.=20Update=20R?= =?UTF-8?q?EADME=20and=20development=20configuration=20to=20reflect=20chan?= =?UTF-8?q?ges.=20Add=20tests=20for=20new=20error=20handling=20mechanisms.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 20 ++++- cmd/main/main.go | 30 +++++++- configs/config.dev.toml | 18 ++++- internal/app/errors.go | 39 ++++++++++ internal/app/errors_test.go | 33 ++++++++ internal/app/processor.go | 6 +- internal/app/processor_test.go | 115 ++++++++++++++++++++++++++++ internal/infra/config/config.go | 101 ++++++++++++++++++++++--- internal/infra/nats/consumer.go | 124 +++++++++++++++++++++++++++---- internal/infra/nats/jetstream.go | 38 ++++++++-- internal/infra/nats/publisher.go | 21 +++++- pkg/di/wire.go | 14 ++-- pkg/di/wire_gen.go | 8 +- 13 files changed, 510 insertions(+), 57 deletions(-) create mode 100644 internal/app/errors.go create mode 100644 internal/app/errors_test.go create mode 100644 internal/app/processor_test.go diff --git a/README.md b/README.md index 339556c..7188e19 100644 --- a/README.md +++ b/README.md @@ -78,11 +78,24 @@ url = "nats://localhost:4222" stream = "TELEGRAM" consumer = "telegram-consumer" +[nats.stream_limits] +max_msgs = 100000 +max_bytes = 67108864 +max_age = "24h" +discard = "old" +storage = "file" +replicas = 1 + +[nats.consumer] +max_deliver = 5 +ack_wait = "30s" +max_ack_pending = 1024 + [subscription] -topic = "Telegram.Serial" +topic = "telegram.serial" [publisher] -topic = "Telegram.Json" +topic = "telegram.json" [postgres] url = "postgres://user:password@localhost:5432/aviation?sslmode=disable" @@ -92,6 +105,7 @@ min_conns = 2 [app] batch_size = 50 batch_timeout = "2s" +monitor_interval = "30s" [log] level = "info" @@ -147,7 +161,7 @@ task run-dev Flags: -n, --nats string Nats server address (default: "nats://localhost:4222") - -t, --topic string Nats topic to listen to (default: "Telegram.Serial") + -t, --topic string Nats topic to listen to (default: "telegram.serial") ``` ## Development diff --git a/cmd/main/main.go b/cmd/main/main.go index 7fd0b6a..a242d80 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -3,10 +3,12 @@ package main import ( "caatsm/pkg/di" "context" + "errors" "fmt" "os" "os/signal" "syscall" + "time" "github.com/urfave/cli/v2" ) @@ -39,7 +41,7 @@ func setupApp() *cli.App { Name: "topic", Aliases: []string{"t"}, Usage: "Nats topic to listen to", - Value: "Telegram.Serial", + Value: "telegram.serial", EnvVars: []string{"NATS_SUBJECT"}, }, }, @@ -72,18 +74,38 @@ func executeListen(c *cli.Context) error { } }() + var runErr error + // Wait for signal or error select { case sig := <-sigChan: fmt.Printf("Received signal: %v, shutting down...\n", sig) cancel() case err := <-errChan: - return err + cancel() + if err != nil && !errors.Is(err, context.Canceled) { + runErr = err + } + } + + waitTimeout := 5 * time.Second + select { + case err := <-errChan: + if err != nil && !errors.Is(err, context.Canceled) { + runErr = err + } + case <-time.After(waitTimeout): + fmt.Printf("Timed out waiting for consumer shutdown after %s\n", waitTimeout) + } + + if err := consumer.Shutdown(context.Background()); err != nil { + if runErr == nil { + runErr = fmt.Errorf("failed to drain NATS connection: %w", err) + } } // Note: processor is initialized but not directly used here - // It's used by the consumer internally _ = processor - return nil + return runErr } diff --git a/configs/config.dev.toml b/configs/config.dev.toml index d40bedb..301a15b 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -5,12 +5,25 @@ cluster = "tele-cluster" stream = "TELEGRAM" consumer = "telegram-consumer" +[nats.stream_limits] +max_msgs = 100000 +max_bytes = 67108864 +max_age = "24h" +discard = "old" +storage = "file" +replicas = 1 + +[nats.consumer] +max_deliver = 5 +ack_wait = "30s" +max_ack_pending = 1024 + [subscription] -topic = "Telegram.Serial" +topic = "telegram.serial" queue = "tele-queue" [publisher] -topic = "Telegram.Json" +topic = "telegram.json" [timeouts] server = "5s" @@ -26,6 +39,7 @@ min_conns = 2 [app] batch_size = 50 batch_timeout = "2s" +monitor_interval = "30s" [log] level = "info" diff --git a/internal/app/errors.go b/internal/app/errors.go new file mode 100644 index 0000000..5a256b7 --- /dev/null +++ b/internal/app/errors.go @@ -0,0 +1,39 @@ +package app + +import "errors" + +// PermanentError indicates a failure that should not be retried. +type PermanentError struct { + err error +} + +// Error implements the error interface. +func (e *PermanentError) Error() string { + if e == nil || e.err == nil { + return "" + } + return e.err.Error() +} + +// Unwrap allows errors.Unwrap/Is/As to inspect the underlying error. +func (e *PermanentError) Unwrap() error { + if e == nil { + return nil + } + return e.err +} + +// Permanent wraps err to mark it as non-retriable. +func Permanent(err error) error { + if err == nil { + return nil + } + return &PermanentError{err: err} +} + +// IsPermanent reports whether the error or any wrapped error is permanent. +func IsPermanent(err error) bool { + var target *PermanentError + return errors.As(err, &target) +} + diff --git a/internal/app/errors_test.go b/internal/app/errors_test.go new file mode 100644 index 0000000..6650529 --- /dev/null +++ b/internal/app/errors_test.go @@ -0,0 +1,33 @@ +package app + +import ( + "errors" + "testing" +) + +func TestPermanentWrapsError(t *testing.T) { + base := errors.New("boom") + perr := Permanent(base) + + if perr == nil { + t.Fatalf("expected wrapped error, got nil") + } + if !IsPermanent(perr) { + t.Fatalf("expected IsPermanent to be true") + } + if !errors.Is(perr, base) { + t.Fatalf("expected wrapped error to unwrap to base") + } + if errors.Is(base, perr) { + t.Fatalf("expected base not to consider wrapper as same") + } +} + +func TestPermanentNil(t *testing.T) { + if Permanent(nil) != nil { + t.Fatalf("Permanent(nil) should return nil") + } + if IsPermanent(nil) { + t.Fatalf("IsPermanent(nil) should be false") + } +} diff --git a/internal/app/processor.go b/internal/app/processor.go index af3a25f..8a6cf15 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -1,9 +1,9 @@ package app import ( - "context" "caatsm/internal/adapter" "caatsm/internal/adapter/parser" + "context" "fmt" "go.uber.org/zap" ) @@ -34,13 +34,13 @@ func NewMessageProcessor( // Handle processes a message func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) error { if raw == nil || len(raw) == 0 { - return fmt.Errorf("empty message") + return Permanent(fmt.Errorf("empty message")) } // Parse the message parsed := p.parser.Parse(string(raw)) if parsed == nil { - return fmt.Errorf("parser returned nil") + return Permanent(fmt.Errorf("parser returned nil")) } // Set the message ID from NATS diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go new file mode 100644 index 0000000..dd01ebf --- /dev/null +++ b/internal/app/processor_test.go @@ -0,0 +1,115 @@ +package app + +import ( + "caatsm/internal/adapter" + "caatsm/internal/adapter/parser" + "caatsm/internal/domain" + "context" + "errors" + "testing" + + "go.uber.org/zap" +) + +type stubParser struct { + value *domain.ParsedMessage +} + +func (s *stubParser) Parse(rawText string) *domain.ParsedMessage { + return s.value +} + +type stubRepository struct { + inserted []*domain.ParsedMessage + err error +} + +func (s *stubRepository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) error { + if s.err != nil { + return s.err + } + s.inserted = append(s.inserted, msg) + return nil +} + +func (s *stubRepository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error { + return errors.New("not implemented") +} + +type stubPublisher struct { + last interface{} + err error +} + +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 TestHandlePublisherErrorIsRetriable(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 not be permanent") + } + if len(repo.inserted) != 1 { + t.Fatalf("expected message to insert before publish failure") + } +} diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index cc273ac..46fa159 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -6,18 +6,18 @@ import ( "strings" "time" - "github.com/knadh/koanf/v2" "github.com/knadh/koanf/parsers/toml" - "github.com/knadh/koanf/providers/file" envprovider "github.com/knadh/koanf/providers/env" + "github.com/knadh/koanf/providers/file" + "github.com/knadh/koanf/v2" ) // Config holds all application configuration type Config struct { - NATS NATSConfig `koanf:"nats"` - Postgres PostgresConfig `koanf:"postgres"` - App AppConfig `koanf:"app"` - Log LogConfig `koanf:"log"` + NATS NATSConfig `koanf:"nats"` + Postgres PostgresConfig `koanf:"postgres"` + App AppConfig `koanf:"app"` + Log LogConfig `koanf:"log"` Publisher PublisherConfig `koanf:"publisher"` // Legacy fields for backward compatibility during migration Subscription SubscriptionConfig `koanf:"subscription"` @@ -26,14 +26,33 @@ type Config struct { // NATSConfig holds NATS/JetStream configuration type NATSConfig struct { - URL string `koanf:"url"` - Stream string `koanf:"stream"` - Consumer string `koanf:"consumer"` + URL string `koanf:"url"` + Stream string `koanf:"stream"` + Consumer string `koanf:"consumer"` + StreamLimits StreamLimitsConfig `koanf:"stream_limits"` + ConsumerRules ConsumerRulesConfig `koanf:"consumer"` // Legacy fields Client string `koanf:"client"` Cluster string `koanf:"cluster"` } +// StreamLimitsConfig defines JetStream retention controls. +type StreamLimitsConfig struct { + MaxMsgs int64 `koanf:"max_msgs"` + MaxBytes int64 `koanf:"max_bytes"` + MaxAge time.Duration `koanf:"max_age"` + Discard string `koanf:"discard"` + Storage string `koanf:"storage"` + Replicas int `koanf:"replicas"` +} + +// ConsumerRulesConfig captures consumer-level options. +type ConsumerRulesConfig struct { + MaxDeliver int `koanf:"max_deliver"` + AckWait time.Duration `koanf:"ack_wait"` + MaxAckPending int `koanf:"max_ack_pending"` +} + // PostgresConfig holds PostgreSQL configuration type PostgresConfig struct { URL string `koanf:"url"` @@ -43,8 +62,9 @@ type PostgresConfig struct { // AppConfig holds application-level configuration type AppConfig struct { - BatchSize int `koanf:"batch_size"` - BatchTimeout time.Duration `koanf:"batch_timeout"` + BatchSize int `koanf:"batch_size"` + BatchTimeout time.Duration `koanf:"batch_timeout"` + MonitorInterval time.Duration `koanf:"monitor_interval"` } // LogConfig holds logging configuration @@ -112,6 +132,9 @@ func LoadConfig() (*Config, error) { if cfg.App.BatchTimeout == 0 { cfg.App.BatchTimeout = 2 * time.Second } + if cfg.App.MonitorInterval == 0 { + cfg.App.MonitorInterval = 30 * time.Second + } if cfg.Postgres.MaxConns == 0 { cfg.Postgres.MaxConns = 10 } @@ -130,6 +153,37 @@ func LoadConfig() (*Config, error) { if cfg.NATS.Consumer == "" { cfg.NATS.Consumer = "telegram-consumer" } + if cfg.NATS.StreamLimits.MaxMsgs == 0 { + cfg.NATS.StreamLimits.MaxMsgs = 100000 + } + if cfg.NATS.StreamLimits.MaxBytes == 0 { + cfg.NATS.StreamLimits.MaxBytes = 64 * 1024 * 1024 + } + if cfg.NATS.StreamLimits.MaxAge == 0 { + cfg.NATS.StreamLimits.MaxAge = 24 * time.Hour + } + if cfg.NATS.StreamLimits.Discard == "" { + cfg.NATS.StreamLimits.Discard = "old" + } + if cfg.NATS.StreamLimits.Storage == "" { + cfg.NATS.StreamLimits.Storage = "file" + } + if cfg.NATS.StreamLimits.Replicas == 0 { + cfg.NATS.StreamLimits.Replicas = 1 + } + if cfg.NATS.ConsumerRules.MaxDeliver == 0 { + cfg.NATS.ConsumerRules.MaxDeliver = 5 + } + 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 + } + } + if cfg.NATS.ConsumerRules.MaxAckPending == 0 { + cfg.NATS.ConsumerRules.MaxAckPending = 1024 + } // Validate configuration if err := cfg.Validate(); err != nil { @@ -159,6 +213,30 @@ func (c *Config) Validate() error { if c.App.BatchTimeout <= 0 { return fmt.Errorf("app.batch_timeout must be greater than 0") } + if c.App.MonitorInterval <= 0 { + return fmt.Errorf("app.monitor_interval must be greater than 0") + } + if c.NATS.StreamLimits.MaxMsgs < 0 { + return fmt.Errorf("nats.stream_limits.max_msgs must be >= 0") + } + if c.NATS.StreamLimits.MaxBytes < 0 { + return fmt.Errorf("nats.stream_limits.max_bytes must be >= 0") + } + if c.NATS.StreamLimits.MaxAge < 0 { + return fmt.Errorf("nats.stream_limits.max_age must be >= 0") + } + if c.NATS.StreamLimits.Replicas <= 0 { + return fmt.Errorf("nats.stream_limits.replicas must be greater than 0") + } + if c.NATS.ConsumerRules.MaxDeliver <= 0 { + return fmt.Errorf("nats.consumer.max_deliver must be greater than 0") + } + if c.NATS.ConsumerRules.AckWait <= 0 { + return fmt.Errorf("nats.consumer.ack_wait must be greater than 0") + } + if c.NATS.ConsumerRules.MaxAckPending < 0 { + return fmt.Errorf("nats.consumer.max_ack_pending must be >= 0") + } return nil } @@ -166,4 +244,3 @@ func (c *Config) Validate() error { func ProvideConfig() (*Config, error) { return LoadConfig() } - diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 5e883ad..0502ed1 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -6,23 +6,25 @@ import ( "context" "errors" "fmt" - "go.uber.org/zap" "github.com/nats-io/nats.go" + "go.uber.org/zap" "time" ) // Consumer handles NATS JetStream message consumption type Consumer struct { - js nats.JetStreamContext - processor *app.MessageProcessor - cfg *config.Config - logger *zap.Logger - subject string + conn *nats.Conn + js nats.JetStreamContext + processor *app.MessageProcessor + cfg *config.Config + logger *zap.Logger + subject string consumerName string } // ProvideConsumer creates a NATS consumer func ProvideConsumer( + conn *nats.Conn, js nats.JetStreamContext, processor *app.MessageProcessor, cfg *config.Config, @@ -39,6 +41,7 @@ func ProvideConsumer( } consumer := &Consumer{ + conn: conn, js: js, processor: processor, cfg: cfg, @@ -62,12 +65,21 @@ func (c *Consumer) ensureConsumer() error { streamName = "TELEGRAM" } + ackWait := c.cfg.NATS.ConsumerRules.AckWait + if ackWait == 0 { + ackWait = c.cfg.Timeouts.AckWait + } + if ackWait == 0 { + ackWait = 30 * time.Second + } + consumerConfig := &nats.ConsumerConfig{ Durable: c.consumerName, DeliverPolicy: nats.DeliverAllPolicy, AckPolicy: nats.AckExplicitPolicy, - AckWait: c.cfg.Timeouts.AckWait, - MaxDeliver: 5, // Maximum number of delivery attempts + AckWait: ackWait, + MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver, + MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending, FilterSubject: c.subject, } @@ -116,6 +128,17 @@ func (c *Consumer) Start(ctx context.Context) error { batchTimeout = 2 * time.Second } + c.logger.Info("Consumer pull configuration", + zap.Int("batch_size", batchSize), + zap.Duration("batch_timeout", batchTimeout), + zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver), + zap.Duration("ack_wait", c.cfg.NATS.ConsumerRules.AckWait), + ) + + statsCtx, statsCancel := context.WithCancel(ctx) + defer statsCancel() + go c.emitConsumerStats(statsCtx, streamName) + for { select { case <-ctx.Done(): @@ -139,24 +162,97 @@ func (c *Consumer) Start(ctx context.Context) error { // Process each message for _, msg := range msgs { if err := c.processMessage(ctx, msg); err != nil { + isPermanent := app.IsPermanent(err) c.logger.Error("Failed to process message", zap.String("subject", msg.Subject), zap.Error(err), + zap.Bool("permanent", isPermanent), ) - // NAK the message to retry + + if isPermanent { + if termErr := msg.Term(); termErr != nil { + c.logger.Error("Failed to TERM message", zap.Error(termErr)) + } + continue + } + + // Transient error: request redelivery if nakErr := msg.Nak(); nakErr != nil { c.logger.Error("Failed to NAK message", zap.Error(nakErr)) } - } else { - // ACK the message - if ackErr := msg.Ack(); ackErr != nil { - c.logger.Error("Failed to ACK message", zap.Error(ackErr)) - } + continue + } + + // ACK the message + if ackErr := msg.Ack(); ackErr != nil { + c.logger.Error("Failed to ACK message", zap.Error(ackErr)) } } } } +func (c *Consumer) emitConsumerStats(ctx context.Context, streamName string) { + interval := c.cfg.App.MonitorInterval + if interval <= 0 { + interval = 30 * time.Second + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + info, err := c.js.ConsumerInfo(streamName, c.consumerName) + if err != nil { + c.logger.Warn("Failed to fetch consumer info", zap.Error(err)) + continue + } + + c.logger.Info("JetStream consumer metrics", + zap.String("stream", 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)), + ) + } + } +} + +// Shutdown drains the underlying NATS connection gracefully. +func (c *Consumer) Shutdown(ctx context.Context) error { + if c.conn == nil { + return nil + } + + timeout := c.cfg.Timeouts.Close + if timeout <= 0 { + timeout = 10 * time.Second + } + + closeCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + errCh := make(chan error, 1) + go func() { + errCh <- c.conn.Drain() + }() + + select { + case err := <-errCh: + c.conn.Close() + return err + case <-closeCtx.Done(): + c.conn.Close() + return fmt.Errorf("nats drain timeout: %w", closeCtx.Err()) + } +} + // processMessage processes a single message func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { msgID := msg.Header.Get("Nats-Msg-Id") diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index b8d0970..a13b5da 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -3,14 +3,14 @@ package nats import ( "caatsm/internal/infra/config" "fmt" - "go.uber.org/zap" + "strings" + "github.com/nats-io/nats.go" - "time" + "go.uber.org/zap" ) -// ProvideJetStream creates a NATS JetStream connection -func ProvideJetStream(cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) { - // Connect to NATS +// ProvideNATSConn creates a reusable NATS connection. +func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) { nc, err := nats.Connect( cfg.NATS.URL, nats.RetryOnFailedConnect(true), @@ -29,6 +29,11 @@ func ProvideJetStream(cfg *config.Config, logger *zap.Logger) (nats.JetStreamCon return nil, fmt.Errorf("failed to connect to NATS: %w", err) } + return nc, nil +} + +// ProvideJetStream creates a NATS JetStream context using an existing connection. +func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) { // Get JetStream context js, err := nc.JetStream() if err != nil { @@ -43,13 +48,30 @@ func ProvideJetStream(cfg *config.Config, logger *zap.Logger) (nats.JetStreamCon subject = "telegram.>" } + streamLimits := cfg.NATS.StreamLimits + storage := nats.FileStorage + switch strings.ToLower(streamLimits.Storage) { + case "memory": + storage = nats.MemoryStorage + case "file": + storage = nats.FileStorage + } + + discard := nats.DiscardOld + if strings.EqualFold(streamLimits.Discard, "new") { + discard = nats.DiscardNew + } + streamConfig := &nats.StreamConfig{ Name: streamName, Subjects: []string{subject}, Retention: nats.LimitsPolicy, - MaxAge: 24 * time.Hour, - Storage: nats.FileStorage, - Replicas: 1, + MaxMsgs: streamLimits.MaxMsgs, + MaxBytes: streamLimits.MaxBytes, + MaxAge: streamLimits.MaxAge, + Discard: discard, + Storage: storage, + Replicas: streamLimits.Replicas, } _, err = js.AddStream(streamConfig) diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index 6e8b63d..21e5fcd 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -2,11 +2,13 @@ package nats import ( "caatsm/internal/adapter" + "caatsm/internal/domain" "caatsm/internal/infra/config" "encoding/json" "fmt" - "go.uber.org/zap" + "github.com/google/uuid" "github.com/nats-io/nats.go" + "go.uber.org/zap" ) // Publisher publishes messages to NATS JetStream @@ -42,8 +44,23 @@ func (p *Publisher) Publish(message interface{}) error { return fmt.Errorf("failed to marshal message: %w", err) } + // Build JetStream message to attach dedup headers + jsMsg := nats.NewMsg(topic) + jsMsg.Data = messageBytes + + switch typed := message.(type) { + case *domain.ParsedMessage: + if typed != nil && typed.Uuid != "" { + jsMsg.Header.Set("Nats-Msg-Id", typed.Uuid) + } else { + jsMsg.Header.Set("Nats-Msg-Id", uuid.NewString()) + } + default: + jsMsg.Header.Set("Nats-Msg-Id", uuid.NewString()) + } + // Publish to JetStream - _, err = p.js.Publish(topic, messageBytes) + _, err = p.js.PublishMsg(jsMsg) if err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/pkg/di/wire.go b/pkg/di/wire.go index 7082dfe..a2e48da 100644 --- a/pkg/di/wire.go +++ b/pkg/di/wire.go @@ -19,27 +19,27 @@ func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { wire.Build( // Config config.ProvideConfig, - + // Logger log.ProvideLogger, - + // Database postgres.ProvideDB, postgres.ProvideRepository, - + // NATS + nats.ProvideNATSConn, nats.ProvideJetStream, nats.ProvidePublisher, - + // Parser parser.ProvideParser, - + // App app.NewMessageProcessor, - + // Consumer nats.ProvideConsumer, ) return nil, nil, nil } - diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index e1128c2..2c2defe 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -32,7 +32,11 @@ func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { if err != nil { return nil, nil, err } - natsJetStreamContext, err := nats.ProvideJetStream(configConfig, zapLogger) + natsConn, err := nats.ProvideNATSConn(configConfig, zapLogger) + if err != nil { + return nil, nil, err + } + natsJetStreamContext, err := nats.ProvideJetStream(natsConn, configConfig, zapLogger) if err != nil { return nil, nil, err } @@ -42,7 +46,7 @@ func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { } parserParser := parser.ProvideParser() appMessageProcessor := app.NewMessageProcessor(parserParser, adapterRepository, adapterPublisher, zapLogger) - natsConsumer, err := nats.ProvideConsumer(natsJetStreamContext, appMessageProcessor, configConfig, zapLogger) + natsConsumer, err := nats.ProvideConsumer(natsConn, natsJetStreamContext, appMessageProcessor, configConfig, zapLogger) if err != nil { return nil, nil, err }