diff --git a/AGENTS.md b/AGENTS.md index 33afd90..e1d1309 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -10,6 +10,8 @@ ## Code Style & Architecture - **Structure**: Clean Architecture (`cmd/`, `internal/{domain,app,adapter,infra}`, `pkg/`). +- **Parsers**: Composite parser pattern with specialized sub-parsers (aviation, weather). +- **Domain**: Core domain types include aviation telegrams and weather reports. - **Formatting**: Run `go fmt ./...` and `goimports` before committing. - **Naming**: `CamelCase` (exported), `camelCase` (private). Package names match dirs. - **Errors**: Wrap with context (`fmt.Errorf("...: %w", err)`). Use `errors.Is`. diff --git a/GEMINI.md b/GEMINI.md new file mode 100644 index 0000000..4f35170 --- /dev/null +++ b/GEMINI.md @@ -0,0 +1,94 @@ +# go-caatsm (Civil Aviation Authority Telegram Message Processor) + +## Project Overview + +`go-caatsm` is a high-performance Go application designed to process aviation telegrams (like FPL, ARR, DEP) and weather reports (METAR, SPECI, TAF) from NATS JetStream, parse them, persist them to PostgreSQL/TimescaleDB, and republish the parsed results. It follows Clean Architecture principles to ensure modularity and testability. + +## Architecture + +The project is structured using Clean Architecture: + +* **`cmd/`**: Application entry points. `cmd/main` is the primary service, `cmd/seed-telegrams` is a utility for generating test data. +* **`internal/domain/`**: Core business logic and types (e.g., `aviation.go`, `fpl.go`, `weather/`). Pure Go, no dependencies on outer layers. +* **`internal/app/`**: Application business rules (use cases). `MessageProcessor` orchestrates the flow between ports. +* **`internal/adapter/`**: Adapters for external interfaces. + * `parser/`: Logic to parse raw telegram text into domain objects. + * `aviation/`: Aviation telegram parser (ARR, DEP, CNL, DLA, FPL) + * `weather/`: Weather report parser (METAR, SPECI, TAF) + * `composite.go`: Composite parser that routes messages to appropriate parser + * `validator/`: AFTN protocol validation. + * `dto/`: Data Transfer Objects. +* **`internal/infra/`**: Infrastructure implementations. + * `nats/`: NATS JetStream consumer and publisher. + * `postgres/`: Database repository using `pgx`. + * `config/`, `log/`, `telemetry/`, `monitoring/`: Cross-cutting concerns. +* **`internal/port/`**: Interfaces defining the contracts for repositories, publishers, and parsers. +* **`pkg/di/`**: Dependency Injection using Google Wire. + +## Tech Stack + +* **Language:** Go 1.24+ +* **Messaging:** NATS JetStream +* **Database:** PostgreSQL (with TimescaleDB extension for time-series data) +* **Observability:** OpenTelemetry (OTLP), Prometheus, Jaeger, Grafana, Zap Logger +* **CLI:** `urfave/cli` +* **DI:** Google Wire +* **Testing:** Ginkgo (BDD), Gomega, Testcontainers (integration tests) + +## Key Commands (Taskfile) + +The project uses `Taskfile.yml` for managing common tasks. + +* **Build:** `task build` (Output: `bin/receiver`) +* **Run (Dev):** `task run-dev` (Connects to local Docker stack) +* **Run (Prod):** `task run-prod` +* **Test (Unit):** `task test` +* **Test (Integration):** `task test-int` (Requires Docker) +* **Lint:** `task lint` +* **Start Infrastructure:** `task up` (Starts Postgres, NATS, Observability stack) +* **Stop Infrastructure:** `task down` +* **Seed Data:** `task seed` (Injects sample telegrams into NATS) + +## Configuration + +Configuration is managed via TOML files in `configs/` and environment variables. +* `configs/config.dev.toml`: Default for development (`GO_ENV=dev`). +* `configs/config.prod.toml`: Production settings (`GO_ENV=prod`). +* Environment Variables: Prefix `CAATSM_` (e.g., `CAATSM_NATS_URL`, `CAATSM_POSTGRES_URL`). + +## Development Workflow + +1. **Start Infrastructure:** + ```bash + task up + ``` +2. **Run Service Locally:** + ```bash + task run-dev + ``` +3. **Generate Traffic:** + ```bash + task seed + # OR for continuous traffic + task seed-slow + ``` +4. **Observe:** + * Grafana: http://localhost:3000 (admin/admin) + * Jaeger: http://localhost:16686 + * Prometheus: http://localhost:9090 + +## Key Files & Directories + +* `cmd/main/main.go`: Application entry point. Sets up config, DI, and starts the listener. +* `internal/app/processor.go`: `MessageProcessor` - The core orchestration logic. +* `internal/adapter/parser/`: Contains parsers for aviation telegrams and weather reports (composite pattern). +* `internal/infra/nats/consumer.go`: JetStream consumer implementation. +* `internal/infra/postgres/telegrams.ddl`: Database schema. +* `docs/`: Extensive documentation (Architecture, NATS, Dev Guide). + +## Notes for AI Agent + +* **Conventions:** Follow existing patterns in `internal/`. Use `internal/port` for interfaces. +* **Testing:** New features must include Ginkgo tests. Integration tests should be added for infrastructure components. +* **DI:** If adding new components, update `pkg/di/wire.go` and run `task wire` (or `task generate`). +* **Safety:** Always check `go.mod` before adding imports. diff --git a/README.md b/README.md index 0318c7b..64110d5 100644 --- a/README.md +++ b/README.md @@ -189,6 +189,7 @@ go run ./cmd/seed-telegrams \ - `docs/migrations.md` - `docs/secret-management.md` - `docs/reliability.md` +- `docs/weather-parser.md` ## Contributing diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 19e64d7..8a7ab24 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -8,15 +8,17 @@ import ( "context" "errors" "fmt" - "strings" - "sync" "time" "github.com/nats-io/nats.go" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" "go.uber.org/zap" ) -// Consumer handles NATS JetStream message consumption with clean separation of concerns +// Consumer handles NATS JetStream message consumption. +// It consolidates stream management, fetching, and processing into a single, cohesive unit. type Consumer struct { // Core dependencies conn *nats.Conn @@ -26,40 +28,24 @@ type Consumer struct { logger *zap.Logger telemetry telemetry.Recorder + // Components + monitor *ConsumerMonitor + dlqHandler DLQHandler + // Configuration - config consumerConfig - - // Collaborators (injected for testability) - fetcher MessageFetcher - batchProcessor MessageProcessor - dlqHandler DLQHandler - - // Resource managers - consumerManager *ConsumerManager - streamManager *StreamManager + streamName string + consumerName string + subject string + batchSize int + batchTimeout time.Duration + ackWait time.Duration + backoff []time.Duration // State - consecutiveProcessErrors int - - // Message tracking for health monitoring - lastMessageTime time.Time - lastMessageSequence uint64 - messageGapMutex sync.RWMutex + consecutiveErrors int } -// consumerConfig holds normalized consumer configuration values. -type consumerConfig struct { - subject string - consumerName string - streamName string - dlqSubject string - ackWait time.Duration - batchSize int - batchTimeout time.Duration - monitorInterval time.Duration -} - -// ProvideConsumer creates a NATS consumer with clean architecture. +// ProvideConsumer initializes a NATS consumer, ensuring infrastructure exists. func ProvideConsumer( conn *nats.Conn, js nats.JetStreamContext, @@ -68,449 +54,302 @@ func ProvideConsumer( rec telemetry.Recorder, logger *zap.Logger, ) (*Consumer, error) { - normCfg := normalizeConsumerConfig(cfg) - - consumer := &Consumer{ - conn: conn, - js: js, - processor: processor, - cfg: cfg, - logger: logger, - telemetry: rec, - config: *normCfg, // dereference the pointer - } - consumer.initCollaborators() - - // Initialize the pending messages metric early (set to 0) so it appears in Prometheus - // even before the consumer starts. This ensures the metric is always visible. - logger.Info("Initializing NATS consumer pending messages metric", - zap.String("stream", normCfg.streamName), - zap.String("consumer", normCfg.consumerName), - zap.Uint64("pending", 0), - zap.Bool("js_available", js != nil), - ) - obsmetrics.RecordNATSConsumerPending(normCfg.streamName, normCfg.consumerName, 0) - - // Initialize managers - consumer.consumerManager = NewConsumerManager(js, normCfg.streamName, normCfg.consumerName, normCfg.subject, logger) - // Use StreamManager with full configuration - streamSubjects := []string{normCfg.subject} - if publisherSubject := strings.TrimSpace(cfg.Publisher.Topic); publisherSubject != "" { - streamSubjects = append(streamSubjects, publisherSubject) - } - // Add DLQ subject to stream if DLQ is enabled - if normCfg.dlqSubject != "" { - streamSubjects = append(streamSubjects, normCfg.dlqSubject) - } - streamSubjects = dedupeSubjects(streamSubjects) - consumer.streamManager = NewStreamManager(js, normCfg.streamName, streamSubjects, logger) - - // Update fetcher with managers now that they're initialized - if fetcher, ok := consumer.fetcher.(*defaultMessageFetcher); ok { - fetcher.consumerManager = consumer.consumerManager - fetcher.streamManager = consumer.streamManager + // 1. Normalize Configuration + c := &Consumer{ + conn: conn, + js: js, + processor: processor, + cfg: cfg, + logger: logger, + telemetry: rec, + streamName: orDefault(cfg.NATS.Stream, "TELEGRAM"), + consumerName: orDefault(cfg.NATS.Consumer, "telegram-consumer"), + subject: cfg.EffectiveSubscriptionTopic(), + batchSize: cfg.App.BatchSize, + batchTimeout: cfg.App.BatchTimeout, + ackWait: orDefaultDuration(cfg.NATS.ConsumerRules.AckWait, 30*time.Second), + backoff: cfg.NATS.ConsumerRules.Backoff, } - // Ensure stream exists before creating consumer - streamCfg := &StreamConfig{ - MaxMsgs: cfg.NATS.StreamLimits.MaxMsgs, - MaxBytes: cfg.NATS.StreamLimits.MaxBytes, - MaxAge: cfg.NATS.StreamLimits.MaxAge, - Discard: cfg.NATS.StreamLimits.Discard, - Storage: cfg.NATS.StreamLimits.Storage, - Replicas: cfg.NATS.StreamLimits.Replicas, + if c.batchSize <= 0 { + c.batchSize = 50 } - if err := consumer.streamManager.EnsureStream(streamCfg); err != nil { - return nil, fmt.Errorf("failed to ensure stream: %w", err) + if c.batchTimeout <= 0 { + c.batchTimeout = 2 * time.Second } - // Create consumer if it doesn't exist - consumerConfig := consumer.buildConsumerConfig() - if err := consumer.consumerManager.EnsureConsumer(consumerConfig); err != nil { - return nil, fmt.Errorf("failed to ensure consumer: %w", err) - } - // Validate DLQ configuration early so misconfiguration is visible at startup - // rather than only when the first poison message appears. - if err := consumer.validateDLQ(); err != nil { - return nil, fmt.Errorf("DLQ validation failed: %w", err) - } - - return consumer, nil -} - -// initCollaborators initializes the collaborator components -func (c *Consumer) initCollaborators() { - c.fetcher = &defaultMessageFetcher{ - batchSize: c.config.batchSize, - batchTimeout: c.config.batchTimeout, - logger: c.logger, - conn: c.conn, - js: c.js, - consumerManager: c.consumerManager, - streamManager: c.streamManager, - config: &c.config, - cfg: c.cfg, - } - - // Initialize DLQ handler first if needed, so batch processor can reference it - if c.config.dlqSubject != "" { + // 2. Initialize Components + c.monitor = NewConsumerMonitor(logger, cfg, js, c.streamName, c.consumerName, cfg.App.MonitorInterval) + if cfg.DLQ.Enabled && cfg.DLQ.Subject != "" { c.dlqHandler = &defaultDLQHandler{ - js: c.js, - dlqSubject: c.config.dlqSubject, - streamName: c.config.streamName, - consumerName: c.config.consumerName, - logger: c.logger, - telemetry: c.telemetry, + js: js, + dlqSubject: cfg.DLQ.Subject, + streamName: c.streamName, + consumerName: c.consumerName, + logger: logger, + telemetry: rec, } } - c.batchProcessor = &defaultBatchProcessor{ - processor: c.processor, - dlqHandler: c.dlqHandler, - logger: c.logger, - telemetry: c.telemetry, - streamName: c.config.streamName, - consumerName: c.config.consumerName, - backoff: c.cfg.NATS.ConsumerRules.Backoff, - consecutiveProcessErrors: &c.consecutiveProcessErrors, + // 3. Ensure Infrastructure (Stream & Consumer) + if err := c.ensureInfrastructure(); err != nil { + return nil, err } + + return c, nil } -// normalizeConsumerConfig extracts and normalizes consumer configuration from the application config. -// This function can be unit-tested without requiring a JetStream context. -func normalizeConsumerConfig(cfg *config.Config) *consumerConfig { - subject := cfg.EffectiveSubscriptionTopic() - - consumerName := cfg.NATS.Consumer - if consumerName == "" { - consumerName = "telegram-consumer" - } - - streamName := cfg.NATS.Stream - if streamName == "" { - streamName = "TELEGRAM" - } - - // DLQ routing is only meaningful in JetStream mode. Respect dlq.enabled to allow - // environments to opt out cleanly even if a subject is configured. - dlqSubject := "" - if cfg.DLQ.Enabled { - dlqSubject = strings.TrimSpace(cfg.DLQ.Subject) - } - - ackWait := cfg.NATS.ConsumerRules.AckWait - if ackWait == 0 { - ackWait = cfg.Timeouts.AckWait - } - if ackWait == 0 { - ackWait = 30 * time.Second - } - - batchSize := cfg.App.BatchSize - if batchSize == 0 { - batchSize = 50 - } - - batchTimeout := cfg.App.BatchTimeout - if batchTimeout == 0 { - batchTimeout = 2 * time.Second - } - - monitorInterval := cfg.App.MonitorInterval - if monitorInterval <= 0 { - monitorInterval = 30 * time.Second - } - - return &consumerConfig{ - subject: subject, - consumerName: consumerName, - streamName: streamName, - dlqSubject: dlqSubject, - ackWait: ackWait, - batchSize: batchSize, - batchTimeout: batchTimeout, - monitorInterval: monitorInterval, - } -} - -// buildConsumerConfig builds the NATS consumer configuration -func (c *Consumer) buildConsumerConfig() *nats.ConsumerConfig { - return &nats.ConsumerConfig{ - Durable: c.config.consumerName, - DeliverPolicy: mapDeliverPolicy(c.cfg.NATS.ConsumerRules.DeliverPolicy), - AckPolicy: nats.AckExplicitPolicy, - AckWait: c.config.ackWait, - ReplayPolicy: mapReplayPolicy(c.cfg.NATS.ConsumerRules.ReplayPolicy), - MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver, - MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending, - FilterSubject: c.config.subject, - BackOff: c.cfg.NATS.ConsumerRules.Backoff, - } -} - -// Start starts consuming messages from JetStream. +// Start begins the main consumption loop. func (c *Consumer) Start(ctx context.Context) error { - return c.startJetStream(ctx) -} - -// RouteToDLQ implements DLQHandler interface -func (c *Consumer) RouteToDLQ(ctx context.Context, msg *nats.Msg, cause error) error { - if c.dlqHandler != nil { - return c.dlqHandler.RouteToDLQ(ctx, msg, cause) - } - return nil -} - -// ValidateDLQ implements DLQHandler interface -func (c *Consumer) ValidateDLQ() error { - if c.dlqHandler != nil { - return c.dlqHandler.ValidateDLQ() - } - return nil -} - -// validateDLQ is a helper for internal use (lowercase) -func (c *Consumer) validateDLQ() error { - return c.ValidateDLQ() -} - -// createPullSubscription creates a pull subscription -func (c *Consumer) createPullSubscription() (*nats.Subscription, error) { - return c.consumerManager.CreatePullSubscription() -} - -// startJetStream starts the JetStream consumer loop. -func (c *Consumer) startJetStream(ctx context.Context) error { - // Create pull subscription - sub, err := c.createPullSubscription() - if err != nil { - return err - } - - // Use a closure that always cleans up the current subscription. - // When subscription is replaced in handleFetchError, this will clean up - // whatever currentSub points to at shutdown time. - var currentSub = sub - cleanupSubscriber := func() { - if currentSub != nil { - if err := currentSub.Unsubscribe(); err != nil { - c.logger.Error("Failed to unsubscribe subscription", zap.Error(err)) - } - currentSub = nil - } - } - defer cleanupSubscriber() - - c.logger.Info("Started consuming messages", - zap.String("subject", c.config.subject), - zap.String("consumer", c.config.consumerName), - zap.String("stream", c.config.streamName), + c.logger.Info("Starting consumer", + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + zap.String("subject", c.subject), ) - // Record initial pending messages metric immediately - // This ensures the metric appears in Prometheus right away - if info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName); err == nil { - c.logger.Info("Recording initial NATS consumer pending messages metric", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Uint64("pending", info.NumPending), - ) - obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) - } else { - c.logger.Warn("Failed to fetch initial consumer info for pending messages metric", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Error(err), - ) + // Start background monitoring + monitorCtx, cancelMonitor := context.WithCancel(ctx) + defer cancelMonitor() + go c.monitor.Start(monitorCtx) + + // Create subscription + sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.BindStream(c.streamName)) + if err != nil { + return fmt.Errorf("failed to subscribe: %w", err) + } + defer sub.Unsubscribe() + + // Initial metric recording + if info, err := c.js.ConsumerInfo(c.streamName, c.consumerName); err == nil { + c.monitor.RecordInitialPending(info.NumPending) } - statsCtx, statsCancel := context.WithCancel(ctx) - defer statsCancel() - go c.emitConsumerStats(statsCtx) - - var fetchErrorStreak int - + // Main Loop for { select { case <-ctx.Done(): - c.logger.Info("Stopping consumer", zap.Error(ctx.Err())) - return ctx.Err() + return nil default: } - // Fetch messages in batch - msgs, err := c.fetcher.FetchBatch(ctx, currentSub) + msgs, err := sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout)) if err != nil { - // If context was cancelled, return immediately + if errors.Is(err, nats.ErrTimeout) { + continue // Normal timeout, just retry + } if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { - c.logger.Info("Stopping consumer due to context cancellation", zap.Error(err)) - return err - } - shouldContinue, handleErr := c.fetcher.HandleFetchError(ctx, err, ¤tSub, &fetchErrorStreak) - if !shouldContinue { - return handleErr + return nil } + // Log other errors but keep loop alive unless critical + c.logger.Warn("Fetch error", zap.Error(err)) + time.Sleep(100 * time.Millisecond) // Slight backoff continue } - // Successful fetch -> reset error streak. - if fetchErrorStreak > 0 { - fetchErrorStreak = 0 - } - - // Update message tracking for health monitoring (track each message) - for _, msg := range msgs { - c.updateMessageTracking(msg) - } - - // Process batch - c.batchProcessor.ProcessBatch(ctx, msgs) + c.processBatch(ctx, msgs) } } -// emitConsumerStats periodically emits basic consumer statistics. -func (c *Consumer) emitConsumerStats(ctx context.Context) { - ticker := time.NewTicker(c.config.monitorInterval) - defer ticker.Stop() - - // Record initial metric (0) to ensure it appears in Prometheus even before first tick - c.logger.Info("Starting NATS consumer stats emission goroutine, recording initial pending metric", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Duration("interval", c.config.monitorInterval), - ) - obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, 0) - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName) - if err != nil { - c.logger.Warn("Failed to fetch consumer info for pending messages metric", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Error(err), - ) - continue - } - // Record pending messages for monitoring - c.logger.Debug("Recording NATS consumer pending messages metric", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Uint64("pending", info.NumPending), - ) - obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) - - // Record AFTN health metrics - gapSeconds := c.getMessageGapSeconds() - healthy := c.isSerialReaderHealthy() - - obsmetrics.RecordMessageGap(c.config.streamName, c.config.consumerName, gapSeconds) - obsmetrics.RecordSerialReaderHealth(c.config.streamName, c.config.consumerName, healthy) - - if !healthy { - c.logger.Warn("Serial reader appears stalled - no messages received recently", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Float64("gap_seconds", gapSeconds), - zap.Duration("threshold", c.cfg.AFTN.MessageGapThreshold), - ) - } - } - } -} - -// Shutdown drains the underlying NATS connection gracefully. +// Shutdown gracefully drains the connection. func (c *Consumer) Shutdown(ctx context.Context) error { if c.conn == nil { return nil } + c.logger.Info("Draining NATS connection...") + return c.conn.Drain() +} - timeout := c.cfg.Timeouts.Close - if timeout <= 0 { - timeout = 2 * time.Second // Reduced from 10s for faster shutdown - } - - 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()) +// processBatch iterates through a batch of messages. +func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) { + for _, msg := range msgs { + select { + case <-ctx.Done(): + return + default: + c.monitor.TrackMessage(msg) + c.processMsg(ctx, msg) + } } } -// updateMessageTracking updates the last message time and sequence number for health monitoring. -// This should be called for every message received to track message flow and detect gaps. -func (c *Consumer) updateMessageTracking(msg *nats.Msg) { - if msg == nil { +// processMsg handles a single message: Trace -> App Logic -> Ack/Nak. +func (c *Consumer) processMsg(ctx context.Context, msg *nats.Msg) { + start := time.Now() + ctx, span := otel.Tracer("caatsm/nats").Start(ctx, "Consumer.processMsg") + defer span.End() + + msgID := c.resolveMsgID(msg) + + // Add metadata to span/logger + span.SetAttributes( + attribute.String("messaging.system", "nats"), + attribute.String("messaging.message_id", msgID), + attribute.String("caatsm.stream", c.streamName), + ) + + // Execute Application Logic + err := c.processor.Handle(ctx, msg.Data, msgID) + + // Handle Result + if err != nil { + c.handleError(ctx, msg, msgID, err) + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, obsmetrics.ResultFail, time.Since(start)) + } else { + // Success + if c.consecutiveErrors > 0 { + c.consecutiveErrors = 0 + } + if ackErr := msg.Ack(); ackErr != nil { + c.logger.Warn("Failed to ACK", zap.String("msg_id", msgID), zap.Error(ackErr)) + } + c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, "ok", time.Since(start)) + } +} + +// handleError decides whether to Ack (Permanent/DLQ) or Nak (Transient). +func (c *Consumer) handleError(ctx context.Context, msg *nats.Msg, msgID string, err error) { + isPermanent := app.IsPermanent(err) + c.logger.Error("Processing failed", + zap.String("msg_id", msgID), + zap.Error(err), + zap.Bool("permanent", isPermanent), + ) + + if isPermanent { + // Poison message: Route to DLQ -> Ack + c.consecutiveErrors = 0 + if c.dlqHandler != nil { + _ = c.dlqHandler.RouteToDLQ(ctx, msg, err) // Logged inside handler + } + _ = msg.Ack() return } - c.messageGapMutex.Lock() - defer c.messageGapMutex.Unlock() + // Transient error: Backpressure -> Nak with Backoff + c.consecutiveErrors++ + c.applyBackpressure(ctx) + + _ = c.nakWithBackoff(msg) +} - now := time.Now() - c.lastMessageTime = now +// nakWithBackoff calculates the appropriate NAK delay based on delivery attempts. +func (c *Consumer) nakWithBackoff(msg *nats.Msg) error { + if len(c.backoff) == 0 { + return msg.Nak() + } + meta, err := msg.Metadata() + if err != nil { + return msg.Nak() + } + + // attempt is 1-based, index is 0-based + attempt := int(meta.NumDelivered) + index := attempt - 1 + if index >= len(c.backoff) { + index = len(c.backoff) - 1 + } else if index < 0 { + index = 0 + } + + return msg.NakWithDelay(c.backoff[index]) +} - // Extract sequence number from message metadata +// applyBackpressure sleeps if error streak is high to protect the system. +func (c *Consumer) applyBackpressure(ctx context.Context) { + if c.consecutiveErrors < 10 { + return + } + delay := time.Duration(c.consecutiveErrors) * 100 * time.Millisecond + if delay > 5*time.Second { + delay = 5 * time.Second + } + + select { + case <-time.After(delay): + case <-ctx.Done(): + } +} + +// ensureInfrastructure creates the Stream and Consumer if they don't exist. +func (c *Consumer) ensureInfrastructure() error { + // 1. Ensure Stream + subjects := []string{c.subject} + if c.cfg.Publisher.Topic != "" { + subjects = append(subjects, c.cfg.Publisher.Topic) + } + if c.cfg.DLQ.Enabled && c.cfg.DLQ.Subject != "" { + subjects = append(subjects, c.cfg.DLQ.Subject) + } + + streamCfg := &nats.StreamConfig{ + Name: c.streamName, + Subjects: dedupeSubjects(subjects), + Retention: nats.WorkQueuePolicy, // Defaulting to WorkQueue for queues + MaxMsgs: c.cfg.NATS.StreamLimits.MaxMsgs, + MaxBytes: c.cfg.NATS.StreamLimits.MaxBytes, + MaxAge: c.cfg.NATS.StreamLimits.MaxAge, + Replicas: c.cfg.NATS.StreamLimits.Replicas, + Storage: nats.FileStorage, + } + if c.cfg.NATS.StreamLimits.Discard == "new" { + streamCfg.Discard = nats.DiscardNew + } + if c.cfg.NATS.StreamLimits.Storage == "memory" { + streamCfg.Storage = nats.MemoryStorage + } + + // Idempotent add/update + if _, err := c.js.AddStream(streamCfg); err != nil { + return fmt.Errorf("ensure stream: %w", err) + } + + // 2. Ensure Consumer + consumerCfg := &nats.ConsumerConfig{ + Durable: c.consumerName, + FilterSubject: c.subject, + AckPolicy: nats.AckExplicitPolicy, + AckWait: c.ackWait, + MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver, + MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending, + ReplayPolicy: nats.ReplayInstantPolicy, + } + if c.cfg.NATS.ConsumerRules.ReplayPolicy == "original" { + consumerCfg.ReplayPolicy = nats.ReplayOriginalPolicy + } + + // Idempotent add/update + if _, err := c.js.AddConsumer(c.streamName, consumerCfg); err != nil { + return fmt.Errorf("ensure consumer: %w", err) + } + + return nil +} + +// resolveMsgID extracts the ID from headers or metadata. +func (c *Consumer) resolveMsgID(msg *nats.Msg) string { + if id := msg.Header.Get("Nats-Msg-Id"); id != "" { + return id + } if meta, err := msg.Metadata(); err == nil { - currentSeq := meta.Sequence.Stream - - // Detect sequence gaps if we have a previous sequence - if c.lastMessageSequence > 0 && c.cfg.AFTN.EnableSequenceGapDetection { - if currentSeq > c.lastMessageSequence+1 { - gapSize := currentSeq - c.lastMessageSequence - 1 - c.logger.Warn("Message sequence gap detected", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Uint64("last_sequence", c.lastMessageSequence), - zap.Uint64("current_sequence", currentSeq), - zap.Uint64("gap_size", gapSize), - ) - obsmetrics.RecordSequenceGap(c.config.streamName, c.config.consumerName, gapSize) - } - } - - c.lastMessageSequence = currentSeq + return fmt.Sprintf("js-%d", meta.Sequence.Stream) } + return "unknown" } -// getMessageGapSeconds returns the number of seconds since the last message was received. -// Returns 0 if no message has been received yet. -func (c *Consumer) getMessageGapSeconds() float64 { - c.messageGapMutex.RLock() - defer c.messageGapMutex.RUnlock() +// --- Helpers --- - if c.lastMessageTime.IsZero() { - return 0 +func orDefault(val, def string) string { + if val != "" { + return val } - - return time.Since(c.lastMessageTime).Seconds() + return def } -// isSerialReaderHealthy returns true if messages are being received within the threshold. -// Returns false if the gap exceeds the configured message gap threshold. -func (c *Consumer) isSerialReaderHealthy() bool { - c.messageGapMutex.RLock() - defer c.messageGapMutex.RUnlock() - - // If we haven't received any messages yet, consider it healthy (initial state) - if c.lastMessageTime.IsZero() { - return true +func orDefaultDuration(val, def time.Duration) time.Duration { + if val > 0 { + return val } - - gap := time.Since(c.lastMessageTime) - return gap < c.cfg.AFTN.MessageGapThreshold -} + return def +} \ No newline at end of file diff --git a/internal/infra/nats/consumer_manager.go b/internal/infra/nats/consumer_manager.go deleted file mode 100644 index f7bec78..0000000 --- a/internal/infra/nats/consumer_manager.go +++ /dev/null @@ -1,100 +0,0 @@ -package nats - -import ( - "errors" - "fmt" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// ConsumerManager handles JetStream consumer lifecycle management -type ConsumerManager struct { - js nats.JetStreamContext - streamName string - consumerName string - subject string - logger *zap.Logger -} - -// NewConsumerManager creates a new consumer manager -func NewConsumerManager(js nats.JetStreamContext, streamName, consumerName, subject string, logger *zap.Logger) *ConsumerManager { - if logger == nil { - logger = zap.NewNop() - } - return &ConsumerManager{ - js: js, - streamName: streamName, - consumerName: consumerName, - subject: subject, - logger: logger, - } -} - -// EnsureConsumer creates the consumer if it doesn't exist; if it already exists, it is reused. -func (cm *ConsumerManager) EnsureConsumer(config *nats.ConsumerConfig) error { - // First check if the consumer already exists to make this initialization idempotent. - info, err := cm.js.ConsumerInfo(cm.streamName, cm.consumerName) - if err == nil && info != nil { - cm.logger.Info("Using existing JetStream consumer", - zap.String("consumer", cm.consumerName), - zap.String("stream", cm.streamName), - zap.String("subject", cm.subject), - ) - return nil - } - if err != nil && !errors.Is(err, nats.ErrConsumerNotFound) { - return fmt.Errorf("failed to fetch consumer info: %w", err) - } - - // Consumer does not exist; create it. - if _, err := cm.js.AddConsumer(cm.streamName, config); err != nil { - return fmt.Errorf("failed to create consumer: %w", err) - } - - cm.logger.Info("Created JetStream consumer", - zap.String("consumer", cm.consumerName), - zap.String("stream", cm.streamName), - zap.String("subject", cm.subject), - zap.Duration("ack_wait", config.AckWait), - ) - - return nil -} - -// CreatePullSubscription creates a pull subscription with recovery logic -func (cm *ConsumerManager) CreatePullSubscription() (*nats.Subscription, error) { - return cm.js.PullSubscribe(cm.subject, cm.consumerName, nats.Bind(cm.streamName, cm.consumerName)) -} - -// CreatePullSubscriptionWithRecovery creates a pull subscription -func (cm *ConsumerManager) CreatePullSubscriptionWithRecovery(streamManager *StreamManager, consumerConfig *nats.ConsumerConfig) (*nats.Subscription, error) { - sub, err := cm.CreatePullSubscription() - if err == nil { - return sub, nil - } - - // Attempt recovery when the consumer or stream is missing. - if !errors.Is(err, nats.ErrConsumerNotFound) && !errors.Is(err, nats.ErrStreamNotFound) { - return nil, fmt.Errorf("create pull subscription: %w", err) - } - - if streamManager != nil { - if streamErr := streamManager.EnsureStream(nil); streamErr != nil { - return nil, fmt.Errorf("recover stream %s: %w", cm.streamName, streamErr) - } - } - - if consumerConfig == nil { - return nil, fmt.Errorf("consumer config is required for recovery") - } - if err := cm.EnsureConsumer(consumerConfig); err != nil { - return nil, fmt.Errorf("recover consumer %s: %w", cm.consumerName, err) - } - - sub, err = cm.CreatePullSubscription() - if err != nil { - return nil, fmt.Errorf("create pull subscription after recovery: %w", err) - } - return sub, nil -} diff --git a/internal/infra/nats/consumer_test.go b/internal/infra/nats/consumer_test.go index 1eda444..a27f500 100644 --- a/internal/infra/nats/consumer_test.go +++ b/internal/infra/nats/consumer_test.go @@ -3,9 +3,6 @@ package nats import ( "errors" "testing" - "time" - - configpkg "caatsm/internal/infra/config" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -48,97 +45,4 @@ var _ = Describe("Consumer helpers", func() { Expect(mapReplayPolicy("")).To(Equal(nats.ReplayInstantPolicy)) }) }) - - Describe("normalizeConsumerConfig", func() { - It("applies default values when config fields are empty", func() { - cfg := &configpkg.Config{ - NATS: configpkg.NATSConfig{ - Mode: "", - }, - App: configpkg.AppConfig{}, - Timeouts: configpkg.TimeoutsConfig{}, - } - - normCfg := normalizeConsumerConfig(cfg) - - Expect(normCfg.consumerName).To(Equal("telegram-consumer")) - Expect(normCfg.streamName).To(Equal("TELEGRAM")) - Expect(normCfg.batchSize).To(Equal(50)) - Expect(normCfg.batchTimeout).To(Equal(2 * time.Second)) - Expect(normCfg.monitorInterval).To(Equal(30 * time.Second)) - Expect(normCfg.ackWait).To(Equal(30 * time.Second)) - }) - - It("uses provided values when config fields are set", func() { - cfg := &configpkg.Config{ - NATS: configpkg.NATSConfig{ - Consumer: "custom-consumer", - Stream: "CUSTOM_STREAM", - ConsumerRules: configpkg.ConsumerRulesConfig{ - AckWait: 60 * time.Second, - }, - }, - App: configpkg.AppConfig{ - BatchSize: 100, - BatchTimeout: 5 * time.Second, - MonitorInterval: 60 * time.Second, - }, - Timeouts: configpkg.TimeoutsConfig{ - AckWait: 45 * time.Second, - }, - } - - normCfg := normalizeConsumerConfig(cfg) - - Expect(normCfg.consumerName).To(Equal("custom-consumer")) - Expect(normCfg.streamName).To(Equal("CUSTOM_STREAM")) - Expect(normCfg.batchSize).To(Equal(100)) - Expect(normCfg.batchTimeout).To(Equal(5 * time.Second)) - Expect(normCfg.monitorInterval).To(Equal(60 * time.Second)) - Expect(normCfg.ackWait).To(Equal(60 * time.Second)) // Uses ConsumerRules.AckWait - }) - - It("falls back to Timeouts.AckWait when ConsumerRules.AckWait is zero", func() { - cfg := &configpkg.Config{ - NATS: configpkg.NATSConfig{ - ConsumerRules: configpkg.ConsumerRulesConfig{ - AckWait: 0, - }, - }, - Timeouts: configpkg.TimeoutsConfig{ - AckWait: 45 * time.Second, - }, - } - - normCfg := normalizeConsumerConfig(cfg) - - Expect(normCfg.ackWait).To(Equal(45 * time.Second)) - }) - - It("sets dlqSubject when DLQ is enabled", func() { - cfg := &configpkg.Config{ - DLQ: configpkg.DLQConfig{ - Enabled: true, - Subject: "caatsm.dlq", - }, - } - - normCfg := normalizeConsumerConfig(cfg) - - Expect(normCfg.dlqSubject).To(Equal("caatsm.dlq")) - }) - - It("clears dlqSubject when DLQ is disabled", func() { - cfg := &configpkg.Config{ - DLQ: configpkg.DLQConfig{ - Enabled: false, - Subject: "caatsm.dlq", - }, - } - - normCfg := normalizeConsumerConfig(cfg) - - Expect(normCfg.dlqSubject).To(Equal("")) - }) - }) -}) +}) \ No newline at end of file diff --git a/internal/infra/nats/message_fetcher.go b/internal/infra/nats/message_fetcher.go deleted file mode 100644 index d37d5ae..0000000 --- a/internal/infra/nats/message_fetcher.go +++ /dev/null @@ -1,90 +0,0 @@ -package nats - -import ( - "caatsm/internal/infra/config" - "context" - "errors" - "time" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// MessageFetcher defines the interface for fetching messages from NATS -type MessageFetcher interface { - FetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) - HandleFetchError(ctx context.Context, err error, sub **nats.Subscription, fetchErrorStreak *int) (bool, error) -} - -// defaultMessageFetcher implements MessageFetcher interface -type defaultMessageFetcher struct { - batchSize int - batchTimeout time.Duration - logger *zap.Logger - conn *nats.Conn - js nats.JetStreamContext - consumerManager *ConsumerManager - streamManager *StreamManager - config *consumerConfig - cfg *config.Config -} - -func (f *defaultMessageFetcher) FetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) { - return f.fetchBatch(ctx, sub) -} - -// fetchBatch fetches a batch of messages from the subscription with context awareness -func (f *defaultMessageFetcher) fetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) { - // Check context before fetching - select { - case <-ctx.Done(): - return nil, ctx.Err() - default: - } - - // Use a shorter timeout for better responsiveness to cancellation - timeout := f.batchTimeout - if timeout > 500*time.Millisecond { - timeout = 500 * time.Millisecond - } - - return sub.Fetch(f.batchSize, nats.MaxWait(timeout)) -} - -func (f *defaultMessageFetcher) HandleFetchError(ctx context.Context, err error, sub **nats.Subscription, fetchErrorStreak *int) (bool, error) { - // Context cancellation - stop processing - if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { - f.logger.Info("Fetch error due to context cancellation", zap.Error(err)) - return false, err - } - - // Timeout is normal - continue - if errors.Is(err, nats.ErrTimeout) { - return true, nil - } - - // Connection issues - apply simple backoff - *fetchErrorStreak++ - backoff := f.calculateExponentialBackoff(*fetchErrorStreak) - f.logger.Warn("Fetch error, applying backoff", - zap.Error(err), - zap.Int("error_streak", *fetchErrorStreak), - zap.Duration("backoff", backoff), - ) - - if !sleepWithContext(ctx, backoff) { - return false, ctx.Err() - } - - return true, nil -} - -// calculateExponentialBackoff calculates exponential backoff duration with a cap -func (f *defaultMessageFetcher) calculateExponentialBackoff(streak int) time.Duration { - if streak <= 0 { - return 0 - } - // Simple exponential backoff: 2^(streak-1) seconds, capped at 30 seconds - backoff := time.Duration(1< 0 { - *p.consecutiveProcessErrors = 0 - } - - elapsed := time.Since(start) - - // ACK the message - if ackErr := msg.Ack(); ackErr != nil { - p.logger.Error("Failed to ACK message", zap.Error(ackErr)) - // Still record metrics even if ACK fails - } - p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed) -} - -// ProcessMessage processes a single message. -func (p *defaultBatchProcessor) ProcessMessage(ctx context.Context, msg *nats.Msg) error { - ctx, span := otel.Tracer("caatsm/nats").Start(ctx, "Consumer.processMessage") - defer span.End() - - // Set semantic messaging attributes - span.SetAttributes( - attribute.String("messaging.system", "nats"), - attribute.String("messaging.operation.name", "receive"), - attribute.String("messaging.destination.name", msg.Subject), - attribute.String("messaging.consumer.group.name", p.consumerName), - attribute.String("caatsm.stream", p.streamName), - ) - - msgID, source, err := p.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" { - p.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), - ) - } - - // Attach structured logging context including stream/consumer and NATS metadata. - jsSeq := uint64(0) - if meta, metaErr := msg.Metadata(); metaErr == nil { - jsSeq = meta.Sequence.Stream - span.SetAttributes( - attribute.Int64("nats.js.stream_seq", int64(meta.Sequence.Stream)), - attribute.Int64("nats.js.consumer_seq", int64(meta.Sequence.Consumer)), - ) - } - - msgLogger := log.WithMessageContext(p.logger, log.MessageFields{ - Service: "caatsm-consumer", - TransportMsgID: msgID, - Stream: p.streamName, - Consumer: p.consumerName, - Subject: msg.Subject, - JSSequence: jsSeq, - }) - - msgLogger.Debug("Processing message", - zap.Int("data_size", len(msg.Data)), - zap.String("msg_id_source", source), - ) - - // Call processor - if err := p.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 -} - -// resolveMsgID extracts or generates a message ID. -func (p *defaultBatchProcessor) 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 -} - -// handleMessageError handles errors that occur during message processing. -func (p *defaultBatchProcessor) handleMessageError(ctx context.Context, msg *nats.Msg, err error, elapsed time.Duration) { - // Check if context is cancelled before processing - select { - case <-ctx.Done(): - p.logger.Warn("Skipping error handling due to context cancellation", - zap.String("subject", msg.Subject), - ) - return - default: - } - - // Extract message ID for better error logging - msgID, _, _ := p.resolveMsgID(msg) - if msgID == "" { - msgID = "unknown" - } - - isPermanent := app.IsPermanent(err) - p.logger.Error("Failed to process message", - zap.String("subject", msg.Subject), - zap.String("msg_id", msgID), - zap.Error(err), - zap.Bool("permanent", isPermanent), - ) - - result := obsmetrics.ResultFail - if isPermanent { - result = obsmetrics.ResultPermanentFail - } - p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, result, elapsed) - - consecutiveErrors := 0 - if p.consecutiveProcessErrors != nil { - consecutiveErrors = *p.consecutiveProcessErrors - } - - processingResult := ProcessingErrorResult{IsPermanent: isPermanent} - if !isPermanent && consecutiveErrors >= 10 { - processingResult.ShouldApplyBackpressure = true - processingResult.BackpressureDelay = time.Duration(consecutiveErrors) * 100 * time.Millisecond - if processingResult.BackpressureDelay > 5*time.Second { - processingResult.BackpressureDelay = 5 * time.Second - } - } - - if processingResult.IsPermanent { - p.handlePermanentError(ctx, msg, err) - return - } - - p.handleTransientError(ctx, msg, processingResult) -} - -// handlePermanentError handles permanent/poison messages. -func (p *defaultBatchProcessor) handlePermanentError(ctx context.Context, msg *nats.Msg, err error) { - if p.consecutiveProcessErrors != nil { - *p.consecutiveProcessErrors = 0 - } - - // Extract message ID for better logging - msgID, _, _ := p.resolveMsgID(msg) - if msgID == "" { - msgID = "unknown" - } - - // Poison/permanent message: route to DLQ if configured, then ACK - dlqRouted := false - if p.dlqHandler != nil { - if dlqErr := p.dlqHandler.RouteToDLQ(ctx, msg, err); dlqErr != nil { - p.logger.Error("Failed to route permanent-error message to DLQ", - zap.String("subject", msg.Subject), - zap.String("msg_id", msgID), - zap.Error(dlqErr), - zap.NamedError("original_error", err), - ) - // Note: We still ACK the message even if DLQ routing fails to prevent - // infinite redelivery of poison messages. The error is logged for manual investigation. - } else { - dlqRouted = true - p.logger.Info("Permanent-error message routed to DLQ", - zap.String("subject", msg.Subject), - zap.String("msg_id", msgID), - ) - } - } else { - p.logger.Warn("Permanent-error message but DLQ handler not configured - message will be ACKed without DLQ routing", - zap.String("subject", msg.Subject), - zap.String("msg_id", msgID), - zap.String("hint", "Enable DLQ by setting dlq.enabled=true and dlq.subject in config to route poison messages for inspection"), - ) - } - - // ACK the message to prevent redelivery - // Even if DLQ routing failed, we ACK to avoid infinite retries of poison messages - if ackErr := msg.Ack(); ackErr != nil { - p.logger.Error("Failed to ACK permanent-error message", - zap.String("subject", msg.Subject), - zap.String("msg_id", msgID), - zap.Bool("dlq_routed", dlqRouted), - zap.Error(ackErr), - ) - } -} - -// handleTransientError handles transient errors with backpressure and redelivery. -func (p *defaultBatchProcessor) handleTransientError(ctx context.Context, msg *nats.Msg, processingResult ProcessingErrorResult) { - // Increment error streak - if p.consecutiveProcessErrors != nil { - if *p.consecutiveProcessErrors < 0 { - *p.consecutiveProcessErrors = 0 - } - *p.consecutiveProcessErrors++ - } - - if processingResult.ShouldApplyBackpressure { - consecutiveErrors := 0 - if p.consecutiveProcessErrors != nil { - consecutiveErrors = *p.consecutiveProcessErrors - } - p.logger.Warn("Applying backpressure due to consecutive processing errors", - zap.Int("consecutive_errors", consecutiveErrors), - zap.Duration("sleep", processingResult.BackpressureDelay), - ) - // Use context-aware sleep instead of blocking time.Sleep - if !sleepWithContext(ctx, processingResult.BackpressureDelay) { - // Context canceled, stop processing - return - } - } - - // Transient error: request redelivery with optional delay - p.telemetry.RecordRetry(ctx, p.streamName, p.consumerName, obsmetrics.RetryReasonProcessorError) - if nakErr := p.nakWithStrategy(msg); nakErr != nil { - p.logger.Error("Failed to NAK message", zap.Error(nakErr)) - } -} - -// nakWithStrategy sends a NAK with appropriate delay based on retry attempt. -func (p *defaultBatchProcessor) nakWithStrategy(msg *nats.Msg) error { - if len(p.backoff) == 0 { - return msg.Nak() - } - - meta, err := msg.Metadata() - if err != nil { - p.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(p.backoff) { - index = len(p.backoff) - 1 - } - delay := p.backoff[index] - if delay <= 0 { - return msg.Nak() - } - - return msg.NakWithDelay(delay) -} diff --git a/internal/infra/nats/metrics_test.go b/internal/infra/nats/metrics_test.go deleted file mode 100644 index 3e94172..0000000 --- a/internal/infra/nats/metrics_test.go +++ /dev/null @@ -1,37 +0,0 @@ -package nats - -import ( - "context" - "time" - - . "github.com/onsi/ginkgo/v2" - "go.uber.org/zap/zaptest" -) - -var _ = Describe("Metrics", func() { - var ( - c *Consumer - ctx context.Context - ) - - BeforeEach(func() { - ctx = context.Background() - c = &Consumer{ - config: consumerConfig{ - streamName: "TEST_STREAM", - consumerName: "test-consumer", - monitorInterval: 30 * time.Second, // Set a valid interval - }, - logger: zaptest.NewLogger(GinkgoT()), - } - }) - - Describe("emitConsumerStats", func() { - It("handles context cancellation", func() { - ctx, cancel := context.WithCancel(ctx) - cancel() - c.emitConsumerStats(ctx) - // Should return without panic - }) - }) -}) diff --git a/internal/infra/nats/monitor.go b/internal/infra/nats/monitor.go new file mode 100644 index 0000000..c34f86f --- /dev/null +++ b/internal/infra/nats/monitor.go @@ -0,0 +1,181 @@ +package nats + +import ( + "caatsm/internal/infra/config" + obsmetrics "caatsm/internal/infra/metrics" + "context" + "sync" + "time" + + "github.com/nats-io/nats.go" + "go.uber.org/zap" +) + +// ConsumerMonitor handles health monitoring and stats emission for the consumer. +type ConsumerMonitor struct { + logger *zap.Logger + cfg *config.Config + js nats.JetStreamContext + + // Configuration + streamName string + consumerName string + monitorInterval time.Duration + + // State + lastMessageTime time.Time + lastMessageSequence uint64 + messageGapMutex sync.RWMutex +} + +// NewConsumerMonitor creates a new ConsumerMonitor. +func NewConsumerMonitor( + logger *zap.Logger, + cfg *config.Config, + js nats.JetStreamContext, + streamName string, + consumerName string, + monitorInterval time.Duration, +) *ConsumerMonitor { + return &ConsumerMonitor{ + logger: logger, + cfg: cfg, + js: js, + streamName: streamName, + consumerName: consumerName, + monitorInterval: monitorInterval, + } +} + +// Start begins the monitoring loop. +func (m *ConsumerMonitor) Start(ctx context.Context) { + ticker := time.NewTicker(m.monitorInterval) + defer ticker.Stop() + + // Record initial metric (0) to ensure it appears in Prometheus even before first tick + m.logger.Info("Starting NATS consumer stats emission goroutine, recording initial pending metric", + zap.String("stream", m.streamName), + zap.String("consumer", m.consumerName), + zap.Duration("interval", m.monitorInterval), + ) + obsmetrics.RecordNATSConsumerPending(m.streamName, m.consumerName, 0) + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + m.emitStats() + } + } +} + +// emitStats gathers and records consumer statistics. +func (m *ConsumerMonitor) emitStats() { + if m.js != nil { + info, err := m.js.ConsumerInfo(m.streamName, m.consumerName) + if err != nil { + m.logger.Warn("Failed to fetch consumer info for pending messages metric", + zap.String("stream", m.streamName), + zap.String("consumer", m.consumerName), + zap.Error(err), + ) + } else { + // Record pending messages for monitoring + m.logger.Debug("Recording NATS consumer pending messages metric", + zap.String("stream", m.streamName), + zap.String("consumer", m.consumerName), + zap.Uint64("pending", info.NumPending), + ) + obsmetrics.RecordNATSConsumerPending(m.streamName, m.consumerName, info.NumPending) + } + } + + // Record AFTN health metrics + gapSeconds := m.GetMessageGapSeconds() + healthy := m.IsHealthy() + + obsmetrics.RecordMessageGap(m.streamName, m.consumerName, gapSeconds) + obsmetrics.RecordSerialReaderHealth(m.streamName, m.consumerName, healthy) + + if !healthy { + m.logger.Warn("Serial reader appears stalled - no messages received recently", + zap.String("stream", m.streamName), + zap.String("consumer", m.consumerName), + zap.Float64("gap_seconds", gapSeconds), + zap.Duration("threshold", m.cfg.AFTN.MessageGapThreshold), + ) + } +} + +// TrackMessage updates health metrics based on a received message. +func (m *ConsumerMonitor) TrackMessage(msg *nats.Msg) { + if msg == nil { + return + } + + m.messageGapMutex.Lock() + defer m.messageGapMutex.Unlock() + + now := time.Now() + m.lastMessageTime = now + + // Extract sequence number from message metadata + if meta, err := msg.Metadata(); err == nil { + currentSeq := meta.Sequence.Stream + + // Detect sequence gaps if we have a previous sequence + if m.lastMessageSequence > 0 && m.cfg.AFTN.EnableSequenceGapDetection { + if currentSeq > m.lastMessageSequence+1 { + gapSize := currentSeq - m.lastMessageSequence - 1 + m.logger.Warn("Message sequence gap detected", + zap.String("stream", m.streamName), + zap.String("consumer", m.consumerName), + zap.Uint64("last_sequence", m.lastMessageSequence), + zap.Uint64("current_sequence", currentSeq), + zap.Uint64("gap_size", gapSize), + ) + obsmetrics.RecordSequenceGap(m.streamName, m.consumerName, gapSize) + } + } + + m.lastMessageSequence = currentSeq + } +} + +// GetMessageGapSeconds returns the number of seconds since the last message was received. +func (m *ConsumerMonitor) GetMessageGapSeconds() float64 { + m.messageGapMutex.RLock() + defer m.messageGapMutex.RUnlock() + + if m.lastMessageTime.IsZero() { + return 0 + } + + return time.Since(m.lastMessageTime).Seconds() +} + +// IsHealthy returns true if messages are being received within the threshold. +func (m *ConsumerMonitor) IsHealthy() bool { + m.messageGapMutex.RLock() + defer m.messageGapMutex.RUnlock() + + // If we haven't received any messages yet, consider it healthy (initial state) + if m.lastMessageTime.IsZero() { + return true + } + + gap := time.Since(m.lastMessageTime) + return gap < m.cfg.AFTN.MessageGapThreshold +} + +// RecordInitialPending logs and records the initial pending messages count. +// This is exposed to allow recording immediately upon startup. +func (m *ConsumerMonitor) RecordInitialPending(pending uint64) { + m.logger.Info("Recording initial NATS consumer pending messages metric", + zap.String("stream", m.streamName), + zap.String("consumer", m.consumerName), + zap.Uint64("pending", pending), + ) + obsmetrics.RecordNATSConsumerPending(m.streamName, m.consumerName, pending) +} diff --git a/internal/infra/nats/stream_manager.go b/internal/infra/nats/stream_manager.go deleted file mode 100644 index 244b0d9..0000000 --- a/internal/infra/nats/stream_manager.go +++ /dev/null @@ -1,134 +0,0 @@ -package nats - -import ( - "errors" - "fmt" - "time" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// StreamManager handles JetStream stream lifecycle management -type StreamManager struct { - js nats.JetStreamContext - streamName string - subjects []string - logger *zap.Logger -} - -// StreamConfig holds configuration for creating a JetStream stream -type StreamConfig struct { - MaxMsgs int64 - MaxBytes int64 - MaxAge time.Duration - Discard string // "old" or "new" - Storage string // "file" or "memory" - Replicas int -} - -// NewStreamManager creates a new stream manager -func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []string, logger *zap.Logger) *StreamManager { - if logger == nil { - logger = zap.NewNop() - } - return &StreamManager{ - js: js, - streamName: streamName, - subjects: subjects, - logger: logger, - } -} - -// EnsureStream ensures that the configured JetStream stream exists, creating it if necessary -func (sm *StreamManager) EnsureStream(cfg *StreamConfig) error { - // Check if stream already exists - info, err := sm.js.StreamInfo(sm.streamName) - if err == nil { - // Stream exists - check if we need to add any missing subjects - existingSubjects := make(map[string]bool) - for _, subj := range info.Config.Subjects { - existingSubjects[subj] = true - } - - // Check if any configured subjects are missing - missingSubjects := []string{} - for _, subj := range sm.subjects { - if !existingSubjects[subj] { - missingSubjects = append(missingSubjects, subj) - } - } - - if len(missingSubjects) > 0 { - // Update stream to include missing subjects - updatedSubjects := info.Config.Subjects - updatedSubjects = append(updatedSubjects, missingSubjects...) - info.Config.Subjects = updatedSubjects - - _, updateErr := sm.js.UpdateStream(&info.Config) - if updateErr != nil { - return fmt.Errorf("failed to update stream %s with new subjects %v: %w", sm.streamName, missingSubjects, updateErr) - } - - sm.logger.Info("Updated JetStream stream with new subjects", - zap.String("stream", sm.streamName), - zap.Strings("added_subjects", missingSubjects), - zap.Strings("all_subjects", updatedSubjects), - ) - } else { - sm.logger.Info("JetStream stream verified", - zap.String("stream", sm.streamName), - zap.Strings("subjects", sm.subjects), - ) - } - return nil - } - - // If stream doesn't exist, create it - if errors.Is(err, nats.ErrStreamNotFound) { - streamCfg := &nats.StreamConfig{ - Name: sm.streamName, - Subjects: sm.subjects, - } - - // Apply limits if provided - if cfg != nil { - if cfg.MaxMsgs > 0 { - streamCfg.MaxMsgs = cfg.MaxMsgs - } - if cfg.MaxBytes > 0 { - streamCfg.MaxBytes = cfg.MaxBytes - } - if cfg.MaxAge > 0 { - streamCfg.MaxAge = cfg.MaxAge - } - if cfg.Discard == "new" { - streamCfg.Discard = nats.DiscardNew - } else { - streamCfg.Discard = nats.DiscardOld - } - if cfg.Storage == "memory" { - streamCfg.Storage = nats.MemoryStorage - } else { - streamCfg.Storage = nats.FileStorage - } - if cfg.Replicas > 0 { - streamCfg.Replicas = cfg.Replicas - } - } - - _, err := sm.js.AddStream(streamCfg) - if err != nil { - return fmt.Errorf("failed to create stream %s: %w", sm.streamName, err) - } - - sm.logger.Info("Created JetStream stream", - zap.String("stream", sm.streamName), - zap.Strings("subjects", sm.subjects), - ) - return nil - } - - // Other error (e.g., permission denied) - return fmt.Errorf("stream %s not found or inaccessible: %w", sm.streamName, err) -}