✨ Update observability configuration by enabling monitoring in the development environment, adding a new Prometheus scrape job for the CAATSM receiver, and enhancing documentation with recommended scrape configurations and validation steps for metrics. Refactor NATS consumer initialization to ensure idempotent consumer creation.

configs/config.dev.toml

@@ -57,6 +57,7 @@ endpoint = "localhost:4318"
 insecure = true
 
 [monitoring]
+disabled = false
 addr = ":2112"
 enable_metrics = true
 enable_health = true

configs/prometheus.dev.yml

@@ -12,3 +12,10 @@ scrape_configs:
       - targets:
           - "nats-exporter:7777"
 
+  - job_name: "caatsm-receiver"
+    static_configs:
+      - targets:
+          # When go-caatsm runs on the host/WSL and Prometheus runs in Docker,
+          # use host.docker.internal to reach the host monitoring server.
+          - "host.docker.internal:2112"
+

docs/observability.md

@@ -39,6 +39,49 @@ These metrics are intended to be scraped by Prometheus (either directly or via t
 - NATS consumer backlog and redelivery counts.  
 - DB query rates and latencies.
 
+#### Prometheus scrape configuration
+
+In the local dev environment, metrics are typically scraped by the Prometheus
+container defined in `docker-compose.dev.yml` using `configs/prometheus.dev.yml`.
+
+A recommended scrape configuration for the receiver is:
+
+```yaml
+scrape_configs:
+  - job_name: "otel-collector"
+    static_configs:
+      - targets:
+          - "otel-collector:8888"
+
+  - job_name: "nats-exporter"
+    static_configs:
+      - targets:
+          - "nats-exporter:7777"
+
+  - job_name: "caatsm-receiver"
+    static_configs:
+      - targets:
+          # go-caatsm running on host/WSL, Prometheus in Docker
+          - "host.docker.internal:2112"
+```
+
+When you run the receiver directly on the host/WSL, ensure the monitoring
+server listens on all interfaces so that Docker can reach it, for example via:
+
+```bash
+export CAATSM_MONITORING_ADDR=0.0.0.0:2112
+export CAATSM_MONITORING_ENABLE_METRICS=true
+export CAATSM_MONITORING_ENABLE_HEALTH=true
+```
+
+Alternative topologies:
+
+- **Receiver and Prometheus in the same Docker network**  
+  Expose the monitoring server via a container port and use the container
+  name as the scrape target, e.g. `caatsm-receiver:2112`.
+- **Receiver behind a reverse proxy / load balancer**  
+  Point Prometheus at the proxy address and path that forwards to `/metrics`.
+
 #### CAATSM – Receiver Overview Dashboard
 
 The `caatsm-overview` Grafana dashboard (provisioned from `configs/grafana-dashboards.dev/caatsm-overview.json`) focuses on the CAATSM receiver service and surfaces:
@@ -50,6 +93,36 @@ The `caatsm-overview` Grafana dashboard (provisioned from `configs/grafana-dashb
 - **Retry and permanent failure rates** – from `caatsm_retries_total` and `caatsm_messages_total{result="permanent_fail"}`.  
 - **Publish failures** – from `caatsm_publish_failures_total`.
 
+To validate that the dashboard is receiving data:
+
+1. Check the monitoring endpoint directly:
+
+   ```bash
+   curl -s http://localhost:2112/metrics | grep caatsm_messages_total || true
+   ```
+
+2. In Prometheus (`http://localhost:9090`), run:
+
+   ```text
+   caatsm_messages_total
+   ```
+
+   and
+
+   ```text
+   rate(caatsm_messages_total[5m])
+   ```
+
+3. In Grafana, open the **CAATSM – Receiver Overview** dashboard and
+   verify that:
+
+   - “Messages by result (5m rate)” shows time series for `ok`, `fail`,
+     and `permanent_fail`.
+   - “Messages per stream/consumer” shows series labelled by `stream`
+     and `consumer`.
+   - DB-related panels show non-zero values based on
+     `caatsm_db_queries_total` and `caatsm_db_query_latency_seconds`.
+
 ### Health and Readiness
 
 The monitoring server exposes:
@@ -72,6 +145,37 @@ Tracing is configured via the `telemetry` section:
 - `telemetry.endpoint` – OTLP HTTP endpoint (e.g. `localhost:4318`).  
 - `telemetry.insecure` – disables TLS for local/dev.
 
+#### OTEL vs Prometheus metrics
+
+The receiver reports two complementary sets of metrics:
+
+- **Prometheus metrics via `/metrics`**  
+  Implemented in `internal/observability/metrics`, covering:
+  - End-to-end message handling (`caatsm_messages_total`,
+    `caatsm_handle_latency_seconds`, `caatsm_retries_total`)
+  - DB activity (`caatsm_db_queries_total`,
+    `caatsm_db_query_latency_seconds`)
+  - Legacy per-telegram metrics
+
+- **OpenTelemetry metrics via OTLP**  
+  Implemented using `otel.Meter` in the NATS consumer and app processor,
+  including:
+  - `caatsm_messages_processed_total`
+  - `caatsm_parse_duration_ms`
+  - `caatsm_publish_failures_total`
+  - `caatsm_nats_consumer_ack_pending`
+  - `caatsm_nats_consumer_redelivered`
+  - `caatsm_nats_consumer_pending`
+  - `caatsm_nats_consumer_delivered`
+
+Prometheus only sees the metrics exposed on `/metrics`. OTEL metrics are
+exported to the configured OTEL collector (`telemetry.endpoint`) via OTLP and
+are, by default, forwarded to Jaeger (traces) and logs (metrics) according to
+`configs/otel-collector.dev.yaml`. If you want OTEL metrics to appear in
+Prometheus as well, you can extend the collector configuration with a
+`prometheus` or `prometheusremotewrite` exporter and add a corresponding
+scrape or remote-write configuration.
+
 Key spans:
 
 - `caatsm/nats`  

internal/infra/nats/consumer.go

@@ -130,7 +130,7 @@ func ProvideConsumer(
 	return consumer, nil
 }
 
-// ensureConsumer creates the consumer if it doesn't exist
+// 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,
@@ -158,21 +158,33 @@ func (c *Consumer) ensureConsumer() error {
 		}
 	}
 
-	_, err := c.js.AddConsumer(c.streamName, consumerConfig)
-	if err != nil && err != nats.ErrConsumerNameAlreadyInUse {
-		return fmt.Errorf("failed to create consumer: %w", err)
-	}
-
-	if err == nil {
-		c.logger.Info("Created JetStream consumer",
+	// 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),
-			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
 	}
+	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
 }