From 02e54a267076e470ab45e96abb2c6ed2af9aa4e4 Mon Sep 17 00:00:00 2001 From: windyboy Date: Fri, 14 Nov 2025 22:43:04 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Refactor=20application=20initializa?= =?UTF-8?q?tion=20to=20load=20configuration=20and=20enhance=20message=20pr?= =?UTF-8?q?ocessing.=20Introduce=20effective=20subscription=20topic=20hand?= =?UTF-8?q?ling=20and=20improve=20error=20management=20in=20NATS=20consume?= =?UTF-8?q?r.=20Update=20tests=20for=20publisher=20error=20handling=20and?= =?UTF-8?q?=20add=20new=20dependency=20injection=20method=20for=20app=20in?= =?UTF-8?q?itialization.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/main/main.go | 16 ++++++- internal/app/processor.go | 19 ++++++-- internal/app/processor_test.go | 6 +-- internal/infra/config/config.go | 13 ++++++ internal/infra/nats/consumer.go | 40 ++++++++++------ internal/infra/nats/jetstream.go | 66 ++++++++++++++++++++++----- internal/infra/postgres/repository.go | 11 ++++- pkg/di/init_manual.go | 60 ++++++++++++++++++++++++ 8 files changed, 196 insertions(+), 35 deletions(-) create mode 100644 pkg/di/init_manual.go diff --git a/cmd/main/main.go b/cmd/main/main.go index a242d80..fe8932a 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -1,6 +1,7 @@ package main import ( + "caatsm/internal/infra/config" "caatsm/pkg/di" "context" "errors" @@ -52,8 +53,21 @@ func setupApp() *cli.App { } func executeListen(c *cli.Context) error { + cfg, err := config.LoadConfig() + if err != nil { + return fmt.Errorf("failed to load config: %w", err) + } + + if flagURL := c.String("nats"); flagURL != "" { + cfg.NATS.URL = flagURL + } + + if flagTopic := c.String("topic"); flagTopic != "" { + cfg.Subscription.Topic = flagTopic + } + // Initialize dependencies using Wire - processor, consumer, err := di.InitializeApp() + processor, consumer, err := di.InitializeAppWithConfig(cfg) if err != nil { return fmt.Errorf("failed to initialize app: %w", err) } diff --git a/internal/app/processor.go b/internal/app/processor.go index 8a6cf15..a4ed3df 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -5,6 +5,7 @@ import ( "caatsm/internal/adapter/parser" "context" "fmt" + "go.uber.org/zap" ) @@ -33,7 +34,7 @@ func NewMessageProcessor( // Handle processes a message func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) error { - if raw == nil || len(raw) == 0 { + if len(raw) == 0 { return Permanent(fmt.Errorf("empty message")) } @@ -50,7 +51,7 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) if !parsed.Parsed { p.logger.Info("Message not parsed", zap.String("msg_id", msgID), - zap.String("content", parsed.Content), + zap.String("content_preview", truncateContent(parsed.Content, 200)), ) } else { p.logger.Info("Message parsed successfully", @@ -72,9 +73,19 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) zap.String("msg_id", msgID), zap.Error(err), ) - // Return error to trigger NAK and retry - return fmt.Errorf("failed to publish message: %w", err) + // Mark as permanent so the consumer will ack instead of retrying + return Permanent(fmt.Errorf("failed to publish message: %w", err)) } return nil } + +func truncateContent(content string, limit int) string { + if limit <= 0 || len(content) <= limit { + return content + } + if limit <= 3 { + return content[:limit] + } + return content[:limit-3] + "..." +} diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go index dd01ebf..36aa8ae 100644 --- a/internal/app/processor_test.go +++ b/internal/app/processor_test.go @@ -94,7 +94,7 @@ func TestHandleSuccessSetsUuidAndPublishes(t *testing.T) { } } -func TestHandlePublisherErrorIsRetriable(t *testing.T) { +func TestHandlePublisherErrorIsPermanent(t *testing.T) { parsed := domain.NewParsedMessage() parsed.Parsed = true @@ -106,8 +106,8 @@ func TestHandlePublisherErrorIsRetriable(t *testing.T) { if err == nil { t.Fatalf("expected error when publisher fails") } - if IsPermanent(err) { - t.Fatalf("publisher failure should not be permanent") + 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/infra/config/config.go b/internal/infra/config/config.go index 46fa159..c2ab16d 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -244,3 +244,16 @@ func (c *Config) Validate() error { func ProvideConfig() (*Config, error) { return LoadConfig() } + +// EffectiveSubscriptionTopic returns the active subscription subject. +// Retains support for legacy config.Subscription fields while allowing +// future consolidation. +func (c *Config) EffectiveSubscriptionTopic() string { + if c == nil { + return "" + } + if topic := c.Subscription.Topic; topic != "" { + return topic + } + return "telegram.>" +} diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 0502ed1..cc4a345 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -30,10 +30,7 @@ func ProvideConsumer( cfg *config.Config, logger *zap.Logger, ) (*Consumer, error) { - subject := cfg.Subscription.Topic - if subject == "" { - subject = "telegram.>" - } + subject := cfg.EffectiveSubscriptionTopic() consumerName := cfg.NATS.Consumer if consumerName == "" { @@ -170,8 +167,8 @@ func (c *Consumer) Start(ctx context.Context) error { ) if isPermanent { - if termErr := msg.Term(); termErr != nil { - c.logger.Error("Failed to TERM message", zap.Error(termErr)) + if ackErr := msg.Ack(); ackErr != nil { + c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) } continue } @@ -255,14 +252,16 @@ func (c *Consumer) Shutdown(ctx context.Context) error { // processMessage processes a single message func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { - msgID := msg.Header.Get("Nats-Msg-Id") - if msgID == "" { - // Use reply subject or generate a simple ID - if msg.Reply != "" { - msgID = msg.Reply - } else { - msgID = fmt.Sprintf("msg-%d", time.Now().UnixNano()) - } + msgID, source, err := c.resolveMsgID(msg) + if err != nil { + 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), + ) } c.logger.Debug("Processing message", @@ -278,3 +277,16 @@ func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { 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 + } + + meta, err := msg.Metadata() + if err != nil { + return "", "", fmt.Errorf("fetch metadata: %w", err) + } + + return fmt.Sprintf("js-%d", meta.Sequence.Stream), "metadata", nil +} diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index a13b5da..930d3dc 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -2,7 +2,9 @@ package nats import ( "caatsm/internal/infra/config" + "errors" "fmt" + "os" "strings" "github.com/nats-io/nats.go" @@ -43,10 +45,7 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na // Create stream if it doesn't exist streamName := cfg.NATS.Stream - subject := cfg.Subscription.Topic - if subject == "" { - subject = "telegram.>" - } + subject := cfg.EffectiveSubscriptionTopic() streamLimits := cfg.NATS.StreamLimits storage := nats.FileStorage @@ -74,15 +73,58 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na Replicas: streamLimits.Replicas, } - _, err = js.AddStream(streamConfig) - if err != nil && err != nats.ErrStreamNameAlreadyInUse { - nc.Close() - return nil, fmt.Errorf("failed to create stream: %w", err) - } - - if err == nil { - logger.Info("Created JetStream", zap.String("stream", streamName), zap.String("subject", subject)) + info, err := js.StreamInfo(streamName) + if err != nil { + if errors.Is(err, nats.ErrStreamNotFound) { + if shouldBootstrapStream() { + if _, err = js.AddStream(streamConfig); err != nil { + nc.Close() + return nil, fmt.Errorf("failed to create stream: %w", err) + } + logger.Info("Created JetStream", zap.String("stream", streamName), zap.String("subject", subject)) + } else { + nc.Close() + return nil, fmt.Errorf("stream %s not found and auto-creation disabled", streamName) + } + } else { + nc.Close() + return nil, fmt.Errorf("failed to fetch stream info: %w", err) + } + } else { + validateStreamConfig(info, subject, logger) } return js, nil } + +func shouldBootstrapStream() bool { + switch strings.ToLower(os.Getenv("GO_ENV")) { + case "", "dev", "development", "test", "testing": + return true + default: + return false + } +} + +func validateStreamConfig(info *nats.StreamInfo, expectedSubject string, logger *zap.Logger) { + if info == nil { + return + } + + if !subjectListContains(info.Config.Subjects, expectedSubject) { + logger.Warn("JetStream stream subjects do not match config", + zap.String("stream", info.Config.Name), + zap.Strings("stream_subjects", info.Config.Subjects), + zap.String("configured_subject", expectedSubject), + ) + } +} + +func subjectListContains(subjects []string, target string) bool { + for _, s := range subjects { + if s == target { + return true + } + } + return false +} diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go index a56ae12..b678a9f 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -6,6 +6,7 @@ import ( "caatsm/internal/domain" "context" "fmt" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "go.uber.org/zap" @@ -45,7 +46,7 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e ON CONFLICT (uuid) DO NOTHING ` - _, err = r.pool.Exec(ctx, query, + 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], ) @@ -53,6 +54,14 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e return fmt.Errorf("failed to insert message: %w", err) } + if tag.RowsAffected() == 0 { + r.logger.Info("Duplicate message skipped", + zap.String("uuid", msg.Uuid), + zap.String("message_id", msg.MessageID), + ) + return nil + } + r.logger.Debug("Inserted message", zap.String("uuid", msg.Uuid), zap.String("message_id", msg.MessageID), diff --git a/pkg/di/init_manual.go b/pkg/di/init_manual.go new file mode 100644 index 0000000..d64eafa --- /dev/null +++ b/pkg/di/init_manual.go @@ -0,0 +1,60 @@ +package di + +import ( + "caatsm/internal/adapter/parser" + "caatsm/internal/app" + "caatsm/internal/infra/config" + "caatsm/internal/infra/log" + "caatsm/internal/infra/nats" + "caatsm/internal/infra/postgres" + "errors" +) + +// InitializeAppWithConfig wires dependencies using the provided config. +func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, error) { + if cfg == nil { + return nil, nil, errors.New("config is required") + } + + logger, err := log.ProvideLogger(cfg) + if err != nil { + return nil, nil, err + } + + pool, err := postgres.ProvideDB(cfg) + if err != nil { + return nil, nil, err + } + + repo, err := postgres.ProvideRepository(pool, logger) + if err != nil { + return nil, nil, err + } + + conn, err := nats.ProvideNATSConn(cfg, logger) + { + if err != nil { + return nil, nil, err + } + } + + js, err := nats.ProvideJetStream(conn, cfg, logger) + if err != nil { + return nil, nil, err + } + + publisher, err := nats.ProvidePublisher(js, cfg, logger) + if err != nil { + return nil, nil, err + } + + messageParser := parser.ProvideParser() + processor := app.NewMessageProcessor(messageParser, repo, publisher, logger) + + consumer, err := nats.ProvideConsumer(conn, js, processor, cfg, logger) + if err != nil { + return nil, nil, err + } + + return processor, consumer, nil +}