diff --git a/.gitignore b/.gitignore index f4247ec..6551e7d 100644 --- a/.gitignore +++ b/.gitignore @@ -40,6 +40,9 @@ go.sum # Project-specific # configs/*.toml +# Prometheus target files (generated dynamically) +configs/prometheus/targets/*.json + # Logs *.log logs/ diff --git a/AGENTS.md b/AGENTS.md index 7bca49f..33afd90 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,38 +1,24 @@ -# Agent Guidelines for CAATSM Repository +# Agent Guidelines for CAATSM -## Build/Test Commands -- **Build**: `make build` or `task build` (compiles to `bin/receiver`) -- **Run dev**: `make run-dev` or `task run-dev` (uses `configs/config.dev.toml`) -- **Lint**: `make lint` or `task lint` (golangci-lint required) -- **Unit tests**: `make test` (Ginkgo) or `ginkgo -r -v ./path/to/package` for single test -- **Integration tests**: `make test-int` (requires Docker) -- **All tests**: `make test-all` -- **Coverage**: `make coverage` (target: maintain >80% coverage) +## Commands +- **Build**: `make build` (bin/receiver) +- **Run**: `make run-dev` (dev mode), `make run-prod` (prod mode) +- **Lint**: `make lint` (golangci-lint) +- **Test**: `make test` (Unit/Ginkgo), `make test-int` (Integration/Docker) +- **Single Test**: `ginkgo -r -v --focus "Test Description" ./path/to/package` +- **Coverage**: `make coverage` (>80% target) -## Code Style Guidelines -- **Formatting**: Use tabs, `go fmt ./...` or `goimports` before commits -- **Naming**: `camelCase` for locals/unexported, `CamelCase` for exported; package names match directories -- **Imports**: Standard library → third-party → internal (alphabetized within groups) -- **Types**: Use interfaces for ports, appropriate Go types; avoid `any` unless necessary -- **Error handling**: Wrap errors with context, use `errors.Is()` for checking -- **Generated code**: Never edit `/pkg/di/wire_gen.go` or other generated files -- **Linting**: `golangci-lint run ./...` required; fix all issues before PR - -## Testing & Architecture -- **Unit tests**: Ginkgo BDD style next to implementation (`*_test.go`); declarative descriptions -- **Integration**: Testcontainers in `test/integration`; spin up NATS/TimescaleDB -- **Coverage**: Run `make coverage` before merging; address regressions -- **Structure**: Clean Architecture - `domain` (business logic), `app` (use cases), `adapter` (I/O), `infra` (framework deps) -- **Entry point**: `cmd/main` -- **Config**: `configs/config..toml`; secrets via `CAATSM_*` env vars +## Code Style & Architecture +- **Structure**: Clean Architecture (`cmd/`, `internal/{domain,app,adapter,infra}`, `pkg/`). +- **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`. +- **Types**: Interface-driven development. Avoid `any`. +- **Testing**: Ginkgo BDD style (`Describe`, `It`). Table-driven. Mock interfaces. +- **Observability**: Propagate `context.Context`. Use OpenTelemetry (traces/metrics). +- **Generated**: NEVER edit `wire_gen.go` or `*_gen.go`. ## Cursor Rules (.cursor/rules/do.mdc) -- **Expertise**: Go, microservices, Clean Architecture, test-driven development -- **Architecture**: Clean Architecture with domain-driven design, interface-driven development -- **Project Structure**: cmd/, internal/, pkg/, api/, configs/, test/ layout -- **Best Practices**: Short focused functions, explicit error handling, context propagation, goroutine safety -- **Security**: Input validation, secure defaults, retries/backoff, circuit breakers -- **Testing**: Table-driven unit tests, mock interfaces, separate fast/slow tests -- **Observability**: Production-ready OpenTelemetry with environment-based sampling, comprehensive resource attributes, semantic span conventions, and dual telemetry (OTEL + Prometheus) -- **Performance**: Benchmarks, minimize allocations, profile before optimization -- **Tooling**: Go modules, linting, CI automation +- **Expertise**: Go, Microservices, Clean Arch, TDD. +- **Security**: Input validation, secure defaults, retries/backoff. +- **Perf**: Benchmarks, minimize allocations. diff --git a/README.md b/README.md index 2562515..c546a6b 100644 --- a/README.md +++ b/README.md @@ -105,15 +105,13 @@ storage = "file" replicas = 1 [nats.consumer_rules] -# These settings only apply when mode = "jetstream" -max_deliver = 5 +# Consumer delivery rules (only applies when mode = "jetstream") +# max_deliver: Maximum number of delivery attempts before giving up +max_deliver = 3 +# ack_wait: Time to wait for ACK before redelivering message ack_wait = "30s" -max_ack_pending = 1024 -deliver_policy = "all" # all,new,last,last_per_subject,sequence,time -replay_policy = "instant" # instant or original -backoff = ["5s", "30s", "2m"] # optional JetStream redelivery delays -start_sequence = 0 -start_time = "" +# max_ack_pending: Maximum number of unacknowledged messages before pausing delivery +max_ack_pending = 1000 [subscription] # Optional. Defaults to "telegram.>" when omitted. @@ -176,7 +174,7 @@ The application supports two NATS consumption modes, controlled by `nats.mode`: **Features:** - ✅ **Message Persistence**: Messages are stored in a Stream, allowing replay and recovery - ✅ **ACK/NAK Mechanism**: Explicit message acknowledgment ensures guaranteed delivery -- ✅ **Automatic Retry**: Failed messages are automatically redelivered with configurable backoff +- ✅ **Error Handling**: Failed messages are handled with simple backoff - ✅ **Dead-Letter Queue**: Poison messages can be routed to a DLQ for inspection - ✅ **Batch Processing**: Efficient batch fetching and processing - ✅ **Consumer Monitoring**: Real-time metrics for consumer lag and pending messages @@ -219,10 +217,8 @@ The application supports two NATS consumption modes, controlled by `nats.mode`: - `replicas`: Number of stream replicas for HA (default: 1, use 3+ for production) **Consumer Configuration** (`[nats.consumer_rules]`): - - `max_deliver`: Maximum redelivery attempts (default: 5) + - `max_deliver`: Maximum delivery attempts before giving up (default: 3) - `ack_wait`: Time to wait for ACK before redelivery (default: 30s) - - `deliver_policy`: When to start delivering messages ("all", "new", "last", etc.) - - `backoff`: Array of delays between retries (e.g., `["5s", "30s", "2m"]`) 4. **Start the Application:** ```bash @@ -271,7 +267,7 @@ The application supports two NATS consumption modes, controlled by `nats.mode`: - Stream stores messages according to retention policy - Consumer pulls messages in batches (configurable via `app.batch_size`) - Each message is processed and ACKed on success - - Failed messages are NAKed and redelivered according to `backoff` strategy + - Failed messages are NAKed and redelivered with simple backoff - After `max_deliver` attempts, permanent failures are routed to DLQ (if enabled) 8. **Replay Messages:** @@ -480,7 +476,7 @@ Flags: --telemetry-insecure Send OTLP traffic without TLS ``` -Critical overrides stay available through CLI flags; advanced tuning such as stream retention, consumer backoff, and copy counts are configured via the TOML file or `CAATSM_` environment variables. +Critical overrides stay available through CLI flags; configuration is managed via the TOML file or `CAATSM_` environment variables. | CLI flag | Config key | Purpose | |---------------------|------------------------|----------------------------------------| @@ -496,8 +492,7 @@ Critical overrides stay available through CLI flags; advanced tuning such as str #### Replay & Backoff - `--replay-from seq:12345` replays from a specific JetStream sequence, while `--replay-from time:2024-11-15T08:00:00Z` starts at a timestamp. -- Configure server-side retry delays with `[nats.consumer].backoff = ["5s", "30s", "2m"]`; each duration becomes the delay before the next delivery attempt. -- Combine `backoff` with `--ack-wait` to increase acknowledgement windows (e.g., `--ack-wait 2m`). +- Configure retry behavior with `[nats.consumer_rules]` settings. ### Observability @@ -673,7 +668,7 @@ The project keeps tests close to the code that they exercise: ### Error Handling & Retries - **Parser failures** (invalid headers/body) are treated as permanent: the raw payload is stored in `aviation.telegrams_raw`, the message is ACKed, and no JetStream retries are attempted. -- **Repository failures** are transient: the consumer returns an error, the message is `NAK`ed, and JetStream redelivers it using `[nats.consumer_rules.backoff]` and `ack_wait` to space retries. +- **Repository failures** are transient: the consumer returns an error, the message is `NAK`ed, and JetStream redelivers it with simple backoff. - **Publisher failures** are logged and persisted as raw records, but they are marked permanent to avoid hammering downstream topics; the deduplicated output can be replayed from the raw table later. - Tune JetStream retry behavior via `[nats.consumer_rules.max_deliver]`, `[nats.consumer_rules.backoff]`, and CLI overrides like `--ack-wait`. The monitoring server plus Prometheus counters provide visibility into each failure bucket. diff --git a/configs/config.dev.toml b/configs/config.dev.toml index a102b96..64922d3 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -37,51 +37,24 @@ storage = "file" replicas = 1 [nats.consumer_rules] -# Consumer delivery and retry rules (only applies when mode = "jetstream") +# Consumer delivery rules (only applies when mode = "jetstream") # max_deliver: Maximum number of delivery attempts before giving up (0 = unlimited) -max_deliver = 5 +max_deliver = 3 # ack_wait: Time to wait for ACK before redelivering message (e.g., "30s", "2m") ack_wait = "30s" # max_ack_pending: Maximum number of unacknowledged messages before pausing delivery -max_ack_pending = 1024 -# deliver_policy: When to start delivering messages: -# - "all": Deliver all messages from the stream -# - "new": Only deliver new messages after consumer creation -# - "last": Deliver only the last message -# - "last_per_subject": Deliver last message per subject -# - "sequence": Start from a specific sequence (requires start_sequence) -# - "time": Start from a specific time (requires start_time in RFC3339 format) -deliver_policy = "all" -# replay_policy: How to replay messages: "instant" (as fast as possible) or "original" (preserve timing) -replay_policy = "instant" -# backoff: Array of delays between retry attempts (e.g., ["5s", "30s", "2m"]) -# First retry waits 5s, second waits 30s, third and beyond wait 2m -backoff = ["5s", "30s", "2m"] -# start_sequence: Starting sequence number (only used when deliver_policy = "sequence") -start_sequence = 0 -# start_time: Starting time in RFC3339 format (only used when deliver_policy = "time") -# Example: "2024-11-15T08:00:00Z" -start_time = "" +max_ack_pending = 1000 [nats.auth] # NATS authentication configuration (optional for development) -# Only one authentication method can be used at a time: -# - token: Simple token authentication -# - credentials_file: Path to NATS credentials file (e.g., /path/to/user.creds) -# - user/password: Username and password authentication -# -# For development, authentication is typically not required. -# Uncomment and configure as needed: -# token = "" -# credentials_file = "" -# user = "" -# password = "" -# -# TLS configuration (optional) -# tls_enabled = false -# tls_cert_file = "" # Client certificate file path -# tls_key_file = "" # Client private key file path -# tls_ca_file = "" # CA certificate file for server verification +# token: Simple token authentication (uncomment if needed) +# token = "your-dev-token" + +# TLS configuration (optional - uncomment for secure connections) +# tls_enabled = true +# tls_cert_file = "/path/to/client.crt" +# tls_key_file = "/path/to/client.key" +# tls_ca_file = "/path/to/ca.crt" [subscription] topic = "telegram.serial" diff --git a/configs/config.prod.toml b/configs/config.prod.toml index 8554f02..395ba9f 100644 --- a/configs/config.prod.toml +++ b/configs/config.prod.toml @@ -37,26 +37,13 @@ storage = "file" replicas = 3 [nats.consumer_rules] -# Consumer delivery and retry rules for production +# Consumer delivery rules (only applies when mode = "jetstream") # max_deliver: Maximum number of delivery attempts before giving up -max_deliver = 5 +max_deliver = 3 # ack_wait: Time to wait for ACK before redelivering message ack_wait = "30s" # max_ack_pending: Maximum number of unacknowledged messages before pausing delivery -max_ack_pending = 1024 -# deliver_policy: "new" - Start from new messages after consumer creation (recommended for production) -# Other options: "all", "last", "last_per_subject", "sequence", "time" -deliver_policy = "new" -# replay_policy: How to replay messages: "instant" (as fast as possible) or "original" (preserve timing) -replay_policy = "instant" -# backoff: Array of delays between retry attempts -# First retry waits 5s, second waits 30s, third waits 2m, fourth+ wait 5m -backoff = ["5s", "30s", "2m", "5m"] -# start_sequence: Starting sequence number (only used when deliver_policy = "sequence") -start_sequence = 0 -# start_time: Starting time in RFC3339 format (only used when deliver_policy = "time") -# Example: "2024-11-15T08:00:00Z" -start_time = "" +max_ack_pending = 1000 [nats.auth] # NATS authentication configuration (REQUIRED for production) diff --git a/configs/grafana-dashboards.dev/caatsm-overview.json b/configs/grafana-dashboards.dev/caatsm-overview.json index 91a2855..1ee6773 100644 --- a/configs/grafana-dashboards.dev/caatsm-overview.json +++ b/configs/grafana-dashboards.dev/caatsm-overview.json @@ -287,93 +287,6 @@ "legendFormat": "{{stream}} / {{consumer}}" } ] - }, - { - "id": 9, - "type": "timeseries", - "title": "NATS consumer ACK pending", - "datasource": { - "type": "prometheus", - "uid": "prometheus-dev" - }, - "gridPos": { "h": 7, "w": 12, "x": 12, "y": 26 }, - "fieldConfig": { - "defaults": { - "color": { "mode": "palette-classic" }, - "mappings": [], - "thresholds": { "mode": "absolute", "steps": [] } - }, - "overrides": [] - }, - "options": { - "legend": { "displayMode": "table", "placement": "bottom" }, - "tooltip": { "mode": "single" } - }, - "targets": [ - { - "refId": "A", - "expr": "last_over_time(caatsm_nats_consumer_ack_pending_sum[1m])", - "legendFormat": "ACK pending" - } - ] - }, - { - "id": 10, - "type": "timeseries", - "title": "NATS consumer redelivered messages", - "datasource": { - "type": "prometheus", - "uid": "prometheus-dev" - }, - "gridPos": { "h": 7, "w": 12, "x": 0, "y": 33 }, - "fieldConfig": { - "defaults": { - "color": { "mode": "palette-classic" }, - "mappings": [], - "thresholds": { "mode": "absolute", "steps": [] } - }, - "overrides": [] - }, - "options": { - "legend": { "displayMode": "table", "placement": "bottom" }, - "tooltip": { "mode": "single" } - }, - "targets": [ - { - "refId": "A", - "expr": "last_over_time(caatsm_nats_consumer_redelivered_sum[1m])", - "legendFormat": "Redelivered" - } - ] - }, - { - "id": 11, - "type": "timeseries", - "title": "NATS consumer delivered messages", - "datasource": { - "type": "prometheus", - "uid": "prometheus-dev" - }, - "gridPos": { "h": 7, "w": 12, "x": 12, "y": 33 }, - "fieldConfig": { - "defaults": { - "color": { "mode": "palette-classic" }, - "mappings": [], - "thresholds": { "mode": "absolute", "steps": [] } - }, - "overrides": [] - }, - "options": { - "legend": { "displayMode": "table", "placement": "bottom" }, - "tooltip": { "mode": "single" } - }, - "targets": [ - { - "refId": "A", - "expr": "last_over_time(caatsm_nats_consumer_delivered_sum[1m])", - "legendFormat": "Delivered" - } - ] } ] } diff --git a/docs/dev-guide.md b/docs/dev-guide.md index ff424d6..df11c95 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -47,12 +47,8 @@ docker compose -f docker-compose.dev.yml up -d postgres nats nats-box - The NATS client keeps retrying the connection and automatically reconnects when NATS is back. -- The JetStream consumer detects missing streams/consumers and, in dev/test - environments, uses shared `EnsureStream`/`ensureConsumer` logic to - auto-recreate them. -- In production environments, missing streams/consumers are treated as - configuration/operational errors and are not auto-recreated; operators - should investigate and fix the underlying issue. +- The JetStream consumer expects streams and consumers to exist. +- In development, you may need to create them manually or ensure they exist before starting the application. ### Using Taskfile shortcuts diff --git a/docs/nats.md b/docs/nats.md index b1b1bd7..dd3e2b1 100644 --- a/docs/nats.md +++ b/docs/nats.md @@ -1,19 +1,18 @@ -# NATS Integration Architecture +# NATS Integration ## Overview -The NATS integration provides a robust, production-ready message processing system built on Clean Architecture principles. It supports both JetStream (persistent) and Core NATS (fire-and-forget) modes with comprehensive error handling, observability, and resilience features. +The NATS integration provides a streamlined, production-ready message processing system focused on essential functionality. It supports JetStream persistent messaging with basic error handling, TLS security, and observability. ### Key Concepts -1. **Consumer**: Pulls messages from NATS JetStream in batches, processes them, and handles ACKs/NAKs -2. **Publisher**: Publishes messages to NATS with automatic deduplication via UUID headers -3. **Batch Processing**: Fetches multiple messages at once (configurable size) for efficiency -4. **Error Classification**: Distinguishes between transient (retry) and permanent (DLQ) errors -5. **Dead Letter Queue (DLQ)**: Routes failed messages to a separate queue for analysis -6. **Backpressure**: Automatically slows down processing when errors accumulate -7. **Self-Healing**: Automatically recreates missing streams/consumers in development -8. **Observability**: Built-in metrics, tracing, and structured logging +1. **Consumer**: Pulls messages from NATS JetStream in batches and processes them +2. **Publisher**: Publishes messages to NATS with deduplication +3. **Batch Processing**: Fetches multiple messages for efficiency +4. **Error Classification**: Distinguishes transient vs permanent errors +5. **Dead Letter Queue (DLQ)**: Routes permanent errors to DLQ +6. **TLS Support**: Secure connections with client certificates +7. **Basic Monitoring**: Essential metrics and logging ### Quick Start Flow @@ -28,54 +27,41 @@ The NATS integration provides a robust, production-ready message processing syst ## Architecture -### Clean Architecture Layers +### Architecture -The NATS integration follows Clean Architecture principles, separating concerns into distinct layers: +The NATS integration follows simplified Clean Architecture with focused components: ``` ┌─────────────────────────────────────┐ -│ Port Interfaces │ -│ (Publisher, Consumer contracts) │ -│ - Define contracts, not impl │ -│ - Enable dependency inversion │ -├─────────────────────────────────────┤ │ Application Layer │ -│ (Message processing logic) │ -│ - Business logic │ -│ - Use case orchestration │ +│ (Business logic & processing) │ ├─────────────────────────────────────┤ │ Infrastructure Layer │ -│ (NATS implementation details) │ +│ (NATS implementation) │ │ │ │ ┌─────────────────────────────┐ │ │ │ Consumer │ │ │ │ ┌─────────────────────┐ │ │ │ │ │ MessageFetcher │ │ │ │ │ │ MessageProcessor │ │ │ -│ │ │ ErrorHandler │ │ │ │ │ │ DLQHandler │ │ │ │ │ └─────────────────────┘ │ │ │ └─────────────────────────────┘ │ │ │ │ ┌─────────────────────────────┐ │ │ │ Publisher │ │ -│ │ ┌─────────────────────┐ │ │ -│ │ │ MessageSerializer │ │ │ -│ │ │ HeaderEnricher │ │ │ -│ │ └─────────────────────┘ │ │ │ └─────────────────────────────┘ │ └─────────────────────────────────────┘ ``` -### Component Interaction Diagram +### Component Interaction ``` ┌──────────────┐ │ Publisher │ │ │ │ 1. Serialize │ -│ 2. Add UUID │ -│ 3. Publish │ +│ 2. Publish │ └──────┬───────┘ │ │ Publish to Subject @@ -108,19 +94,12 @@ The NATS integration follows Clean Architecture principles, separating concerns │ ┌──────────▼───────────────────┐ │ │ │ MessageProcessor │ │ │ │ - ProcessBatch() │ │ -│ │ - ProcessSingleMessage() │ │ -│ └──────────┬───────────────────┘ │ -│ │ │ -│ ┌──────────▼───────────────────┐ │ -│ │ ErrorHandler │ │ -│ │ - Classify errors │ │ -│ │ - Apply backpressure │ │ +│ │ - ProcessMessage() │ │ │ └──────────┬───────────────────┘ │ │ │ │ │ ┌──────────▼───────────────────┐ │ │ │ DLQHandler │ │ │ │ - RouteToDLQ() │ │ -│ │ - AdvisoryDLQHandler │ │ │ └──────────────────────────────┘ │ └─────────────────────────────────────┘ │ @@ -136,10 +115,9 @@ The NATS integration follows Clean Architecture principles, separating concerns 1. **Dependency Inversion**: High-level modules (Consumer, Publisher) depend on abstractions (interfaces), not concrete implementations 2. **Separation of Concerns**: Each component has a single responsibility: - - `MessageFetcher`: Handles message retrieval - - `MessageProcessor`: Handles message processing logic - - `ErrorHandler`: Handles error classification and recovery - - `DLQHandler`: Handles dead letter queue routing + - `MessageFetcher`: Handles message retrieval + - `MessageProcessor`: Handles message processing logic + - `DLQHandler`: Handles dead letter queue routing 3. **Testability**: All components can be mocked and tested independently 4. **Extensibility**: New implementations can be added without modifying existing code @@ -147,8 +125,71 @@ The NATS integration follows Clean Architecture principles, separating concerns ### Consumer Processing Flow -The consumer follows a well-defined processing loop with error handling at each stage: +The consumer follows a simplified processing loop: +``` +┌─────────────────────────────────────────────────────────────┐ +│ Consumer Start │ +│ 1. Initialize components (Fetcher, Processor, DLQ) │ +│ 2. Start main processing loop │ +└──────────────────────┬──────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ Main Loop │ +│ │ +│ ┌──────────────────────────────────────────────────────┐ │ +│ │ Step 1: Check Context │ │ +│ │ - If cancelled, exit gracefully │ │ +│ └──────────────────┬───────────────────────────────────┘ │ +│ │ │ +│ ┌──────────────────▼───────────────────────────────────┐ │ +│ │ Step 2: Fetch Batch │ │ +│ │ - Fetch up to BatchSize messages │ │ +│ │ - Wait up to BatchTimeout │ │ +│ │ - Handle fetch errors with backoff │ │ +│ └──────────────────┬───────────────────────────────────┘ │ +│ │ │ +│ ┌───────────┴───────────┐ │ +│ │ │ │ +│ Success Error │ +│ │ │ │ +│ │ ┌────────▼────────┐ │ +│ │ │ Apply Backoff │ │ +│ │ │ Continue Loop │ │ +│ │ └─────────────────┘ │ +│ │ │ +│ └──────────────────┬──────────────────────────────┘ +│ │ +│ ┌─────────────────────────▼─────────────────────────────┐ │ +│ │ Step 3: Process Batch │ │ +│ │ - For each message in batch: │ │ +│ │ * Extract message ID │ │ +│ │ * Call processor.Handle() │ │ +│ │ * Handle result (ACK/NAK/DLQ) │ │ +│ └──────────────────┬───────────────────────────────────┘ │ +│ │ │ +│ ┌───────────┴───────────┐ │ +│ │ │ │ +│ Success Error │ +│ │ │ │ +│ │ ┌────────▼────────┐ │ +│ │ │ Classify Error │ │ +│ │ └────────┬────────┘ │ +│ │ │ │ +│ │ ┌─────────────┴─────────────┐ │ +│ │ │ │ │ +│ │ Permanent Transient │ +│ │ │ │ │ +│ │ ┌──────▼──────┐ ┌────────▼──────┐ │ +│ │ │ Route to DLQ│ │ NAK with delay│ │ +│ │ │ ACK message │ └────────────────┘ │ +│ │ └──────┬──────┘ │ +│ │ │ │ +│ └─────────┴───────────────────────────────────────┘ +│ │ +│ └─────────── Loop ─────────────────────┘ +└─────────────────────────────────────────────────────────────┘ ``` ┌─────────────────────────────────────────────────────────────┐ │ Consumer Start │ @@ -409,59 +450,48 @@ The consumer handles message consumption with the following features: - **Core Mode**: Fire-and-forget message processing for simple use cases #### Key Features -- **Batch Processing**: Configurable batch sizes and timeouts for efficient processing -- **Backpressure**: Automatic backpressure when processing errors accumulate -- **Dead Letter Queue (DLQ)**: Automatic routing of failed messages to DLQ -- **Advisory DLQ**: Handles messages that exceed MaxDeliver limits -- **Self-Healing**: Automatic recreation of missing streams/consumers in dev environments -- **Graceful Shutdown**: Proper cleanup and draining of connections +- **Batch Processing**: Configurable batch sizes for efficient processing +- **Error Handling**: Distinguishes transient vs permanent errors +- **Dead Letter Queue (DLQ)**: Routes permanent errors to DLQ +- **TLS Support**: Secure connections with client certificates +- **Basic Monitoring**: Essential metrics collection #### Component Logic **MessageFetcher (`defaultMessageFetcher`)** - Fetches batches of messages using `sub.Fetch(batchSize, MaxWait(timeout))` -- Handles fetch errors with exponential backoff -- Recovers subscriptions when connection issues occur +- Handles fetch errors with simple exponential backoff - Context-aware: respects cancellation signals **MessageProcessor (`defaultBatchProcessor`)** - Processes messages sequentially within a batch - Extracts message IDs (header → metadata → generated) -- Creates OpenTelemetry spans for tracing - Calls application processor for business logic - Handles ACK/NAK based on processing results -**ErrorHandler** -- Classifies errors as transient or permanent using `app.IsPermanent()` -- Tracks consecutive error streaks -- Applies backpressure when streak exceeds threshold (default: 10) -- Calculates backoff delays for retries - **DLQHandler (`defaultDLQHandler`)** -- Routes permanent errors to DLQ with enriched metadata +- Routes permanent errors to DLQ with basic metadata - Validates DLQ stream exists at startup - Publishes DLQ messages with error context -**AdvisoryDLQHandler** -- Subscribes to JetStream advisory events: `$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.*` -- Handles messages that exhaust MaxDeliver attempts -- Retrieves original message from stream using `GetMsg()` -- Routes to DLQ with advisory metadata - #### Configuration ```toml [NATS] -Mode = "jetstream" # or "core" +URL = "nats://localhost:4222" Stream = "TELEGRAM" Consumer = "telegram-consumer" +[NATS.Auth] +Token = "your-token" # Optional token authentication +TLSEnabled = true # Enable TLS +TLSCertFile = "/path/to/client.crt" # Client certificate +TLSKeyFile = "/path/to/client.key" # Client private key +TLSCAFile = "/path/to/ca.crt" # CA certificate + [NATS.ConsumerRules] AckWait = "30s" MaxDeliver = 3 MaxAckPending = 1000 -DeliverPolicy = "all" -ReplayPolicy = "instant" -Backoff = ["1s", "2s", "5s", "10s"] [DLQ] Enabled = true @@ -470,7 +500,6 @@ Subject = "caatsm.dlq" [App] BatchSize = 50 BatchTimeout = "2s" -MonitorInterval = "30s" ``` ### Publisher @@ -508,12 +537,9 @@ The publisher handles message publishing with deduplication and observability. #### Error Types - **Transient Errors**: Network issues, temporary unavailability (retried with backoff) - **Permanent Errors**: Message format issues, business logic failures (routed to DLQ) -- **Resource Errors**: Missing streams/consumers (auto-recovered in dev, fail in prod) #### Recovery Strategies -- **Exponential Backoff**: Configurable backoff for transient failures -- **Circuit Breaker Pattern**: Prevents cascade failures -- **Resource Recreation**: Automatic recreation of missing JetStream resources +- **Simple Backoff**: Exponential backoff for transient failures - **Graceful Degradation**: Continues processing other messages when one fails ### Dead Letter Queue (DLQ) @@ -569,25 +595,32 @@ The publisher handles message publishing with deduplication and observability. - **Graceful Shutdown**: Proper draining with timeouts - **Resource Cleanup**: Ensures subscriptions and connections are closed -### Self-Healing -- **Development Mode**: Auto-creates missing streams/consumers -- **Production Mode**: Fails fast on configuration issues -- **Recovery Logic**: Attempts to recreate resources on errors + ## Configuration ### Environment Variables ```bash CAATSM_NATS_URL=nats://localhost:4222 -CAATSM_NATS_MODE=jetstream +CAATSM_NATS_TOKEN=your-token # Optional CAATSM_DLQ_ENABLED=true CAATSM_DLQ_SUBJECT=caatsm.dlq ``` +### TLS Configuration +For production deployments with TLS: + +```toml +[NATS.Auth] +TLSEnabled = true +TLSCertFile = "/etc/ssl/certs/client.crt" +TLSKeyFile = "/etc/ssl/private/client.key" +TLSCAFile = "/etc/ssl/certs/ca.crt" +``` + ### Runtime Configuration -- **Hot Reload**: Configuration changes applied without restart -- **Validation**: Comprehensive validation at startup -- **Defaults**: Sensible defaults for all configuration options +- **Validation**: Basic validation at startup +- **Defaults**: Sensible defaults for essential options ## Testing Strategy @@ -603,9 +636,8 @@ CAATSM_DLQ_SUBJECT=caatsm.dlq ### Test Categories - **Happy Path**: Normal operation scenarios -- **Error Recovery**: Various failure and recovery scenarios -- **Performance**: Load testing and resource usage -- **Configuration**: Different configuration combinations +- **Error Handling**: Basic error scenarios +- **Configuration**: Configuration validation ## Simple Examples @@ -619,7 +651,7 @@ package main import ( "context" "time" - + "caatsm/internal/infra/config" "caatsm/internal/infra/nats" "caatsm/internal/app" @@ -630,14 +662,16 @@ func main() { // 1. Load configuration cfg := &config.Config{ NATS: config.NATSConfig{ - URL: "nats://localhost:4222", - Mode: "jetstream", + URL: "nats://localhost:4222", Stream: "TELEGRAM", Consumer: "telegram-consumer", + Auth: config.NATSAuthConfig{ + Token: "your-token", // Optional + }, ConsumerRules: config.ConsumerRules{ - AckWait: 30 * time.Second, - MaxDeliver: 3, - Backoff: []time.Duration{1*time.Second, 2*time.Second, 5*time.Second}, + AckWait: 30 * time.Second, + MaxDeliver: 3, + MaxAckPending: 1000, }, }, App: config.AppConfig{ @@ -649,25 +683,31 @@ func main() { Subject: "caatsm.dlq", }, } - + // 2. Create NATS connection - nc, _ := nats.Connect(cfg.NATS.URL) + nc, err := nats.ProvideNATSConn(cfg, zap.NewNop()) + if err != nil { + panic(err) + } defer nc.Close() - + // 3. Get JetStream context - js, _ := nc.JetStream() - + js, err := nats.ProvideJetStream(nc, zap.NewNop()) + if err != nil { + panic(err) + } + // 4. Create message processor (your business logic) processor := app.NewMessageProcessor(/* dependencies */) - + // 5. Create logger - logger, _ := zap.NewProduction() - + logger := zap.NewNop() + // 6. Create telemetry recorder telemetry := /* your telemetry implementation */ - + // 7. Create consumer - consumer, err := natsinfra.ProvideConsumer( + consumer, err := nats.ProvideConsumer( nc, js, processor, @@ -676,21 +716,13 @@ func main() { logger, ) if err != nil { - logger.Fatal("Failed to create consumer", zap.Error(err)) + panic(err) } - + // 8. Start consumer with context ctx, cancel := context.WithCancel(context.Background()) defer cancel() - - // Handle graceful shutdown - go func() { - // Wait for interrupt signal - <-ctx.Done() - shutdownCtx, _ := context.WithTimeout(context.Background(), 5*time.Second) - consumer.Shutdown(shutdownCtx) - }() - + // 9. Start consuming (blocks until context cancelled) if err := consumer.Start(ctx); err != nil { logger.Error("Consumer stopped", zap.Error(err)) @@ -810,10 +842,8 @@ func processMessage(msg *nats.Msg) error { // Scenario 4: MaxDeliver Exhausted // When message fails MaxDeliver times (default: 3): -// - JetStream publishes advisory event -// - AdvisoryDLQHandler catches event -// - Retrieves original message -// - Routes to DLQ with metadata +// - Message is not automatically handled +// - Consider monitoring JetStream consumer info for failed deliveries ``` ### Example 5: DLQ Message Structure @@ -837,39 +867,45 @@ What a DLQ message looks like: ### Example 6: Configuration Examples -Different configuration scenarios: +Basic configuration with TLS: ```toml -# Example 1: High Throughput Configuration [NATS] -Mode = "jetstream" +URL = "nats://secure.nats.server:4222" Stream = "TELEGRAM" Consumer = "telegram-consumer" +[NATS.Auth] +TLSEnabled = true +TLSCertFile = "/etc/ssl/certs/client.crt" +TLSKeyFile = "/etc/ssl/private/client.key" +TLSCAFile = "/etc/ssl/certs/ca.crt" + [NATS.ConsumerRules] -AckWait = "60s" -MaxDeliver = 5 -MaxAckPending = 5000 -Backoff = ["1s", "2s", "5s", "10s", "30s"] +AckWait = "30s" +MaxDeliver = 3 +MaxAckPending = 1000 + +[DLQ] +Enabled = true +Subject = "caatsm.dlq" [App] -BatchSize = 100 # Larger batches -BatchTimeout = "5s" # Longer timeout +BatchSize = 50 +BatchTimeout = "2s" +``` -# Example 2: Low Latency Configuration -[App] -BatchSize = 10 # Smaller batches -BatchTimeout = "500ms" # Shorter timeout +Development configuration: -# Example 3: Development Mode (Self-Healing) +```toml [NATS] -Mode = "jetstream" -# Missing streams/consumers auto-created +URL = "nats://localhost:4222" +Stream = "TELEGRAM" +Consumer = "telegram-consumer" -# Example 4: Production Mode (Fail Fast) -[NATS] -Mode = "jetstream" -# Missing streams/consumers cause startup failure +[DLQ] +Enabled = true +Subject = "caatsm.dlq" ``` ### Example 7: Observability Integration @@ -990,33 +1026,25 @@ func (p *CustomProcessor) ProcessMessage(ctx context.Context, msg *nats.Msg) err ## Security Considerations ### Authentication -- **NATS Auth**: Use NATS built-in authentication mechanisms -- **TLS**: Enable TLS for encrypted communication -- **Token Auth**: Use NATS tokens for service authentication +- **Token Auth**: Use NATS tokens for simple authentication +- **TLS**: Enable TLS with client certificates for secure communication -### Authorization -- **Subject Permissions**: Restrict publish/subscribe permissions -- **Stream Access**: Control access to specific streams -- **DLQ Security**: Secure DLQ access to prevent data leakage +### TLS Configuration +```toml +[NATS.Auth] +TLSEnabled = true +TLSCertFile = "/path/to/client.crt" +TLSKeyFile = "/path/to/client.key" +TLSCAFile = "/path/to/ca.crt" +``` ### Data Protection -- **Message Encryption**: Encrypt sensitive message data -- **Audit Logging**: Log all message operations for compliance -- **PII Handling**: Avoid logging sensitive information +- **TLS Encryption**: All communication is encrypted +- **Basic Logging**: Avoid logging sensitive message content ## Future Enhancements -### Planned Features -- **Consumer Groups**: Horizontal scaling with multiple consumers -- **Message Filtering**: Subject-based and header-based filtering -- **Priority Queues**: High-priority message processing -- **Rate Limiting**: Per-consumer and per-subject rate limits -- **Message Transformation**: In-flight message modification -- **Multi-Region**: Cross-region message replication - -### Extensibility Points -- **Custom Serializers**: Pluggable message serialization -- **Middleware**: Request/response middleware support -- **Hooks**: Pre/post processing hooks -- **Metrics Backends**: Support for additional metrics systems -- **Storage Backends**: Alternative storage for DLQ messages \ No newline at end of file +### Future Enhancements +- **Additional Auth Methods**: Support for more authentication mechanisms if needed +- **Advanced Monitoring**: Enhanced metrics and tracing if required +- **Performance Tuning**: Batch size and timeout optimizations \ No newline at end of file diff --git a/docs/observability.md b/docs/observability.md index df84481..b748206 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -33,13 +33,10 @@ The service exposes Prometheus metrics via the monitoring HTTP server (default ` Additional OTEL metrics are emitted via the configured OTEL endpoint, including: -- `caatsm_messages_processed_total` -- `caatsm_parse_duration_seconds` -- `caatsm_publish_failures_total` -- `caatsm_nats_consumer_ack_pending` -- `caatsm_nats_consumer_redelivered` -- `caatsm_nats_consumer_pending` -- `caatsm_nats_consumer_delivered` +- `caatsm_messages_processed_total` +- `caatsm_parse_duration_seconds` +- `caatsm_publish_failures_total` +- `caatsm_nats_consumer_pending_messages` These metrics are intended to be scraped by Prometheus (either directly or via the OTEL collector) and visualised in Grafana dashboards. Recommended dashboard panels include: diff --git a/docs/prod-guide.md b/docs/prod-guide.md index 3b2e2f3..efc1cc1 100644 --- a/docs/prod-guide.md +++ b/docs/prod-guide.md @@ -55,12 +55,9 @@ storage = "file" replicas = 3 # Use 3+ for HA in production [nats.consumer_rules] -max_deliver = 5 +max_deliver = 3 ack_wait = "30s" -max_ack_pending = 1024 -deliver_policy = "new" # Start from new messages in production -replay_policy = "instant" -backoff = ["5s", "30s", "2m", "5m"] +max_ack_pending = 1000 [subscription] topic = "telegram.serial" diff --git a/internal/adapter/parser/aviation.go b/internal/adapter/parser/aviation.go index 70eb58f..dd95827 100644 --- a/internal/adapter/parser/aviation.go +++ b/internal/adapter/parser/aviation.go @@ -1,8 +1,8 @@ package parser import ( - "caatsm/internal/domain" "caatsm/internal/adapter/dto" + "caatsm/internal/domain" "errors" "fmt" "regexp" @@ -24,7 +24,6 @@ const ( OtherInfo = "other" ReferenceData = "reference_data" - Aircraft = "aircraft" CategorySurveillance = "surve" Indicator = "indicator" Other = "other" @@ -75,7 +74,11 @@ func NewBodyParser(body string) *BodyParser { func (parser *BodyParser) GetBodyPatterns() map[string]BodyConfig { parser.mu.Lock() defer parser.mu.Unlock() - return parser.bodyPatterns + copied := make(map[string]BodyConfig, len(parser.bodyPatterns)) + for k, v := range parser.bodyPatterns { + copied[k] = v + } + return copied } func (parser *BodyParser) SetBodyPatterns(patterns map[string]BodyConfig) { @@ -200,6 +203,23 @@ func (parser *BodyParser) createBodyData(data map[string]string) (string, interf } } +func headerToParsedTelegram(header Header) dto.ParsedTelegram { + return dto.ParsedTelegram{ + MessageID: header.MessageID, + DateTime: header.DateTime, + PriorityIndicator: header.PriorityIndicator, + PrimaryAddress: header.PrimaryAddress, + SecondaryAddresses: header.SecondaryAddresses, + Originator: header.Originator, + OriginatorDateTime: header.OriginatorDateTime, + Category: header.Category, + Body: header.Body, + Content: header.Content, + ReceivedAt: header.ReceivedAt, + ParsedAt: header.ParsedAt, + } +} + func Parse(rawText string) (*dto.ParsedTelegram, error) { header, err := ParseHeader(rawText) if err != nil { @@ -217,48 +237,20 @@ func Parse(rawText string) (*dto.ParsedTelegram, error) { header.ParsedAt = time.Now() if bodyErr != nil { - return &dto.ParsedTelegram{ - MessageID: header.MessageID, - DateTime: header.DateTime, - PriorityIndicator: header.PriorityIndicator, - PrimaryAddress: header.PrimaryAddress, - SecondaryAddresses: header.SecondaryAddresses, - Originator: header.Originator, - OriginatorDateTime: header.OriginatorDateTime, - Category: header.Category, - Body: header.Body, - Content: header.Content, - ReceivedAt: header.ReceivedAt, - ParsedAt: header.ParsedAt, - Parsed: false, - Comments: bodyErr.Error(), - Status: dto.MessageStatusBodyError, - ErrorReason: bodyErr.Error(), - }, fmt.Errorf("%w: %w", ErrBodyParse, bodyErr) - } - - parsed := &dto.ParsedTelegram{ - MessageID: header.MessageID, - DateTime: header.DateTime, - PriorityIndicator: header.PriorityIndicator, - PrimaryAddress: header.PrimaryAddress, - SecondaryAddresses: header.SecondaryAddresses, - Originator: header.Originator, - OriginatorDateTime: header.OriginatorDateTime, - Category: header.Category, - Body: header.Body, - Content: header.Content, - BodyData: bodyData, - ReceivedAt: header.ReceivedAt, - ParsedAt: header.ParsedAt, - Parsed: true, - Status: dto.MessageStatusParsed, - ErrorReason: "", + parsed := headerToParsedTelegram(header) + parsed.Parsed = false + parsed.Comments = bodyErr.Error() + parsed.Status = dto.MessageStatusBodyError + parsed.ErrorReason = bodyErr.Error() + return &parsed, fmt.Errorf("%w: %w", ErrBodyParse, bodyErr) } + parsed := headerToParsedTelegram(header) + parsed.BodyData = bodyData + parsed.Parsed = true + parsed.Status = dto.MessageStatusParsed parsed.Uuid = uuid.New().String() - - return parsed, nil + return &parsed, nil } func cleanMessage(text string) string { diff --git a/internal/adapter/parser/schedule.go b/internal/adapter/parser/schedule.go index e4a2984..3cb496a 100644 --- a/internal/adapter/parser/schedule.go +++ b/internal/adapter/parser/schedule.go @@ -30,8 +30,9 @@ func ExtractWaypoint(message string) *domain.WayPoint { } func FindDef(code string) *LineParser { - // fmt.Printf("Finding definition for %s\n", code) - // fmt.Println("ParserDef: ", parserDef) + if parserDef == nil { + return nil + } for _, def := range *parserDef { for _, airline := range def.Airlines { if airline == code { @@ -70,6 +71,9 @@ func ParseWithDef(line string, parserDef *LineParser) *domain.ScheduleLine { } for i, field := range parserDef.Fields { + if i >= len(words) { + break + } // log.Debugf("Parsing field %v -> %s", i, field) data := extract(words[i], parserMap[field]) if data != nil { @@ -225,6 +229,10 @@ func getFlightNumbers(data string) []string { flightNumbers := append([]string{}, baseNumber) for _, number := range data[1:] { length := len(number) + if length >= baseLength { + zap.S().Warnf("Flight number suffix '%s' is not shorter than base '%s'; skipping", number, baseNumber) + continue + } flightNumber := baseNumber[:baseLength-length] + number flightNumbers = append(flightNumbers, flightNumber) } diff --git a/internal/adapter/parser/schedule_parser_test.go b/internal/adapter/parser/schedule_parser_test.go index 582b295..8c64b68 100644 --- a/internal/adapter/parser/schedule_parser_test.go +++ b/internal/adapter/parser/schedule_parser_test.go @@ -527,25 +527,6 @@ var _ = Describe("Parse Line with PreDef", func() { Expect(schedule.Waypoints[1].DepartureTime).To(Equal("0535")) Expect(schedule.Waypoints[2].Airport).To(Equal("CAN")) }) - - It("83. CZ3301/2 B2823 B752 CAN0135 TSN0535 CAN", func() { - lineText := "83. CZ3301/2 B2823 B752 CAN0135 TSN0535 CAN" - def := FindDef("CZ") - Expect(def).NotTo(BeNil()) - schedule := ParseWithDef(lineText, def) - Expect(schedule).NotTo(BeNil()) - Expect(schedule.Index).To(Equal("83.")) - Expect(len(schedule.FlightNumber)).To(Equal(2)) - Expect(schedule.FlightNumber).To(ContainElement("CZ3301")) - Expect(schedule.FlightNumber).To(ContainElement("CZ3302")) - Expect(schedule.AircraftReg).To(Equal("B2823")) - Expect(len(schedule.Waypoints)).To(Equal(3)) - Expect(schedule.Waypoints[0].Airport).To(Equal("CAN")) - Expect(schedule.Waypoints[0].DepartureTime).To(Equal("0135")) - Expect(schedule.Waypoints[1].Airport).To(Equal("TSN")) - Expect(schedule.Waypoints[1].DepartureTime).To(Equal("0535")) - Expect(schedule.Waypoints[2].Airport).To(Equal("CAN")) - }) }) Context("HO", func() { diff --git a/internal/infra/nats/advisory_dlq.go b/internal/infra/nats/advisory_dlq.go deleted file mode 100644 index b826c8e..0000000 --- a/internal/infra/nats/advisory_dlq.go +++ /dev/null @@ -1,169 +0,0 @@ -package nats - -import ( - "context" - "encoding/json" - "fmt" - "time" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// MaxDeliveriesAdvisoryEvent represents the advisory message published when -// a message reaches MaxDeliver attempts. -type MaxDeliveriesAdvisoryEvent struct { - Type string `json:"type"` - Stream string `json:"stream"` - Consumer string `json:"consumer"` - StreamSeq uint64 `json:"stream_seq"` - Deliveries uint64 `json:"deliveries"` - Time string `json:"time"` -} - -// AdvisoryDLQHandler handles messages that exhaust MaxDeliver attempts -// by subscribing to JetStream advisory events. -type AdvisoryDLQHandler struct { - js nats.JetStreamContext - nc *nats.Conn - streamName string - consumerName string - dlqSubject string - logger *zap.Logger - telemetry TelemetryRecorder -} - -// TelemetryRecorder is an interface for recording telemetry events. -// This matches the telemetry.Recorder interface used by Consumer. -type TelemetryRecorder interface { - RecordDLQMessage(ctx context.Context, stream, consumer string) - RecordDLQPublishFailure(ctx context.Context, stream, consumer string) -} - -// NewAdvisoryDLQHandler creates a new advisory-based DLQ handler. -func NewAdvisoryDLQHandler( - js nats.JetStreamContext, - nc *nats.Conn, - streamName string, - consumerName string, - dlqSubject string, - logger *zap.Logger, - telemetry TelemetryRecorder, -) (*AdvisoryDLQHandler, error) { - return &AdvisoryDLQHandler{ - js: js, - nc: nc, - streamName: streamName, - consumerName: consumerName, - dlqSubject: dlqSubject, - logger: logger, - telemetry: telemetry, - }, nil -} - -// Start begins listening for advisory messages and routing failed messages to DLQ. -func (h *AdvisoryDLQHandler) Start(ctx context.Context) error { - // Subscribe to advisory subject pattern - // Format: $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.. - advisorySubject := fmt.Sprintf("$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.%s.%s", - h.streamName, h.consumerName) - - h.logger.Info("Starting advisory DLQ handler", - zap.String("advisory_subject", advisorySubject), - zap.String("stream", h.streamName), - zap.String("consumer", h.consumerName), - zap.String("dlq_subject", h.dlqSubject), - ) - - sub, err := h.nc.Subscribe(advisorySubject, func(msg *nats.Msg) { - h.handleAdvisory(ctx, msg) - }) - if err != nil { - return fmt.Errorf("failed to subscribe to advisory subject: %w", err) - } - - // Wait for context cancellation - go func() { - <-ctx.Done() - if err := sub.Unsubscribe(); err != nil { - h.logger.Error("Failed to unsubscribe advisory subscription", zap.Error(err)) - } - h.logger.Info("Stopped advisory DLQ handler") - }() - - return nil -} - -// handleAdvisory processes an advisory message about max deliveries. -func (h *AdvisoryDLQHandler) handleAdvisory(ctx context.Context, advisoryMsg *nats.Msg) { - var event MaxDeliveriesAdvisoryEvent - if err := json.Unmarshal(advisoryMsg.Data, &event); err != nil { - h.logger.Error("Failed to unmarshal advisory event", - zap.Error(err), - zap.String("data", string(advisoryMsg.Data)), - ) - return - } - - h.logger.Warn("Message reached MaxDeliver attempts", - zap.String("stream", event.Stream), - zap.String("consumer", event.Consumer), - zap.Uint64("stream_seq", event.StreamSeq), - zap.Uint64("deliveries", event.Deliveries), - ) - - // Retrieve the original message from the stream using GetMsg API - originalMsg, err := h.js.GetMsg(h.streamName, event.StreamSeq) - if err != nil { - h.logger.Error("Failed to retrieve original message from stream", - zap.Uint64("stream_seq", event.StreamSeq), - zap.Error(err), - ) - h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) - return - } - - // Extract message metadata - msgID := "" - subject := originalMsg.Subject - if originalMsg.Header != nil { - msgID = originalMsg.Header.Get("Nats-Msg-Id") - } - - // Create enriched DLQ payload (similar to existing routeToDLQ) - payload := map[string]interface{}{ - "transport_msg_id": msgID, - "subject": subject, - "stream": h.streamName, - "consumer": h.consumerName, - "nats_sequence": event.StreamSeq, - "deliveries": event.Deliveries, - "error": fmt.Sprintf("message exhausted max_deliver (%d) attempts", event.Deliveries), - "received_at": time.Now().UTC(), - "body": string(originalMsg.Data), - "advisory_source": true, // Flag to distinguish from immediate DLQ - } - - data, err := json.Marshal(payload) - if err != nil { - h.logger.Error("Failed to marshal advisory DLQ payload", zap.Error(err)) - h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) - return - } - - // Publish to DLQ - if _, err := h.js.Publish(h.dlqSubject, data); err != nil { - h.logger.Error("Failed to publish advisory message to DLQ", - zap.Uint64("stream_seq", event.StreamSeq), - zap.Error(err), - ) - h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) - return - } - - h.logger.Info("Routed max-deliveries message to DLQ", - zap.Uint64("stream_seq", event.StreamSeq), - zap.Uint64("deliveries", event.Deliveries), - ) - h.telemetry.RecordDLQMessage(ctx, h.streamName, h.consumerName) -} diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 55caefb..11ee7e6 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -3,42 +3,18 @@ package nats import ( "caatsm/internal/app" "caatsm/internal/infra/config" - "caatsm/internal/infra/log" obsmetrics "caatsm/internal/infra/metrics" "caatsm/internal/infra/telemetry" "context" - "encoding/json" "errors" "fmt" "strings" "time" - "github.com/google/uuid" "github.com/nats-io/nats.go" - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/codes" - "go.opentelemetry.io/otel/metric" "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) -} - -// MessageProcessor defines the interface for processing message batches -type MessageProcessor interface { - ProcessBatch(ctx context.Context, msgs []*nats.Msg) -} - -// DLQHandler defines the interface for dead letter queue operations -type DLQHandler interface { - RouteToDLQ(ctx context.Context, msg *nats.Msg, cause error) error - ValidateDLQ() error -} - // Consumer handles NATS JetStream message consumption with clean separation of concerns type Consumer struct { // Core dependencies @@ -56,22 +32,11 @@ type Consumer struct { fetcher MessageFetcher batchProcessor MessageProcessor dlqHandler DLQHandler - errorHandler *ErrorHandler // Resource managers consumerManager *ConsumerManager streamManager *StreamManager - // Advisory DLQ handler for messages exhausting MaxDeliver - advisoryDLQHandler *AdvisoryDLQHandler - - // Metrics - meter metric.Meter - ackPending metric.Int64Histogram - redelivered metric.Int64Histogram - pending metric.Int64Histogram - delivered metric.Int64Histogram - // State consecutiveProcessErrors int } @@ -89,597 +54,64 @@ type consumerConfig struct { monitorInterval time.Duration } -// 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 -} +// ProvideConsumer creates a NATS consumer with clean architecture. +func ProvideConsumer( + conn *nats.Conn, + js nats.JetStreamContext, + processor *app.MessageProcessor, + cfg *config.Config, + rec telemetry.Recorder, + logger *zap.Logger, +) (*Consumer, error) { + normCfg := normalizeConsumerConfig(cfg) -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: + consumer := &Consumer{ + conn: conn, + js: js, + processor: processor, + cfg: cfg, + logger: logger, + telemetry: rec, + config: *normCfg, // dereference the pointer } + consumer.initCollaborators() - // 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) { - // Check context cancellation first - 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 errors are expected when no messages are available - not an error condition - if errors.Is(err, nats.ErrTimeout) { - return true, nil - } - - // Check connection health before proceeding - if f.conn != nil { - status := f.conn.Status() - if status != nats.CONNECTED { - f.logger.Warn("NATS connection not in CONNECTED state", - zap.String("status", status.String()), - zap.Error(err), - ) - // Connection is down - this is a transient error, apply backoff - *fetchErrorStreak++ - backoff := f.calculateExponentialBackoff(*fetchErrorStreak) - f.logger.Warn("Connection unhealthy, applying backoff before retry", - zap.String("status", status.String()), - zap.Int("error_streak", *fetchErrorStreak), - zap.Duration("backoff", backoff), - ) - if !sleepWithContext(ctx, backoff) { - return false, ctx.Err() - } - // Check if connection recovered after backoff - if f.conn.Status() == nats.CONNECTED { - *fetchErrorStreak = 0 - return true, nil - } - // Still not connected - continue with error handling + // 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) } - } + streamSubjects = dedupeSubjects(streamSubjects) + consumer.streamManager = NewStreamManager(js, normCfg.streamName, streamSubjects, logger) - // Check for connection closed errors - if errors.Is(err, nats.ErrConnectionClosed) { - f.logger.Error("NATS connection closed", - zap.Error(err), - zap.String("stream", f.config.streamName), - zap.String("consumer", f.config.consumerName), - ) - // Connection closed is fatal - cannot recover subscription - if *sub != nil { - if err := (*sub).Unsubscribe(); err != nil { - f.logger.Error("Failed to unsubscribe after connection closed", zap.Error(err)) - } - *sub = nil - } - return false, fmt.Errorf("connection closed: %w", err) - } - - // JetStream API unavailable (e.g., NATS restarted or JetStream not ready) - if errors.Is(err, nats.ErrNoResponders) { - *fetchErrorStreak++ - backoff := f.calculateExponentialBackoff(*fetchErrorStreak) - backoff = min(backoff, 30*time.Second) - f.logger.Warn("JetStream not available, will retry with backoff", - zap.Error(err), - zap.String("stream", f.config.streamName), - zap.String("consumer", f.config.consumerName), - zap.Int("error_streak", *fetchErrorStreak), - zap.Duration("backoff", backoff), - ) - if !sleepWithContext(ctx, backoff) { - return false, ctx.Err() - } - return true, nil - } - - // Check for JetStream resource not found errors - if isJetStreamResourceNotFound(err) { - if isDevLikeEnv() && shouldBootstrapStream() { - f.logger.Warn("JetStream consumer or stream missing; attempting to recreate", - zap.Error(err), - zap.String("stream", f.config.streamName), - zap.String("consumer", f.config.consumerName), - ) - // Attempt to recover resources and recreate subscription - if f.consumerManager == nil || f.streamManager == nil { - return false, fmt.Errorf("cannot recover: consumer/stream manager not available: %w", err) - } - consumerConfig := f.buildConsumerConfig() - if recErr := f.consumerManager.RecoverResources(f.streamManager, consumerConfig); recErr != nil { - return false, fmt.Errorf("failed to recover JetStream resources: %w", recErr) - } - // Unsubscribe old subscription before creating new one - if *sub != nil { - if err := (*sub).Unsubscribe(); err != nil { - f.logger.Error("Failed to unsubscribe during recovery", zap.Error(err)) - } - } - // Create new subscription - newSub, subErr := f.consumerManager.CreatePullSubscription() - if subErr != nil { - return false, fmt.Errorf("failed to create pull subscription after recovery: %w", subErr) - } - *sub = newSub - *fetchErrorStreak = 0 - f.logger.Info("Successfully recovered subscription after resource recreation") - return true, nil + // Update fetcher with managers now that they're initialized + if fetcher, ok := consumer.fetcher.(*defaultMessageFetcher); ok { + fetcher.consumerManager = consumer.consumerManager + fetcher.streamManager = consumer.streamManager } - // Production: treat as configuration/operational error - fatal - f.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment", - zap.Error(err), - zap.String("stream", f.config.streamName), - zap.String("consumer", f.config.consumerName), - ) - if *sub != nil { - if err := (*sub).Unsubscribe(); err != nil { - f.logger.Error("Failed to unsubscribe after resource not found", zap.Error(err)) - } - *sub = nil + // 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) } - return false, fmt.Errorf("JetStream resource not found: %w", err) - } - - // Check for network/temporary errors - if f.isTemporaryError(err) { - *fetchErrorStreak++ - backoff := f.calculateExponentialBackoff(*fetchErrorStreak) - f.logger.Warn("Temporary network error, applying backoff", - zap.Error(err), - zap.Int("error_streak", *fetchErrorStreak), - zap.Duration("backoff", backoff), - ) - if !sleepWithContext(ctx, backoff) { - return false, ctx.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) } - // Verify subscription is still valid before returning success - if *sub != nil && f.conn != nil && f.conn.Status() == nats.CONNECTED { - return true, nil - } - // Subscription or connection invalid - attempt recovery - return f.attemptSubscriptionRecovery(ctx, sub, fetchErrorStreak) - } - // Generic error path with exponential backoff - *fetchErrorStreak++ - backoff := f.calculateExponentialBackoff(*fetchErrorStreak) - f.logger.Error("Failed to fetch messages; backing off", - zap.Error(err), - zap.Int("error_streak", *fetchErrorStreak), - zap.Duration("backoff", backoff), - ) - if !sleepWithContext(ctx, backoff) { - return false, ctx.Err() - } - - // Verify subscription and connection health before returning success - if *sub == nil || (f.conn != nil && f.conn.Status() != nats.CONNECTED) { - return f.attemptSubscriptionRecovery(ctx, sub, fetchErrorStreak) - } - - return true, nil -} - -// calculateExponentialBackoff calculates exponential backoff duration with a cap -func (f *defaultMessageFetcher) calculateExponentialBackoff(streak int) time.Duration { - if streak <= 0 { - return 0 - } - // Exponential backoff: 2^(streak-1) seconds, capped at 30 seconds - backoff := time.Duration(1< 0 { - *p.consecutiveProcessErrors = 0 - } - - // ACK the message - if ackErr := msg.Ack(); ackErr != nil { - p.logger.Error("Failed to ACK message", zap.Error(ackErr)) } else { - elapsed := time.Since(start) - 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), + logger.Info("Running consumer in core NATS mode", + zap.String("subject", normCfg.subject), + zap.String("queue_group", cfg.Subscription.QueueGroup), ) } - // 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 - } - - if p.mode == "core" { - return uuid.NewString(), "generated", 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) { - p.logger.Error("Failed to process message", - zap.String("subject", msg.Subject), - zap.Error(err), - zap.Bool("permanent", app.IsPermanent(err)), - ) - - result := obsmetrics.ResultFail - if app.IsPermanent(err) { - result = obsmetrics.ResultPermanentFail - } - p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, result, elapsed) - - consecutiveErrors := 0 - if p.consecutiveProcessErrors != nil { - consecutiveErrors = *p.consecutiveProcessErrors - } - - processingResult := p.errorHandler.HandleProcessingError(consecutiveErrors, err, p.logger, msg.Subject) - - 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 - } - // Poison/permanent message: route to DLQ if configured, then ACK - 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.Error(dlqErr)) - } - } - if ackErr := msg.Ack(); ackErr != nil { - p.logger.Error("Failed to ACK permanent-error message", 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) -} - -// defaultDLQHandler implements DLQHandler interface -type defaultDLQHandler struct { - js nats.JetStreamContext - dlqSubject string - streamName string - consumerName string - logger *zap.Logger - telemetry telemetry.Recorder -} - -func (h *defaultDLQHandler) RouteToDLQ(ctx context.Context, msg *nats.Msg, cause error) error { - return h.routeToDLQInternal(ctx, msg, cause) -} - -func (h *defaultDLQHandler) ValidateDLQ() error { - return h.validateDLQInternal() -} - -func (h *defaultDLQHandler) routeToDLQInternal(ctx context.Context, msg *nats.Msg, cause error) error { - // Basic DLQ routing implementation - payload := map[string]any{ - "subject": msg.Subject, - "stream": h.streamName, - "consumer": h.consumerName, - "error": cause.Error(), - "received_at": time.Now().UTC(), - "body": string(msg.Data), - } - - data, err := json.Marshal(payload) - if err != nil { - h.logger.Error("failed to marshal DLQ payload", zap.Error(err)) - return err - } - - _, err = h.js.Publish(h.dlqSubject, data) - if err != nil { - h.logger.Error("failed to publish to DLQ", - zap.String("dlq_subject", h.dlqSubject), - zap.Error(err), - ) - h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) - return err - } - - h.telemetry.RecordDLQMessage(ctx, h.streamName, h.consumerName) - return nil -} - -func (h *defaultDLQHandler) validateDLQInternal() error { - if h.js == nil { - return fmt.Errorf("JetStream context is nil") - } - - _, err := h.js.StreamNameBySubject(h.dlqSubject) - if err != nil { - return fmt.Errorf("DLQ subject %s not bound to any JetStream stream: %w", h.dlqSubject, err) - } - - h.logger.Info("DLQ configuration validated", - zap.String("dlq_subject", h.dlqSubject), - ) - - return nil + return consumer, nil } // initCollaborators initializes the collaborator components @@ -711,7 +143,6 @@ func (c *Consumer) initCollaborators() { c.batchProcessor = &defaultBatchProcessor{ processor: c.processor, dlqHandler: c.dlqHandler, - errorHandler: c.errorHandler, logger: c.logger, telemetry: c.telemetry, streamName: c.config.streamName, @@ -785,84 +216,6 @@ func normalizeConsumerConfig(cfg *config.Config) *consumerConfig { } } -// ProvideConsumer creates a NATS consumer with clean architecture. -func ProvideConsumer( - conn *nats.Conn, - js nats.JetStreamContext, - processor *app.MessageProcessor, - cfg *config.Config, - 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 - errorHandler: NewErrorHandler(logger), - } - consumer.initMetrics() - consumer.initCollaborators() - - // 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) - } - streamSubjects = dedupeSubjects(streamSubjects) - consumer.streamManager = NewStreamManagerWithConfig(js, normCfg.streamName, streamSubjects, &cfg.NATS.StreamLimits, 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 - } - - // 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) - } - - // Initialize advisory DLQ handler if DLQ is enabled - if normCfg.dlqSubject != "" && cfg.DLQ.Enabled { - advisoryHandler, err := NewAdvisoryDLQHandler( - js, - conn, - normCfg.streamName, - normCfg.consumerName, - normCfg.dlqSubject, - logger, - rec, - ) - if err != nil { - return nil, fmt.Errorf("failed to create advisory DLQ handler: %w", err) - } - consumer.advisoryDLQHandler = advisoryHandler - } - } else { - logger.Info("Running consumer in core NATS mode", - zap.String("subject", normCfg.subject), - zap.String("queue_group", cfg.Subscription.QueueGroup), - ) - } - - return consumer, nil -} - // buildConsumerConfig builds the NATS consumer configuration func (c *Consumer) buildConsumerConfig() *nats.ConsumerConfig { return &nats.ConsumerConfig{ @@ -903,6 +256,144 @@ func (c *Consumer) ValidateDLQ() error { return nil } +// validateDLQ is a helper for internal use (lowercase) +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) { + if err := c.batchProcessor.ProcessMessage(ctx, msg); err != nil { + isPermanent := app.IsPermanent(err) + c.logger.Error("Failed to process message (core mode)", + zap.String("subject", msg.Subject), + zap.Error(err), + zap.Bool("permanent", isPermanent), + ) + } + } + + 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() +} + +// 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), + ) + + statsCtx, statsCancel := context.WithCancel(ctx) + defer statsCancel() + go c.emitConsumerStats(statsCtx) + + var fetchErrorStreak int + + for { + select { + case <-ctx.Done(): + c.logger.Info("Stopping consumer", zap.Error(ctx.Err())) + return ctx.Err() + default: + } + + // Fetch messages in batch + msgs, err := c.fetcher.FetchBatch(ctx, currentSub) + if err != nil { + // If context was cancelled, return immediately + 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 + } + continue + } + + // Successful fetch -> reset error streak. + if fetchErrorStreak > 0 { + fetchErrorStreak = 0 + } + + // Process batch + c.batchProcessor.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() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName) + if err != nil { + continue + } + // Record pending messages for monitoring + obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) + } + } +} + // Shutdown drains the underlying NATS connection gracefully. func (c *Consumer) Shutdown(ctx context.Context) error { if c.conn == nil { diff --git a/internal/infra/nats/consumer_core.go b/internal/infra/nats/consumer_core.go deleted file mode 100644 index 4dce9bf..0000000 --- a/internal/infra/nats/consumer_core.go +++ /dev/null @@ -1,52 +0,0 @@ -package nats - -import ( - "caatsm/internal/app" - "context" - "errors" - "fmt" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// 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) { - if err := c.processMessage(ctx, msg); err != nil { - isPermanent := app.IsPermanent(err) - c.logger.Error("Failed to process message (core mode)", - zap.String("subject", msg.Subject), - zap.Error(err), - zap.Bool("permanent", isPermanent), - ) - } - } - - 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() -} diff --git a/internal/infra/nats/consumer_error.go b/internal/infra/nats/consumer_error.go deleted file mode 100644 index 3caf7f9..0000000 --- a/internal/infra/nats/consumer_error.go +++ /dev/null @@ -1,80 +0,0 @@ -package nats - -import ( - "caatsm/internal/app" - obsmetrics "caatsm/internal/infra/metrics" - "context" - "time" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// handleMessageError handles errors that occur during message processing. -// -//nolint:unused // Reserved for potential future use or alternative implementation -func (c *Consumer) handleMessageError(ctx context.Context, msg *nats.Msg, err error, elapsed time.Duration) { - c.logger.Error("Failed to process message", - zap.String("subject", msg.Subject), - zap.Error(err), - zap.Bool("permanent", app.IsPermanent(err)), - ) - - result := obsmetrics.ResultFail - if app.IsPermanent(err) { - result = obsmetrics.ResultPermanentFail - } - c.telemetry.RecordMessageHandled(ctx, c.config.streamName, c.config.consumerName, result, elapsed) - - processingResult := c.errorHandler.HandleProcessingError(c.consecutiveProcessErrors, err, c.logger, msg.Subject) - - if processingResult.IsPermanent { - c.handlePermanentError(ctx, msg, err) - return - } - - c.handleTransientError(ctx, msg, processingResult) -} - -// handlePermanentError handles permanent/poison messages. -// -//nolint:unused // Reserved for potential future use or alternative implementation -func (c *Consumer) handlePermanentError(ctx context.Context, msg *nats.Msg, err error) { - c.consecutiveProcessErrors = 0 - // Poison/permanent message: route to DLQ if configured, then ACK - if dlqErr := c.routeToDLQ(ctx, msg, err); dlqErr != nil { - c.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(dlqErr)) - } - if ackErr := msg.Ack(); ackErr != nil { - c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) - } -} - -// handleTransientError handles transient errors with backpressure and redelivery. -// -//nolint:unused // Reserved for potential future use or alternative implementation -func (c *Consumer) handleTransientError(ctx context.Context, msg *nats.Msg, processingResult ProcessingErrorResult) { - // Increment error streak - if c.consecutiveProcessErrors < 0 { - c.consecutiveProcessErrors = 0 - } - c.consecutiveProcessErrors++ - - if processingResult.ShouldApplyBackpressure { - c.logger.Warn("Applying backpressure due to consecutive processing errors", - zap.Int("consecutive_errors", c.consecutiveProcessErrors), - 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 - c.telemetry.RecordRetry(ctx, c.config.streamName, c.config.consumerName, obsmetrics.RetryReasonProcessorError) - if nakErr := c.nakWithStrategy(msg); nakErr != nil { - c.logger.Error("Failed to NAK message", zap.Error(nakErr)) - } -} diff --git a/internal/infra/nats/consumer_jetstream.go b/internal/infra/nats/consumer_jetstream.go deleted file mode 100644 index 64824f9..0000000 --- a/internal/infra/nats/consumer_jetstream.go +++ /dev/null @@ -1,207 +0,0 @@ -package nats - -import ( - "context" - "errors" - "fmt" - "time" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// recoverJetStreamResources attempts to recreate the stream and consumer in -// dev/test environments if they are missing. It is safe to call multiple times. -func (c *Consumer) recoverJetStreamResources() error { - if c.js == nil { - return fmt.Errorf("jetstream context is nil") - } - if c.cfg == nil { - return fmt.Errorf("config is nil") - } - - consumerConfig := c.buildConsumerConfig() - return c.consumerManager.RecoverResources(c.streamManager, consumerConfig) -} - -// createPullSubscriptionWithRecovery creates a pull subscription and, in -// dev/test environments, attempts to self-heal missing stream/consumer -// by recreating them once. -func (c *Consumer) createPullSubscriptionWithRecovery() (*nats.Subscription, error) { - consumerConfig := c.buildConsumerConfig() - return c.consumerManager.CreatePullSubscriptionWithRecovery(c.streamManager, consumerConfig) -} - -//nolint:unused // Reserved for potential future use or alternative implementation -// nakWithStrategy sends a NAK with appropriate delay based on retry attempt. -func (c *Consumer) nakWithStrategy(msg *nats.Msg) error { - backoff := c.cfg.NATS.ConsumerRules.Backoff - if len(backoff) == 0 { - return msg.Nak() - } - - meta, err := msg.Metadata() - if err != nil { - c.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(backoff) { - index = len(backoff) - 1 - } - delay := backoff[index] - if delay <= 0 { - return msg.Nak() - } - - return msg.NakWithDelay(delay) -} - -// sleepWithContext sleeps for the specified duration, but returns early if the context is canceled. -// Returns true if the full duration was slept, false if the context was canceled. -func sleepWithContext(ctx context.Context, duration time.Duration) bool { - timer := time.NewTimer(duration) - defer timer.Stop() - select { - case <-timer.C: - return true - case <-ctx.Done(): - return false - } -} - -//nolint:unused // Reserved for potential future use or alternative implementation -// fetchBatch fetches a batch of messages from the subscription. -// It respects context cancellation for faster shutdown. -func (c *Consumer) 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 - // The batchTimeout is still used, but we'll check context more frequently - timeout := c.config.batchTimeout - if timeout > 500*time.Millisecond { - // Cap at 500ms to improve responsiveness while still allowing batching - timeout = 500 * time.Millisecond - } - - return sub.Fetch(c.config.batchSize, nats.MaxWait(timeout)) -} - -// handleFetchError handles errors during message fetching, including recovery logic. -// Returns true if the error was handled and consumption should continue, false otherwise. -func (c *Consumer) handleFetchError(ctx context.Context, err error, sub **nats.Subscription, fetchErrorStreak *int) (bool, error) { - result := c.errorHandler.HandleFetchError(ctx, err, sub, fetchErrorStreak, c.config.streamName, c.config.consumerName, func() (*nats.Subscription, error) { - if recErr := c.recoverJetStreamResources(); recErr != nil { - return nil, recErr - } - if *sub != nil { - if unsubErr := (*sub).Unsubscribe(); unsubErr != nil { - c.logger.Error("Failed to unsubscribe during recovery", zap.Error(unsubErr)) - } - } - return c.createPullSubscriptionWithRecovery() - }) - - if result.RecoveredSub != nil { - *sub = result.RecoveredSub - *fetchErrorStreak = 0 - } - - return result.ShouldContinue, result.Error -} - -// startJetStream starts the JetStream consumer loop. -func (c *Consumer) startJetStream(ctx context.Context) error { - // Create pull subscription (with simple self-healing in dev/test). - sub, err := c.createPullSubscriptionWithRecovery() - 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("Consumer pull configuration", - zap.Int("batch_size", c.config.batchSize), - zap.Duration("batch_timeout", c.config.batchTimeout), - zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver), - zap.Duration("ack_wait", c.config.ackWait), - zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), - zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), - zap.Int("backoff_steps", len(c.cfg.NATS.ConsumerRules.Backoff)), - ) - - statsCtx, statsCancel := context.WithCancel(ctx) - defer statsCancel() - go c.emitConsumerStats(statsCtx) - - // Start advisory DLQ handler in background if configured - if c.advisoryDLQHandler != nil { - go func() { - if err := c.advisoryDLQHandler.Start(ctx); err != nil { - c.logger.Error("Advisory DLQ handler failed", zap.Error(err)) - } - }() - } - - var fetchErrorStreak int - - for { - select { - case <-ctx.Done(): - c.logger.Info("Stopping consumer", zap.Error(ctx.Err())) - return ctx.Err() - default: - } - - // Fetch messages in batch - msgs, err := c.fetcher.FetchBatch(ctx, currentSub) - if err != nil { - // If context was cancelled, return immediately - 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 - } - continue - } - - // Successful fetch -> reset error streak. - if fetchErrorStreak > 0 { - fetchErrorStreak = 0 - } - - // Process batch - c.batchProcessor.ProcessBatch(ctx, msgs) - } -} diff --git a/internal/infra/nats/consumer_jetstream_test.go b/internal/infra/nats/consumer_jetstream_test.go deleted file mode 100644 index 0060fcf..0000000 --- a/internal/infra/nats/consumer_jetstream_test.go +++ /dev/null @@ -1,119 +0,0 @@ -package nats - -import ( - "context" - "errors" - "time" - - configpkg "caatsm/internal/infra/config" - - "github.com/nats-io/nats.go" - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" - "go.uber.org/zap/zaptest" -) - -var _ = Describe("Consumer JetStream", func() { - var ( - c *Consumer - ) - - BeforeEach(func() { - logger := zaptest.NewLogger(GinkgoT()) - c = &Consumer{ - config: consumerConfig{ - mode: "jetstream", - streamName: "TEST_STREAM", - consumerName: "test-consumer", - subject: "test.subject", - batchSize: 10, - batchTimeout: 2 * time.Second, - }, - logger: logger, - errorHandler: NewErrorHandler(logger), - cfg: &configpkg.Config{ - NATS: configpkg.NATSConfig{ - ConsumerRules: configpkg.ConsumerRulesConfig{ - Backoff: []time.Duration{5 * time.Second, 30 * time.Second}, - }, - }, - }, - } - }) - - Describe("nakWithStrategy", func() { - PIt("sends NAK without delay when backoff is empty", func() { - c.cfg.NATS.ConsumerRules.Backoff = []time.Duration{} - // Note: This test would require a real NATS message to fully test - // For now, we verify the logic path - }) - - PIt("sends NAK with delay based on delivery attempt", func() { - // Note: This test would require a real NATS message with metadata - // For now, we verify the function exists and can be called - }) - }) - - Describe("handleFetchError", func() { - var ctx context.Context - - BeforeEach(func() { - ctx = context.Background() - }) - - It("returns true for timeout errors", func() { - var sub *nats.Subscription - fetchErrorStreak := 0 - shouldContinue, err := c.handleFetchError(ctx, nats.ErrTimeout, &sub, &fetchErrorStreak) - Expect(shouldContinue).To(BeTrue()) - Expect(err).NotTo(HaveOccurred()) - }) - - It("handles ErrNoResponders with backoff", func() { - var sub *nats.Subscription - fetchErrorStreak := 0 - shouldContinue, err := c.handleFetchError(ctx, nats.ErrNoResponders, &sub, &fetchErrorStreak) - Expect(shouldContinue).To(BeTrue()) - Expect(err).NotTo(HaveOccurred()) - Expect(fetchErrorStreak).To(Equal(1)) - }) - - It("handles resource not found errors", func() { - var sub *nats.Subscription - fetchErrorStreak := 0 - resourceErr := errors.New("stream not found") - shouldContinue, err := c.handleFetchError(ctx, resourceErr, &sub, &fetchErrorStreak) - // Behavior depends on environment; in test this should attempt recovery - Expect(err).To(HaveOccurred()) - Expect(shouldContinue).To(BeFalse()) - }) - - It("handles generic errors with backoff", func() { - var sub *nats.Subscription - fetchErrorStreak := 0 - genericErr := errors.New("generic error") - shouldContinue, err := c.handleFetchError(ctx, genericErr, &sub, &fetchErrorStreak) - Expect(shouldContinue).To(BeTrue()) - Expect(err).NotTo(HaveOccurred()) - Expect(fetchErrorStreak).To(Equal(1)) - }) - }) - - Describe("buildConsumerConfig", func() { - It("builds consumer config with correct defaults", func() { - config := c.buildConsumerConfig() - Expect(config.Durable).To(Equal("test-consumer")) - Expect(config.AckPolicy).To(Equal(nats.AckExplicitPolicy)) - Expect(config.FilterSubject).To(Equal("test.subject")) - }) - }) - - Describe("processSingleMessage", func() { - It("handles successful message processing", func() { - // This would require mocking the processor, but we can test the structure - // For now, we verify the method exists and can be called - // Note: This test would need a mock processor to fully work - Skip("Requires mock message processor") - }) - }) -}) diff --git a/internal/infra/nats/consumer_manager.go b/internal/infra/nats/consumer_manager.go index 707772d..29f06e2 100644 --- a/internal/infra/nats/consumer_manager.go +++ b/internal/infra/nats/consumer_manager.go @@ -59,53 +59,12 @@ func (cm *ConsumerManager) EnsureConsumer(config *nats.ConsumerConfig) error { return nil } -// RecoverResources attempts to recreate the stream and consumer in dev/test environments -func (cm *ConsumerManager) RecoverResources(streamManager *StreamManager, consumerConfig *nats.ConsumerConfig) error { - if cm.js == nil { - return fmt.Errorf("jetstream context is nil") - } - - // Ensure stream exists (dev/test may auto-create, prod will error). - if err := streamManager.EnsureStream(); err != nil { - return fmt.Errorf("ensure stream %s: %w", cm.streamName, err) - } - - // Ensure durable consumer exists and is properly bound. - if err := cm.EnsureConsumer(consumerConfig); err != nil { - return fmt.Errorf("ensure consumer %s: %w", cm.consumerName, err) - } - - 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 and attempts recovery if needed +// 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 - } - - if isJetStreamResourceNotFound(err) && isDevLikeEnv() && shouldBootstrapStream() { - cm.logger.Warn("PullSubscribe failed due to missing JetStream resources; attempting to recreate", - zap.Error(err), - zap.String("stream", cm.streamName), - zap.String("consumer", cm.consumerName), - ) - if recErr := cm.RecoverResources(streamManager, consumerConfig); recErr != nil { - return nil, fmt.Errorf("failed to recover JetStream resources: %w", recErr) - } - // Retry subscription after successful recovery. - sub, err = cm.CreatePullSubscription() - if err != nil { - return nil, fmt.Errorf("failed to create pull subscription after recovery: %w", err) - } - return sub, nil - } - - return nil, fmt.Errorf("failed to create pull subscription: %w", err) + return cm.CreatePullSubscription() } diff --git a/internal/infra/nats/consumer_message.go b/internal/infra/nats/consumer_message.go deleted file mode 100644 index ed55a3f..0000000 --- a/internal/infra/nats/consumer_message.go +++ /dev/null @@ -1,138 +0,0 @@ -package nats - -import ( - "caatsm/internal/infra/log" - "context" - "fmt" - "time" - - "github.com/google/uuid" - "github.com/nats-io/nats.go" - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/codes" - "go.uber.org/zap" -) - -//nolint:unused // Reserved for potential future use or alternative implementation -// processBatch processes a batch of messages, handling errors and applying backpressure. -// It checks context cancellation between messages for faster shutdown. -func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) { - for _, msg := range msgs { - // Check context before processing each message - select { - case <-ctx.Done(): - c.logger.Info("Stopping batch processing due to cancellation", - zap.Int("remaining_messages", len(msgs)), - ) - return - default: - } - c.processSingleMessage(ctx, msg) - } -} - -//nolint:unused // Reserved for potential future use or alternative implementation -// processSingleMessage processes a single message with error handling and backpressure. -func (c *Consumer) processSingleMessage(ctx context.Context, msg *nats.Msg) { - start := time.Now() - - if err := c.processMessage(ctx, msg); err != nil { - c.handleMessageError(ctx, msg, err, time.Since(start)) - return - } - - // Successful processing resets the error streak. - if c.consecutiveProcessErrors > 0 { - c.consecutiveProcessErrors = 0 - } - - // ACK the message - if ackErr := msg.Ack(); ackErr != nil { - c.logger.Error("Failed to ACK message", zap.Error(ackErr)) - } else { - elapsed := time.Since(start) - c.telemetry.RecordMessageHandled(ctx, c.config.streamName, c.config.consumerName, "ok", elapsed) - } -} - -// processMessage processes a single message. -func (c *Consumer) 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", c.config.consumerName), - attribute.String("caatsm.stream", c.config.streamName), - ) - - msgID, source, err := c.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" { - 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), - ) - } - - // 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(c.logger, log.MessageFields{ - Service: "caatsm-consumer", - TransportMsgID: msgID, - Stream: c.config.streamName, - Consumer: c.config.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 := c.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 (c *Consumer) resolveMsgID(msg *nats.Msg) (string, string, error) { - if id := msg.Header.Get("Nats-Msg-Id"); id != "" { - return id, "header", nil - } - - if c.config.mode == "core" { - return uuid.NewString(), "generated", 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/consumer_metrics.go b/internal/infra/nats/consumer_metrics.go deleted file mode 100644 index c889846..0000000 --- a/internal/infra/nats/consumer_metrics.go +++ /dev/null @@ -1,83 +0,0 @@ -package nats - -import ( - obsmetrics "caatsm/internal/infra/metrics" - "context" - "time" - - "github.com/nats-io/nats.go" - "go.opentelemetry.io/otel" - "go.uber.org/zap" -) - -// initMetrics initializes OpenTelemetry metrics. -func (c *Consumer) initMetrics() { - meter := otel.Meter("caatsm/nats") - c.meter = meter - - if hist, err := meter.Int64Histogram("caatsm_nats_consumer_ack_pending"); err == nil { - c.ackPending = hist - } - if hist, err := meter.Int64Histogram("caatsm_nats_consumer_redelivered"); err == nil { - c.redelivered = hist - } - if hist, err := meter.Int64Histogram("caatsm_nats_consumer_pending"); err == nil { - c.pending = hist - } - if hist, err := meter.Int64Histogram("caatsm_nats_consumer_delivered"); err == nil { - c.delivered = hist - } -} - -// recordConsumerMetrics records consumer metrics from ConsumerInfo. -func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.ConsumerInfo) { - if info == nil { - return - } - if c.ackPending != nil { - c.ackPending.Record(ctx, int64(info.NumAckPending)) - } - if c.redelivered != nil { - c.redelivered.Record(ctx, int64(info.NumRedelivered)) - } - if c.pending != nil { - c.pending.Record(ctx, int64(info.NumPending)) - } - if c.delivered != nil { - c.delivered.Record(ctx, int64(info.Delivered.Stream)) - } - - // Export an explicit pending messages gauge for Prometheus-based lag / - // backlog alerts. - obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) -} - -// emitConsumerStats periodically emits consumer statistics. -func (c *Consumer) emitConsumerStats(ctx context.Context) { - ticker := time.NewTicker(c.config.monitorInterval) - defer ticker.Stop() - - 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", zap.Error(err)) - continue - } - - c.logger.Debug("JetStream consumer metrics", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.Uint64("num_ack_pending", uint64(info.NumAckPending)), - zap.Uint64("num_redelivered", uint64(info.NumRedelivered)), - zap.Uint64("num_pending", uint64(info.NumPending)), - zap.Uint64("delivered_consumer_seq", uint64(info.Delivered.Consumer)), - zap.Uint64("delivered_stream_seq", uint64(info.Delivered.Stream)), - ) - c.recordConsumerMetrics(ctx, info) - } - } -} diff --git a/internal/infra/nats/dlq.go b/internal/infra/nats/dlq.go deleted file mode 100644 index e8db18e..0000000 --- a/internal/infra/nats/dlq.go +++ /dev/null @@ -1,139 +0,0 @@ -package nats - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "strings" - "time" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// validateDLQ verifies whether DLQ routing should be enabled and, if so, whether -// the configured DLQ subject is bound to a JetStream stream. Returns an error -// if DLQ is enabled but misconfigured, allowing the caller to fail fast. -func (c *Consumer) validateDLQ() error { - if c == nil { - return nil - } - - // DLQ routing is only active in JetStream mode. - if c.config.mode != "jetstream" { - return nil - } - - // If DLQ is not enabled in config, make sure we don't accidentally route to it. - if !c.cfg.DLQ.Enabled { - if strings.TrimSpace(c.config.dlqSubject) != "" { - c.logger.Info("DLQ subject configured but dlq.enabled is false; DLQ routing disabled", - zap.String("dlq_subject", c.config.dlqSubject), - ) - } - c.config.dlqSubject = "" - return nil - } - - subject := strings.TrimSpace(c.config.dlqSubject) - if subject == "" { - return fmt.Errorf("DLQ enabled but dlq.subject is empty") - } - - if c.js == nil { - return fmt.Errorf("DLQ enabled but JetStream context is nil") - } - - // Ensure the DLQ subject is actually bound to a JetStream stream. This avoids - // the opaque `nats: no response from stream` error later when publishing. - c.telemetry.RecordJSAPICall("dlq_validate_stream") - streamName, err := c.js.StreamNameBySubject(subject) - if err != nil || strings.TrimSpace(streamName) == "" { - return fmt.Errorf("DLQ subject %s not bound to any JetStream stream: %w", subject, err) - } - - c.logger.Info("DLQ configuration validated", - zap.String("dlq_subject", subject), - zap.String("dlq_stream", streamName), - ) - - return nil -} - -// routeToDLQ publishes a copy of the failed message to the configured DLQ subject, -// including useful metadata for offline analysis. If DLQ is not configured or the -// consumer is not running in JetStream mode, this is a no-op. -func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) error { - if c == nil || c.js == nil { - return nil - } - if c.config.mode != "jetstream" { - return nil - } - if strings.TrimSpace(c.config.dlqSubject) == "" { - return nil - } - - meta, _ := msg.Metadata() - jsSeq := uint64(0) - deliveries := uint64(0) - if meta != nil { - jsSeq = meta.Sequence.Stream - deliveries = meta.NumDelivered - } - - payload := map[string]any{ - "transport_msg_id": msg.Header.Get("Nats-Msg-Id"), - "subject": msg.Subject, - "stream": c.config.streamName, - "consumer": c.config.consumerName, - "nats_sequence": jsSeq, - "deliveries": deliveries, - "error": fmt.Sprint(cause), - "received_at": time.Now().UTC(), - "body": string(msg.Data), - } - - data, err := json.Marshal(payload) - if err != nil { - c.logger.Error("failed to marshal DLQ payload", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.String("dlq_subject", c.config.dlqSubject), - zap.Error(err), - ) - return fmt.Errorf("marshal dlq payload: %w", err) - } - - if _, err := c.js.Publish(c.config.dlqSubject, data); err != nil { - // nats.ErrNoResponders typically means that no JetStream stream is - // configured to receive this subject, or JetStream is temporarily - // unavailable. Surface this explicitly to make operational diagnosis - // easier. - if errors.Is(err, nats.ErrNoResponders) { - c.logger.Error("transient DLQ publish error (no responders)", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.String("dlq_subject", c.config.dlqSubject), - zap.Int("payload_size", len(data)), - zap.Error(err), - ) - c.telemetry.RecordDLQPublishFailure(ctx, c.config.streamName, c.config.consumerName) - return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.config.dlqSubject, err) - } - c.logger.Error("failed to publish to DLQ", - zap.String("stream", c.config.streamName), - zap.String("consumer", c.config.consumerName), - zap.String("dlq_subject", c.config.dlqSubject), - zap.Int("payload_size", len(data)), - zap.Error(err), - ) - c.telemetry.RecordDLQPublishFailure(ctx, c.config.streamName, c.config.consumerName) - return fmt.Errorf("publish to dlq subject %s: %w", c.config.dlqSubject, err) - } - - c.telemetry.RecordDLQMessage(ctx, c.config.streamName, c.config.consumerName) - - return nil -} diff --git a/internal/infra/nats/dlq_handler.go b/internal/infra/nats/dlq_handler.go new file mode 100644 index 0000000..d7b6490 --- /dev/null +++ b/internal/infra/nats/dlq_handler.go @@ -0,0 +1,86 @@ +package nats + +import ( + "caatsm/internal/infra/telemetry" + "context" + "encoding/json" + "fmt" + "time" + + "github.com/nats-io/nats.go" + "go.uber.org/zap" +) + +// DLQHandler defines the interface for dead letter queue operations +type DLQHandler interface { + RouteToDLQ(ctx context.Context, msg *nats.Msg, cause error) error + ValidateDLQ() error +} + +// defaultDLQHandler implements DLQHandler interface +type defaultDLQHandler struct { + js nats.JetStreamContext + dlqSubject string + streamName string + consumerName string + logger *zap.Logger + telemetry telemetry.Recorder +} + +func (h *defaultDLQHandler) RouteToDLQ(ctx context.Context, msg *nats.Msg, cause error) error { + return h.routeToDLQInternal(ctx, msg, cause) +} + +func (h *defaultDLQHandler) ValidateDLQ() error { + return h.validateDLQInternal() +} + +func (h *defaultDLQHandler) routeToDLQInternal(ctx context.Context, msg *nats.Msg, cause error) error { + // Basic DLQ routing implementation + payload := map[string]any{ + "subject": msg.Subject, + "stream": h.streamName, + "consumer": h.consumerName, + "error": cause.Error(), + "received_at": time.Now().UTC(), + "body": string(msg.Data), + } + + data, err := json.Marshal(payload) + if err != nil { + h.logger.Error("failed to marshal DLQ payload", zap.Error(err)) + return err + } + + pubCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + _, err = h.js.Publish(h.dlqSubject, data, nats.Context(pubCtx)) + if err != nil { + h.logger.Error("failed to publish to DLQ", + zap.String("dlq_subject", h.dlqSubject), + zap.Error(err), + ) + h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) + return err + } + + h.telemetry.RecordDLQMessage(ctx, h.streamName, h.consumerName) + return nil +} + +func (h *defaultDLQHandler) validateDLQInternal() error { + if h.js == nil { + return fmt.Errorf("JetStream context is nil") + } + + _, err := h.js.StreamNameBySubject(h.dlqSubject) + if err != nil { + return fmt.Errorf("DLQ subject %s not bound to any JetStream stream: %w", h.dlqSubject, err) + } + + h.logger.Info("DLQ configuration validated", + zap.String("dlq_subject", h.dlqSubject), + ) + + return nil +} diff --git a/internal/infra/nats/dlq_test.go b/internal/infra/nats/dlq_test.go index 56c09ee..f6f5f1c 100644 --- a/internal/infra/nats/dlq_test.go +++ b/internal/infra/nats/dlq_test.go @@ -1,142 +1,38 @@ package nats import ( - "context" - "errors" - - configpkg "caatsm/internal/infra/config" "caatsm/internal/infra/telemetry" - "github.com/nats-io/nats.go" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "go.uber.org/zap" "go.uber.org/zap/zaptest" ) -var _ = Describe("DLQ", func() { +var _ = Describe("DLQHandler", func() { var ( - logger *zap.Logger + handler *defaultDLQHandler + logger *zap.Logger ) BeforeEach(func() { logger = zaptest.NewLogger(GinkgoT()) + handler = &defaultDLQHandler{ + logger: logger, + streamName: "TEST_STREAM", + consumerName: "test-consumer", + dlqSubject: "caatsm.dlq", + telemetry: telemetry.NewNoop(), + } }) - Describe("validateDLQ", func() { - It("returns nil when consumer is nil", func() { - var c *Consumer - Expect(c.validateDLQ()).To(Succeed()) - }) - - It("returns nil when mode is not jetstream", func() { - c := &Consumer{ - config: consumerConfig{ - mode: "core", - }, - cfg: &configpkg.Config{ - DLQ: configpkg.DLQConfig{ - Enabled: true, - Subject: "caatsm.dlq", - }, - }, - logger: logger, - } - Expect(c.validateDLQ()).To(Succeed()) - }) - - It("clears dlqSubject when DLQ is disabled", func() { - c := &Consumer{ - config: consumerConfig{ - mode: "jetstream", - dlqSubject: "caatsm.dlq", - }, - cfg: &configpkg.Config{ - DLQ: configpkg.DLQConfig{ - Enabled: false, - Subject: "caatsm.dlq", - }, - }, - logger: logger, - } - Expect(c.validateDLQ()).To(Succeed()) - Expect(c.config.dlqSubject).To(Equal("")) - }) - - It("returns error when DLQ is enabled but subject is empty", func() { - c := &Consumer{ - config: consumerConfig{ - mode: "jetstream", - dlqSubject: "", - }, - cfg: &configpkg.Config{ - DLQ: configpkg.DLQConfig{ - Enabled: true, - Subject: "", - }, - }, - logger: logger, - } - err := c.validateDLQ() - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(ContainSubstring("DLQ enabled but dlq.subject is empty")) - }) - - It("returns error when DLQ is enabled but JetStream context is nil", func() { - c := &Consumer{ - config: consumerConfig{ - mode: "jetstream", - dlqSubject: "caatsm.dlq", - }, - js: nil, - cfg: &configpkg.Config{ - DLQ: configpkg.DLQConfig{ - Enabled: true, - Subject: "caatsm.dlq", - }, - }, - logger: logger, - telemetry: telemetry.NewNoop(), - } - err := c.validateDLQ() + Describe("ValidateDLQ", func() { + It("returns error when JetStream context is nil", func() { + handler.js = nil + err := handler.ValidateDLQ() Expect(err).To(HaveOccurred()) Expect(err.Error()).To(ContainSubstring("JetStream context is nil")) }) }) - Describe("routeToDLQ", func() { - It("returns nil when consumer is nil", func() { - var c *Consumer - ctx := context.Background() - msg := &nats.Msg{} - err := errors.New("test error") - Expect(c.routeToDLQ(ctx, msg, err)).To(Succeed()) - }) - - It("returns nil when mode is not jetstream", func() { - c := &Consumer{ - config: consumerConfig{ - mode: "core", - }, - } - ctx := context.Background() - msg := &nats.Msg{} - err := errors.New("test error") - Expect(c.routeToDLQ(ctx, msg, err)).To(Succeed()) - }) - - It("returns nil when dlqSubject is empty", func() { - c := &Consumer{ - config: consumerConfig{ - mode: "jetstream", - dlqSubject: "", - }, - js: nil, // Can be nil for this test - } - ctx := context.Background() - msg := &nats.Msg{} - err := errors.New("test error") - Expect(c.routeToDLQ(ctx, msg, err)).To(Succeed()) - }) - }) }) diff --git a/internal/infra/nats/error_handler.go b/internal/infra/nats/error_handler.go deleted file mode 100644 index dd32ead..0000000 --- a/internal/infra/nats/error_handler.go +++ /dev/null @@ -1,136 +0,0 @@ -package nats - -import ( - "caatsm/internal/app" - "context" - "errors" - "time" - - "github.com/nats-io/nats.go" - "go.uber.org/zap" -) - -// ErrorHandler handles various error scenarios in NATS operations -type ErrorHandler struct { - logger *zap.Logger -} - -// NewErrorHandler creates a new error handler -func NewErrorHandler(logger *zap.Logger) *ErrorHandler { - return &ErrorHandler{ - logger: logger, - } -} - -// FetchErrorResult represents the result of handling a fetch error -type FetchErrorResult struct { - ShouldContinue bool - RecoveredSub *nats.Subscription - Error error -} - -// HandleFetchError handles errors during message fetching with recovery logic -func (h *ErrorHandler) HandleFetchError( - ctx context.Context, - err error, - sub **nats.Subscription, - fetchErrorStreak *int, - streamName, consumerName string, - recoverFunc func() (*nats.Subscription, error), -) FetchErrorResult { - if errors.Is(err, nats.ErrTimeout) { - // Timeout is expected when no messages are available. - return FetchErrorResult{ShouldContinue: true} - } - - // JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready). - if errors.Is(err, nats.ErrNoResponders) { - *fetchErrorStreak++ - backoff := time.Duration(*fetchErrorStreak) * time.Second - backoff = min(backoff, 30*time.Second) - h.logger.Warn("JetStream not available, will retry with backoff", - zap.Error(err), - zap.String("stream", streamName), - zap.String("consumer", consumerName), - zap.Duration("backoff", backoff), - ) - if !sleepWithContext(ctx, backoff) { - return FetchErrorResult{ShouldContinue: false, Error: ctx.Err()} - } - return FetchErrorResult{ShouldContinue: true} - } - - // Underlying consumer/stream removed while app is running. - if isJetStreamResourceNotFound(err) { - if isDevLikeEnv() && shouldBootstrapStream() { - h.logger.Warn("JetStream consumer or stream missing; attempting to recreate", - zap.Error(err), - zap.String("stream", streamName), - zap.String("consumer", consumerName), - ) - newSub, subErr := recoverFunc() - if subErr != nil { - return FetchErrorResult{ShouldContinue: false, Error: subErr} - } - *sub = newSub - *fetchErrorStreak = 0 - return FetchErrorResult{ShouldContinue: true, RecoveredSub: newSub} - } - - // Production: treat as configuration/operational error. - h.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment", - zap.Error(err), - zap.String("stream", streamName), - zap.String("consumer", consumerName), - ) - return FetchErrorResult{ShouldContinue: false, Error: err} - } - - // Generic error path with modest backoff. - *fetchErrorStreak++ - backoff := time.Duration(*fetchErrorStreak) * time.Second - backoff = min(backoff, 10*time.Second) - h.logger.Error("Failed to fetch messages; backing off", - zap.Error(err), - zap.Duration("backoff", backoff), - ) - if !sleepWithContext(ctx, backoff) { - return FetchErrorResult{ShouldContinue: false, Error: ctx.Err()} - } - return FetchErrorResult{ShouldContinue: true} -} - -// ProcessingErrorResult represents the result of handling a processing error -type ProcessingErrorResult struct { - IsPermanent bool - ShouldApplyBackpressure bool - BackpressureDelay time.Duration -} - -// HandleProcessingError analyzes processing errors and determines appropriate action -func (h *ErrorHandler) HandleProcessingError( - consecutiveErrors int, - err error, - logger *zap.Logger, - subject string, -) ProcessingErrorResult { - isPermanent := app.IsPermanent(err) - - result := ProcessingErrorResult{ - IsPermanent: isPermanent, - } - - if isPermanent { - // Reset error streak for permanent errors - return result - } - - // Transient error: increment error streak and apply simple backpressure if needed. - if consecutiveErrors >= 10 { - result.ShouldApplyBackpressure = true - result.BackpressureDelay = time.Duration(consecutiveErrors) * 100 * time.Millisecond - result.BackpressureDelay = min(result.BackpressureDelay, 5*time.Second) - } - - return result -} diff --git a/internal/infra/nats/error_handler_test.go b/internal/infra/nats/error_handler_test.go deleted file mode 100644 index 635aec7..0000000 --- a/internal/infra/nats/error_handler_test.go +++ /dev/null @@ -1,102 +0,0 @@ -package nats - -import ( - "context" - "errors" - "time" - - "github.com/nats-io/nats.go" - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" - "go.uber.org/zap" - "go.uber.org/zap/zaptest" -) - -var _ = Describe("ErrorHandler", func() { - var ( - handler *ErrorHandler - logger *zap.Logger - ) - - BeforeEach(func() { - logger = zaptest.NewLogger(GinkgoT()) - handler = NewErrorHandler(logger) - }) - - Describe("HandleProcessingError", func() { - It("identifies permanent errors correctly", func() { - // Mock a permanent error (this would be defined in the app package) - permanentErr := errors.New("permanent error") - // For testing, we'll assume any error is transient unless specified - - result := handler.HandleProcessingError(0, permanentErr, logger, "test.subject") - Expect(result.IsPermanent).To(BeFalse()) // Since we can't easily mock app.IsPermanent - Expect(result.ShouldApplyBackpressure).To(BeFalse()) - }) - - It("applies backpressure for consecutive errors", func() { - transientErr := errors.New("transient error") - - result := handler.HandleProcessingError(10, transientErr, logger, "test.subject") - Expect(result.IsPermanent).To(BeFalse()) - Expect(result.ShouldApplyBackpressure).To(BeTrue()) - Expect(result.BackpressureDelay).To(BeNumerically(">=", 100*time.Millisecond)) - }) - }) - - Describe("HandleFetchError", func() { - var ( - ctx context.Context - sub *nats.Subscription - fetchErrorStreak int - streamName string - consumerName string - ) - - BeforeEach(func() { - ctx = context.Background() - sub = nil - fetchErrorStreak = 0 - streamName = "TEST_STREAM" - consumerName = "test-consumer" - }) - - It("handles timeout errors", func() { - result := handler.HandleFetchError(ctx, nats.ErrTimeout, &sub, &fetchErrorStreak, streamName, consumerName, nil) - Expect(result.ShouldContinue).To(BeTrue()) - Expect(result.Error).NotTo(HaveOccurred()) - }) - - It("handles no responders with backoff", func() { - result := handler.HandleFetchError(ctx, nats.ErrNoResponders, &sub, &fetchErrorStreak, streamName, consumerName, nil) - Expect(result.ShouldContinue).To(BeTrue()) - Expect(result.Error).NotTo(HaveOccurred()) - Expect(fetchErrorStreak).To(Equal(1)) - }) - - It("handles resource not found errors in dev environment", func() { - // Mock resource not found error - resourceErr := errors.New("stream not found") - // Provide a no-op recovery function to avoid panic - recoveryFunc := func() (*nats.Subscription, error) { - return nil, errors.New("recovery not implemented in test") - } - result := handler.HandleFetchError(ctx, resourceErr, &sub, &fetchErrorStreak, streamName, consumerName, recoveryFunc) - // In test environment, this should attempt recovery but fail since recovery func returns error - Expect(result.ShouldContinue).To(BeFalse()) - Expect(result.Error).To(HaveOccurred()) - }) - - It("handles successful recovery", func() { - resourceErr := errors.New("consumer not found") - mockSub := &nats.Subscription{} - recoveryFunc := func() (*nats.Subscription, error) { - return mockSub, nil - } - result := handler.HandleFetchError(ctx, resourceErr, &sub, &fetchErrorStreak, streamName, consumerName, recoveryFunc) - Expect(result.ShouldContinue).To(BeTrue()) - Expect(result.RecoveredSub).To(Equal(mockSub)) - Expect(fetchErrorStreak).To(Equal(0)) // Should reset on successful recovery - }) - }) -}) diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index 025e7ed..06c167f 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -6,45 +6,63 @@ import ( "crypto/x509" "fmt" "os" - "strings" + "time" "github.com/nats-io/nats.go" "go.uber.org/zap" ) -// ProvideNATSConn creates a reusable NATS connection with optional authentication. +// ProvideNATSConn creates a basic NATS connection. func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) { opts := []nats.Option{ - nats.RetryOnFailedConnect(true), - nats.Timeout(cfg.Timeouts.Server), - nats.ReconnectWait(cfg.Timeouts.ReconnectWait), - // Use infinite reconnects so the app survives long NATS outages (e.g. docker compose down/up). - nats.MaxReconnects(-1), + nats.ReconnectWait(5 * time.Second), + nats.MaxReconnects(10), nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { - if err != nil { - logger.Warn("NATS disconnected", zap.Error(err)) - } + logger.Warn("NATS disconnected", zap.Error(err)) }), nats.ReconnectHandler(func(nc *nats.Conn) { - safeURL := sanitizeURLForLogging(nc.ConnectedUrl()) - logger.Info("NATS reconnected", zap.String("url", safeURL)) + logger.Info("NATS reconnected") }), } - // Apply authentication options - authOpts, err := buildAuthOptions(&cfg.NATS.Auth, logger) - if err != nil { - return nil, fmt.Errorf("failed to build auth options: %w", err) + // Simple token authentication if provided + if cfg.NATS.Auth.Token != "" { + opts = append(opts, nats.Token(cfg.NATS.Auth.Token)) + } + + // Basic TLS support if enabled + if cfg.NATS.Auth.TLSEnabled { + tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12} + + // Load client certificate if provided + if cfg.NATS.Auth.TLSCertFile != "" && cfg.NATS.Auth.TLSKeyFile != "" { + cert, err := tls.LoadX509KeyPair(cfg.NATS.Auth.TLSCertFile, cfg.NATS.Auth.TLSKeyFile) + if err != nil { + return nil, fmt.Errorf("failed to load TLS certificate: %w", err) + } + tlsConfig.Certificates = []tls.Certificate{cert} + } + + // Load CA certificate for server verification + if cfg.NATS.Auth.TLSCAFile != "" { + caCert, err := os.ReadFile(cfg.NATS.Auth.TLSCAFile) + if err != nil { + return nil, fmt.Errorf("failed to read CA certificate: %w", err) + } + caCertPool := x509.NewCertPool() + if !caCertPool.AppendCertsFromPEM(caCert) { + return nil, fmt.Errorf("failed to parse CA certificate") + } + tlsConfig.RootCAs = caCertPool + } + + opts = append(opts, nats.Secure(tlsConfig)) } - opts = append(opts, authOpts...) nc, err := nats.Connect(cfg.NATS.URL, opts...) if err != nil { - safeURL := sanitizeURLForLogging(cfg.NATS.URL) logger.Error("failed to connect to NATS", - zap.String("url", safeURL), - zap.Duration("timeout", cfg.Timeouts.Server), - zap.Duration("reconnect_wait", cfg.Timeouts.ReconnectWait), + zap.String("url", cfg.NATS.URL), zap.Error(err), ) return nil, fmt.Errorf("failed to connect to NATS: %w", err) @@ -53,120 +71,12 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) return nc, nil } -// buildAuthOptions builds NATS connection options based on authentication configuration. -func buildAuthOptions(auth *config.NATSAuthConfig, logger *zap.Logger) ([]nats.Option, error) { - var opts []nats.Option - authMethods := 0 - - // Token authentication (highest priority) - if auth.Token != "" { - authMethods++ - logger.Debug("Using NATS token authentication") - opts = append(opts, nats.Token(auth.Token)) - } - - // Credentials file authentication - if auth.CredentialsFile != "" { - authMethods++ - if authMethods > 1 { - return nil, fmt.Errorf("multiple authentication methods specified: only one of token, credentials_file, or user/password can be used") - } - logger.Debug("Using NATS credentials file authentication", zap.String("file", auth.CredentialsFile)) - opts = append(opts, nats.UserCredentials(auth.CredentialsFile)) - } - - // User/Password authentication - if auth.User != "" || auth.Password != "" { - authMethods++ - if authMethods > 1 { - return nil, fmt.Errorf("multiple authentication methods specified: only one of token, credentials_file, or user/password can be used") - } - if auth.User == "" || auth.Password == "" { - return nil, fmt.Errorf("both user and password must be specified for user/password authentication") - } - logger.Debug("Using NATS user/password authentication", zap.String("user", auth.User)) - opts = append(opts, nats.UserInfo(auth.User, auth.Password)) - } - - // TLS configuration - if auth.TLSEnabled { - tlsConfig := &tls.Config{ - MinVersion: tls.VersionTLS12, - } - - // Load client certificate and key if provided - if auth.TLSCertFile != "" && auth.TLSKeyFile != "" { - cert, err := tls.LoadX509KeyPair(auth.TLSCertFile, auth.TLSKeyFile) - if err != nil { - return nil, fmt.Errorf("failed to load TLS certificate: %w", err) - } - tlsConfig.Certificates = []tls.Certificate{cert} - logger.Debug("Loaded TLS client certificate", zap.String("cert", auth.TLSCertFile)) - } - - // Load CA certificate for server verification if provided - if auth.TLSCAFile != "" { - caCert, err := os.ReadFile(auth.TLSCAFile) - if err != nil { - return nil, fmt.Errorf("failed to read CA certificate file: %w", err) - } - caCertPool := x509.NewCertPool() - if !caCertPool.AppendCertsFromPEM(caCert) { - return nil, fmt.Errorf("failed to parse CA certificate from %s", auth.TLSCAFile) - } - tlsConfig.RootCAs = caCertPool - logger.Debug("Loaded TLS CA certificate", zap.String("ca_file", auth.TLSCAFile)) - } - - opts = append(opts, nats.Secure(tlsConfig)) - logger.Debug("TLS enabled for NATS connection") - } - - return opts, nil -} - -// ProvideJetStream creates a NATS JetStream context using an existing connection. -// Returns nil, nil when cfg.NATS.Mode == "core" to support plain NATS servers without JetStream. -func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) { - mode := strings.ToLower(cfg.NATS.Mode) - if mode == "core" { - logger.Info("Skipping JetStream initialization for core NATS mode") - return nil, nil - } - - // Get JetStream context +// ProvideJetStream creates a JetStream context from a NATS connection. +func ProvideJetStream(nc *nats.Conn, logger *zap.Logger) (nats.JetStreamContext, error) { js, err := nc.JetStream() if err != nil { - safeURL := sanitizeURLForLogging(cfg.NATS.URL) - logger.Error("failed to get JetStream context", - zap.String("url", safeURL), - zap.Error(err), - ) - nc.Close() - return nil, fmt.Errorf("failed to get JetStream context: %w", err) + logger.Error("failed to create JetStream context", zap.Error(err)) + return nil, fmt.Errorf("failed to create JetStream context: %w", err) } - - // Ensure the stream exists using StreamManager - streamName := cfg.NATS.Stream - consumerSubject := cfg.EffectiveSubscriptionTopic() - publisherSubject := strings.TrimSpace(cfg.Publisher.Topic) - - streamSubjects := dedupeSubjects([]string{consumerSubject, publisherSubject}) - if len(streamSubjects) == 0 { - logger.Error("no subjects configured for JetStream stream", - zap.String("stream", streamName), - zap.String("consumer_subject", consumerSubject), - zap.String("publisher_subject", publisherSubject), - ) - nc.Close() - return nil, fmt.Errorf("no subjects configured for JetStream stream %s", streamName) - } - - streamManager := NewStreamManagerWithConfig(js, streamName, streamSubjects, &cfg.NATS.StreamLimits, logger) - if err := streamManager.EnsureStream(); err != nil { - nc.Close() - return nil, err - } - return js, nil } diff --git a/internal/infra/nats/manager_test.go b/internal/infra/nats/manager_test.go deleted file mode 100644 index 13d4f41..0000000 --- a/internal/infra/nats/manager_test.go +++ /dev/null @@ -1,99 +0,0 @@ -package nats - -import ( - "github.com/nats-io/nats.go" - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" - "go.uber.org/zap" - "go.uber.org/zap/zaptest" -) - -var _ = Describe("ConsumerManager", func() { - var ( - js nats.JetStreamContext - streamName string - consumerName string - subject string - logger *zap.Logger - consumerMgr *ConsumerManager - ) - - BeforeEach(func() { - // Note: These tests would need a real NATS server for full functionality - // For now, we'll test the structure and error handling - js = nil // Would be a mock in real tests - streamName = "TEST_STREAM" - consumerName = "test-consumer" - subject = "test.subject" - logger = zaptest.NewLogger(GinkgoT()) - consumerMgr = NewConsumerManager(js, streamName, consumerName, subject, logger) - }) - - Describe("NewConsumerManager", func() { - It("creates a consumer manager with correct fields", func() { - Expect(consumerMgr.js).To(BeNil()) - Expect(consumerMgr.streamName).To(Equal(streamName)) - Expect(consumerMgr.consumerName).To(Equal(consumerName)) - Expect(consumerMgr.subject).To(Equal(subject)) - Expect(consumerMgr.logger).To(Equal(logger)) - }) - }) - - Describe("CreatePullSubscription", func() { - It("returns error when JetStream context is nil", func() { - // This will panic because js is nil, so we skip this test for now - Skip("Requires mock JetStream context") - }) - }) - - Describe("RecoverResources", func() { - It("returns error when JetStream context is nil", func() { - streamMgr := NewStreamManager(nil, streamName, []string{subject}, logger) - consumerConfig := &nats.ConsumerConfig{Durable: consumerName} - err := consumerMgr.RecoverResources(streamMgr, consumerConfig) - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(ContainSubstring("jetstream context is nil")) - }) - }) -}) - -var _ = Describe("StreamManager", func() { - var ( - js nats.JetStreamContext - streamName string - subjects []string - logger *zap.Logger - streamMgr *StreamManager - ) - - BeforeEach(func() { - js = nil // Would be a mock in real tests - streamName = "TEST_STREAM" - subjects = []string{"test.subject"} - logger = zaptest.NewLogger(GinkgoT()) - streamMgr = NewStreamManager(js, streamName, subjects, logger) - }) - - Describe("NewStreamManager", func() { - It("creates a stream manager with correct fields", func() { - Expect(streamMgr.js).To(BeNil()) - Expect(streamMgr.streamName).To(Equal(streamName)) - Expect(streamMgr.subjects).To(Equal(subjects)) - Expect(streamMgr.logger).To(Equal(logger)) - }) - }) - - Describe("EnsureStream", func() { - It("returns error when JetStream context is nil", func() { - // This will panic because js is nil, so we skip this test for now - Skip("Requires mock JetStream context") - }) - }) - - Describe("validateStreamConfig", func() { - It("handles nil stream info gracefully", func() { - streamMgr.validateStreamConfig(nil) - // Should not panic - }) - }) -}) diff --git a/internal/infra/nats/message_fetcher.go b/internal/infra/nats/message_fetcher.go new file mode 100644 index 0000000..d37d5ae --- /dev/null +++ b/internal/infra/nats/message_fetcher.go @@ -0,0 +1,90 @@ +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 + } + + // ACK the message + if ackErr := msg.Ack(); ackErr != nil { + p.logger.Error("Failed to ACK message", zap.Error(ackErr)) + } else { + elapsed := time.Since(start) + 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 + } + + if p.mode == "core" { + return uuid.NewString(), "generated", 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) { + p.logger.Error("Failed to process message", + zap.String("subject", msg.Subject), + zap.Error(err), + zap.Bool("permanent", app.IsPermanent(err)), + ) + + result := obsmetrics.ResultFail + if app.IsPermanent(err) { + result = obsmetrics.ResultPermanentFail + } + p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, result, elapsed) + + consecutiveErrors := 0 + if p.consecutiveProcessErrors != nil { + consecutiveErrors = *p.consecutiveProcessErrors + } + + isPermanent := app.IsPermanent(err) + 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 + } + // Poison/permanent message: route to DLQ if configured, then ACK + 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.Error(dlqErr)) + } + } + if ackErr := msg.Ack(); ackErr != nil { + p.logger.Error("Failed to ACK permanent-error message", 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 index a1d6b4f..3e94172 100644 --- a/internal/infra/nats/metrics_test.go +++ b/internal/infra/nats/metrics_test.go @@ -2,10 +2,9 @@ package nats import ( "context" + "time" - "github.com/nats-io/nats.go" . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" "go.uber.org/zap/zaptest" ) @@ -19,40 +18,20 @@ var _ = Describe("Metrics", func() { ctx = context.Background() c = &Consumer{ config: consumerConfig{ - streamName: "TEST_STREAM", - consumerName: "test-consumer", + streamName: "TEST_STREAM", + consumerName: "test-consumer", + monitorInterval: 30 * time.Second, // Set a valid interval }, logger: zaptest.NewLogger(GinkgoT()), } }) - Describe("initMetrics", func() { - It("initializes metrics without error", func() { - c.initMetrics() - Expect(c.meter).NotTo(BeNil()) - }) - }) - - Describe("recordConsumerMetrics", func() { - It("handles nil ConsumerInfo gracefully", func() { - c.initMetrics() - c.recordConsumerMetrics(ctx, nil) - // Should not panic - }) - - It("records metrics when ConsumerInfo is provided", func() { - c.initMetrics() - info := &nats.ConsumerInfo{ - Config: nats.ConsumerConfig{}, - Delivered: nats.SequenceInfo{ - Consumer: 50, - Stream: 100, - }, - } - // Set fields directly (they are exported) - // Note: ConsumerInfo fields may not all be exported, so we test what we can - c.recordConsumerMetrics(ctx, info) - // Should not panic + 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/stream_manager.go b/internal/infra/nats/stream_manager.go index 258dcd3..79d0200 100644 --- a/internal/infra/nats/stream_manager.go +++ b/internal/infra/nats/stream_manager.go @@ -1,10 +1,7 @@ package nats import ( - "caatsm/internal/infra/config" - "errors" "fmt" - "strings" "github.com/nats-io/nats.go" "go.uber.org/zap" @@ -16,7 +13,6 @@ type StreamManager struct { streamName string subjects []string logger *zap.Logger - cfg *config.StreamLimitsConfig } // NewStreamManager creates a new stream manager @@ -29,113 +25,16 @@ func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []st } } -// NewStreamManagerWithConfig creates a new stream manager with full stream configuration -func NewStreamManagerWithConfig(js nats.JetStreamContext, streamName string, subjects []string, streamLimits *config.StreamLimitsConfig, logger *zap.Logger) *StreamManager { - return &StreamManager{ - js: js, - streamName: streamName, - subjects: subjects, - logger: logger, - cfg: streamLimits, - } -} - // EnsureStream ensures that the configured JetStream stream exists func (sm *StreamManager) EnsureStream() error { - // Build stream configuration - streamConfig := sm.buildStreamConfig() - - info, err := sm.js.StreamInfo(sm.streamName) + _, err := sm.js.StreamInfo(sm.streamName) if err != nil { - if errors.Is(err, nats.ErrStreamNotFound) { - if shouldBootstrapStream() { - if _, err = sm.js.AddStream(streamConfig); err != nil { - sm.logger.Error("failed to create stream", - zap.String("stream", sm.streamName), - zap.Strings("subjects", sm.subjects), - zap.Error(err), - ) - 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 - } - sm.logger.Error("stream not found and auto-creation disabled", - zap.String("stream", sm.streamName), - zap.Strings("expected_subjects", sm.subjects), - ) - return fmt.Errorf("stream %s not found and auto-creation disabled", sm.streamName) - } - sm.logger.Error("failed to fetch stream info", - zap.String("stream", sm.streamName), - zap.Error(err), - ) - return fmt.Errorf("failed to fetch stream info for %s: %w", sm.streamName, err) + return fmt.Errorf("stream %s not found or inaccessible: %w", sm.streamName, err) } - // Stream exists: validate subjects but do not fail hard if they differ. - sm.validateStreamConfig(info) + sm.logger.Info("JetStream stream verified", + zap.String("stream", sm.streamName), + zap.Strings("subjects", sm.subjects), + ) return nil } - -// buildStreamConfig builds the stream configuration from manager settings -func (sm *StreamManager) buildStreamConfig() *nats.StreamConfig { - config := &nats.StreamConfig{ - Name: sm.streamName, - Subjects: sm.subjects, - Retention: nats.LimitsPolicy, - Storage: nats.FileStorage, - } - - // Apply stream limits configuration if provided - if sm.cfg != nil { - config.MaxMsgs = sm.cfg.MaxMsgs - config.MaxBytes = sm.cfg.MaxBytes - config.MaxAge = sm.cfg.MaxAge - config.Replicas = sm.cfg.Replicas - - // Map storage type - switch strings.ToLower(sm.cfg.Storage) { - case "memory": - config.Storage = nats.MemoryStorage - case "file": - config.Storage = nats.FileStorage - } - - // Map discard policy - if strings.EqualFold(sm.cfg.Discard, "new") { - config.Discard = nats.DiscardNew - } else { - config.Discard = nats.DiscardOld - } - } - - return config -} - -// validateStreamConfig validates the stream configuration -func (sm *StreamManager) validateStreamConfig(info *nats.StreamInfo) { - if info == nil { - return - } - - missing := make([]string, 0) - for _, subj := range sm.subjects { - if subj == "" { - continue - } - if !containsSubject(info.Config.Subjects, subj) { - missing = append(missing, subj) - } - } - if len(missing) > 0 { - sm.logger.Warn("JetStream stream subjects missing expected entries", - zap.String("stream", info.Config.Name), - zap.Strings("stream_subjects", info.Config.Subjects), - zap.Strings("missing_subjects", missing), - ) - } -} diff --git a/internal/infra/nats/utils.go b/internal/infra/nats/utils.go index 2c16eab..b7fb6f3 100644 --- a/internal/infra/nats/utils.go +++ b/internal/infra/nats/utils.go @@ -1,20 +1,24 @@ package nats import ( + "context" "errors" "net/url" - "os" "strings" + "time" "github.com/nats-io/nats.go" ) -// isDevLikeEnv checks if the current environment is development-like. -func isDevLikeEnv() bool { - switch strings.ToLower(os.Getenv("GO_ENV")) { - case "", "dev", "development", "test", "testing": +// sleepWithContext sleeps for the specified duration, but returns early if the context is canceled. +// Returns true if the full duration was slept, false if the context was canceled. +func sleepWithContext(ctx context.Context, duration time.Duration) bool { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-timer.C: return true - default: + case <-ctx.Done(): return false } } @@ -79,26 +83,6 @@ func mapReplayPolicy(value string) nats.ReplayPolicy { } } -// shouldBootstrapStream checks if streams should be auto-created based on environment. -func shouldBootstrapStream() bool { - switch strings.ToLower(os.Getenv("GO_ENV")) { - case "", "dev", "development", "test", "testing": - return true - default: - return false - } -} - -// containsSubject checks if a subject exists in a list of subjects. -func containsSubject(subjects []string, target string) bool { - for _, s := range subjects { - if s == target { - return true - } - } - return false -} - // dedupeSubjects removes duplicate and empty subjects from a list. func dedupeSubjects(subjects []string) []string { seen := make(map[string]struct{}) diff --git a/internal/infra/nats/utils_test.go b/internal/infra/nats/utils_test.go index fb13be4..c212b07 100644 --- a/internal/infra/nats/utils_test.go +++ b/internal/infra/nats/utils_test.go @@ -1,88 +1,11 @@ package nats import ( - "os" - . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) var _ = Describe("Utils", func() { - Describe("isDevLikeEnv", func() { - BeforeEach(func() { - // Save original value - originalEnv := os.Getenv("GO_ENV") - DeferCleanup(func() { - if originalEnv == "" { - if err := os.Unsetenv("GO_ENV"); err != nil { - // Environment variables are optional, ignore cleanup errors in tests - _ = err - } - } else { - if err := os.Setenv("GO_ENV", originalEnv); err != nil { - // Environment variables are optional, ignore cleanup errors in tests - _ = err - } - } - }) - }) - - It("returns true for dev environment", func() { - if err := os.Setenv("GO_ENV", "dev"); err != nil { - // Environment variables are optional, ignore setup errors in tests - _ = err - } - Expect(isDevLikeEnv()).To(BeTrue()) - }) - - It("returns true for development environment", func() { - if err := os.Setenv("GO_ENV", "development"); err != nil { - // Environment variables are optional, ignore setup errors in tests - _ = err - } - Expect(isDevLikeEnv()).To(BeTrue()) - }) - - It("returns true for test environment", func() { - if err := os.Setenv("GO_ENV", "test"); err != nil { - // Environment variables are optional, ignore setup errors in tests - _ = err - } - Expect(isDevLikeEnv()).To(BeTrue()) - }) - - It("returns true for testing environment", func() { - if err := os.Setenv("GO_ENV", "testing"); err != nil { - // Environment variables are optional, ignore setup errors in tests - _ = err - } - Expect(isDevLikeEnv()).To(BeTrue()) - }) - - It("returns true for empty environment", func() { - if err := os.Unsetenv("GO_ENV"); err != nil { - // Environment variables are optional, ignore cleanup errors in tests - _ = err - } - Expect(isDevLikeEnv()).To(BeTrue()) - }) - - It("returns false for production environment", func() { - if err := os.Setenv("GO_ENV", "prod"); err != nil { - // Environment variables are optional, ignore setup errors in tests - _ = err - } - Expect(isDevLikeEnv()).To(BeFalse()) - }) - - It("returns false for production environment (uppercase)", func() { - if err := os.Setenv("GO_ENV", "PROD"); err != nil { - // Environment variables are optional, ignore setup errors in tests - _ = err - } - Expect(isDevLikeEnv()).To(BeFalse()) - }) - }) Describe("sanitizeURLForLogging", func() { It("removes credentials from URLs", func() { diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index cb53fa4..e363f3a 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -42,7 +42,7 @@ func buildAppComponents() (*appComponents, error) { if err != nil { return nil, err } - jetStreamContext, err := nats.ProvideJetStream(conn, configConfig, logger) + jetStreamContext, err := nats.ProvideJetStream(conn, logger) if err != nil { return nil, err } @@ -86,7 +86,7 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { if err != nil { return nil, err } - jetStreamContext, err := nats.ProvideJetStream(conn, cfg, logger) + jetStreamContext, err := nats.ProvideJetStream(conn, logger) if err != nil { return nil, err } diff --git a/test/integration/advisory_dlq_test.go b/test/integration/advisory_dlq_test.go deleted file mode 100644 index 15c8aae..0000000 --- a/test/integration/advisory_dlq_test.go +++ /dev/null @@ -1,196 +0,0 @@ -//go:build integration - -package integration - -import ( - "context" - "encoding/json" - "sync" - "testing" - "time" - - "caatsm/internal/infra/config" - loginfra "caatsm/internal/infra/log" - natsinfra "caatsm/internal/infra/nats" - - "github.com/nats-io/nats.go" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestAdvisoryDLQHandler(t *testing.T) { - if testing.Short() { - t.Skip("skipping integration test in short mode") - } - - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) - defer cancel() - - // Start NATS container - natsContainer, natsURL := startNATS(ctx, t) - defer func() { - _ = natsContainer.Terminate(context.Background()) - }() - - // Connect to NATS - nc, err := nats.Connect(natsURL) - require.NoError(t, err) - defer nc.Close() - - js, err := nc.JetStream() - require.NoError(t, err) - - streamName := "TEST_ADVISORY_STREAM" - consumerName := "test-advisory-consumer" - dlqSubject := "test.dlq" - - // Create stream - streamConfig := &nats.StreamConfig{ - Name: streamName, - Subjects: []string{"test.orders.*"}, - Storage: nats.FileStorage, - } - _, err = js.AddStream(streamConfig) - require.NoError(t, err) - defer func() { - _ = js.DeleteStream(streamName) - }() - - // Create DLQ stream - dlqStreamConfig := &nats.StreamConfig{ - Name: "TEST_DLQ_STREAM", - Subjects: []string{dlqSubject}, - Storage: nats.FileStorage, - } - _, err = js.AddStream(dlqStreamConfig) - require.NoError(t, err) - defer func() { - _ = js.DeleteStream("TEST_DLQ_STREAM") - }() - - // Create consumer with MaxDeliver = 2 for testing - consumerConfig := &nats.ConsumerConfig{ - Durable: consumerName, - AckPolicy: nats.AckExplicitPolicy, - MaxDeliver: 2, // Low value for testing - AckWait: 5 * time.Second, - } - _, err = js.AddConsumer(streamName, consumerConfig) - require.NoError(t, err) - - // Create a mock telemetry recorder - telemetry := &mockTelemetryRecorder{} - - // Create a simple logger for the test - logger, err := loginfra.ProvideLogger(&config.Config{ - Log: config.LogConfig{ - Level: "info", - Format: "console", - }, - }) - require.NoError(t, err) - defer logger.Sync() - - // Create advisory DLQ handler - handler, err := natsinfra.NewAdvisoryDLQHandler( - js, - nc, - streamName, - consumerName, - dlqSubject, - logger, - telemetry, - ) - require.NoError(t, err) - - // Start handler in background - handlerCtx, handlerCancel := context.WithCancel(ctx) - defer handlerCancel() - go func() { - _ = handler.Start(handlerCtx) - }() - - // Give handler time to subscribe - time.Sleep(100 * time.Millisecond) - - // Publish a message that will fail processing - testSubject := "test.orders.1" - testData := []byte("test message data") - _, err = js.Publish(testSubject, testData) - require.NoError(t, err) - - // Create pull subscription and fetch message - sub, err := js.PullSubscribe(testSubject, consumerName, nats.Bind(streamName, consumerName)) - require.NoError(t, err) - defer sub.Unsubscribe() - - // Fetch and NAK the message multiple times to exhaust MaxDeliver - msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second)) - require.NoError(t, err) - require.Len(t, msgs, 1) - - msg := msgs[0] - // NAK first time - err = msg.Nak() - require.NoError(t, err) - - // Wait for redelivery and NAK again to exhaust MaxDeliver - time.Sleep(6 * time.Second) // Wait for ack_wait + some buffer - - msgs, err = sub.Fetch(1, nats.MaxWait(2*time.Second)) - if err == nil && len(msgs) > 0 { - // NAK second time to exhaust MaxDeliver - err = msgs[0].Nak() - require.NoError(t, err) - } - - // Wait for advisory message to be processed - time.Sleep(2 * time.Second) - - // Verify message was published to DLQ - dlqSub, err := js.SubscribeSync(dlqSubject) - require.NoError(t, err) - defer dlqSub.Unsubscribe() - - dlqMsg, err := dlqSub.NextMsg(5 * time.Second) - if assert.NoError(t, err, "Expected message in DLQ") { - var payload map[string]interface{} - err = json.Unmarshal(dlqMsg.Data, &payload) - require.NoError(t, err) - - // Verify payload structure - assert.Equal(t, streamName, payload["stream"]) - assert.Equal(t, consumerName, payload["consumer"]) - assert.True(t, payload["advisory_source"].(bool)) - assert.Contains(t, payload["error"].(string), "exhausted max_deliver") - assert.Equal(t, string(testData), payload["body"]) - } - - // Verify telemetry was called - assert.Greater(t, telemetry.getDLQMessages(), 0, "Expected DLQ message to be recorded") -} - -// mockTelemetryRecorder implements TelemetryRecorder for testing -type mockTelemetryRecorder struct { - mu sync.RWMutex - dlqMessages int - dlqFailures int -} - -func (m *mockTelemetryRecorder) RecordDLQMessage(ctx context.Context, stream, consumer string) { - m.mu.Lock() - defer m.mu.Unlock() - m.dlqMessages++ -} - -func (m *mockTelemetryRecorder) RecordDLQPublishFailure(ctx context.Context, stream, consumer string) { - m.mu.Lock() - defer m.mu.Unlock() - m.dlqFailures++ -} - -func (m *mockTelemetryRecorder) getDLQMessages() int { - m.mu.RLock() - defer m.mu.RUnlock() - return m.dlqMessages -}