diff --git a/Makefile b/Makefile index 937e8fe..d5df59d 100644 --- a/Makefile +++ b/Makefile @@ -29,8 +29,8 @@ build: ## Build the receiver binary run: run-dev ## Alias for run-dev .PHONY: run-dev -run-dev: build ## Run the receiver in development mode (uses core NATS mode by default). Use MONITORING_ADDR=ip:port to set monitoring address - @echo "Running receiver in development mode (NATS mode: core by default)..." +run-dev: build ## Run the receiver in development mode (uses JetStream mode). Use MONITORING_ADDR=ip:port to set monitoring address + @echo "Running receiver in development mode (NATS mode: JetStream)..." @if [ -n "$(MONITORING_ADDR)" ]; then \ echo "Monitoring address: $(MONITORING_ADDR)"; \ GO_ENV=dev $(BINARY) listen --monitoring-addr $(MONITORING_ADDR); \ @@ -54,8 +54,8 @@ run-test: build ## Run the receiver in test mode @GO_ENV=test $(BINARY) listen .PHONY: run-local -run-local: ## Run receiver directly via go run (uses core NATS mode by default in dev). Use MONITORING_ADDR=ip:port to set monitoring address - @echo "Running receiver via go run (GO_ENV=$(GO_ENV), NATS mode: core by default in dev)..." +run-local: ## Run receiver directly via go run (uses JetStream mode). Use MONITORING_ADDR=ip:port to set monitoring address + @echo "Running receiver via go run (GO_ENV=$(GO_ENV), NATS mode: JetStream)..." @if [ -n "$(MONITORING_ADDR)" ]; then \ echo "Monitoring address: $(MONITORING_ADDR)"; \ GO_ENV=$(GO_ENV) go run $(CMD) listen --monitoring-addr $(MONITORING_ADDR); \ @@ -121,7 +121,7 @@ clean: ## Clean build artifacts and coverage files @rm -rf $(BUILD_DIR) coverage .PHONY: seed -seed: ## Generate sample telegrams (publishes to Core NATS by default) +seed: ## Generate sample telegrams (publishes to JetStream) @GO_ENV=dev \ NATS_URL=$${CAATSM_NATS_URL:-nats://localhost:4222} \ SUBJECT=$${CAATSM_NATS_SUBJECT:-telegram.serial} \ diff --git a/README.md b/README.md index f55cd76..83f0e1a 100644 --- a/README.md +++ b/README.md @@ -87,8 +87,7 @@ Configuration is loaded from TOML files and environment variables. The configura ```toml [nats] url = "nats://localhost:4222" -# mode: "jetstream" (default) or "core" -# See "NATS Mode Selection" section below for detailed comparison +# mode: "jetstream" (required - only JetStream mode is supported) mode = "jetstream" stream = "TELEGRAM" consumer = "telegram-consumer" @@ -96,7 +95,7 @@ client = "serial-client" cluster = "tele-cluster" [nats.stream_limits] -# These settings only apply when mode = "jetstream" +# Stream retention and storage limits max_msgs = 100000 max_bytes = 67108864 max_age = "24h" @@ -105,7 +104,7 @@ storage = "file" replicas = 1 [nats.consumer_rules] -# Consumer delivery rules (only applies when mode = "jetstream") +# Consumer delivery rules # max_deliver: Maximum number of delivery attempts before giving up max_deliver = 3 # ack_wait: Time to wait for ACK before redelivering message @@ -116,7 +115,7 @@ max_ack_pending = 1000 [subscription] # Optional. Defaults to "telegram.>" when omitted. topic = "telegram.serial" -# queue_group: Used in both core and jetstream modes for load balancing +# queue_group: Queue group for load balancing (JetStream uses durable consumers) queue_group = "tele-queue" [publisher] @@ -295,76 +294,6 @@ The application supports two NATS consumption modes, controlled by `nats.mode`: - **High pending count**: Increase `batch_size` or add more consumer instances - **Messages being redelivered**: Check processing logs for errors; adjust `ack_wait` if processing takes longer -#### Core NATS Mode (Default for Development) - -**Configuration:** `nats.mode = "core"` (default in `config.dev.toml`) - -**Features:** -- ⚡ **Simple Pub/Sub**: Basic publish/subscribe messaging -- ⚡ **Queue Groups**: Load balancing across multiple consumers -- ⚡ **Low Latency**: No persistence overhead -- ⚡ **Fast Startup**: No stream/consumer setup required -- ❌ **No Persistence**: Messages are lost if no consumer is available -- ❌ **No ACK**: No delivery guarantees -- ❌ **No Retry**: Processing failures are logged but not retried -- ❌ **No DLQ**: Failed messages cannot be routed to a dead-letter queue - -**Use Cases:** -- **Local development** (recommended default) -- Quick testing and iteration -- Real-time monitoring/logging where message loss is acceptable -- Simple pub/sub scenarios without reliability requirements -- Performance testing without persistence overhead - -**Configuration Requirements:** -- Works with any NATS server (JetStream not required) -- Only `[subscription]` settings are used (queue_group for load balancing) -- `[nats.consumer_rules]` and `[nats.stream_limits]` are ignored - -**How to Use Core Mode:** - -1. **Start NATS Server** (JetStream not required, but can be enabled): - ```bash - # Simple NATS server - nats-server - - # Or with Docker Compose (JetStream enabled but not required for core mode) - docker compose -f docker-compose.dev.yml up -d nats - ``` - -2. **Start the Application** (Core mode is default in dev config): - ```bash - # Core mode is default, no need to specify - GO_ENV=dev \ - CAATSM_POSTGRES_URL=postgres://user:pass@localhost:5432/aviation \ - go run ./cmd/main listen - ``` - -3. **Publish Messages** (use standard NATS publish): - ```bash - # Using nats-box - docker compose exec nats-box nats pub telegram.serial "ZCZC TEST123 150631..." - - # Or use seed-telegrams without --jetstream flag - go run ./cmd/seed-telegrams \ - --nats-url nats://localhost:4222 \ - --subject telegram.serial \ - --count 10 - ``` - -**Switching Modes:** - -```bash -# Use Core NATS mode (default for development) -CAATSM_NATS_MODE=core go run ./cmd/main listen -# Or simply (core is default in config.dev.toml) -go run ./cmd/main listen - -# Use JetStream mode (for production or integration testing) -CAATSM_NATS_MODE=jetstream go run ./cmd/main listen -``` - -**Note:** The publisher always uses JetStream for deduplicated fan-out, regardless of the consumer mode. If you need pure Core NATS, ensure publishers also use Core NATS subjects. ## Usage @@ -416,7 +345,7 @@ Build information is automatically populated from: Use Make targets (binary mode): ```bash -make run-dev # GO_ENV=dev (uses core NATS mode by default) +make run-dev # GO_ENV=dev (uses JetStream mode) make run-local # go run ./cmd/main listen (honors GO_ENV) ``` @@ -425,7 +354,7 @@ Task equivalents: ```bash task run-dev task run-local # go run ./cmd/main listen -task dev-run # boots docker-compose dev stack + go run (core NATS mode) +task dev-run # boots docker-compose dev stack + go run (JetStream mode) ``` #### Production Mode @@ -567,7 +496,7 @@ docker compose -f docker-compose.dev.yml up -d otel-collector jaeger prometheus Run the processor locally while the infra runs in Docker: ```bash -# Core NATS mode (default for development, fast and lightweight) +# Development mode (JetStream enabled) GO_ENV=dev \ CAATSM_POSTGRES_URL=postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable \ go run ./cmd/main listen @@ -655,9 +584,8 @@ The project keeps tests close to the code that they exercise: ## Message Flow -1. **NATS Consumer** receives raw telegram messages from NATS: - - **JetStream mode** (default): Uses durable pull consumer with batch processing, ACK/NAK, and retry logic - - **Core mode**: Uses `QueueSubscribe` for simple pub/sub with queue group load balancing (no persistence or retries) +1. **NATS Consumer** receives raw telegram messages from NATS JetStream: + - Uses durable pull consumer with batch processing, ACK/NAK, and retry logic 2. **MessageProcessor** orchestrates the processing: - Parses the message using the Parser adapter - Stores the parsed message in PostgreSQL via Repository diff --git a/Taskfile.yml b/Taskfile.yml index 2c0344e..1b0ad4f 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -85,13 +85,13 @@ tasks: - GO_ENV=test {{.binary}} listen run-local: - desc: 'Run receiver via go run (default GO_ENV=dev, uses core NATS mode by default). Use MONITORING_ADDR=ip:port to set monitoring address' + desc: 'Run receiver via go run (default GO_ENV=dev, uses JetStream mode). Use MONITORING_ADDR=ip:port to set monitoring address' vars: MONITORING_ADDR: sh: echo "${MONITORING_ADDR:-}" cmds: - | - echo "Running receiver via go run (GO_ENV=${GO_ENV:-dev}, NATS mode: core by default in dev)..." + echo "Running receiver via go run (GO_ENV=${GO_ENV:-dev}, NATS mode: JetStream)..." - | cmd="GO_ENV=${GO_ENV:-dev} go run {{.cmd}} listen" if [ -n "${MONITORING_ADDR:-}" ]; then diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 9c8787f..333e70e 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -1,20 +1,12 @@ [nats] url = "nats://localhost:4222" -# mode: "core" (default for dev) or "jetstream" (recommended for production) -# - "core": Uses Core NATS for simple pub/sub messaging with: -# * No message persistence (messages lost if consumer offline) -# * No ACK mechanism (fire-and-forget delivery) -# * No automatic retry on processing failures -# * Queue groups for load balancing only -# * Suitable for development/testing or real-time scenarios where message loss is acceptable -# * Recommended for local development and testing -# - "jetstream": Uses NATS JetStream for persistent message streaming with: +# mode: "jetstream" (required - only JetStream mode is supported) +# JetStream provides: # * Message persistence and replay capability # * ACK/NAK mechanism for guaranteed delivery # * Automatic retry with configurable backoff # * Dead-letter queue (DLQ) support # * Batch processing and consumer monitoring -# * Recommended for production environments mode = "jetstream" client = "serial-client" cluster = "tele-cluster" @@ -22,7 +14,7 @@ stream = "TELEGRAM" consumer = "telegram-consumer" [nats.stream_limits] -# Stream retention and storage limits (only applies when mode = "jetstream") +# Stream retention and storage limits # max_msgs: Maximum number of messages to keep in the stream (0 = unlimited) max_msgs = 100000 # max_bytes: Maximum total size of messages in bytes (0 = unlimited, 67108864 = 64MB) @@ -37,7 +29,7 @@ storage = "file" replicas = 1 [nats.consumer_rules] -# Consumer delivery rules (only applies when mode = "jetstream") +# Consumer delivery rules # max_deliver: Maximum number of delivery attempts before giving up (0 = unlimited) max_deliver = 3 # ack_wait: Time to wait for ACK before redelivering message (e.g., "30s", "2m") @@ -75,7 +67,7 @@ max_conns = 10 min_conns = 2 [app] -# Batch processing configuration (applies to both core and jetstream modes) +# Batch processing configuration # batch_size: Number of messages to fetch in each batch (JetStream pull batch size) batch_size = 50 # batch_timeout: Maximum time to wait when fetching a batch (e.g., "2s") @@ -121,7 +113,7 @@ write_timeout = "5s" health_timeout = "2s" [dlq] -# Dead-Letter Queue configuration (only applies when mode = "jetstream") +# Dead-Letter Queue configuration # enabled: Enable DLQ routing for poison messages (messages that fail after max_deliver attempts) enabled = true # Set to true when switching to JetStream mode # subject: NATS subject where failed messages will be published for manual inspection diff --git a/configs/config.testdefaults.toml b/configs/config.testdefaults.toml index 69ec52b..0433a13 100644 --- a/configs/config.testdefaults.toml +++ b/configs/config.testdefaults.toml @@ -1,5 +1,6 @@ [nats] url = "nats://localhost:4222" +mode = "jetstream" stream = "TELEGRAM" [publisher] diff --git a/docs/dev-guide.md b/docs/dev-guide.md index 9e44b44..6866c4d 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -10,11 +10,7 @@ Spin up PostgreSQL/TimescaleDB and NATS JetStream in the background: docker compose -f docker-compose.dev.yml up -d postgres nats nats-box ``` -> **NATS Mode Selection:** The application supports two consumption modes: -> - **Core NATS mode** (default for development): Simple pub/sub without persistence or retry mechanisms. Recommended for local development and testing where message loss is acceptable. Fast and lightweight. -> - **JetStream mode**: Provides message persistence, ACK/NAK, automatic retries, and DLQ support. Recommended for production environments and integration testing. -> -> Development mode defaults to `nats.mode = "core"` in `config.dev.toml`. To use JetStream in development, set `CAATSM_NATS_MODE=jetstream` or change the config file. **Note:** The publisher always targets JetStream for deduplicated fan-out, so if you use Core mode for consumption, ensure your publishers align with your messaging strategy. See the README.md "NATS Mode Selection" section for a detailed comparison. +> **NATS Mode:** The application uses JetStream mode exclusively for message persistence, ACK/NAK, automatic retries, and DLQ support. This ensures consistent behavior across development, testing, and production environments. - `postgres` seeds the `aviation` schema using `internal/infra/postgres/telegrams.ddl` and exposes port `5432`. - `nats` enables JetStream with client port `4222` and monitoring/UI on `8222`. @@ -55,7 +51,7 @@ docker compose -f docker-compose.dev.yml up -d postgres nats nats-box The `Taskfile.yml` includes helper targets that wrap the commands above: - `task up` – starts PostgreSQL, NATS (JetStream, toolbox, and Prometheus exporter), and the observability stack (OpenTelemetry Collector, Jaeger, Prometheus, Grafana) using Docker Compose. -- `task dev-run` – ensures `task up` has run, exports the necessary `CAATSM_*` environment variables (defaults to `CAATSM_NATS_MODE=core` for development), and executes `go run ./cmd/main listen` with telemetry enabled. +- `task dev-run` – ensures `task up` has run, exports the necessary `CAATSM_*` environment variables, and executes `go run ./cmd/main listen` with telemetry enabled. - `task down` – stops the entire stack and removes containers/volumes. Use these tasks if you prefer a one-command workflow instead of invoking `docker compose` and environment exports manually. @@ -80,31 +76,17 @@ GO_ENV=dev go run ./cmd/seed-telegrams \ --status random ``` -### Publishing to Core NATS - -For Core NATS mode, use standard publish: - -```bash -# Publish to Core NATS (no persistence) -GO_ENV=dev go run ./cmd/seed-telegrams \ - --nats-url nats://127.0.0.1:4222 \ - --subject telegram.serial \ - --count 20 \ - --category mixed \ - --status random -``` - ### Using Makefile/Taskfile Tasks For convenience, you can use the provided tasks: ```bash -# Quick seed (10 messages, Core NATS) +# Quick seed (10 messages, JetStream) make seed # or task seed -# Continuous slow seeding (until Ctrl-C, JetStream by default) +# Continuous slow seeding (until Ctrl-C, JetStream) make seed-slow # or task seed-slow @@ -125,8 +107,6 @@ make seed-slow # Custom interval and category make seed-slow INTERVAL_MIN=5s INTERVAL_MAX=10s CATEGORY=DEP -# Use Core NATS instead of JetStream -make seed-slow USE_JS=false INTERVAL_MIN=1s INTERVAL_MAX=3s ``` This is equivalent to running: @@ -164,7 +144,7 @@ Setting `--count 0` makes it run indefinitely until interrupted. # View messages in JetStream stream docker compose exec nats-box nats stream view TELEGRAM -# Subscribe to messages (Core NATS or JetStream) +# Subscribe to messages (JetStream) docker compose exec nats-box nats sub 'telegram.>' # View consumer status and pending messages @@ -184,22 +164,12 @@ The main processor keeps consuming `subscription.topic` (defaults to `telegram.> 2. **Run the processor with telemetry enabled** ```bash - # Using Core NATS mode (default for development) CAATSM_TELEMETRY_ENABLED=true \ CAATSM_TELEMETRY_ENDPOINT=localhost:4318 \ CAATSM_TELEMETRY_INSECURE=true \ GO_ENV=dev \ CAATSM_POSTGRES_URL=postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable \ go run ./cmd/main listen - - # Or use JetStream mode for integration testing - CAATSM_TELEMETRY_ENABLED=true \ - CAATSM_TELEMETRY_ENDPOINT=localhost:4318 \ - CAATSM_TELEMETRY_INSECURE=true \ - GO_ENV=dev \ - CAATSM_NATS_MODE=jetstream \ - CAATSM_POSTGRES_URL=postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable \ - go run ./cmd/main listen ``` - The service name reported to Jaeger is `caatsm`. diff --git a/docs/nats.md b/docs/nats.md index f9ff8e0..4caebc1 100644 --- a/docs/nats.md +++ b/docs/nats.md @@ -445,9 +445,7 @@ Success The consumer handles message consumption with the following features: -#### Modes -- **JetStream Mode**: Persistent, durable message processing with acknowledgments -- **Core Mode**: Fire-and-forget message processing for simple use cases +The consumer uses JetStream mode exclusively for persistent, durable message processing with acknowledgments. #### Key Features - **Batch Processing**: Configurable batch sizes for efficient processing @@ -508,7 +506,7 @@ The publisher handles message publishing with deduplication and observability. #### Features - **Message Deduplication**: Automatic UUID-based deduplication headers -- **JetStream/Core Mode Support**: Adapts based on available JetStream context +- **JetStream Publishing**: Uses JetStream for reliable message delivery - **Structured Logging**: Comprehensive logging of publish operations - **Error Classification**: Distinguishes between transient and permanent errors diff --git a/docs/reliability.md b/docs/reliability.md index 0b93efc..a9957ac 100644 --- a/docs/reliability.md +++ b/docs/reliability.md @@ -15,11 +15,7 @@ subject = "caatsm.dlq" - When `dlq.enabled` is `true` and `dlq.subject` is non-empty, **permanent** failures are routed to the DLQ subject. - A permanent failure is indicated by wrapping an error with `app.Permanent` and is treated as a **poison message**. -> Note: DLQ routing only applies when the NATS mode is `jetstream`. In `core` -> mode, the consumer does not attempt to publish to the DLQ even if it is -> configured. - -Behaviour (JetStream mode): +Behaviour: 1. The NATS consumer calls `processor.Handle`. 2. If an error is returned and `app.IsPermanent(err)` is true: diff --git a/go.mod b/go.mod index 5eb28ab..588bafb 100644 --- a/go.mod +++ b/go.mod @@ -24,7 +24,7 @@ require ( go.opentelemetry.io/otel/sdk v1.38.0 go.opentelemetry.io/otel/sdk/metric v1.38.0 go.opentelemetry.io/otel/trace v1.38.0 - go.uber.org/zap v1.27.0 + go.uber.org/zap v1.27.1 gopkg.in/natefinch/lumberjack.v2 v2.2.1 ) @@ -100,7 +100,7 @@ require ( go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.3 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/crypto v0.44.0 // indirect + golang.org/x/crypto v0.45.0 // indirect golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect golang.org/x/mod v0.30.0 // indirect golang.org/x/net v0.47.0 // indirect diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index a55471d..7f36b17 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -330,10 +330,10 @@ func (c *Config) Validate() error { if c.NATS.URL == "" { return fmt.Errorf("nats.url is required") } - switch strings.ToLower(c.NATS.Mode) { - case "", "jetstream", "core": - default: - return fmt.Errorf("nats.mode must be 'jetstream' or 'core'") + // Validate NATS mode - only JetStream is supported + mode := strings.ToLower(c.NATS.Mode) + if mode != "" && mode != "jetstream" { + return fmt.Errorf("nats.mode must be 'jetstream' or empty (defaults to 'jetstream'), got: %s", c.NATS.Mode) } if c.NATS.Stream == "" { return fmt.Errorf("nats.stream is required") diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index e83011d..abeadc5 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -45,7 +45,6 @@ type Consumer struct { type consumerConfig struct { subject string consumerName string - mode string streamName string dlqSubject string ackWait time.Duration @@ -78,75 +77,56 @@ func ProvideConsumer( // 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. - // We do this first, before any operations that might fail, to ensure the metric exists. - // Initialize the metric unconditionally when in JetStream mode, even if js is nil, - // as it will be updated later when js becomes available. - if normCfg.mode == "jetstream" { - 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), - ) - // Always initialize the metric in JetStream mode to ensure it appears in Prometheus - // The metric will be updated with actual values when the consumer starts - obsmetrics.RecordNATSConsumerPending(normCfg.streamName, normCfg.consumerName, 0) - } else { - logger.Debug("Skipping NATS consumer pending messages metric initialization (not JetStream mode)", - zap.String("mode", normCfg.mode), - ) - } + 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 - if consumer.config.mode == "jetstream" { - 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) + 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 - } + // Update fetcher with managers now that they're initialized + if fetcher, ok := consumer.fetcher.(*defaultMessageFetcher); ok { + fetcher.consumerManager = consumer.consumerManager + fetcher.streamManager = consumer.streamManager + } - // 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 err := consumer.streamManager.EnsureStream(streamCfg); err != nil { - return nil, fmt.Errorf("failed to ensure stream: %w", err) - } + // 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 err := consumer.streamManager.EnsureStream(streamCfg); err != nil { + return nil, fmt.Errorf("failed to ensure stream: %w", err) + } - // 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) - } - - } else { - logger.Info("Running consumer in core NATS mode", - zap.String("subject", normCfg.subject), - zap.String("queue_group", cfg.Subscription.QueueGroup), - ) + // 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 @@ -185,7 +165,6 @@ func (c *Consumer) initCollaborators() { telemetry: c.telemetry, streamName: c.config.streamName, consumerName: c.config.consumerName, - mode: c.config.mode, backoff: c.cfg.NATS.ConsumerRules.Backoff, consecutiveProcessErrors: &c.consecutiveProcessErrors, } @@ -201,11 +180,6 @@ func normalizeConsumerConfig(cfg *config.Config) *consumerConfig { consumerName = "telegram-consumer" } - mode := strings.ToLower(cfg.NATS.Mode) - if mode == "" { - mode = "jetstream" - } - streamName := cfg.NATS.Stream if streamName == "" { streamName = "TELEGRAM" @@ -244,7 +218,6 @@ func normalizeConsumerConfig(cfg *config.Config) *consumerConfig { return &consumerConfig{ subject: subject, consumerName: consumerName, - mode: mode, streamName: streamName, dlqSubject: dlqSubject, ackWait: ackWait, @@ -269,12 +242,8 @@ func (c *Consumer) buildConsumerConfig() *nats.ConsumerConfig { } } -// Start starts consuming messages. +// Start starts consuming messages from JetStream. func (c *Consumer) Start(ctx context.Context) error { - if c.config.mode == "core" { - return c.startCore(ctx) - } - return c.startJetStream(ctx) } @@ -299,42 +268,6 @@ func (c *Consumer) validateDLQ() error { return c.ValidateDLQ() } -// startCore starts the Core NATS consumer loop. -func (c *Consumer) startCore(ctx context.Context) error { - queueGroup := c.cfg.Subscription.QueueGroup - if queueGroup == "" { - queueGroup = c.config.consumerName - } - - handler := func(msg *nats.Msg) { - // Use ProcessBatch to ensure metrics are recorded via processSingleMessage - // ProcessBatch handles error recording and metrics for both success and failure cases - c.batchProcessor.ProcessBatch(ctx, []*nats.Msg{msg}) - } - - sub, err := c.conn.QueueSubscribe(c.config.subject, queueGroup, handler) - if err != nil { - return fmt.Errorf("failed to subscribe to %s: %w", c.config.subject, err) - } - if err := c.conn.Flush(); err != nil { - return fmt.Errorf("failed to flush NATS connection: %w", err) - } - - c.logger.Info("Started core NATS subscription", - zap.String("subject", c.config.subject), - zap.String("queue_group", queueGroup), - ) - - <-ctx.Done() - c.logger.Info("Stopping core NATS consumer", zap.Error(ctx.Err())) - - if err := sub.Drain(); err != nil && !errors.Is(err, nats.ErrConnectionClosed) { - return fmt.Errorf("failed to drain core subscription: %w", err) - } - - return ctx.Err() -} - // createPullSubscription creates a pull subscription func (c *Consumer) createPullSubscription() (*nats.Subscription, error) { return c.consumerManager.CreatePullSubscription() diff --git a/internal/infra/nats/consumer_test.go b/internal/infra/nats/consumer_test.go index 8c365c6..1eda444 100644 --- a/internal/infra/nats/consumer_test.go +++ b/internal/infra/nats/consumer_test.go @@ -62,7 +62,6 @@ var _ = Describe("Consumer helpers", func() { normCfg := normalizeConsumerConfig(cfg) Expect(normCfg.consumerName).To(Equal("telegram-consumer")) - Expect(normCfg.mode).To(Equal("jetstream")) Expect(normCfg.streamName).To(Equal("TELEGRAM")) Expect(normCfg.batchSize).To(Equal(50)) Expect(normCfg.batchTimeout).To(Equal(2 * time.Second)) @@ -74,7 +73,6 @@ var _ = Describe("Consumer helpers", func() { cfg := &configpkg.Config{ NATS: configpkg.NATSConfig{ Consumer: "custom-consumer", - Mode: "core", Stream: "CUSTOM_STREAM", ConsumerRules: configpkg.ConsumerRulesConfig{ AckWait: 60 * time.Second, @@ -93,7 +91,6 @@ var _ = Describe("Consumer helpers", func() { normCfg := normalizeConsumerConfig(cfg) Expect(normCfg.consumerName).To(Equal("custom-consumer")) - Expect(normCfg.mode).To(Equal("core")) Expect(normCfg.streamName).To(Equal("CUSTOM_STREAM")) Expect(normCfg.batchSize).To(Equal(100)) Expect(normCfg.batchTimeout).To(Equal(5 * time.Second)) diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index 13fa5d1..8358f9a 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -72,14 +72,7 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) } // ProvideJetStream creates a JetStream context from a NATS connection. -// Returns nil when mode is "core" to indicate JetStream should not be used. func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) { - // In core mode, return nil so that publishers/consumers use core NATS - if cfg.NATS.Mode == "core" { - logger.Debug("Skipping JetStream initialization (core mode)") - return nil, nil - } - js, err := nc.JetStream() if err != nil { logger.Error("failed to create JetStream context", zap.Error(err)) diff --git a/internal/infra/nats/message_handler_test.go b/internal/infra/nats/message_handler_test.go index 86d6ff0..daf246b 100644 --- a/internal/infra/nats/message_handler_test.go +++ b/internal/infra/nats/message_handler_test.go @@ -17,7 +17,6 @@ var _ = Describe("MessageHandler", func() { logger: zaptest.NewLogger(GinkgoT()), streamName: "TEST_STREAM", consumerName: "test-consumer", - mode: "jetstream", } }) @@ -34,20 +33,7 @@ var _ = Describe("MessageHandler", func() { Expect(source).To(Equal("header")) }) - It("generates UUID for core mode when header is missing", func() { - processor.mode = "core" - msg := &nats.Msg{ - Header: nats.Header{}, - } - - id, source, err := processor.resolveMsgID(msg) - Expect(err).NotTo(HaveOccurred()) - Expect(id).NotTo(BeEmpty()) - Expect(source).To(Equal("generated")) - }) - - It("returns error for JetStream mode when header and metadata are missing", func() { - processor.mode = "jetstream" + It("returns error when header and metadata are missing", func() { msg := &nats.Msg{ Header: nats.Header{}, } diff --git a/internal/infra/nats/message_processor.go b/internal/infra/nats/message_processor.go index dbb0760..3bfc7b2 100644 --- a/internal/infra/nats/message_processor.go +++ b/internal/infra/nats/message_processor.go @@ -9,7 +9,6 @@ import ( "fmt" "time" - "github.com/google/uuid" "github.com/nats-io/nats.go" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -39,7 +38,6 @@ type defaultBatchProcessor struct { // Configuration needed for processing streamName string consumerName string - mode string backoff []time.Duration // Pointer to consecutive errors counter (shared with Consumer) consecutiveProcessErrors *int @@ -76,19 +74,12 @@ func (p *defaultBatchProcessor) processSingleMessage(ctx context.Context, msg *n elapsed := time.Since(start) - // ACK the message (only in JetStream mode; Core NATS doesn't support ACK) - if p.mode == "jetstream" { - 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) - } else { - p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed) - } - } else { - // Core NATS mode: record metrics without ACK (ACK not supported) - p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed) + // 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. @@ -160,10 +151,6 @@ func (p *defaultBatchProcessor) resolveMsgID(msg *nats.Msg) (string, string, err return id, "header", nil } - if p.mode == "core" { - return uuid.NewString(), "generated", nil - } - meta, err := msg.Metadata() if err != nil { return "", "", fmt.Errorf("fetch metadata: %w", err) @@ -232,13 +219,6 @@ func (p *defaultBatchProcessor) handlePermanentError(ctx context.Context, msg *n *p.consecutiveProcessErrors = 0 } - if p.mode != "jetstream" { - p.logger.Debug("Permanent-error message in core mode; skipping DLQ/ACK (not supported)", - zap.String("subject", msg.Subject), - ) - return - } - // Extract message ID for better logging msgID, _, _ := p.resolveMsgID(msg) if msgID == "" { @@ -310,13 +290,6 @@ func (p *defaultBatchProcessor) handleTransientError(ctx context.Context, msg *n } } - if p.mode != "jetstream" { - p.logger.Debug("Transient-error message in core mode; skipping retry (ACK/NAK unsupported)", - zap.String("subject", msg.Subject), - ) - 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 { diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index c6afe2a..a420296 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -20,8 +20,7 @@ type Publisher struct { logger *zap.Logger } -// ProvidePublisher creates a NATS publisher. -// When js is nil (core mode), returns a CorePublisher that uses plain NATS. +// ProvidePublisher creates a NATS JetStream publisher. func ProvidePublisher( js nats.JetStreamContext, nc *nats.Conn, @@ -29,7 +28,7 @@ func ProvidePublisher( logger *zap.Logger, ) (port.Publisher, error) { if js == nil { - return ProvideCorePublisher(nc, cfg, logger) + return nil, fmt.Errorf("JetStream context is required") } return &Publisher{ js: js, @@ -38,63 +37,6 @@ func ProvidePublisher( }, nil } -// CorePublisher publishes messages to plain NATS (non-JetStream) -type CorePublisher struct { - conn *nats.Conn - cfg *config.Config - logger *zap.Logger -} - -// ProvideCorePublisher creates a NATS publisher for core mode -func ProvideCorePublisher( - conn *nats.Conn, - cfg *config.Config, - logger *zap.Logger, -) (port.Publisher, error) { - return &CorePublisher{ - conn: conn, - cfg: cfg, - logger: logger, - }, nil -} - -// Publish publishes a message using plain NATS -func (p *CorePublisher) Publish(message any) error { - topic := p.cfg.Publisher.Topic - if topic == "" { - p.logger.Error("publisher topic is not configured") - return fmt.Errorf("publisher topic is not configured") - } - - // Marshal message to JSON - messageBytes, err := json.Marshal(message) - if err != nil { - p.logger.Error("failed to marshal message", - zap.String("topic", topic), - zap.Error(err), - ) - return fmt.Errorf("failed to marshal message: %w", err) - } - - // Publish to plain NATS - err = p.conn.Publish(topic, messageBytes) - if err != nil { - p.logger.Error("failed to publish message", - zap.String("topic", topic), - zap.Int("message_size", len(messageBytes)), - zap.Error(err), - ) - return fmt.Errorf("failed to publish message: %w", err) - } - - p.logger.Debug("Published message", - zap.String("topic", topic), - zap.Int("size", len(messageBytes)), - ) - - return nil -} - // Publish publishes a message func (p *Publisher) Publish(message any) error { topic := p.cfg.Publisher.Topic