From 704c7b80f68ba303aaabbf5eba5176dd8f1f6643 Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 17 Nov 2025 16:25:23 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20repository=20guidelines=20and?= =?UTF-8?q?=20enhance=20documentation=20for=20project=20structure,=20build?= =?UTF-8?q?=20commands,=20coding=20standards,=20and=20testing=20practices.?= =?UTF-8?q?=20Introduce=20`AGENTS.md`=20for=20contributor=20guidance,=20up?= =?UTF-8?q?date=20`README.md`=20to=20reference=20new=20guidelines,=20and?= =?UTF-8?q?=20improve=20configuration=20documentation=20for=20NATS=20modes?= =?UTF-8?q?.=20Update=20Makefile=20and=20Taskfile=20with=20clearer=20run?= =?UTF-8?q?=20commands=20and=20requirements=20for=20development=20and=20pr?= =?UTF-8?q?oduction=20modes.=20Add=20production=20deployment=20guide=20and?= =?UTF-8?q?=20improve=20logging=20configuration=20for=20better=20observabi?= =?UTF-8?q?lity.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 23 + Makefile | 15 +- README.md | 271 +++++- Taskfile.yml | 24 +- cmd/main/main.go | 8 +- configs/config.dev.toml | 50 +- configs/config.prod.toml | 143 ++++ .../caatsm-overview.json | 89 +- configs/otel-collector.dev.yaml | 6 +- docs/dev-guide.md | 70 +- docs/prod-guide.md | 458 +++++++++++ go.mod | 2 +- internal/adapter/dto/telegram_test.go | 34 + internal/infra/log/logger_test.go | 62 ++ internal/infra/metrics/metrics_test.go | 53 ++ internal/infra/monitoring/server_test.go | 95 +++ internal/infra/nats/consumer.go | 775 ++---------------- internal/infra/nats/consumer_core.go | 52 ++ internal/infra/nats/consumer_jetstream.go | 399 +++++++++ .../infra/nats/consumer_jetstream_test.go | 97 +++ internal/infra/nats/consumer_test.go | 147 ++++ internal/infra/nats/dlq.go | 139 ++++ internal/infra/nats/dlq_test.go | 131 +++ internal/infra/nats/jetstream.go | 16 +- internal/infra/nats/message_handler.go | 88 ++ internal/infra/nats/message_handler_test.go | 62 ++ internal/infra/nats/metrics.go | 83 ++ internal/infra/nats/metrics_test.go | 57 ++ internal/infra/nats/publisher.go | 64 +- internal/infra/nats/utils.go | 80 ++ internal/infra/nats/utils_test.go | 87 ++ internal/infra/postgres/db.go | 39 +- internal/infra/postgres/db_test.go | 27 + pkg/di/wire_gen.go | 12 +- 34 files changed, 2976 insertions(+), 782 deletions(-) create mode 100644 AGENTS.md create mode 100644 configs/config.prod.toml create mode 100644 docs/prod-guide.md create mode 100644 internal/adapter/dto/telegram_test.go create mode 100644 internal/infra/log/logger_test.go create mode 100644 internal/infra/metrics/metrics_test.go create mode 100644 internal/infra/monitoring/server_test.go create mode 100644 internal/infra/nats/consumer_core.go create mode 100644 internal/infra/nats/consumer_jetstream.go create mode 100644 internal/infra/nats/consumer_jetstream_test.go create mode 100644 internal/infra/nats/consumer_test.go create mode 100644 internal/infra/nats/dlq.go create mode 100644 internal/infra/nats/dlq_test.go create mode 100644 internal/infra/nats/message_handler.go create mode 100644 internal/infra/nats/message_handler_test.go create mode 100644 internal/infra/nats/metrics.go create mode 100644 internal/infra/nats/metrics_test.go create mode 100644 internal/infra/nats/utils.go create mode 100644 internal/infra/nats/utils_test.go create mode 100644 internal/infra/postgres/db_test.go diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..3579ab3 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,23 @@ +# Repository Guidelines + +## Project Structure & Module Organization +Application entry lives in `cmd/main`, while Clean Architecture layers live under `internal` (`domain`, `app`, `adapter`, and `infra`). Shared wiring and compiled providers sit in `pkg/di`, configs in `configs/config..toml`, docs in `docs`, and reusable test fixtures in `test`. Keep new assets near the layer they extend (e.g., new parsers in `internal/adapter/parser`). + +## Build, Test, and Development Commands +- `make build` / `task build` — compile `./cmd/main` into `bin/receiver` with Wire-generated deps. +- `make run-dev` / `task run-dev` — run with `GO_ENV=dev`, respecting `configs/config.dev.toml`. +- `make lint` / `task lint` — execute `golangci-lint` with the repository config. +- `make test`, `make test-int`, `make test-all` — run Ginkgo unit suites, integration suites (`test/integration`), or both. +- `make coverage` — produce `coverage/coverage.html`; open it before merging substantial changes. + +## Coding Style & Naming Conventions +Stick to idiomatic Go: tabs for indentation, `camelCase` for locals, `CamelCase` for exported APIs, and package names that match their directory. Always run `gofmt`/`goimports` (or rely on `go fmt ./...`) before opening a PR. Generated files belong under `/pkg/di` (Wire) or the directory they serve; never hand-edit `wire_gen.go`. Linting via `golangci-lint` is required before submission. + +## Testing Guidelines +Unit specs live next to implementation files as `*_test.go` and rely on Ginkgo; keep descriptions declarative ("should parse DEP messages"). Integration suites in `test/integration` spin up NATS and TimescaleDB via Testcontainers; run them locally with Docker. Target coverage is whatever `make coverage` reports for the touched packages—raise regressions above 80% when practical. + +## Commit & Pull Request Guidelines +Follow the existing history style: optional emoji prefix + imperative summary (e.g., `✨ Add telemetry recorder`). Reference tickets in the body (`Refs #123`) and explain config or schema migrations explicitly. Pull requests must describe the change, include relevant commands/logs, attach screenshots for dashboard updates, and call out any new flags or environment variables. + +## Security & Configuration Tips +Store secrets in environment variables (`CAATSM_*`) rather than committing them. When introducing new configuration keys, update the matching `configs/config..toml` and document overrides in `README.md`. Review `docker-compose.dev.yml` before running integration tests to ensure local services are isolated from production infrastructure. diff --git a/Makefile b/Makefile index 04449cc..eba1f63 100644 --- a/Makefile +++ b/Makefile @@ -18,13 +18,18 @@ build: ## Build the receiver binary run: run-dev ## Alias for run-dev .PHONY: run-dev -run-dev: build ## Run the receiver in development mode - @echo "Running receiver in development mode..." +run-dev: build ## Run the receiver in development mode (uses core NATS mode by default) + @echo "Running receiver in development mode (NATS mode: core by default)..." @GO_ENV=dev $(BINARY) listen .PHONY: run-prod -run-prod: build ## Run the receiver in production mode +run-prod: build ## Run the receiver in production mode (requires config.prod.toml and JetStream mode) @echo "Running receiver in production mode..." + @echo "Note: Production mode requires:" + @echo " - configs/config.prod.toml file (or CAATSM_* environment variables)" + @echo " - NATS JetStream enabled (nats.mode = jetstream)" + @echo " - Stream and Consumer must exist (not auto-created in prod)" + @echo " - PostgreSQL connection configured" @GO_ENV=prod $(BINARY) listen .PHONY: run-test @@ -33,8 +38,8 @@ run-test: build ## Run the receiver in test mode @GO_ENV=test $(BINARY) listen .PHONY: run-local -run-local: ## Run receiver directly via go run - @echo "Running receiver via go run..." +run-local: ## Run receiver directly via go run (uses core NATS mode by default in dev) + @echo "Running receiver via go run (GO_ENV=$(GO_ENV), NATS mode: core by default in dev)..." @GO_ENV=$(GO_ENV) go run $(CMD) listen .PHONY: test diff --git a/README.md b/README.md index 4aa2696..85533b2 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,10 @@ This project follows Clean Architecture principles with clear separation of conc - **Structured Logging**: Zap logger with configurable levels and formats - **Batch Processing**: Efficient batch message processing and database inserts +## Contributor Guide + +For coding standards, test expectations, and release hygiene, read [AGENTS.md](AGENTS.md). + ## Prerequisites - Go 1.22+ @@ -83,12 +87,16 @@ Configuration is loaded from TOML files and environment variables. The configura ```toml [nats] url = "nats://localhost:4222" +# mode: "jetstream" (default) or "core" +# See "NATS Mode Selection" section below for detailed comparison +mode = "jetstream" stream = "TELEGRAM" consumer = "telegram-consumer" client = "serial-client" cluster = "tele-cluster" [nats.stream_limits] +# These settings only apply when mode = "jetstream" max_msgs = 100000 max_bytes = 67108864 max_age = "24h" @@ -97,6 +105,7 @@ storage = "file" replicas = 1 [nats.consumer_rules] +# These settings only apply when mode = "jetstream" max_deliver = 5 ack_wait = "30s" max_ack_pending = 1024 @@ -109,6 +118,7 @@ start_time = "" [subscription] # Optional. Defaults to "telegram.>" when omitted. topic = "telegram.serial" +# queue_group: Used in both core and jetstream modes for load balancing queue_group = "tele-queue" [publisher] @@ -150,6 +160,211 @@ export CAATSM_LOG_LEVEL="debug" Environment variable names are converted from `CAATSM_NATS_URL` to `nats.url` in the configuration. +### NATS Mode Selection + +The application supports two NATS consumption modes, controlled by `nats.mode`: + +#### JetStream Mode (Recommended for Production) + +**Configuration:** `nats.mode = "jetstream"` (default) + +**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 +- ✅ **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 +- ✅ **At-Least-Once Delivery**: Messages are guaranteed to be delivered at least once + +**Use Cases:** +- Production environments requiring message reliability +- Scenarios where message loss is unacceptable +- Systems needing message replay capabilities +- Applications requiring retry logic for transient failures + +**Configuration Requirements:** +- Requires JetStream to be enabled on the NATS server +- Stream must be created (auto-created in dev/test environments) +- Consumer configuration via `[nats.consumer_rules]` section + +**How to Use JetStream Mode:** + +1. **Prerequisites:** + - Ensure NATS server has JetStream enabled (default in `docker-compose.dev.yml`) + - Set `nats.mode = "jetstream"` in your config file (or use `CAATSM_NATS_MODE=jetstream`) + +2. **Start NATS with JetStream:** + ```bash + # Using Docker Compose (recommended for development) + docker compose -f docker-compose.dev.yml up -d nats + + # Or start NATS server manually with JetStream enabled: + # nats-server -js + ``` + +3. **Configure Stream and Consumer:** + The application automatically creates the Stream and Consumer on startup in dev/test environments (`GO_ENV=dev` or `GO_ENV=test`). In production, you may need to create them manually or ensure they exist. + + **Stream Configuration** (`[nats.stream_limits]`): + - `max_msgs`: Maximum number of messages in the stream (default: 100000) + - `max_bytes`: Maximum total size of messages (default: 64MB) + - `max_age`: Maximum age of messages before deletion (default: 24h) + - `storage`: "file" (persistent) or "memory" (ephemeral) + - `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) + - `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 + # Development mode (auto-creates stream/consumer) + GO_ENV=dev \ + CAATSM_NATS_MODE=jetstream \ + CAATSM_POSTGRES_URL=postgres://user:pass@localhost:5432/aviation \ + go run ./cmd/main listen + + # Production mode (requires stream/consumer to exist) + GO_ENV=prod \ + CAATSM_NATS_MODE=jetstream \ + ./bin/receiver listen + ``` + +5. **Publish Messages to JetStream:** + ```bash + # Using nats-box (included in docker-compose.dev.yml) + docker compose exec nats-box nats pub telegram.serial "ZCZC TEST123 150631..." + + # Or use the seed-telegrams tool with JetStream + go run ./cmd/seed-telegrams \ + --nats-url nats://localhost:4222 \ + --jetstream \ + --stream TELEGRAM \ + --js-subject telegram.serial \ + --count 10 + ``` + +6. **Monitor JetStream:** + ```bash + # View stream info + docker compose exec nats-box nats stream info TELEGRAM + + # View consumer info + docker compose exec nats-box nats consumer info TELEGRAM telegram-consumer + + # View pending messages + docker compose exec nats-box nats consumer next TELEGRAM telegram-consumer + + # Or use NATS monitoring UI at http://localhost:8222 + ``` + +7. **Message Processing Flow:** + - Messages are published to the configured subject (e.g., `telegram.serial`) + - 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 + - After `max_deliver` attempts, permanent failures are routed to DLQ (if enabled) + +8. **Replay Messages:** + ```bash + # Replay from a specific sequence + ./bin/receiver listen --replay-from seq:12345 + + # Replay from a specific time + ./bin/receiver listen --replay-from time:2024-11-15T08:00:00Z + ``` + +9. **Dead-Letter Queue (DLQ):** + Enable DLQ in config to route poison messages: + ```toml + [dlq] + enabled = true + subject = "caatsm.dlq" + ``` + Messages that fail after `max_deliver` attempts are published to the DLQ subject for manual inspection. + +10. **Troubleshooting:** + - **Stream not found**: Ensure `GO_ENV=dev` for auto-creation, or create manually in production + - **Consumer not found**: Application auto-creates consumer on startup + - **Messages not being consumed**: Check consumer info for pending messages and delivery status + - **High pending count**: Increase `batch_size` or add more consumer instances + - **Messages being redelivered**: Check processing logs for errors; adjust `ack_wait` if processing takes longer + +#### Core NATS Mode (Default for Development) + +**Configuration:** `nats.mode = "core"` (default in `config.dev.toml`) + +**Features:** +- ⚡ **Simple Pub/Sub**: Basic publish/subscribe messaging +- ⚡ **Queue Groups**: Load balancing across multiple consumers +- ⚡ **Low Latency**: No persistence overhead +- ⚡ **Fast Startup**: No stream/consumer setup required +- ❌ **No Persistence**: Messages are lost if no consumer is available +- ❌ **No ACK**: No delivery guarantees +- ❌ **No Retry**: Processing failures are logged but not retried +- ❌ **No DLQ**: Failed messages cannot be routed to a dead-letter queue + +**Use Cases:** +- **Local development** (recommended default) +- Quick testing and iteration +- Real-time monitoring/logging where message loss is acceptable +- Simple pub/sub scenarios without reliability requirements +- Performance testing without persistence overhead + +**Configuration Requirements:** +- Works with any NATS server (JetStream not required) +- Only `[subscription]` settings are used (queue_group for load balancing) +- `[nats.consumer_rules]` and `[nats.stream_limits]` are ignored + +**How to Use Core Mode:** + +1. **Start NATS Server** (JetStream not required, but can be enabled): + ```bash + # Simple NATS server + nats-server + + # Or with Docker Compose (JetStream enabled but not required for core mode) + docker compose -f docker-compose.dev.yml up -d nats + ``` + +2. **Start the Application** (Core mode is default in dev config): + ```bash + # Core mode is default, no need to specify + GO_ENV=dev \ + CAATSM_POSTGRES_URL=postgres://user:pass@localhost:5432/aviation \ + go run ./cmd/main listen + ``` + +3. **Publish Messages** (use standard NATS publish): + ```bash + # Using nats-box + docker compose exec nats-box nats pub telegram.serial "ZCZC TEST123 150631..." + + # Or use seed-telegrams without --jetstream flag + go run ./cmd/seed-telegrams \ + --nats-url nats://localhost:4222 \ + --subject telegram.serial \ + --count 10 + ``` + +**Switching Modes:** + +```bash +# Use Core NATS mode (default for development) +CAATSM_NATS_MODE=core go run ./cmd/main listen +# Or simply (core is default in config.dev.toml) +go run ./cmd/main listen + +# Use JetStream mode (for production or integration testing) +CAATSM_NATS_MODE=jetstream go run ./cmd/main listen +``` + +**Note:** The publisher always uses JetStream for deduplicated fan-out, regardless of the consumer mode. If you need pure Core NATS, ensure publishers also use Core NATS subjects. + ## Usage ### Build @@ -174,12 +389,12 @@ go build -o bin/receiver ./cmd/main ### Run +#### Development Mode + Use Make targets (binary mode): ```bash -make run-dev # GO_ENV=dev -make run-prod # GO_ENV=prod -make run-test # GO_ENV=test +make run-dev # GO_ENV=dev (uses core NATS mode by default) make run-local # go run ./cmd/main listen (honors GO_ENV) ``` @@ -187,12 +402,32 @@ Task equivalents: ```bash task run-dev -task run-prod -task run-test task run-local # go run ./cmd/main listen -task dev-run # boots docker-compose dev stack + go run +task dev-run # boots docker-compose dev stack + go run (core NATS mode) ``` +#### Production Mode + +For production deployment, see the comprehensive guide: **[Production Deployment Guide](docs/prod-guide.md)** + +Quick start: + +```bash +# Build the binary +make build + +# Run in production mode +make run-prod # GO_ENV=prod (requires config.prod.toml) +``` + +**Key requirements:** +- JetStream mode (mandatory) +- Stream and Consumer must be created manually +- Production configuration file: `configs/config.prod.toml` +- SSL/TLS for secure connections + +See `docs/prod-guide.md` for complete production deployment instructions. + ### Command Line Options ```bash @@ -295,11 +530,17 @@ docker compose -f docker-compose.dev.yml up -d postgres nats nats-box docker compose -f docker-compose.dev.yml up -d otel-collector jaeger prometheus grafana ``` -Run the processor locally while the infra runs in Docker (default mode is JetStream; switch to core only if you explicitly set `CAATSM_NATS_MODE=core`): +Run the processor locally while the infra runs in Docker: ```bash +# Core NATS mode (default for development, fast and lightweight) GO_ENV=dev \ -CAATSM_NATS_MODE=core \ +CAATSM_POSTGRES_URL=postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable \ + go run ./cmd/main listen + +# JetStream mode (for integration testing or production-like behavior) +GO_ENV=dev \ +CAATSM_NATS_MODE=jetstream \ CAATSM_POSTGRES_URL=postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable \ go run ./cmd/main listen ``` @@ -308,8 +549,13 @@ Tear everything down with `docker compose -f docker-compose.dev.yml down -v`. ## Deployment Examples -- `docs/deploy-systemd.md` shows a minimal systemd unit that wires configuration via environment files and restarts on failure. -- `docs/deploy-k8s.md` provides a reference Deployment + ConfigMap/Secret with liveness/readiness probes hitting `/healthz` and `/metrics`. +For production deployment, see the comprehensive guide: **[Production Deployment Guide](docs/prod-guide.md)** + +Additional deployment-specific guides: + +- **`docs/prod-guide.md`** - Complete production deployment guide with configuration, setup, and operations +- **`docs/deploy-systemd.md`** - Systemd service deployment with environment file configuration +- **`docs/deploy-k8s.md`** - Kubernetes deployment with ConfigMap/Secret and health probes ### Project Structure @@ -355,7 +601,9 @@ The project keeps tests close to the code that they exercise: ## Message Flow -1. **NATS Consumer** receives raw telegram messages from NATS (JetStream durable pull by default; plain `nc.Subscribe` only when you opt into `nats.mode=core`) +1. **NATS Consumer** receives raw telegram messages from NATS: + - **JetStream mode** (default): Uses durable pull consumer with batch processing, ACK/NAK, and retry logic + - **Core mode**: Uses `QueueSubscribe` for simple pub/sub with queue group load balancing (no persistence or retries) 2. **MessageProcessor** orchestrates the processing: - Parses the message using the Parser adapter - Stores the parsed message in PostgreSQL via Repository @@ -769,4 +1017,3 @@ This repository has not declared a public license yet. ## Contributing Contribution guidelines are not published; please coordinate changes via pull requests or direct maintainers. - diff --git a/Taskfile.yml b/Taskfile.yml index c5913f5..c1fe6fa 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -24,19 +24,26 @@ tasks: - task: run-dev run-dev: - desc: Run the receiver in development mode + desc: Run the receiver in development mode (uses core NATS mode by default) deps: - build cmds: - - echo "Running receiver in development mode..." + - | + echo "Running receiver in development mode (NATS mode: core by default)..." - GO_ENV=dev {{.binary}} listen run-prod: - desc: Run the receiver in production mode + desc: Run the receiver in production mode (requires config.prod.toml and JetStream mode) deps: - build cmds: - - echo "Running receiver in production mode..." + - | + echo "Running receiver in production mode..." + echo "Note: Production mode requires:" + echo " - configs/config.prod.toml file (or CAATSM_* environment variables)" + echo " - NATS JetStream enabled (nats.mode = jetstream)" + echo " - Stream and Consumer must exist (not auto-created in prod)" + echo " - PostgreSQL connection configured" - GO_ENV=prod {{.binary}} listen run-test: @@ -48,10 +55,11 @@ tasks: - GO_ENV=test {{.binary}} listen run-local: - desc: Run receiver via go run (default GO_ENV=dev) + desc: Run receiver via go run (default GO_ENV=dev, uses core NATS mode by default) cmds: - | - echo "Running receiver via go run (GO_ENV=${GO_ENV:-dev})..." + echo "Running receiver via go run (GO_ENV=${GO_ENV:-dev}, NATS mode: core by default in dev)..." + - | GO_ENV=${GO_ENV:-dev} go run {{.cmd}} listen test: @@ -177,7 +185,7 @@ tasks: - up env: CAATSM_NATS_URL: nats://localhost:4222 - CAATSM_NATS_MODE: jetstream + CAATSM_NATS_MODE: core CAATSM_POSTGRES_URL: postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable CAATSM_TELEMETRY_ENABLED: "true" CAATSM_TELEMETRY_ENDPOINT: localhost:4318 @@ -195,7 +203,7 @@ tasks: go run ./cmd/main listen seed: - desc: Generate sample telegrams (publish to NATS) + desc: Generate sample telegrams (publishes to Core NATS by default, compatible with dev mode) cmds: - | GO_ENV=dev \ diff --git a/cmd/main/main.go b/cmd/main/main.go index 8fd8223..5305807 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -13,6 +13,8 @@ import ( "syscall" "time" + "go.uber.org/zap" + "github.com/urfave/cli/v2" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -181,7 +183,8 @@ func runListen(parentCtx context.Context, cfg *config.Config) error { // Wait for shutdown signal or consumer error select { case <-ctx.Done(): - fmt.Printf("Received shutdown signal: %v, shutting down...\n", ctx.Err()) + zap.L().Info("Received shutdown signal, shutting down", + zap.Error(ctx.Err())) case err := <-errChan: if err != nil { runErr = err @@ -198,7 +201,8 @@ func runListen(parentCtx context.Context, cfg *config.Config) error { runErr = err } case <-time.After(waitTimeout): - fmt.Printf("Timed out waiting for consumer shutdown after %s\n", waitTimeout) + zap.L().Warn("Timed out waiting for consumer shutdown", + zap.Duration("timeout", waitTimeout)) } if err := consumer.Shutdown(context.Background()); err != nil { diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 88df636..f31b8af 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -1,5 +1,20 @@ [nats] url = "nats://localhost:4222" +# mode: "core" (default for dev) or "jetstream" (recommended for production) +# - "core": Uses Core NATS for simple pub/sub messaging with: +# * No message persistence (messages lost if consumer offline) +# * No ACK mechanism (fire-and-forget delivery) +# * No automatic retry on processing failures +# * Queue groups for load balancing only +# * Suitable for development/testing or real-time scenarios where message loss is acceptable +# * Recommended for local development and testing +# - "jetstream": Uses NATS JetStream for persistent message streaming with: +# * Message persistence and replay capability +# * ACK/NAK mechanism for guaranteed delivery +# * Automatic retry with configurable backoff +# * Dead-letter queue (DLQ) support +# * Batch processing and consumer monitoring +# * Recommended for production environments mode = "core" client = "serial-client" cluster = "tele-cluster" @@ -7,21 +22,45 @@ stream = "TELEGRAM" consumer = "telegram-consumer" [nats.stream_limits] +# Stream retention and storage limits (only applies when mode = "jetstream") +# max_msgs: Maximum number of messages to keep in the stream (0 = unlimited) max_msgs = 100000 +# max_bytes: Maximum total size of messages in bytes (0 = unlimited, 67108864 = 64MB) max_bytes = 67108864 +# max_age: Maximum age of messages before automatic deletion (e.g., "24h", "7d") max_age = "24h" +# discard: What to do when limits are reached: "old" (delete oldest) or "new" (reject new) discard = "old" +# storage: "file" (persistent to disk) or "memory" (ephemeral, faster but lost on restart) storage = "file" +# replicas: Number of stream replicas for high availability (1 = single node, 3+ for production cluster) replicas = 1 [nats.consumer_rules] +# Consumer delivery and retry rules (only applies when mode = "jetstream") +# max_deliver: Maximum number of delivery attempts before giving up (0 = unlimited) max_deliver = 5 +# 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 = "" [subscription] @@ -43,8 +82,12 @@ max_conns = 10 min_conns = 2 [app] +# Batch processing configuration (applies to both core and jetstream modes) +# batch_size: Number of messages to fetch in each batch (JetStream pull batch size) batch_size = 50 +# batch_timeout: Maximum time to wait when fetching a batch (e.g., "2s") batch_timeout = "2s" +# monitor_interval: How often to emit consumer statistics and metrics monitor_interval = "30s" [log] @@ -85,5 +128,8 @@ write_timeout = "5s" health_timeout = "2s" [dlq] -enabled = true -subject = "caatsm.dlq" \ No newline at end of file +# Dead-Letter Queue configuration (only applies when mode = "jetstream") +# enabled: Enable DLQ routing for poison messages (messages that fail after max_deliver attempts) +enabled = false # Set to true when switching to JetStream mode +# subject: NATS subject where failed messages will be published for manual inspection +subject = "caatsm.dlq" diff --git a/configs/config.prod.toml b/configs/config.prod.toml new file mode 100644 index 0000000..4f91a25 --- /dev/null +++ b/configs/config.prod.toml @@ -0,0 +1,143 @@ +# Production Configuration for CAATSM +# +# This configuration is optimized for production environments. +# Key differences from dev config: +# - Uses JetStream mode (required for production) +# - Higher resource limits and connection pools +# - JSON logging (for log aggregation) +# - SSL/TLS enabled for secure connections +# - Higher stream replicas for HA (3+) +# - Longer retention periods +# +# IMPORTANT: Stream and Consumer must be created manually in production. +# The application does NOT auto-create them in production mode. + +[nats] +url = "nats://nats.prod:4222" +# Production MUST use JetStream mode for message reliability +mode = "jetstream" +client = "caatsm-prod-client" +cluster = "prod-cluster" +stream = "TELEGRAM" +consumer = "telegram-consumer" + +[nats.stream_limits] +# Production stream limits - adjust based on your requirements +# max_msgs: Maximum number of messages to keep in the stream (0 = unlimited) +max_msgs = 1000000 +# max_bytes: Maximum total size of messages in bytes (1GB = 1073741824) +max_bytes = 1073741824 +# max_age: Maximum age of messages before automatic deletion (7 days) +max_age = "168h" +# discard: What to do when limits are reached: "old" (delete oldest) or "new" (reject new) +discard = "old" +# storage: "file" (persistent to disk) - REQUIRED for production +storage = "file" +# replicas: Number of stream replicas for high availability (3+ for production cluster) +replicas = 3 + +[nats.consumer_rules] +# Consumer delivery and retry rules for production +# max_deliver: Maximum number of delivery attempts before giving up +max_deliver = 5 +# 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 = "" + +[subscription] +topic = "telegram.serial" +queue_group = "tele-queue" + +[publisher] +topic = "telegram.json" + +[timeouts] +server = "10s" +reconnect_wait = "5s" +close = "30s" +ack_wait = "30s" + +[postgres] +# Production PostgreSQL connection - USE SSL/TLS +# Replace with your production database URL +url = "postgres://user:password@db.prod:5432/aviation?sslmode=require" +# Higher connection pool for production workloads +max_conns = 20 +min_conns = 5 + +[app] +# Production batch processing configuration +# batch_size: Larger batch size for better throughput +batch_size = 100 +# batch_timeout: Maximum time to wait when fetching a batch +batch_timeout = "2s" +# monitor_interval: How often to emit consumer statistics and metrics +monitor_interval = "30s" + +[log] +# Production logging configuration +# level: Use "info" or "warn" in production (avoid "debug") +level = "info" +# format: "json" for log aggregation systems (ELK, Loki, etc.) +format = "json" +# output: Only stdout in production (let container/logging system handle file rotation) +output = ["stdout"] +# file: Not used in production (logging to stdout) +# file = "logs/caatsm.log" + +# File rotation settings (not used when output = ["stdout"]) +# max_size = 100 # MB +# max_backups = 7 # Keep 7 rotated files +# max_age = 30 # Keep logs for 30 days +# compress = true # Compress old log files + +# Advanced options +disable_caller = false +disable_stacktrace = false +development = false + +# Sampling configuration (optional, for high-volume scenarios) +# [log.sampling] +# initial = 100 # Log first 100 messages +# thereafter = 100 # Then log every 100th message +# tick = "1s" # Per second + +[telemetry] +# Production telemetry configuration +enabled = true +# Replace with your production OTLP collector endpoint +endpoint = "otel-collector.prod:4318" +# Use TLS in production (set to false) +insecure = false + +[monitoring] +# Production monitoring configuration +disabled = false +addr = ":2112" +enable_metrics = true +enable_health = true +read_timeout = "5s" +write_timeout = "5s" +health_timeout = "2s" + +[dlq] +# Dead-Letter Queue configuration (REQUIRED for production) +# enabled: Enable DLQ routing for poison messages +enabled = true +# subject: NATS subject where failed messages will be published for manual inspection +subject = "caatsm.dlq" + diff --git a/configs/grafana-dashboards.dev/caatsm-overview.json b/configs/grafana-dashboards.dev/caatsm-overview.json index 0ff49a0..91a2855 100644 --- a/configs/grafana-dashboards.dev/caatsm-overview.json +++ b/configs/grafana-dashboards.dev/caatsm-overview.json @@ -267,7 +267,7 @@ "type": "prometheus", "uid": "prometheus-dev" }, - "gridPos": { "h": 7, "w": 24, "x": 0, "y": 26 }, + "gridPos": { "h": 7, "w": 12, "x": 0, "y": 26 }, "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, @@ -287,6 +287,93 @@ "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/configs/otel-collector.dev.yaml b/configs/otel-collector.dev.yaml index 6ce9efd..387f108 100644 --- a/configs/otel-collector.dev.yaml +++ b/configs/otel-collector.dev.yaml @@ -13,6 +13,10 @@ exporters: endpoint: jaeger:4317 tls: insecure: true + prometheus: + endpoint: "0.0.0.0:8888" + const_labels: + source: "otel-collector" service: pipelines: @@ -21,5 +25,5 @@ service: exporters: [logging, otlp/jaeger] metrics: receivers: [otlp] - exporters: [logging] + exporters: [logging, prometheus] diff --git a/docs/dev-guide.md b/docs/dev-guide.md index 1001a89..ff424d6 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -10,7 +10,11 @@ Spin up PostgreSQL/TimescaleDB and NATS JetStream in the background: docker compose -f docker-compose.dev.yml up -d postgres nats nats-box ``` -> Development mode defaults to `nats.mode = "core"`, so the processor consumes directly from the configured subject (`subscription.topic`). **However, the publisher always targets JetStream for deduplicated fan-out, so the provided Taskfile (and most examples below) override the mode to `jetstream`.** If you truly need core mode, set `CAATSM_NATS_MODE=core` manually and ensure any publishers use core subjects. +> **NATS Mode Selection:** The application supports two consumption modes: +> - **Core NATS mode** (default for development): Simple pub/sub without persistence or retry mechanisms. Recommended for local development and testing where message loss is acceptable. Fast and lightweight. +> - **JetStream mode**: Provides message persistence, ACK/NAK, automatic retries, and DLQ support. Recommended for production environments and integration testing. +> +> Development mode defaults to `nats.mode = "core"` in `config.dev.toml`. To use JetStream in development, set `CAATSM_NATS_MODE=jetstream` or change the config file. **Note:** The publisher always targets JetStream for deduplicated fan-out, so if you use Core mode for consumption, ensure your publishers align with your messaging strategy. See the README.md "NATS Mode Selection" section for a detailed comparison. - `postgres` seeds the `aviation` schema using `internal/infra/postgres/telegrams.ddl` and exposes port `5432`. - `nats` enables JetStream with client port `4222` and monitoring/UI on `8222`. @@ -55,7 +59,7 @@ docker compose -f docker-compose.dev.yml up -d postgres nats nats-box The `Taskfile.yml` includes helper targets that wrap the commands above: - `task up` – starts PostgreSQL, NATS (JetStream, toolbox, and Prometheus exporter), and the observability stack (OpenTelemetry Collector, Jaeger, Prometheus, Grafana) using Docker Compose. -- `task dev-run` – ensures `task up` has run, exports the necessary `CAATSM_*` environment variables (including `CAATSM_NATS_MODE=jetstream`), and executes `go run ./cmd/main listen` with telemetry enabled. +- `task dev-run` – ensures `task up` has run, exports the necessary `CAATSM_*` environment variables (defaults to `CAATSM_NATS_MODE=core` for development), and executes `go run ./cmd/main listen` with telemetry enabled. - `task down` – stops the entire stack and removes containers/volumes. Use these tasks if you prefer a one-command workflow instead of invoking `docker compose` and environment exports manually. @@ -64,10 +68,29 @@ Use these tasks if you prefer a one-command workflow instead of invoking `docker Use the helper CLI in `cmd/seed-telegrams` to push realistic payloads onto NATS (mirrors the fixtures in `internal/adapter/parser/aviation_parser_test.go`): +### Publishing to JetStream (Recommended) + +When using JetStream mode, publish messages to the JetStream stream: + ```bash -# Insert rows into aviation.telegrams_raw and publish to NATS simultaneously +# Publish to JetStream stream (messages are persisted) +GO_ENV=dev go run ./cmd/seed-telegrams \ + --nats-url nats://127.0.0.1:4222 \ + --jetstream \ + --stream TELEGRAM \ + --js-subject telegram.serial \ + --count 20 \ + --category mixed \ + --status random +``` + +### Publishing to Core NATS + +For Core NATS mode, use standard publish: + +```bash +# Publish to Core NATS (no persistence) GO_ENV=dev go run ./cmd/seed-telegrams \ - --postgres-url postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable \ --nats-url nats://127.0.0.1:4222 \ --subject telegram.serial \ --count 20 \ @@ -75,13 +98,29 @@ GO_ENV=dev go run ./cmd/seed-telegrams \ --status random ``` -- `--postgres-url` controls database insertion (omit to skip DB writes); metadata lands in `aviation.telegrams_raw.metadata`. -- `--dry-run` prints telegrams without touching NATS/Postgres. -- `--category` chooses ARR/DEP/CNL/DLA/FPL or `mixed`. -- `--status` controls stored/published status (`parsed|header_error|body_error|publish_error|repository_error|random`). -- `--no-nats` disables publishing; `--jetstream`, `--stream`, `--js-subject` toggle JetStream publishing. -- Inspect deliveries with `docker compose exec nats-box nats sub 'telegram.>'`. -- When running in core mode (default), the seeder publishes via standard `nc.Publish` and sets `Nats-Msg-Id` headers so the processor can derive message IDs. +### Common Options + +- `--postgres-url`: Insert rows into `aviation.telegrams_raw` (omit to skip DB writes) +- `--dry-run`: Print telegrams without publishing to NATS/Postgres +- `--category`: Choose message type (`ARR|DEP|CNL|DLA|FPL|mixed`) +- `--status`: Control stored/published status (`parsed|header_error|body_error|publish_error|repository_error|random`) +- `--no-nats`: Disable publishing to NATS +- `--jetstream`: Enable JetStream publishing (requires `--stream` and `--js-subject`) +- `--stream`: JetStream stream name (default: `TELEGRAM`) +- `--js-subject`: Subject within the JetStream stream + +### Inspecting Messages + +```bash +# View messages in JetStream stream +docker compose exec nats-box nats stream view TELEGRAM + +# Subscribe to messages (Core NATS or JetStream) +docker compose exec nats-box nats sub 'telegram.>' + +# View consumer status and pending messages +docker compose exec nats-box nats consumer info TELEGRAM telegram-consumer +``` The main processor keeps consuming `subscription.topic` (defaults to `telegram.>`). Use the seeder to simulate parser failures, publish errors, or replay raw telegrams directly from the database. @@ -96,6 +135,15 @@ The main processor keeps consuming `subscription.topic` (defaults to `telegram.> 2. **Run the processor with telemetry enabled** ```bash + # Using Core NATS mode (default for development) + CAATSM_TELEMETRY_ENABLED=true \ + CAATSM_TELEMETRY_ENDPOINT=localhost:4318 \ + CAATSM_TELEMETRY_INSECURE=true \ + GO_ENV=dev \ + CAATSM_POSTGRES_URL=postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=disable \ + go run ./cmd/main listen + + # Or use JetStream mode for integration testing CAATSM_TELEMETRY_ENABLED=true \ CAATSM_TELEMETRY_ENDPOINT=localhost:4318 \ CAATSM_TELEMETRY_INSECURE=true \ diff --git a/docs/prod-guide.md b/docs/prod-guide.md new file mode 100644 index 0000000..e417266 --- /dev/null +++ b/docs/prod-guide.md @@ -0,0 +1,458 @@ +# Production Deployment Guide + +This guide covers deploying and running the CAATSM application in production environments. + +## Overview + +Production deployments require: +- **JetStream mode** (mandatory) - for message reliability and persistence +- **Manual Stream/Consumer creation** - not auto-created in production +- **Proper configuration** - using `configs/config.prod.toml` +- **SSL/TLS connections** - for secure database and telemetry connections +- **High availability** - stream replicas set to 3+ for HA + +## Prerequisites + +Before deploying to production: + +1. **Build the binary:** + ```bash + make build + # Binary will be at bin/receiver + ``` + +2. **NATS JetStream cluster** - must be running and accessible +3. **PostgreSQL/TimescaleDB** - database must be accessible with SSL +4. **OpenTelemetry collector** (optional) - for observability +5. **Prometheus** (optional) - for metrics scraping + +## Configuration + +### 1. Production Configuration File + +The repository includes a production configuration template at `configs/config.prod.toml`. Copy and customize it: + +```bash +cp configs/config.prod.toml configs/config.prod.toml.local +# Edit config.prod.toml.local with your production values +``` + +**Key configuration sections:** + +```toml +[nats] +url = "nats://nats.prod:4222" +mode = "jetstream" # Production MUST use JetStream +stream = "TELEGRAM" +consumer = "telegram-consumer" + +[nats.stream_limits] +max_msgs = 1000000 # Adjust based on requirements +max_bytes = 1073741824 # 1GB +max_age = "168h" # 7 days +discard = "old" +storage = "file" +replicas = 3 # Use 3+ for HA in production + +[nats.consumer_rules] +max_deliver = 5 +ack_wait = "30s" +max_ack_pending = 1024 +deliver_policy = "new" # Start from new messages in production +replay_policy = "instant" +backoff = ["5s", "30s", "2m", "5m"] + +[subscription] +topic = "telegram.serial" +queue_group = "tele-queue" + +[postgres] +url = "postgres://user:password@db.prod:5432/aviation?sslmode=require" +max_conns = 20 +min_conns = 5 + +[app] +batch_size = 100 +batch_timeout = "2s" +monitor_interval = "30s" + +[log] +level = "info" +format = "json" +output = ["stdout"] + +[telemetry] +enabled = true +endpoint = "otel-collector.prod:4318" +insecure = false # Use TLS in production + +[monitoring] +disabled = false +addr = ":2112" +enable_metrics = true +enable_health = true + +[dlq] +enabled = true +subject = "caatsm.dlq" +``` + +### 2. Environment Variables + +Alternatively, you can use environment variables instead of a config file: + +```bash +export GO_ENV=prod +export CAATSM_NATS_URL=nats://nats.prod:4222 +export CAATSM_NATS_MODE=jetstream +export CAATSM_POSTGRES_URL=postgres://user:pass@db:5432/aviation?sslmode=require +export CAATSM_LOG_LEVEL=info +export CAATSM_TELEMETRY_ENABLED=true +export CAATSM_TELEMETRY_ENDPOINT=otel-collector.prod:4318 +export CAATSM_TELEMETRY_INSECURE=false +``` + +## JetStream Setup + +### Pre-create Stream and Consumer + +**IMPORTANT:** The application does NOT auto-create streams/consumers in production. You must create them manually before starting the application. + +#### Using NATS CLI + +```bash +# Create stream +nats stream add TELEGRAM \ + --subjects "telegram.serial,telegram.json" \ + --storage file \ + --replicas 3 \ + --max-msgs 1000000 \ + --max-bytes 1GB \ + --max-age 7d \ + --discard old + +# Create consumer +nats consumer add TELEGRAM telegram-consumer \ + --filter "telegram.serial" \ + --ack explicit \ + --deliver new \ + --max-deliver 5 \ + --ack-wait 30s \ + --max-pending 1024 +``` + +#### Using NATS Management API + +You can also create streams/consumers programmatically using the NATS management API or configuration files. + +### Verify Setup + +```bash +# Check stream exists +nats stream info TELEGRAM + +# Check consumer exists +nats consumer info TELEGRAM telegram-consumer + +# Test connection +nats pub telegram.serial "ZCZC TEST 150631..." +``` + +## Running the Application + +### Using Make + +```bash +make run-prod # GO_ENV=prod (requires config.prod.toml) +``` + +### Using Task + +```bash +task run-prod # GO_ENV=prod (requires config.prod.toml) +``` + +### Direct Execution + +```bash +# Using binary with config file +GO_ENV=prod ./bin/receiver listen + +# Or with environment variables (no config file needed) +GO_ENV=prod \ +CAATSM_NATS_URL=nats://nats.prod:4222 \ +CAATSM_NATS_MODE=jetstream \ +CAATSM_POSTGRES_URL=postgres://user:pass@db:5432/aviation?sslmode=require \ +./bin/receiver listen +``` + +## Production Checklist + +Before deploying to production, verify: + +- ✅ `nats.mode = "jetstream"` in config (mandatory) +- ✅ Stream and Consumer created manually +- ✅ Stream replicas set to 3+ for high availability +- ✅ PostgreSQL connection configured with SSL (`sslmode=require`) +- ✅ Log level set to `info` or `warn` (not `debug`) +- ✅ Log format set to `json` for log aggregation +- ✅ Telemetry endpoint configured (if using observability) +- ✅ Telemetry TLS enabled (`insecure = false`) +- ✅ Monitoring endpoints exposed for Prometheus scraping +- ✅ DLQ enabled for poison message handling +- ✅ Appropriate retention limits configured (max_msgs, max_bytes, max_age) +- ✅ Connection pool sizes appropriate for workload +- ✅ Batch sizes tuned for throughput + +## Monitoring and Observability + +### Health Endpoints + +The application exposes health and readiness endpoints: + +- `GET /livez` - Liveness endpoint (process health) +- `GET /readyz` - Readiness endpoint (checks PostgreSQL and NATS) +- `GET /healthz` - Alias for `/readyz` +- `GET /metrics` - Prometheus metrics + +Configure Prometheus to scrape metrics: + +```yaml +scrape_configs: + - job_name: 'caatsm' + static_configs: + - targets: ['caatsm:2112'] +``` + +### Key Metrics + +Monitor these metrics in production: + +- `caatsm_messages_total{stream,consumer,result}` - Message throughput and results +- `caatsm_handle_latency_seconds_bucket` - Processing latency +- `caatsm_retries_total{stream,consumer,reason}` - Retry counts +- `caatsm_db_queries_total{operation,result}` - Database activity +- `caatsm_dlq_messages_total` - Dead-letter queue messages +- `caatsm_nats_consumer_pending_messages` - Consumer backlog/lag + +### Logging + +Production logs are in JSON format for easy parsing by log aggregation systems: + +```json +{ + "level": "info", + "ts": 1234567890.123, + "caller": "nats/consumer.go:123", + "msg": "Started consuming messages", + "subject": "telegram.serial", + "consumer": "telegram-consumer", + "stream": "TELEGRAM" +} +``` + +## High Availability + +### Multiple Instances + +Run multiple instances of the application for high availability: + +- All instances use the same durable consumer name +- JetStream distributes messages across instances +- Each instance independently fetches messages +- If an instance fails, others continue processing + +### Stream Replication + +Configure stream with 3+ replicas for HA: + +```toml +[nats.stream_limits] +replicas = 3 # Minimum 3 for HA, 5 for better distribution +``` + +### Database Connection Pooling + +Configure appropriate connection pool sizes: + +```toml +[postgres] +max_conns = 20 # Adjust based on number of instances +min_conns = 5 +``` + +## Troubleshooting + +### Stream Not Found + +**Error:** `stream TELEGRAM not found` + +**Solution:** Create the stream manually before starting the application (see "JetStream Setup" above). + +### Consumer Not Found + +**Error:** `consumer telegram-consumer not found in stream TELEGRAM` + +**Solution:** Create the consumer manually before starting the application (see "JetStream Setup" above). + +### Messages Not Being Consumed + +**Symptoms:** High pending count, no messages processed + +**Check:** +1. Verify consumer exists: `nats consumer info TELEGRAM telegram-consumer` +2. Check pending messages: `nats consumer next TELEGRAM telegram-consumer` +3. Verify application is running and connected +4. Check logs for errors + +**Solutions:** +- Increase `batch_size` if processing is slow +- Add more consumer instances +- Check for processing errors in logs + +### High Pending Count + +**Symptoms:** Consumer has many pending messages + +**Solutions:** +- Increase `batch_size` in config +- Add more application instances +- Check processing latency +- Verify database performance + +### Messages Being Redelivered + +**Symptoms:** Same messages processed multiple times + +**Check:** +- Processing logs for errors +- `ack_wait` timeout may be too short +- Processing may be taking longer than `ack_wait` + +**Solutions:** +- Increase `ack_wait` if processing takes longer +- Fix processing errors +- Check database connection and performance + +### Connection Issues + +**NATS Connection:** +- Verify NATS server is accessible +- Check network connectivity +- Verify NATS URL in config + +**PostgreSQL Connection:** +- Verify database is accessible +- Check SSL certificate configuration +- Verify connection string format +- Check firewall rules + +## Deployment Options + +### Systemd Deployment + +See `docs/deploy-systemd.md` for a complete systemd service deployment example. + +### Kubernetes Deployment + +See `docs/deploy-k8s.md` for Kubernetes deployment with ConfigMap/Secret and health probes. + +## Performance Tuning + +### Batch Processing + +Adjust batch size based on message size and processing time: + +```toml +[app] +batch_size = 100 # Increase for higher throughput +batch_timeout = "2s" # Adjust based on latency requirements +``` + +### Connection Pools + +Tune database connection pool: + +```toml +[postgres] +max_conns = 20 # Total connections across all instances +min_conns = 5 # Keep-alive connections +``` + +### Stream Retention + +Configure retention based on requirements: + +```toml +[nats.stream_limits] +max_msgs = 1000000 # Maximum messages +max_bytes = 1073741824 # Maximum size (1GB) +max_age = "168h" # Maximum age (7 days) +``` + +### Consumer Settings + +Tune consumer for your workload: + +```toml +[nats.consumer_rules] +max_ack_pending = 1024 # Increase for higher throughput +ack_wait = "30s" # Adjust based on processing time +backoff = ["5s", "30s", "2m", "5m"] # Retry delays +``` + +## Security Considerations + +1. **Use SSL/TLS** for all connections: + - PostgreSQL: `sslmode=require` + - Telemetry: `insecure = false` + +2. **Secure secrets** - Use environment variables or secret management: + - Database passwords + - NATS credentials + - API keys + +3. **Network security**: + - Use private networks for internal services + - Restrict access to monitoring endpoints + - Use firewall rules appropriately + +4. **Logging** - Avoid logging sensitive data: + - Don't log message payloads in production + - Use appropriate log levels + +## Backup and Recovery + +### Database Backups + +Ensure regular backups of PostgreSQL/TimescaleDB: +- Use pg_dump or TimescaleDB backup tools +- Test restore procedures regularly + +### JetStream State + +JetStream state is stored in NATS: +- Ensure NATS cluster has proper backup procedures +- Stream data is replicated across cluster nodes +- Test disaster recovery procedures + +### Message Replay + +If needed, messages can be replayed from JetStream: + +```bash +# Replay from a specific sequence +./bin/receiver listen --replay-from seq:12345 + +# Replay from a specific time +./bin/receiver listen --replay-from time:2024-11-15T08:00:00Z +``` + +## Support + +For issues or questions: +- Check logs: `journalctl -u caatsm` (systemd) or container logs +- Review metrics in Prometheus/Grafana +- Check health endpoints: `curl http://localhost:2112/readyz` +- Consult deployment-specific documentation + diff --git a/go.mod b/go.mod index 879cd19..cc169a2 100644 --- a/go.mod +++ b/go.mod @@ -24,6 +24,7 @@ require ( go.opentelemetry.io/otel/sdk/metric v1.38.0 go.opentelemetry.io/otel/trace v1.38.0 go.uber.org/zap v1.27.0 + gopkg.in/natefinch/lumberjack.v2 v2.2.1 ) require ( @@ -107,5 +108,4 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect google.golang.org/grpc v1.76.0 // indirect google.golang.org/protobuf v1.36.10 // indirect - gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect ) diff --git a/internal/adapter/dto/telegram_test.go b/internal/adapter/dto/telegram_test.go new file mode 100644 index 0000000..de91b8f --- /dev/null +++ b/internal/adapter/dto/telegram_test.go @@ -0,0 +1,34 @@ +package dto + +import "testing" + +func TestNewParsedTelegramDefaults(t *testing.T) { + pt := NewParsedTelegram() + + if pt == nil { + t.Fatal("expected NewParsedTelegram to return a non-nil pointer") + } + + if pt.Parsed { + t.Fatalf("expected Parsed to be false, got %v", pt.Parsed) + } + + if pt.Status != MessageStatusUnknown { + t.Fatalf("expected default status %q, got %q", MessageStatusUnknown, pt.Status) + } +} + +func TestMessageStatusStringValues(t *testing.T) { + cases := map[MessageStatus]string{ + MessageStatusUnknown: "unknown", + MessageStatusParsed: "parsed", + MessageStatusHeaderError: "header_error", + MessageStatusBodyError: "body_error", + } + + for status, want := range cases { + if string(status) != want { + t.Fatalf("expected %q for status %v, got %q", want, status, status) + } + } +} diff --git a/internal/infra/log/logger_test.go b/internal/infra/log/logger_test.go new file mode 100644 index 0000000..a30f06a --- /dev/null +++ b/internal/infra/log/logger_test.go @@ -0,0 +1,62 @@ +package log + +import ( + "os" + "path/filepath" + "testing" + + configpkg "caatsm/internal/infra/config" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestLog(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Log Suite") +} + +var _ = Describe("ProvideLogger", func() { + Context("when running in development console mode", func() { + It("returns a functional logger", func() { + cfg := &configpkg.Config{ + Log: configpkg.LogConfig{ + Level: "debug", + Format: "console", + Development: true, + Output: []string{"stdout"}, + }, + } + + logger, err := ProvideLogger(cfg) + Expect(err).NotTo(HaveOccurred()) + Expect(logger).ToNot(BeNil()) + }) + }) + + Context("when file output is configured", func() { + It("creates the directory before writing logs", func() { + tmpDir := filepath.Join(os.TempDir(), "caatsm-log-test") + defer os.RemoveAll(tmpDir) + logPath := filepath.Join(tmpDir, "child", "app.log") + cfg := &configpkg.Config{ + Log: configpkg.LogConfig{ + Level: "info", + Format: "json", + Development: false, + Output: []string{"file"}, + File: logPath, + MaxSize: 1, + MaxBackups: 1, + MaxAge: 1, + Compress: false, + }, + } + + logger, err := ProvideLogger(cfg) + Expect(err).NotTo(HaveOccurred()) + Expect(logger).ToNot(BeNil()) + Expect(filepath.Dir(logPath)).To(BeADirectory()) + }) + }) +}) diff --git a/internal/infra/metrics/metrics_test.go b/internal/infra/metrics/metrics_test.go new file mode 100644 index 0000000..dc1278e --- /dev/null +++ b/internal/infra/metrics/metrics_test.go @@ -0,0 +1,53 @@ +package metrics + +import ( + "net/http" + "net/http/httptest" + "testing" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestMetrics(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Metrics Suite") +} + +var _ = Describe("Metrics", func() { + Describe("Handler", func() { + It("serves metrics with the correct content type", func() { + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + resp := httptest.NewRecorder() + + Handler().ServeHTTP(resp, req) + + Expect(resp.Code).To(Equal(http.StatusOK)) + Expect(resp.Header().Get("Content-Type")).To(ContainSubstring("text/plain")) + }) + }) + + Describe("labelValue helper", func() { + It("returns unknown for empty values and lowercases input", func() { + Expect(labelValue(" ")).To(Equal("unknown")) + Expect(labelValue("SOME_VALUE")).To(Equal("some_value")) + }) + }) + + Describe("record helpers", func() { + It("can be invoked without panicking", func() { + Expect(func() { + RecordProcessed("parsed", "ARR", 150*time.Millisecond) + RecordFailure("parser") + RecordMessageHandled("TEST", "consumer", ResultOK, 205*time.Millisecond) + RecordRetry("TEST", "consumer", RetryReasonProcessorError) + RecordDLQMessage("TEST", "consumer") + RecordDLQPublishFailure("TEST", "consumer") + RecordDBQuery("insert", DBResultOK, 10*time.Millisecond) + RecordJSAPICall("publish") + RecordNATSConsumerPending("TEST", "consumer", 7) + }).NotTo(Panic()) + }) + }) +}) diff --git a/internal/infra/monitoring/server_test.go b/internal/infra/monitoring/server_test.go new file mode 100644 index 0000000..ca505ea --- /dev/null +++ b/internal/infra/monitoring/server_test.go @@ -0,0 +1,95 @@ +package monitoring + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "caatsm/internal/infra/buildinfo" + "caatsm/internal/infra/config" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "go.uber.org/zap" +) + +func TestMonitoring(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Monitoring Suite") +} + +var _ = Describe("Server", func() { + Context("when monitoring is disabled", func() { + It("returns nil from ProvideServer", func() { + cfg := &config.Config{ + Monitoring: config.MonitoringConfig{ + Disabled: true, + Addr: ":0", + }, + } + + server, err := ProvideServer(cfg, zap.NewNop(), nil, nil) + Expect(err).NotTo(HaveOccurred()) + Expect(server).To(BeNil()) + }) + }) + + Context("health and metrics handlers", func() { + BeforeEach(func() { + buildinfo.Version = "v-test" + buildinfo.Commit = "abc" + buildinfo.BuiltAt = "now" + }) + + It("returns build info on /livez", func() { + server := &Server{ + cfg: config.MonitoringConfig{}, + logger: zap.NewNop(), + } + + req := httptest.NewRequest(http.MethodGet, "/livez", nil) + resp := httptest.NewRecorder() + + server.handleLive(resp, req) + + Expect(resp.Code).To(Equal(http.StatusOK)) + + payload := map[string]interface{}{} + Expect(json.NewDecoder(resp.Body).Decode(&payload)).To(Succeed()) + Expect(payload["status"]).To(Equal("ok")) + }) + + It("reports unconfigured dependencies as unhealthy", func() { + server := &Server{ + cfg: config.MonitoringConfig{ + EnableHealth: true, + HealthTimeout: time.Second, + }, + logger: zap.NewNop(), + } + + req := httptest.NewRequest(http.MethodGet, "/healthz", nil) + resp := httptest.NewRecorder() + + server.handleHealth(resp, req) + + Expect(resp.Code).To(Equal(http.StatusServiceUnavailable)) + + payload := map[string]interface{}{} + Expect(json.NewDecoder(resp.Body).Decode(&payload)).To(Succeed()) + + deps := payload["dependencies"].(map[string]interface{}) + Expect(deps["postgres"].(map[string]interface{})["status"]).To(Equal("unconfigured")) + Expect(deps["nats"].(map[string]interface{})["status"]).To(Equal("unconfigured")) + }) + }) + + Describe("httpStatusLabel", func() { + It("classifies status codes", func() { + Expect(httpStatusLabel(200)).To(Equal("ok")) + Expect(httpStatusLabel(502)).To(Equal("error")) + }) + }) +}) diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 157a8b3..709ac94 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -3,22 +3,13 @@ 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" - "os" "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" ) @@ -50,41 +41,22 @@ type Consumer struct { consecutiveProcessErrors int } -func isDevLikeEnv() bool { - switch strings.ToLower(os.Getenv("GO_ENV")) { - case "", "dev", "development", "test", "testing": - return true - default: - return false - } +// consumerConfig holds normalized consumer configuration values. +type consumerConfig struct { + subject string + consumerName string + mode string + streamName string + dlqSubject string + ackWait time.Duration + batchSize int + batchTimeout time.Duration + monitorInterval time.Duration } -func isJetStreamResourceNotFound(err error) bool { - if err == nil { - return false - } - if errors.Is(err, nats.ErrStreamNotFound) || errors.Is(err, nats.ErrConsumerNotFound) { - return true - } - - // Some JetStream API errors are only exposed via error strings. - msg := strings.ToLower(err.Error()) - if strings.Contains(msg, "stream not found") || strings.Contains(msg, "consumer not found") { - return true - } - - return false -} - -// ProvideConsumer creates a NATS consumer -func ProvideConsumer( - conn *nats.Conn, - js nats.JetStreamContext, - processor *app.MessageProcessor, - cfg *config.Config, - rec telemetry.Recorder, - logger *zap.Logger, -) (*Consumer, error) { +// normalizeConsumerConfig extracts and normalizes consumer configuration from the application config. +// This function can be unit-tested without requiring a JetStream context. +func normalizeConsumerConfig(cfg *config.Config) *consumerConfig { subject := cfg.EffectiveSubscriptionTopic() consumerName := cfg.NATS.Consumer @@ -132,13 +104,7 @@ func ProvideConsumer( monitorInterval = 30 * time.Second } - consumer := &Consumer{ - conn: conn, - js: js, - processor: processor, - cfg: cfg, - logger: logger, - telemetry: rec, + return &consumerConfig{ subject: subject, consumerName: consumerName, mode: mode, @@ -149,6 +115,36 @@ func ProvideConsumer( batchTimeout: batchTimeout, monitorInterval: monitorInterval, } +} + +// ProvideConsumer creates a NATS consumer. +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, + subject: normCfg.subject, + consumerName: normCfg.consumerName, + mode: normCfg.mode, + streamName: normCfg.streamName, + dlqSubject: normCfg.dlqSubject, + ackWait: normCfg.ackWait, + batchSize: normCfg.batchSize, + batchTimeout: normCfg.batchTimeout, + monitorInterval: normCfg.monitorInterval, + } consumer.initMetrics() if consumer.mode == "jetstream" { @@ -158,10 +154,12 @@ func ProvideConsumer( } // Validate DLQ configuration early so misconfiguration is visible at startup // rather than only when the first poison message appears. - consumer.validateDLQ() + if err := consumer.validateDLQ(); err != nil { + return nil, fmt.Errorf("DLQ validation failed: %w", err) + } } else { logger.Info("Running consumer in core NATS mode", - zap.String("subject", subject), + zap.String("subject", normCfg.subject), zap.String("queue_group", cfg.Subscription.QueueGroup), ) } @@ -169,176 +167,7 @@ func ProvideConsumer( return consumer, nil } -// ensureConsumer creates the consumer if it doesn't exist; if it already exists, it is reused. -func (c *Consumer) ensureConsumer() error { - consumerConfig := &nats.ConsumerConfig{ - Durable: c.consumerName, - DeliverPolicy: mapDeliverPolicy(c.cfg.NATS.ConsumerRules.DeliverPolicy), - AckPolicy: nats.AckExplicitPolicy, - AckWait: c.ackWait, - ReplayPolicy: mapReplayPolicy(c.cfg.NATS.ConsumerRules.ReplayPolicy), - MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver, - MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending, - FilterSubject: c.subject, - BackOff: c.cfg.NATS.ConsumerRules.Backoff, - } - if consumerConfig.DeliverPolicy == nats.DeliverByStartSequencePolicy && c.cfg.NATS.ConsumerRules.StartSequence > 0 { - consumerConfig.OptStartSeq = c.cfg.NATS.ConsumerRules.StartSequence - } - if consumerConfig.DeliverPolicy == nats.DeliverByStartTimePolicy && strings.TrimSpace(c.cfg.NATS.ConsumerRules.StartTime) != "" { - startTime, err := time.Parse(time.RFC3339, c.cfg.NATS.ConsumerRules.StartTime) - if err != nil { - c.logger.Warn("Invalid start time, falling back to deliver policy defaults", - zap.String("start_time", c.cfg.NATS.ConsumerRules.StartTime), - zap.Error(err), - ) - } else { - consumerConfig.OptStartTime = &startTime - } - } - - // First check if the consumer already exists to make this initialization idempotent. - info, err := c.js.ConsumerInfo(c.streamName, c.consumerName) - if err == nil && info != nil { - c.logger.Info("Using existing JetStream consumer", - zap.String("consumer", c.consumerName), - zap.String("stream", c.streamName), - zap.String("subject", c.subject), - ) - return nil - } - if err != nil && !errors.Is(err, nats.ErrConsumerNotFound) { - return fmt.Errorf("failed to fetch consumer info: %w", err) - } - - // Consumer does not exist; create it. - if _, err := c.js.AddConsumer(c.streamName, consumerConfig); err != nil { - return fmt.Errorf("failed to create consumer: %w", err) - } - - c.logger.Info("Created JetStream consumer", - zap.String("consumer", c.consumerName), - zap.String("stream", c.streamName), - zap.String("subject", c.subject), - zap.Duration("ack_wait", c.ackWait), - zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), - zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), - ) - - return nil -} - -// 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") - } - - // Ensure stream exists (dev/test may auto-create, prod will error). - if err := EnsureStream(c.js, c.cfg, c.logger); err != nil { - return fmt.Errorf("ensure stream %s: %w", c.streamName, err) - } - - // Ensure durable consumer exists and is properly bound. - if err := c.ensureConsumer(); err != nil { - return fmt.Errorf("ensure consumer %s: %w", c.consumerName, err) - } - - return nil -} - -// 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) { - sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName)) - if err == nil { - return sub, nil - } - - if isJetStreamResourceNotFound(err) && isDevLikeEnv() && shouldBootstrapStream() { - c.logger.Warn("PullSubscribe failed due to missing JetStream resources; attempting to recreate", - zap.Error(err), - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - ) - if recErr := c.recoverJetStreamResources(); recErr != nil { - return nil, fmt.Errorf("failed to recover JetStream resources: %w", recErr) - } - // Retry subscription after successful recovery. - sub, err = c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName)) - 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) -} - -// validateDLQ verifies whether DLQ routing should be enabled and, if so, whether -// the configured DLQ subject is bound to a JetStream stream. If validation fails, -// DLQ routing is disabled (by clearing c.dlqSubject) and a warning is logged, -// but the consumer is still allowed to start. -func (c *Consumer) validateDLQ() { - if c == nil { - return - } - - // DLQ routing is only active in JetStream mode. - if c.mode != "jetstream" { - return - } - - // 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.dlqSubject) != "" { - c.logger.Info("DLQ subject configured but dlq.enabled is false; DLQ routing disabled", - zap.String("dlq_subject", c.dlqSubject), - ) - } - c.dlqSubject = "" - return - } - - subject := strings.TrimSpace(c.dlqSubject) - if subject == "" { - c.logger.Warn("DLQ enabled but dlq.subject is empty; DLQ routing disabled") - return - } - - if c.js == nil { - c.logger.Warn("DLQ enabled but JetStream context is nil; DLQ routing disabled", - zap.String("dlq_subject", subject), - ) - c.dlqSubject = "" - return - } - - // 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) == "" { - c.logger.Warn("DLQ subject not bound to any JetStream stream; DLQ routing disabled", - zap.String("dlq_subject", subject), - zap.Error(err), - ) - c.dlqSubject = "" - return - } - - c.logger.Info("DLQ configuration validated", - zap.String("dlq_subject", subject), - zap.String("dlq_stream", streamName), - ) -} - -// Start starts consuming messages +// Start starts consuming messages. func (c *Consumer) Start(ctx context.Context) error { if c.mode == "core" { return c.startCore(ctx) @@ -347,267 +176,6 @@ func (c *Consumer) Start(ctx context.Context) error { return c.startJetStream(ctx) } -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 - } - defer sub.Unsubscribe() - - c.logger.Info("Started consuming messages", - zap.String("subject", c.subject), - zap.String("consumer", c.consumerName), - zap.String("stream", c.streamName), - ) - - c.logger.Info("Consumer pull configuration", - zap.Int("batch_size", c.batchSize), - zap.Duration("batch_timeout", c.batchTimeout), - zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver), - zap.Duration("ack_wait", c.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) - - 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 := sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout)) - if err != nil { - if errors.Is(err, nats.ErrTimeout) { - // Timeout is expected when no messages are available. - continue - } - - // 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 - if backoff > 30*time.Second { - backoff = 30 * time.Second - } - c.logger.Warn("JetStream not available, will retry with backoff", - zap.Error(err), - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - zap.Duration("backoff", backoff), - ) - time.Sleep(backoff) - continue - } - - // Underlying consumer/stream removed while app is running. - if isJetStreamResourceNotFound(err) { - if isDevLikeEnv() && shouldBootstrapStream() { - c.logger.Warn("JetStream consumer or stream missing; attempting to recreate", - zap.Error(err), - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - ) - if recErr := c.recoverJetStreamResources(); recErr != nil { - c.logger.Error("Failed to recover JetStream resources", zap.Error(recErr)) - return recErr - } - - // Recreate subscription after successful recovery. - sub.Unsubscribe() - sub, err = c.createPullSubscriptionWithRecovery() - if err != nil { - return err - } - - // Reset error streak after successful recovery. - fetchErrorStreak = 0 - continue - } - - // Production: treat as configuration/operational error. - c.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment", - zap.Error(err), - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - ) - return err - } - - // Generic error path with modest backoff. - fetchErrorStreak++ - backoff := time.Duration(fetchErrorStreak) * time.Second - if backoff > 10*time.Second { - backoff = 10 * time.Second - } - c.logger.Error("Failed to fetch messages; backing off", - zap.Error(err), - zap.Duration("backoff", backoff), - ) - time.Sleep(backoff) - continue - } - - // Successful fetch -> reset error streak. - if fetchErrorStreak > 0 { - fetchErrorStreak = 0 - } - - // Process each message - // TODO: consider buffering messages to take advantage of Repository.InsertBatch for higher throughput. - for _, msg := range msgs { - start := time.Now() - - if err := c.processMessage(ctx, msg); err != nil { - isPermanent := app.IsPermanent(err) - elapsed := time.Since(start) - - c.logger.Error("Failed to process message", - zap.String("subject", msg.Subject), - zap.Error(err), - zap.Bool("permanent", isPermanent), - ) - - result := obsmetrics.ResultFail - if isPermanent { - result = obsmetrics.ResultPermanentFail - } - c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, result, elapsed) - - if isPermanent { - c.consecutiveProcessErrors = 0 - // Poison/permanent message: route to DLQ if configured, then ACK - if err := c.routeToDLQ(ctx, msg, err); err != nil { - c.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(err)) - } - if ackErr := msg.Ack(); ackErr != nil { - c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) - } - continue - } - - // Transient error: increment error streak and apply simple backpressure if needed. - if c.consecutiveProcessErrors < 0 { - c.consecutiveProcessErrors = 0 - } - c.consecutiveProcessErrors++ - if c.consecutiveProcessErrors >= 10 { - // Apply a brief sleep to slow down consumption when the system - // is failing many messages in a row (e.g. DB unavailable). - backoff := time.Duration(c.consecutiveProcessErrors) * 100 * time.Millisecond - if backoff > 5*time.Second { - backoff = 5 * time.Second - } - c.logger.Warn("Applying backpressure due to consecutive processing errors", - zap.Int("consecutive_errors", c.consecutiveProcessErrors), - zap.Duration("sleep", backoff), - ) - time.Sleep(backoff) - } - - // Transient error: request redelivery with optional delay - c.telemetry.RecordRetry(ctx, c.streamName, c.consumerName, obsmetrics.RetryReasonProcessorError) - if nakErr := c.nakWithStrategy(msg); nakErr != nil { - c.logger.Error("Failed to NAK message", zap.Error(nakErr)) - } - continue - } - - // 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.streamName, c.consumerName, "ok", elapsed) - } - } - } -} - -func (c *Consumer) startCore(ctx context.Context) error { - queueGroup := c.cfg.Subscription.QueueGroup - if queueGroup == "" { - queueGroup = c.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.subject, queueGroup, handler) - if err != nil { - return fmt.Errorf("failed to subscribe to %s: %w", c.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.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() -} - -func (c *Consumer) emitConsumerStats(ctx context.Context) { - ticker := time.NewTicker(c.monitorInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - info, err := c.js.ConsumerInfo(c.streamName, c.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.streamName), - zap.String("consumer", c.consumerName), - zap.Uint64("num_ack_pending", uint64(info.NumAckPending)), - zap.Uint64("num_redelivered", uint64(info.NumRedelivered)), - zap.Uint64("num_pending", uint64(info.NumPending)), - zap.Uint64("delivered_consumer_seq", uint64(info.Delivered.Consumer)), - zap.Uint64("delivered_stream_seq", uint64(info.Delivered.Stream)), - ) - c.recordConsumerMetrics(ctx, info) - } - } -} - // Shutdown drains the underlying NATS connection gracefully. func (c *Consumer) Shutdown(ctx context.Context) error { if c.conn == nil { @@ -636,246 +204,3 @@ func (c *Consumer) Shutdown(ctx context.Context) error { return fmt.Errorf("nats drain timeout: %w", closeCtx.Err()) } } - -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 - } -} - -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.streamName, c.consumerName, info.NumPending) -} - -// 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.mode != "jetstream" { - return nil - } - if strings.TrimSpace(c.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]interface{}{ - "transport_msg_id": msg.Header.Get("Nats-Msg-Id"), - "subject": msg.Subject, - "stream": c.streamName, - "consumer": c.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.streamName), - zap.String("consumer", c.consumerName), - zap.String("dlq_subject", c.dlqSubject), - zap.Error(err), - ) - return fmt.Errorf("marshal dlq payload: %w", err) - } - - if _, err := c.js.Publish(c.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.streamName), - zap.String("consumer", c.consumerName), - zap.String("dlq_subject", c.dlqSubject), - zap.Int("payload_size", len(data)), - zap.Error(err), - ) - c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) - return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.dlqSubject, err) - } - c.logger.Error("failed to publish to DLQ", - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - zap.String("dlq_subject", c.dlqSubject), - zap.Int("payload_size", len(data)), - zap.Error(err), - ) - c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) - return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err) - } - - c.telemetry.RecordDLQMessage(ctx, c.streamName, c.consumerName) - - return nil -} - -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) -} - -// 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() - span.SetAttributes(attribute.String("nats.subject", msg.Subject)) - - 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.streamName, - Consumer: c.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 -} - -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.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 -} - -func mapDeliverPolicy(value string) nats.DeliverPolicy { - switch strings.ToLower(value) { - case "new": - return nats.DeliverNewPolicy - case "last": - return nats.DeliverLastPolicy - case "last_per_subject": - return nats.DeliverLastPerSubjectPolicy - case "sequence": - return nats.DeliverByStartSequencePolicy - case "time": - return nats.DeliverByStartTimePolicy - default: - return nats.DeliverAllPolicy - } -} - -func mapReplayPolicy(value string) nats.ReplayPolicy { - switch strings.ToLower(value) { - case "original": - return nats.ReplayOriginalPolicy - default: - return nats.ReplayInstantPolicy - } -} diff --git a/internal/infra/nats/consumer_core.go b/internal/infra/nats/consumer_core.go new file mode 100644 index 0000000..0082b42 --- /dev/null +++ b/internal/infra/nats/consumer_core.go @@ -0,0 +1,52 @@ +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.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.subject, queueGroup, handler) + if err != nil { + return fmt.Errorf("failed to subscribe to %s: %w", c.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.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_jetstream.go b/internal/infra/nats/consumer_jetstream.go new file mode 100644 index 0000000..2fd2c44 --- /dev/null +++ b/internal/infra/nats/consumer_jetstream.go @@ -0,0 +1,399 @@ +package nats + +import ( + "caatsm/internal/app" + obsmetrics "caatsm/internal/infra/metrics" + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" + "go.uber.org/zap" +) + +// ensureConsumer creates the consumer if it doesn't exist; if it already exists, it is reused. +func (c *Consumer) ensureConsumer() error { + consumerConfig := &nats.ConsumerConfig{ + Durable: c.consumerName, + DeliverPolicy: mapDeliverPolicy(c.cfg.NATS.ConsumerRules.DeliverPolicy), + AckPolicy: nats.AckExplicitPolicy, + AckWait: c.ackWait, + ReplayPolicy: mapReplayPolicy(c.cfg.NATS.ConsumerRules.ReplayPolicy), + MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver, + MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending, + FilterSubject: c.subject, + BackOff: c.cfg.NATS.ConsumerRules.Backoff, + } + if consumerConfig.DeliverPolicy == nats.DeliverByStartSequencePolicy && c.cfg.NATS.ConsumerRules.StartSequence > 0 { + consumerConfig.OptStartSeq = c.cfg.NATS.ConsumerRules.StartSequence + } + if consumerConfig.DeliverPolicy == nats.DeliverByStartTimePolicy && strings.TrimSpace(c.cfg.NATS.ConsumerRules.StartTime) != "" { + startTime, err := time.Parse(time.RFC3339, c.cfg.NATS.ConsumerRules.StartTime) + if err != nil { + c.logger.Warn("Invalid start time, falling back to deliver policy defaults", + zap.String("start_time", c.cfg.NATS.ConsumerRules.StartTime), + zap.Error(err), + ) + } else { + consumerConfig.OptStartTime = &startTime + } + } + + // First check if the consumer already exists to make this initialization idempotent. + info, err := c.js.ConsumerInfo(c.streamName, c.consumerName) + if err == nil && info != nil { + c.logger.Info("Using existing JetStream consumer", + zap.String("consumer", c.consumerName), + zap.String("stream", c.streamName), + zap.String("subject", c.subject), + ) + return nil + } + if err != nil && !errors.Is(err, nats.ErrConsumerNotFound) { + return fmt.Errorf("failed to fetch consumer info: %w", err) + } + + // Consumer does not exist; create it. + if _, err := c.js.AddConsumer(c.streamName, consumerConfig); err != nil { + return fmt.Errorf("failed to create consumer: %w", err) + } + + c.logger.Info("Created JetStream consumer", + zap.String("consumer", c.consumerName), + zap.String("stream", c.streamName), + zap.String("subject", c.subject), + zap.Duration("ack_wait", c.ackWait), + zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), + zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), + ) + + return nil +} + +// 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") + } + + // Ensure stream exists (dev/test may auto-create, prod will error). + if err := EnsureStream(c.js, c.cfg, c.logger); err != nil { + return fmt.Errorf("ensure stream %s: %w", c.streamName, err) + } + + // Ensure durable consumer exists and is properly bound. + if err := c.ensureConsumer(); err != nil { + return fmt.Errorf("ensure consumer %s: %w", c.consumerName, err) + } + + return nil +} + +// 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) { + sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName)) + if err == nil { + return sub, nil + } + + if isJetStreamResourceNotFound(err) && isDevLikeEnv() && shouldBootstrapStream() { + c.logger.Warn("PullSubscribe failed due to missing JetStream resources; attempting to recreate", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + ) + if recErr := c.recoverJetStreamResources(); recErr != nil { + return nil, fmt.Errorf("failed to recover JetStream resources: %w", recErr) + } + // Retry subscription after successful recovery. + sub, err = c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName)) + 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) +} + +// 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 + } +} + + +// fetchBatch fetches a batch of messages from the subscription. +func (c *Consumer) fetchBatch(sub *nats.Subscription) ([]*nats.Msg, error) { + return sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout)) +} + +// 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) { + if errors.Is(err, nats.ErrTimeout) { + // Timeout is expected when no messages are available. + return true, nil + } + + // 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 + if backoff > 30*time.Second { + backoff = 30 * time.Second + } + c.logger.Warn("JetStream not available, will retry with backoff", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + zap.Duration("backoff", backoff), + ) + // Use context-aware sleep instead of blocking time.Sleep + if !sleepWithContext(ctx, backoff) { + return false, ctx.Err() + } + return true, nil + } + + // Underlying consumer/stream removed while app is running. + if isJetStreamResourceNotFound(err) { + if isDevLikeEnv() && shouldBootstrapStream() { + c.logger.Warn("JetStream consumer or stream missing; attempting to recreate", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + ) + if recErr := c.recoverJetStreamResources(); recErr != nil { + return false, recErr + } + + // Recreate subscription after successful recovery. + (*sub).Unsubscribe() + newSub, subErr := c.createPullSubscriptionWithRecovery() + if subErr != nil { + return false, subErr + } + *sub = newSub + *fetchErrorStreak = 0 + return true, nil + } + + // Production: treat as configuration/operational error. + c.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + ) + return false, err + } + + // Generic error path with modest backoff. + *fetchErrorStreak++ + backoff := time.Duration(*fetchErrorStreak) * time.Second + if backoff > 10*time.Second { + backoff = 10 * time.Second + } + c.logger.Error("Failed to fetch messages; backing off", + zap.Error(err), + zap.Duration("backoff", backoff), + ) + // Use context-aware sleep instead of blocking time.Sleep + if !sleepWithContext(ctx, backoff) { + return false, ctx.Err() + } + return true, nil +} + +// processBatch processes a batch of messages, handling errors and applying backpressure. +func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) { + for _, msg := range msgs { + start := time.Now() + + if err := c.processMessage(ctx, msg); err != nil { + isPermanent := app.IsPermanent(err) + elapsed := time.Since(start) + + c.logger.Error("Failed to process message", + zap.String("subject", msg.Subject), + zap.Error(err), + zap.Bool("permanent", isPermanent), + ) + + result := obsmetrics.ResultFail + if isPermanent { + result = obsmetrics.ResultPermanentFail + } + c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, result, elapsed) + + if isPermanent { + c.consecutiveProcessErrors = 0 + // Poison/permanent message: route to DLQ if configured, then ACK + if err := c.routeToDLQ(ctx, msg, err); err != nil { + c.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(err)) + } + if ackErr := msg.Ack(); ackErr != nil { + c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) + } + continue + } + + // Transient error: increment error streak and apply simple backpressure if needed. + if c.consecutiveProcessErrors < 0 { + c.consecutiveProcessErrors = 0 + } + c.consecutiveProcessErrors++ + if c.consecutiveProcessErrors >= 10 { + // Apply a brief sleep to slow down consumption when the system + // is failing many messages in a row (e.g. DB unavailable). + backoff := time.Duration(c.consecutiveProcessErrors) * 100 * time.Millisecond + if backoff > 5*time.Second { + backoff = 5 * time.Second + } + c.logger.Warn("Applying backpressure due to consecutive processing errors", + zap.Int("consecutive_errors", c.consecutiveProcessErrors), + zap.Duration("sleep", backoff), + ) + // Use context-aware sleep instead of blocking time.Sleep + if !sleepWithContext(ctx, backoff) { + // Context canceled, stop processing batch + return + } + } + + // Transient error: request redelivery with optional delay + c.telemetry.RecordRetry(ctx, c.streamName, c.consumerName, obsmetrics.RetryReasonProcessorError) + if nakErr := c.nakWithStrategy(msg); nakErr != nil { + c.logger.Error("Failed to NAK message", zap.Error(nakErr)) + } + continue + } + + // 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.streamName, c.consumerName, "ok", elapsed) + } + } +} + +// 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 *nats.Subscription = sub + cleanupSubscriber := func() { + if currentSub != nil { + currentSub.Unsubscribe() + currentSub = nil + } + } + defer cleanupSubscriber() + + c.logger.Info("Started consuming messages", + zap.String("subject", c.subject), + zap.String("consumer", c.consumerName), + zap.String("stream", c.streamName), + ) + + c.logger.Info("Consumer pull configuration", + zap.Int("batch_size", c.batchSize), + zap.Duration("batch_timeout", c.batchTimeout), + zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver), + zap.Duration("ack_wait", c.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) + + 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.fetchBatch(currentSub) + if err != nil { + shouldContinue, handleErr := c.handleFetchError(ctx, err, ¤tSub, &fetchErrorStreak) + if !shouldContinue { + return handleErr + } + continue + } + + // Successful fetch -> reset error streak. + if fetchErrorStreak > 0 { + fetchErrorStreak = 0 + } + + // Process batch + c.processBatch(ctx, msgs) + } +} diff --git a/internal/infra/nats/consumer_jetstream_test.go b/internal/infra/nats/consumer_jetstream_test.go new file mode 100644 index 0000000..19bca4e --- /dev/null +++ b/internal/infra/nats/consumer_jetstream_test.go @@ -0,0 +1,97 @@ +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() { + c = &Consumer{ + mode: "jetstream", + streamName: "TEST_STREAM", + consumerName: "test-consumer", + subject: "test.subject", + batchSize: 10, + batchTimeout: 2 * time.Second, + logger: zaptest.NewLogger(GinkgoT()), + 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)) + }) + }) +}) diff --git a/internal/infra/nats/consumer_test.go b/internal/infra/nats/consumer_test.go new file mode 100644 index 0000000..8c365c6 --- /dev/null +++ b/internal/infra/nats/consumer_test.go @@ -0,0 +1,147 @@ +package nats + +import ( + "errors" + "testing" + "time" + + configpkg "caatsm/internal/infra/config" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/nats-io/nats.go" +) + +func TestNATS(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "NATS Suite") +} + +var _ = Describe("Consumer helpers", func() { + Describe("isJetStreamResourceNotFound", func() { + It("detects missing resources for known errors", func() { + Expect(isJetStreamResourceNotFound(nil)).To(BeFalse()) + Expect(isJetStreamResourceNotFound(nats.ErrStreamNotFound)).To(BeTrue()) + Expect(isJetStreamResourceNotFound(errors.New("consumer not found in stream not found"))).To(BeTrue()) + }) + }) + + Describe("policy mapping", func() { + It("maps deliver policies to NATS constants", func() { + tests := map[string]nats.DeliverPolicy{ + "": nats.DeliverAllPolicy, + "new": nats.DeliverNewPolicy, + "LAST": nats.DeliverLastPolicy, + "last_per_subject": nats.DeliverLastPerSubjectPolicy, + "sequence": nats.DeliverByStartSequencePolicy, + "time": nats.DeliverByStartTimePolicy, + "unknown-so-far": nats.DeliverAllPolicy, + } + for input, want := range tests { + Expect(mapDeliverPolicy(input)).To(Equal(want)) + } + }) + + It("maps replay policy to instant by default", func() { + Expect(mapReplayPolicy("original")).To(Equal(nats.ReplayOriginalPolicy)) + Expect(mapReplayPolicy("")).To(Equal(nats.ReplayInstantPolicy)) + }) + }) + + Describe("normalizeConsumerConfig", func() { + It("applies default values when config fields are empty", func() { + cfg := &configpkg.Config{ + NATS: configpkg.NATSConfig{ + Mode: "", + }, + App: configpkg.AppConfig{}, + Timeouts: configpkg.TimeoutsConfig{}, + } + + normCfg := normalizeConsumerConfig(cfg) + + Expect(normCfg.consumerName).To(Equal("telegram-consumer")) + Expect(normCfg.mode).To(Equal("jetstream")) + Expect(normCfg.streamName).To(Equal("TELEGRAM")) + Expect(normCfg.batchSize).To(Equal(50)) + Expect(normCfg.batchTimeout).To(Equal(2 * time.Second)) + Expect(normCfg.monitorInterval).To(Equal(30 * time.Second)) + Expect(normCfg.ackWait).To(Equal(30 * time.Second)) + }) + + It("uses provided values when config fields are set", func() { + cfg := &configpkg.Config{ + NATS: configpkg.NATSConfig{ + Consumer: "custom-consumer", + Mode: "core", + Stream: "CUSTOM_STREAM", + ConsumerRules: configpkg.ConsumerRulesConfig{ + AckWait: 60 * time.Second, + }, + }, + App: configpkg.AppConfig{ + BatchSize: 100, + BatchTimeout: 5 * time.Second, + MonitorInterval: 60 * time.Second, + }, + Timeouts: configpkg.TimeoutsConfig{ + AckWait: 45 * time.Second, + }, + } + + normCfg := normalizeConsumerConfig(cfg) + + Expect(normCfg.consumerName).To(Equal("custom-consumer")) + Expect(normCfg.mode).To(Equal("core")) + Expect(normCfg.streamName).To(Equal("CUSTOM_STREAM")) + Expect(normCfg.batchSize).To(Equal(100)) + Expect(normCfg.batchTimeout).To(Equal(5 * time.Second)) + Expect(normCfg.monitorInterval).To(Equal(60 * time.Second)) + Expect(normCfg.ackWait).To(Equal(60 * time.Second)) // Uses ConsumerRules.AckWait + }) + + It("falls back to Timeouts.AckWait when ConsumerRules.AckWait is zero", func() { + cfg := &configpkg.Config{ + NATS: configpkg.NATSConfig{ + ConsumerRules: configpkg.ConsumerRulesConfig{ + AckWait: 0, + }, + }, + Timeouts: configpkg.TimeoutsConfig{ + AckWait: 45 * time.Second, + }, + } + + normCfg := normalizeConsumerConfig(cfg) + + Expect(normCfg.ackWait).To(Equal(45 * time.Second)) + }) + + It("sets dlqSubject when DLQ is enabled", func() { + cfg := &configpkg.Config{ + DLQ: configpkg.DLQConfig{ + Enabled: true, + Subject: "caatsm.dlq", + }, + } + + normCfg := normalizeConsumerConfig(cfg) + + Expect(normCfg.dlqSubject).To(Equal("caatsm.dlq")) + }) + + It("clears dlqSubject when DLQ is disabled", func() { + cfg := &configpkg.Config{ + DLQ: configpkg.DLQConfig{ + Enabled: false, + Subject: "caatsm.dlq", + }, + } + + normCfg := normalizeConsumerConfig(cfg) + + Expect(normCfg.dlqSubject).To(Equal("")) + }) + }) +}) diff --git a/internal/infra/nats/dlq.go b/internal/infra/nats/dlq.go new file mode 100644 index 0000000..3f753ae --- /dev/null +++ b/internal/infra/nats/dlq.go @@ -0,0 +1,139 @@ +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.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.dlqSubject) != "" { + c.logger.Info("DLQ subject configured but dlq.enabled is false; DLQ routing disabled", + zap.String("dlq_subject", c.dlqSubject), + ) + } + c.dlqSubject = "" + return nil + } + + subject := strings.TrimSpace(c.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.mode != "jetstream" { + return nil + } + if strings.TrimSpace(c.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]interface{}{ + "transport_msg_id": msg.Header.Get("Nats-Msg-Id"), + "subject": msg.Subject, + "stream": c.streamName, + "consumer": c.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.streamName), + zap.String("consumer", c.consumerName), + zap.String("dlq_subject", c.dlqSubject), + zap.Error(err), + ) + return fmt.Errorf("marshal dlq payload: %w", err) + } + + if _, err := c.js.Publish(c.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.streamName), + zap.String("consumer", c.consumerName), + zap.String("dlq_subject", c.dlqSubject), + zap.Int("payload_size", len(data)), + zap.Error(err), + ) + c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) + return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.dlqSubject, err) + } + c.logger.Error("failed to publish to DLQ", + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + zap.String("dlq_subject", c.dlqSubject), + zap.Int("payload_size", len(data)), + zap.Error(err), + ) + c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) + return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err) + } + + c.telemetry.RecordDLQMessage(ctx, c.streamName, c.consumerName) + + return nil +} diff --git a/internal/infra/nats/dlq_test.go b/internal/infra/nats/dlq_test.go new file mode 100644 index 0000000..b5ef616 --- /dev/null +++ b/internal/infra/nats/dlq_test.go @@ -0,0 +1,131 @@ +package nats + +import ( + "context" + "errors" + + configpkg "caatsm/internal/infra/config" + "caatsm/internal/infra/telemetry" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/nats-io/nats.go" + "go.uber.org/zap" + "go.uber.org/zap/zaptest" +) + +var _ = Describe("DLQ", func() { + var ( + logger *zap.Logger + ) + + BeforeEach(func() { + logger = zaptest.NewLogger(GinkgoT()) + }) + + 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{ + 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{ + 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.dlqSubject).To(Equal("")) + }) + + It("returns error when DLQ is enabled but subject is empty", func() { + c := &Consumer{ + 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{ + 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() + 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{ + 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{ + 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/jetstream.go b/internal/infra/nats/jetstream.go index 8a2654b..62719e7 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -26,12 +26,14 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) } }), nats.ReconnectHandler(func(nc *nats.Conn) { - logger.Info("NATS reconnected", zap.String("url", nc.ConnectedUrl())) + safeURL := sanitizeURLForLogging(nc.ConnectedUrl()) + logger.Info("NATS reconnected", zap.String("url", safeURL)) }), ) if err != nil { + safeURL := sanitizeURLForLogging(cfg.NATS.URL) logger.Error("failed to connect to NATS", - zap.String("url", cfg.NATS.URL), + zap.String("url", safeURL), zap.Duration("timeout", cfg.Timeouts.Server), zap.Duration("reconnect_wait", cfg.Timeouts.ReconnectWait), zap.Error(err), @@ -43,12 +45,20 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) } // 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 js, err := nc.JetStream() if err != nil { + safeURL := sanitizeURLForLogging(cfg.NATS.URL) logger.Error("failed to get JetStream context", - zap.String("url", cfg.NATS.URL), + zap.String("url", safeURL), zap.Error(err), ) nc.Close() diff --git a/internal/infra/nats/message_handler.go b/internal/infra/nats/message_handler.go new file mode 100644 index 0000000..117dc6b --- /dev/null +++ b/internal/infra/nats/message_handler.go @@ -0,0 +1,88 @@ +package nats + +import ( + "caatsm/internal/infra/log" + "context" + "fmt" + + "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" +) + +// 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() + span.SetAttributes(attribute.String("nats.subject", msg.Subject)) + + 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.streamName, + Consumer: c.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.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/message_handler_test.go b/internal/infra/nats/message_handler_test.go new file mode 100644 index 0000000..d5a411e --- /dev/null +++ b/internal/infra/nats/message_handler_test.go @@ -0,0 +1,62 @@ +package nats + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/nats-io/nats.go" + "go.uber.org/zap/zaptest" +) + +var _ = Describe("MessageHandler", func() { + var ( + c *Consumer + ) + + BeforeEach(func() { + c = &Consumer{ + mode: "jetstream", + streamName: "TEST_STREAM", + consumerName: "test-consumer", + logger: zaptest.NewLogger(GinkgoT()), + } + }) + + Describe("resolveMsgID", func() { + It("extracts message ID from header", func() { + msg := &nats.Msg{ + Header: nats.Header{}, + } + msg.Header.Set("Nats-Msg-Id", "msg-123") + + id, source, err := c.resolveMsgID(msg) + Expect(err).NotTo(HaveOccurred()) + Expect(id).To(Equal("msg-123")) + Expect(source).To(Equal("header")) + }) + + It("generates UUID for core mode when header is missing", func() { + c.mode = "core" + msg := &nats.Msg{ + Header: nats.Header{}, + } + + id, source, err := c.resolveMsgID(msg) + Expect(err).NotTo(HaveOccurred()) + Expect(id).NotTo(BeEmpty()) + Expect(source).To(Equal("generated")) + }) + + It("returns error for JetStream mode when header and metadata are missing", func() { + c.mode = "jetstream" + msg := &nats.Msg{ + Header: nats.Header{}, + } + + // Without metadata, this should return an error + _, _, err := c.resolveMsgID(msg) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("fetch metadata")) + }) + }) +}) + diff --git a/internal/infra/nats/metrics.go b/internal/infra/nats/metrics.go new file mode 100644 index 0000000..2f113d1 --- /dev/null +++ b/internal/infra/nats/metrics.go @@ -0,0 +1,83 @@ +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.streamName, c.consumerName, info.NumPending) +} + +// emitConsumerStats periodically emits consumer statistics. +func (c *Consumer) emitConsumerStats(ctx context.Context) { + ticker := time.NewTicker(c.monitorInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + info, err := c.js.ConsumerInfo(c.streamName, c.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.streamName), + zap.String("consumer", c.consumerName), + zap.Uint64("num_ack_pending", uint64(info.NumAckPending)), + zap.Uint64("num_redelivered", uint64(info.NumRedelivered)), + zap.Uint64("num_pending", uint64(info.NumPending)), + zap.Uint64("delivered_consumer_seq", uint64(info.Delivered.Consumer)), + zap.Uint64("delivered_stream_seq", uint64(info.Delivered.Stream)), + ) + c.recordConsumerMetrics(ctx, info) + } + } +} diff --git a/internal/infra/nats/metrics_test.go b/internal/infra/nats/metrics_test.go new file mode 100644 index 0000000..9508e98 --- /dev/null +++ b/internal/infra/nats/metrics_test.go @@ -0,0 +1,57 @@ +package nats + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/nats-io/nats.go" + "go.uber.org/zap/zaptest" +) + +var _ = Describe("Metrics", func() { + var ( + c *Consumer + ctx context.Context + ) + + BeforeEach(func() { + ctx = context.Background() + c = &Consumer{ + streamName: "TEST_STREAM", + consumerName: "test-consumer", + 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 + }) + }) +}) + diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index 080594a..d8608fd 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -20,12 +20,17 @@ type Publisher struct { logger *zap.Logger } -// ProvidePublisher creates a NATS publisher +// ProvidePublisher creates a NATS publisher. +// When js is nil (core mode), returns a CorePublisher that uses plain NATS. func ProvidePublisher( js nats.JetStreamContext, + nc *nats.Conn, cfg *config.Config, logger *zap.Logger, ) (port.Publisher, error) { + if js == nil { + return ProvideCorePublisher(nc, cfg, logger) + } return &Publisher{ js: js, cfg: cfg, @@ -33,6 +38,63 @@ func ProvidePublisher( }, nil } +// CorePublisher publishes messages to plain NATS (non-JetStream) +type CorePublisher struct { + conn *nats.Conn + cfg *config.Config + logger *zap.Logger +} + +// ProvideCorePublisher creates a NATS publisher for core mode +func ProvideCorePublisher( + conn *nats.Conn, + cfg *config.Config, + logger *zap.Logger, +) (port.Publisher, error) { + return &CorePublisher{ + conn: conn, + cfg: cfg, + logger: logger, + }, nil +} + +// Publish publishes a message using plain NATS +func (p *CorePublisher) Publish(message interface{}) error { + topic := p.cfg.Publisher.Topic + if topic == "" { + p.logger.Error("publisher topic is not configured") + return fmt.Errorf("publisher topic is not configured") + } + + // Marshal message to JSON + messageBytes, err := json.Marshal(message) + if err != nil { + p.logger.Error("failed to marshal message", + zap.String("topic", topic), + zap.Error(err), + ) + return fmt.Errorf("failed to marshal message: %w", err) + } + + // Publish to plain NATS + err = p.conn.Publish(topic, messageBytes) + if err != nil { + p.logger.Error("failed to publish message", + zap.String("topic", topic), + zap.Int("message_size", len(messageBytes)), + zap.Error(err), + ) + return fmt.Errorf("failed to publish message: %w", err) + } + + p.logger.Debug("Published message", + zap.String("topic", topic), + zap.Int("size", len(messageBytes)), + ) + + return nil +} + // Publish publishes a message func (p *Publisher) Publish(message interface{}) error { topic := p.cfg.Publisher.Topic diff --git a/internal/infra/nats/utils.go b/internal/infra/nats/utils.go new file mode 100644 index 0000000..3c7da7e --- /dev/null +++ b/internal/infra/nats/utils.go @@ -0,0 +1,80 @@ +package nats + +import ( + "errors" + "net/url" + "os" + "strings" + + "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": + return true + default: + return false + } +} + +// isJetStreamResourceNotFound checks if an error indicates missing JetStream resources. +func isJetStreamResourceNotFound(err error) bool { + if err == nil { + return false + } + if errors.Is(err, nats.ErrStreamNotFound) || errors.Is(err, nats.ErrConsumerNotFound) { + return true + } + + // Some JetStream API errors are only exposed via error strings. + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "stream not found") || strings.Contains(msg, "consumer not found") +} + +// sanitizeURLForLogging removes credentials from URLs for safe logging. +func sanitizeURLForLogging(rawURL string) string { + if rawURL == "" { + return "" + } + + // Parse URL + u, err := url.Parse(rawURL) + if err != nil { + // If parsing fails, return a safe placeholder + return "***" + } + + // Rebuild without credentials + u.User = nil + return u.String() +} + +// mapDeliverPolicy maps string configuration to NATS DeliverPolicy. +func mapDeliverPolicy(value string) nats.DeliverPolicy { + switch strings.ToLower(value) { + case "new": + return nats.DeliverNewPolicy + case "last": + return nats.DeliverLastPolicy + case "last_per_subject": + return nats.DeliverLastPerSubjectPolicy + case "sequence": + return nats.DeliverByStartSequencePolicy + case "time": + return nats.DeliverByStartTimePolicy + default: + return nats.DeliverAllPolicy + } +} + +// mapReplayPolicy maps string configuration to NATS ReplayPolicy. +func mapReplayPolicy(value string) nats.ReplayPolicy { + switch strings.ToLower(value) { + case "original": + return nats.ReplayOriginalPolicy + default: + return nats.ReplayInstantPolicy + } +} diff --git a/internal/infra/nats/utils_test.go b/internal/infra/nats/utils_test.go new file mode 100644 index 0000000..50f12e7 --- /dev/null +++ b/internal/infra/nats/utils_test.go @@ -0,0 +1,87 @@ +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 == "" { + os.Unsetenv("GO_ENV") + } else { + os.Setenv("GO_ENV", originalEnv) + } + }) + }) + + It("returns true for dev environment", func() { + os.Setenv("GO_ENV", "dev") + Expect(isDevLikeEnv()).To(BeTrue()) + }) + + It("returns true for development environment", func() { + os.Setenv("GO_ENV", "development") + Expect(isDevLikeEnv()).To(BeTrue()) + }) + + It("returns true for test environment", func() { + os.Setenv("GO_ENV", "test") + Expect(isDevLikeEnv()).To(BeTrue()) + }) + + It("returns true for testing environment", func() { + os.Setenv("GO_ENV", "testing") + Expect(isDevLikeEnv()).To(BeTrue()) + }) + + It("returns true for empty environment", func() { + os.Unsetenv("GO_ENV") + Expect(isDevLikeEnv()).To(BeTrue()) + }) + + It("returns false for production environment", func() { + os.Setenv("GO_ENV", "prod") + Expect(isDevLikeEnv()).To(BeFalse()) + }) + + It("returns false for production environment (uppercase)", func() { + os.Setenv("GO_ENV", "PROD") + Expect(isDevLikeEnv()).To(BeFalse()) + }) + }) + + Describe("sanitizeURLForLogging", func() { + It("removes credentials from URLs", func() { + url := "nats://user:pass@localhost:4222" + Expect(sanitizeURLForLogging(url)).To(Equal("nats://localhost:4222")) + }) + + It("handles URLs without credentials", func() { + url := "nats://localhost:4222" + Expect(sanitizeURLForLogging(url)).To(Equal("nats://localhost:4222")) + }) + + It("handles empty strings", func() { + Expect(sanitizeURLForLogging("")).To(Equal("")) + }) + + It("handles invalid URLs", func() { + url := "://invalid" + result := sanitizeURLForLogging(url) + Expect(result).To(Equal("***")) + }) + + It("handles URLs with user but no password", func() { + url := "nats://user@localhost:4222" + Expect(sanitizeURLForLogging(url)).To(Equal("nats://localhost:4222")) + }) + }) +}) + diff --git a/internal/infra/postgres/db.go b/internal/infra/postgres/db.go index e0e8227..061773e 100644 --- a/internal/infra/postgres/db.go +++ b/internal/infra/postgres/db.go @@ -10,6 +10,30 @@ import ( "go.uber.org/zap" ) +// sanitizePostgresURLForLogging creates a safe version of the connection string for logging +// by removing credentials and only showing host, port, and database +func sanitizePostgresURLForLogging(poolConfig *pgxpool.Config) string { + if poolConfig == nil || poolConfig.ConnConfig == nil { + return "postgres://***@***/***" + } + + host := poolConfig.ConnConfig.Host + port := poolConfig.ConnConfig.Port + database := poolConfig.ConnConfig.Database + + if host == "" { + host = "***" + } + if port == 0 { + port = 5432 + } + if database == "" { + database = "***" + } + + return fmt.Sprintf("postgres://***@%s:%d/%s", host, port, database) +} + // ProvideDB creates a PostgreSQL connection pool func ProvideDB(cfg *config.Config, logger *zap.Logger) (*pgxpool.Pool, error) { ctx := context.Background() @@ -17,12 +41,15 @@ func ProvideDB(cfg *config.Config, logger *zap.Logger) (*pgxpool.Pool, error) { poolConfig, err := pgxpool.ParseConfig(cfg.Postgres.URL) if err != nil { logger.Error("failed to parse postgres URL", - zap.String("url", cfg.Postgres.URL), + zap.String("host", "unknown"), zap.Error(err), ) return nil, fmt.Errorf("failed to parse postgres URL: %w", err) } + // Extract safe connection info for logging + safeURL := sanitizePostgresURLForLogging(poolConfig) + poolConfig.MaxConns = int32(cfg.Postgres.MaxConns) poolConfig.MinConns = int32(cfg.Postgres.MinConns) poolConfig.MaxConnLifetime = time.Hour @@ -31,7 +58,7 @@ func ProvideDB(cfg *config.Config, logger *zap.Logger) (*pgxpool.Pool, error) { pool, err := pgxpool.NewWithConfig(ctx, poolConfig) if err != nil { logger.Error("failed to create connection pool", - zap.String("url", cfg.Postgres.URL), + zap.String("url", safeURL), zap.Int32("max_conns", cfg.Postgres.MaxConns), zap.Int32("min_conns", cfg.Postgres.MinConns), zap.Error(err), @@ -42,13 +69,17 @@ func ProvideDB(cfg *config.Config, logger *zap.Logger) (*pgxpool.Pool, error) { // Test connection if err := pool.Ping(ctx); err != nil { logger.Error("failed to ping database", - zap.String("url", cfg.Postgres.URL), + zap.String("url", safeURL), zap.Error(err), ) return nil, fmt.Errorf("failed to ping database: %w", err) } // Log the connection pool - logger.Info("Connected to PostgreSQL", zap.Int32("max_conns", cfg.Postgres.MaxConns), zap.Int32("min_conns", cfg.Postgres.MinConns), zap.String("url", cfg.Postgres.URL)) + logger.Info("Connected to PostgreSQL", + zap.String("url", safeURL), + zap.Int32("max_conns", cfg.Postgres.MaxConns), + zap.Int32("min_conns", cfg.Postgres.MinConns), + ) return pool, nil } diff --git a/internal/infra/postgres/db_test.go b/internal/infra/postgres/db_test.go new file mode 100644 index 0000000..8a700cc --- /dev/null +++ b/internal/infra/postgres/db_test.go @@ -0,0 +1,27 @@ +package postgres + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestPostgres(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Postgres Suite") +} + +var _ = Describe("sanitizePostgresURLForLogging", func() { + It("returns a masked URL when config is empty", func() { + Expect(sanitizePostgresURLForLogging(nil)).To(Equal("postgres://***@***/***")) + }) + + It("returns a sanitized URL when fields are populated", func() { + cfg, err := pgxpool.ParseConfig("postgres://user:secret@db.local:6543/telegrams") + Expect(err).NotTo(HaveOccurred()) + Expect(sanitizePostgresURLForLogging(cfg)).To(Equal("postgres://***@db.local:6543/telegrams")) + }) +}) diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index 833bd80..cb53fa4 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -26,11 +26,11 @@ func buildAppComponents() (*appComponents, error) { if err != nil { return nil, err } - pool, err := postgres.ProvideDB(configConfig) + logger, err := log.ProvideLogger(configConfig) if err != nil { return nil, err } - logger, err := log.ProvideLogger(configConfig) + pool, err := postgres.ProvideDB(configConfig, logger) if err != nil { return nil, err } @@ -46,7 +46,7 @@ func buildAppComponents() (*appComponents, error) { if err != nil { return nil, err } - publisher, err := nats.ProvidePublisher(jetStreamContext, configConfig, logger) + publisher, err := nats.ProvidePublisher(jetStreamContext, conn, configConfig, logger) if err != nil { return nil, err } @@ -70,11 +70,11 @@ func buildAppComponents() (*appComponents, error) { func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { parserParser := parser.ProvideParser() - pool, err := postgres.ProvideDB(cfg) + logger, err := log.ProvideLogger(cfg) if err != nil { return nil, err } - logger, err := log.ProvideLogger(cfg) + pool, err := postgres.ProvideDB(cfg, logger) if err != nil { return nil, err } @@ -90,7 +90,7 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { if err != nil { return nil, err } - publisher, err := nats.ProvidePublisher(jetStreamContext, cfg, logger) + publisher, err := nats.ProvidePublisher(jetStreamContext, conn, cfg, logger) if err != nil { return nil, err }