✨ Implement dead-letter queue (DLQ) functionality for handling permanent failures in message processing. Update configuration to enable DLQ and specify the subject for routing failed messages. Enhance observability by adding metrics for message handling, retries, and database operations. Introduce structured logging for better traceability of message processing events.

configs/config.dev.toml

@@ -63,3 +63,7 @@ enable_health = true
 read_timeout = "5s"
 write_timeout = "5s"
 health_timeout = "2s"
+
+[dlq]
+enabled = true
+subject = "caatsm.dlq"

configs/grafana-dashboards.dev/caatsm-overview.json

@@ -0,0 +1,265 @@
+{
+  "id": null,
+  "uid": "caatsm-overview",
+  "title": "CAATSM – Receiver Overview",
+  "tags": ["caatsm", "receiver", "dev"],
+  "schemaVersion": 38,
+  "version": 1,
+  "refresh": "10s",
+  "time": {
+    "from": "now-15m",
+    "to": "now"
+  },
+  "timepicker": {},
+  "annotations": {
+    "list": [
+      {
+        "builtIn": 1,
+        "datasource": "-- Grafana --",
+        "enable": true,
+        "hide": true,
+        "iconColor": "rgba(0, 211, 255, 1)",
+        "name": "Annotations & Alerts",
+        "type": "dashboard"
+      }
+    ]
+  },
+  "templating": {
+    "list": []
+  },
+  "panels": [
+    {
+      "id": 1,
+      "type": "stat",
+      "title": "Messages by result (5m rate)",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus-dev"
+      },
+      "gridPos": { "h": 4, "w": 8, "x": 0, "y": 0 },
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": ["lastNotNull"],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              { "color": "green", "value": null },
+              { "color": "red", "value": 0 }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "sum by (result) (rate(caatsm_messages_total[5m]))",
+          "legendFormat": "{{result}}"
+        }
+      ]
+    },
+    {
+      "id": 2,
+      "type": "timeseries",
+      "title": "Messages per stream/consumer",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus-dev"
+      },
+      "gridPos": { "h": 8, "w": 16, "x": 8, "y": 0 },
+      "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": "sum by (stream, consumer) (rate(caatsm_messages_total[5m]))",
+          "legendFormat": "{{stream}} / {{consumer}}"
+        }
+      ]
+    },
+    {
+      "id": 3,
+      "type": "timeseries",
+      "title": "End-to-end handle latency (P50/P95/P99)",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus-dev"
+      },
+      "gridPos": { "h": 8, "w": 24, "x": 0, "y": 4 },
+      "fieldConfig": {
+        "defaults": {
+          "color": { "mode": "palette-classic" },
+          "mappings": [],
+          "thresholds": { "mode": "absolute", "steps": [] }
+        },
+        "overrides": []
+      },
+      "options": {
+        "legend": { "displayMode": "table", "placement": "bottom" },
+        "tooltip": { "mode": "single" }
+      },
+      "targets": [
+        {
+          "refId": "P50",
+          "expr": "histogram_quantile(0.50, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))",
+          "legendFormat": "P50"
+        },
+        {
+          "refId": "P95",
+          "expr": "histogram_quantile(0.95, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))",
+          "legendFormat": "P95"
+        },
+        {
+          "refId": "P99",
+          "expr": "histogram_quantile(0.99, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))",
+          "legendFormat": "P99"
+        }
+      ]
+    },
+    {
+      "id": 4,
+      "type": "timeseries",
+      "title": "DB query rate by operation/result",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus-dev"
+      },
+      "gridPos": { "h": 7, "w": 12, "x": 0, "y": 12 },
+      "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": "sum by (operation, result) (rate(caatsm_db_queries_total[5m]))",
+          "legendFormat": "{{operation}} / {{result}}"
+        }
+      ]
+    },
+    {
+      "id": 5,
+      "type": "timeseries",
+      "title": "DB query latency P95 by operation",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus-dev"
+      },
+      "gridPos": { "h": 7, "w": 12, "x": 12, "y": 12 },
+      "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": "histogram_quantile(0.95, sum by (le, operation) (rate(caatsm_db_query_latency_seconds_bucket[5m])))",
+          "legendFormat": "{{operation}}"
+        }
+      ]
+    },
+    {
+      "id": 6,
+      "type": "timeseries",
+      "title": "Retries and permanent failures",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus-dev"
+      },
+      "gridPos": { "h": 7, "w": 12, "x": 0, "y": 19 },
+      "fieldConfig": {
+        "defaults": {
+          "color": { "mode": "palette-classic" },
+          "mappings": [],
+          "thresholds": { "mode": "absolute", "steps": [] }
+        },
+        "overrides": []
+      },
+      "options": {
+        "legend": { "displayMode": "table", "placement": "bottom" },
+        "tooltip": { "mode": "single" }
+      },
+      "targets": [
+        {
+          "refId": "retries",
+          "expr": "sum(rate(caatsm_retries_total[5m]))",
+          "legendFormat": "retries"
+        },
+        {
+          "refId": "permanent_fail",
+          "expr": "sum(rate(caatsm_messages_total{result=\"permanent_fail\"}[5m]))",
+          "legendFormat": "permanent_fail"
+        }
+      ]
+    },
+    {
+      "id": 7,
+      "type": "timeseries",
+      "title": "Publish failures by category",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus-dev"
+      },
+      "gridPos": { "h": 7, "w": 12, "x": 12, "y": 19 },
+      "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": "sum by (message_category) (rate(caatsm_publish_failures_total[5m]))",
+          "legendFormat": "{{message_category}}"
+        }
+      ]
+    }
+  ]
+}
+
+

docs/architecture-ha.md

@@ -0,0 +1,74 @@
+## High-Level Architecture and HA / Scaling
+
+### Components
+
+- **NATS / JetStream** – message broker providing durable storage and redelivery:
+  - Stream: `TELEGRAM` (configurable via `nats.stream`).  
+  - Consumer: durable pull consumer per deployment (`nats.consumer`).  
+  - Subjects:
+    - Inbound telegrams: `subscription.topic` (e.g. `telegram.serial`).  
+    - Parsed telegrams (publisher): `publisher.topic` (e.g. `telegram.json`).  
+    - Dead-letter: `dlq.subject` (e.g. `caatsm.dlq`).
+
+- **Receiver service (`caatsm`)**:
+  - NATS JetStream **pull consumer** (`internal/infra/nats/consumer.go`).  
+  - Telegram parser and domain model (`internal/app`, `internal/parsers`, `internal/domain`).  
+  - PostgreSQL repository (`internal/infra/postgres`).  
+  - Monitoring/observability server (`internal/infra/monitoring`).
+
+### HA and Failover
+
+- NATS/JetStream is expected to run as a **cluster** with `replicas` configured on the stream to ensure message durability.  
+- The receiver service is stateless aside from DB side effects and can be deployed with multiple replicas:
+  - Each replica connects to the same NATS cluster and JetStream stream.  
+  - Durability and at-least-once semantics are handled by JetStream.
+
+Consumer behaviour:
+
+- Pull-based consumption with configurable batch size/timeout (`app.batch_size`, `app.batch_timeout`).  
+- When a receiver instance stops or crashes:
+  - Its NATS connection is drained and closed.  
+  - Remaining messages remain in the stream.  
+  - Another healthy instance continues pulling from the durable consumer.
+
+### Scaling and Rebalancing
+
+Scaling out:
+
+- Increase the number of receiver replicas.  
+- All replicas share the same durable consumer name; for pull-based consumption, each instance independently fetches messages.  
+- JetStream distributes messages across fetch calls; with more instances, aggregate throughput increases.
+
+Scaling in / failure:
+
+- When replicas are reduced or fail, the remaining instances continue to fetch messages.  
+- No explicit rebalancing logic is required in the application; JetStream manages which messages are available for pull.
+
+Tuning:
+
+- **Per-instance throughput** is primarily influenced by:
+  - `app.batch_size`  
+  - `app.batch_timeout`  
+  - the number of concurrent instances
+
+- **Backpressure** is provided through:
+  - JetStream `backoff` and `max_deliver` settings.  
+  - Additional sleeps in the consumer when many consecutive errors occur.  
+  - Readiness checks exposing DB/NATS health.
+
+### Failure Scenarios
+
+1. **DB outage**:
+   - Insert operations fail and are treated as transient.  
+   - Messages are NAKed with delay and the error streak causes additional consumer sleep.  
+   - `/readyz` returns 503, signalling this instance should be removed from traffic.
+
+2. **NATS outage**:
+   - Connection events are logged via `ProvideNATSConn` callbacks.  
+   - The consumer will stop fetching; once NATS is back and reconnected, consumption resumes.
+
+3. **Single instance crash**:
+   - Other instances continue consuming from JetStream.  
+   - No messages are lost; unacked messages remain pending and will be fetched by surviving instances.
+
+

docs/observability.md

@@ -0,0 +1,111 @@
+## Observability
+
+### Metrics
+
+The service exposes Prometheus metrics via the monitoring HTTP server (default `:2112`):
+
+- `caatsm_messages_total{stream,consumer,result}`  
+  Total number of messages handled by the receiver, labelled by stream/consumer and result (`ok`, `fail`, `permanent_fail`, `retry`).
+
+- `caatsm_handle_latency_seconds{stream,consumer}`  
+  End-to-end handling latency from NATS receive to handler completion.
+
+- `caatsm_retries_total{stream,consumer,reason}`  
+  Number of retries (NAKs) issued by the consumer, labelled by reason (e.g. `processor_error`).
+
+- `caatsm_js_api_calls_total{operation}`  
+  JetStream API calls performed by the service.
+
+- `caatsm_db_queries_total{operation,result}`  
+  Database operations grouped by operation (`insert_one`, `insert_batch`, `insert_raw`) and result (`ok`, `error`).
+
+- `caatsm_db_query_latency_seconds{operation}`  
+  DB operation latency.
+
+Additional OTEL metrics are emitted via the configured OTEL endpoint, 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`
+
+These metrics are intended to be scraped by Prometheus (either directly or via the OTEL collector) and visualised in Grafana dashboards. Recommended dashboard panels include:
+
+- Per-stream/consumer message rate and error rate.  
+- Handling latency P50/P95/P99.  
+- NATS consumer backlog and redelivery counts.  
+- DB query rates and latencies.
+
+#### 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:
+
+- **Message throughput by result** – derived from `caatsm_messages_total{result}`.  
+- **Per stream/consumer rates** – `caatsm_messages_total{stream,consumer}`.  
+- **End-to-end handle latency** – P50/P95/P99 from `caatsm_handle_latency_seconds_bucket`.  
+- **DB query rate and latency** – from `caatsm_db_queries_total` and `caatsm_db_query_latency_seconds_bucket`.  
+- **Retry and permanent failure rates** – from `caatsm_retries_total` and `caatsm_messages_total{result="permanent_fail"}`.  
+- **Publish failures** – from `caatsm_publish_failures_total`.
+
+### Health and Readiness
+
+The monitoring server exposes:
+
+- `/healthz` – basic liveness and dependency check.  
+- `/readyz` – readiness endpoint with the same logic as `/healthz`, intended for load balancers / orchestrators.
+
+Checks performed:
+
+- PostgreSQL: `pgxpool.Pool.Ping` with configurable timeout (`monitoring.health_timeout`).  
+- NATS: connection status must be `CONNECTED`.
+
+A non-200 response indicates the service is not healthy/ready and should be removed from traffic.
+
+### Tracing
+
+Tracing is configured via the `telemetry` section:
+
+- `telemetry.enabled` – enables OTEL exporters.  
+- `telemetry.endpoint` – OTLP HTTP endpoint (e.g. `localhost:4318`).  
+- `telemetry.insecure` – disables TLS for local/dev.
+
+Key spans:
+
+- `caatsm/nats`  
+  - `Consumer.processMessage`
+- `caatsm/app`  
+  - `MessageProcessor.Handle`  
+  - `Publisher.Publish`
+- `caatsm/postgres`  
+  - `Repository.InsertOne`  
+  - `Repository.InsertBatch`  
+  - `Repository.InsertRaw`
+
+Important attributes:
+
+- `nats.subject`, `nats.msg_id`, `nats.js.stream_seq`, `nats.js.consumer_seq`  
+- `telegram.message_id`, `telegram.category`, `telegram.status`  
+- `db.table`, `db.inserted`
+
+### Structured Logging Contract
+
+Logging is done with Zap. The `internal/observability/logging` package standardises fields via `MessageFields`:
+
+- `service` – logical component (`caatsm-consumer`, `caatsm-processor` etc.).  
+- `transport_msg_id` – NATS/envelope message ID (derived from `Nats-Msg-Id` or JetStream sequence).  
+- `telegram_message_id` – business telegram message ID from the payload.  
+- `category` – telegram category (ARR, DEP, FPL, etc.).  
+- `stream`, `consumer`, `subject` – JetStream context.  
+- `nats_sequence` – JetStream stream sequence, when available.  
+- `request_id`, `trace_id` – correlation identifiers.  
+- `error_type` – high-level classification:
+  - `business` – payload/validation/domain issues; not suitable for retry.
+  - `transient` – network/DB/NATS glitches that may succeed on retry.
+  - `fatal` – programming errors, schema mismatches, or configuration issues requiring operator attention.
+
+Handler and consumer logs should always be emitted through `WithMessageContext` to ensure these fields are present where applicable.
+
+

docs/reliability.md

@@ -0,0 +1,119 @@
+## Reliability and Fault Handling
+
+This document summarises how the service handles failures, provides resilience, and avoids data loss or duplication.
+
+### Dead-Letter Queue (DLQ) and Poison Messages
+
+Configuration is defined under `dlq`:
+
+```toml
+[dlq]
+enabled = true
+subject = "caatsm.dlq"
+```
+
+- When `dlq.enabled` is `true` and `dlq.subject` is non-empty, **permanent** failures are routed to the DLQ subject.  
+- A permanent failure is indicated by wrapping an error with `app.Permanent` and is treated as a **poison message**.
+
+Behaviour (JetStream mode):
+
+1. The NATS consumer calls `processor.Handle`.  
+2. If an error is returned and `app.IsPermanent(err)` is true:
+   - The original NATS message is copied into a DLQ payload with metadata:
+     - `transport_msg_id` (NATS message ID)  
+     - `subject`, `stream`, `consumer`  
+     - `nats_sequence`, `deliveries`  
+     - `error` (stringified cause)  
+     - `received_at` (DLQ event time)  
+     - `body` (raw message body)
+   - The payload is published to `dlq.subject` using JetStream.  
+   - The original message is **ACKed**, so it will not be redelivered.
+
+The DLQ subject should be consumed by an offline repair/analysis tool or operational dashboard that can:
+
+- Inspect poison messages.  
+- Decide whether to fix and re-publish, or discard with justification.  
+- Track DLQ volume over time for alerting.
+
+### Transient Errors and Backoff
+
+Transient errors (not marked permanent) result in:
+
+- Negative acknowledgements with delay (`NakWithDelay`) according to `nats.consumer_rules.backoff`.  
+- A retry streak counter inside the consumer:
+  - Each transient error increases `consecutiveProcessErrors`.  
+  - After 10 or more consecutive errors, the consumer applies an additional **sleep**:
+    - `backoff = min(consecutive_errors * 100ms, 5s)`.  
+    - A warning log with the sleep duration and error count is emitted.
+
+This combination provides **backpressure** when downstream systems (especially the DB) are in trouble, slowing down consumption instead of aggressively retrying.
+
+### Persistence and Idempotency
+
+The primary persistence path is `Repository.InsertOne` into `aviation.telegrams`. To avoid applying the same business event multiple times, a **minimal idempotency check** is implemented:
+
+- If both `message_id` and `date_time` are non-empty:
+  - `InsertOne` first calls `messageExists(message_id, date_time)`.  
+  - If a row already exists, the insert is **skipped** and an informational log is written.  
+  - Otherwise, the insert proceeds.
+
+This makes repeated delivery of the same telegram (same `message_id`/`date_time`) safe from a business perspective, even if JetStream redelivers messages or upstream replays.
+
+For higher guarantees in production environments, you may:
+
+- Add a unique index on `(message_id, date_time)` at the DB level, and treat any conflict as a duplicate.  
+- Extend the idempotency key with additional fields (e.g. originator, category) if required by the business model.
+
+### DB Degradation and Backpressure
+
+Database write failures in `Repository.InsertOne` and related methods are treated as **transient** by default:
+
+- Errors propagate back to the NATS consumer.  
+- The consumer issues a NAK (with delay) and increases the transient error counter.  
+- When errors persist, the added sleep in the consumer reduces message throughput and gives the DB time to recover.
+
+DB health also feeds into readiness:
+
+- The monitoring server hits `pgxpool.Pool.Ping` on `/readyz` and `/healthz`.  
+- If the DB is not reachable, the endpoints return `503`, signalling to orchestrators that this instance should be drained from traffic.
+
+Together, this yields:
+
+- **Backpressure** via reduced consumption rate and NATS-level backoff.  
+- **Degradation signalling** via health probes for external systems to act upon.
+
+### Retry and Max Deliver
+
+JetStream consumer configuration (via `nats.consumer_rules`) controls:
+
+- `max_deliver` – maximum number of redeliveries before JetStream gives up.  
+- `ack_wait` – how long JetStream waits for an ACK before considering the message pending.  
+- `backoff` – per-attempt delays for `NakWithDelay`.
+
+Recommended pattern:
+
+- Keep `max_deliver` modest (e.g. 5).  
+- Use a backoff array such as `[5s, 30s, 2m]`.  
+- Treat messages that still fail after `max_deliver` as candidates for DLQ, via the permanent error/poison message path where applicable.
+
+### Alerts and Dashboards
+
+Prometheus alert suggestions:
+
+- High failure rate:
+  - `rate(caatsm_messages_total{result!="ok"}[5m])` above a small threshold.  
+  - `rate(caatsm_retries_total[5m])` above a threshold.
+
+- DLQ growth:
+  - Alerts on DLQ stream message count, using NATS/JetStream exporter metrics.
+
+- Readiness / health:
+  - Alert when `/readyz` fails or when DB/NATS checks start failing consistently.
+
+Dashboards should combine:
+
+- Message rates, error rates, and DLQ rates.  
+- NATS consumer statistics (pending, redelivered, ack_pending).  
+- DB health indicators (latency, error counts, connection usage).
+
+

internal/app/processor.go

@@ -4,6 +4,7 @@ import (
 	"caatsm/internal/adapter"
 	"caatsm/internal/adapter/parser"
 	"caatsm/internal/domain"
+	obslogging "caatsm/internal/observability/logging"
 	obsmetrics "caatsm/internal/observability/metrics"
 	"context"
 	"fmt"
@@ -110,6 +111,20 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string)
 		}
 	}
 
+	// Enrich span with parsed telegram information as soon as we have it.
+	span.SetAttributes(
+		attribute.String("telegram.message_id", parsed.MessageID),
+		attribute.String("telegram.category", parsed.Category),
+		attribute.String("telegram.status", string(parsed.Status)),
+	)
+
+	msgLogger := obslogging.WithMessageContext(p.logger, obslogging.MessageFields{
+		Service:        "caatsm-processor",
+		TransportMsgID: msgID,
+		BusinessMsgID:  parsed.MessageID,
+		Category:       parsed.Category,
+	})
+
 	if parseErr != nil || !parsed.Parsed {
 		if parsed.ErrorReason == "" && parseErr != nil {
 			parsed.ErrorReason = parseErr.Error()
@@ -117,9 +132,8 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string)
 		span.RecordError(parseErr)
 		span.SetStatus(codes.Error, parseErr.Error())
 		p.persistRaw(ctx, parsed)
-		p.logger.Warn("Message failed to parse",
+		msgLogger.With(zap.String("status", string(parsed.Status))).
-			zap.String("msg_id", msgID),
+			Warn("Message failed to parse",
-			zap.String("status", string(parsed.Status)),
 				zap.String("content_preview", truncateContent(parsed.Content, 256)),
 				zap.Error(parseErr),
 			)
@@ -143,10 +157,7 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string)
 		attribute.String("telegram.category", parsed.Category),
 	)
 
-	p.logger.Info("Message parsed successfully",
+	msgLogger.Info("Message parsed successfully",
-		zap.String("msg_id", msgID),
-		zap.String("message_id", parsed.MessageID),
-		zap.String("category", parsed.Category),
 		zap.Time("received_at", parsed.ReceivedAt),
 		zap.Time("parsed_at", parsed.ParsedAt),
 	)
@@ -172,8 +183,8 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string)
 	_, pubSpan := tracer.Start(ctx, "Publisher.Publish")
 	if err := p.publisher.Publish(parsed); err != nil {
 		// Log error but don't fail the entire operation
-		p.logger.Error("Failed to publish message",
+		msgLogger.With(zap.String("status", string(parsed.Status))).
-			zap.String("msg_id", msgID),
+			Error("Failed to publish message",
 				zap.Error(err),
 			)
 		pubSpan.RecordError(err)

internal/infra/config/config.go

@@ -21,6 +21,7 @@ type Config struct {
 	Publisher  PublisherConfig  `koanf:"publisher"`
 	Telemetry  TelemetryConfig  `koanf:"telemetry"`
 	Monitoring MonitoringConfig `koanf:"monitoring"`
+	DLQ        DLQConfig        `koanf:"dlq"`
 	// Legacy fields for backward compatibility during migration
 	Subscription SubscriptionConfig `koanf:"subscription"`
 	Timeouts     TimeoutsConfig     `koanf:"timeouts"`
@@ -93,6 +94,14 @@ type TelemetryConfig struct {
 	Insecure bool   `koanf:"insecure"`
 }
 
+// DLQConfig defines the dead-letter queue routing for poison/permanent messages.
+// If Enabled is true and Subject is non-empty, permanent failures will be published
+// to the configured subject for offline processing.
+type DLQConfig struct {
+	Enabled bool   `koanf:"enabled"`
+	Subject string `koanf:"subject"`
+}
+
 // MonitoringConfig controls the lightweight HTTP server that exposes health and metrics endpoints.
 type MonitoringConfig struct {
 	Disabled      bool          `koanf:"disabled"`

internal/infra/monitoring/server.go

@@ -47,7 +47,12 @@ func ProvideServer(
 
 	routes := 0
 	if cfg.Monitoring.EnableHealth {
+		// Liveness: basic process check. For now this reuses the same implementation
+		// as readiness but can diverge in the future if we need a cheaper liveness probe.
 		mux.HandleFunc("/healthz", server.handleHealth)
+		// Readiness: alias to the same implementation so consumers can adopt /readyz
+		// without breaking existing /healthz users.
+		mux.HandleFunc("/readyz", server.handleHealth)
 		routes++
 	}
 	if cfg.Monitoring.EnableMetrics {

internal/infra/nats/consumer.go

@@ -3,7 +3,10 @@ package nats
 import (
 	"caatsm/internal/app"
 	"caatsm/internal/infra/config"
+	obslogging "caatsm/internal/observability/logging"
+	obsmetrics "caatsm/internal/observability/metrics"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
@@ -29,6 +32,7 @@ type Consumer struct {
 	consumerName    string
 	mode            string
 	streamName      string
+	dlqSubject      string
 	ackWait         time.Duration
 	batchSize       int
 	batchTimeout    time.Duration
@@ -38,6 +42,9 @@ type Consumer struct {
 	redelivered     metric.Int64Histogram
 	pending         metric.Int64Histogram
 	delivered       metric.Int64Histogram
+
+	// simple backpressure / degradation state
+	consecutiveProcessErrors int
 }
 
 // ProvideConsumer creates a NATS consumer
@@ -65,6 +72,8 @@ func ProvideConsumer(
 		streamName = "TELEGRAM"
 	}
 
+	dlqSubject := strings.TrimSpace(cfg.DLQ.Subject)
+
 	ackWait := cfg.NATS.ConsumerRules.AckWait
 	if ackWait == 0 {
 		ackWait = cfg.Timeouts.AckWait
@@ -98,6 +107,7 @@ func ProvideConsumer(
 		consumerName:    consumerName,
 		mode:            mode,
 		streamName:      streamName,
+		dlqSubject:      dlqSubject,
 		ackWait:         ackWait,
 		batchSize:       batchSize,
 		batchTimeout:    batchTimeout,
@@ -227,31 +237,74 @@ func (c *Consumer) startJetStream(ctx context.Context) error {
 		// 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
+				}
+				obsmetrics.RecordMessageHandled(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
+				obsmetrics.RecordRetry(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)
+				obsmetrics.RecordMessageHandled(c.streamName, c.consumerName, "ok", elapsed)
 			}
 		}
 	}
@@ -359,16 +412,16 @@ func (c *Consumer) initMetrics() {
 	meter := otel.Meter("caatsm/nats")
 	c.meter = meter
 
-	if hist, err := meter.Int64Histogram("nats.consumer.ack_pending"); err == nil {
+	if hist, err := meter.Int64Histogram("caatsm_nats_consumer_ack_pending"); err == nil {
 		c.ackPending = hist
 	}
-	if hist, err := meter.Int64Histogram("nats.consumer.redelivered"); err == nil {
+	if hist, err := meter.Int64Histogram("caatsm_nats_consumer_redelivered"); err == nil {
 		c.redelivered = hist
 	}
-	if hist, err := meter.Int64Histogram("nats.consumer.pending"); err == nil {
+	if hist, err := meter.Int64Histogram("caatsm_nats_consumer_pending"); err == nil {
 		c.pending = hist
 	}
-	if hist, err := meter.Int64Histogram("nats.consumer.delivered"); err == nil {
+	if hist, err := meter.Int64Histogram("caatsm_nats_consumer_delivered"); err == nil {
 		c.delivered = hist
 	}
 }
@@ -391,6 +444,52 @@ func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.Consume
 	}
 }
 
+// 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 {
+		return fmt.Errorf("marshal dlq payload: %w", err)
+	}
+
+	if _, err := c.js.Publish(c.dlqSubject, data); err != nil {
+		return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err)
+	}
+
+	return nil
+}
+
 func (c *Consumer) nakWithStrategy(msg *nats.Msg) error {
 	backoff := c.cfg.NATS.ConsumerRules.Backoff
 	if len(backoff) == 0 {
@@ -439,10 +538,28 @@ func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error {
 		)
 	}
 
-	c.logger.Debug("Processing message",
+	// Attach structured logging context including stream/consumer and NATS metadata.
-		zap.String("subject", msg.Subject),
+	jsSeq := uint64(0)
-		zap.String("msg_id", msgID),
+	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 := obslogging.WithMessageContext(c.logger, obslogging.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

internal/infra/postgres/repository.go

@@ -4,6 +4,7 @@ import (
 	"caatsm/internal/adapter"
 	"caatsm/internal/adapter/mapper"
 	"caatsm/internal/domain"
+	obsmetrics "caatsm/internal/observability/metrics"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -40,6 +41,25 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e
 	defer span.End()
 	span.SetAttributes(attribute.String("db.table", "aviation.telegrams"))
 
+	// Optional idempotency check based on business message identity. If we have a
+	// non-empty message ID and date/time, we can cheaply skip duplicates here to
+	// avoid applying the same business event multiple times.
+	if msg != nil && msg.MessageID != "" && msg.DateTime != "" {
+		exists, err := r.messageExists(ctx, msg.MessageID, msg.DateTime)
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+			return fmt.Errorf("failed to check existing message: %w", err)
+		}
+		if exists {
+			r.logger.Info("Duplicate message detected by message_id/date_time; skipping insert",
+				zap.String("message_id", msg.MessageID),
+				zap.String("date_time", msg.DateTime),
+			)
+			return nil
+		}
+	}
+
 	row, err := r.mapper.ToDBRow(msg)
 	if err != nil {
 		span.RecordError(err)
@@ -59,16 +79,24 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e
 		ON CONFLICT (uuid, received_at) DO NOTHING
 	`
 
+	start := time.Now()
 	tag, err := r.pool.Exec(ctx, query,
 		row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[8],
 		row[9], row[10], row[11], row[12], row[13], row[14], row[15], row[16],
 	)
+	elapsed := time.Since(start)
+
+	result := obsmetrics.DBResultOK
 	if err != nil {
+		result = obsmetrics.DBResultError
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
+		obsmetrics.RecordDBQuery("insert_one", result, elapsed)
 		return fmt.Errorf("failed to insert message: %w", err)
 	}
 
+	obsmetrics.RecordDBQuery("insert_one", result, elapsed)
+
 	if tag.RowsAffected() == 0 {
 		r.logger.Info("Duplicate message skipped",
 			zap.String("uuid", msg.Uuid),
@@ -106,6 +134,7 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa
 	}
 
 	// Use CopyFrom for efficient batch insert
+	start := time.Now()
 	copyCount, err := r.pool.CopyFrom(
 		ctx,
 		pgx.Identifier{"aviation", "telegrams"},
@@ -117,12 +146,19 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa
 		},
 		pgx.CopyFromRows(rows),
 	)
+	elapsed := time.Since(start)
+
+	result := obsmetrics.DBResultOK
 	if err != nil {
+		result = obsmetrics.DBResultError
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
+		obsmetrics.RecordDBQuery("insert_batch", result, elapsed)
 		return fmt.Errorf("failed to batch insert messages: %w", err)
 	}
 
+	obsmetrics.RecordDBQuery("insert_batch", result, elapsed)
+
 	span.SetAttributes(attribute.Int64("db.inserted", copyCount))
 	r.logger.Info("Batch inserted messages",
 		zap.Int("count", int(copyCount)),
@@ -174,6 +210,7 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) e
 			metadata = EXCLUDED.metadata
 	`
 
+	start := time.Now()
 	_, err = r.pool.Exec(ctx, query,
 		msg.Uuid,
 		string(msg.Status),
@@ -182,12 +219,19 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) e
 		msg.ReceivedAt,
 		metadataJSON,
 	)
+	elapsed := time.Since(start)
+
+	result := obsmetrics.DBResultOK
 	if err != nil {
+		result = obsmetrics.DBResultError
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
+		obsmetrics.RecordDBQuery("insert_raw", result, elapsed)
 		return fmt.Errorf("failed to insert raw telegram: %w", err)
 	}
 
+	obsmetrics.RecordDBQuery("insert_raw", result, elapsed)
+
 	span.SetAttributes(attribute.String("telegram.uuid", msg.Uuid), attribute.String("telegram.status", string(msg.Status)))
 	r.logger.Debug("Persisted raw telegram",
 		zap.String("uuid", msg.Uuid),
@@ -196,3 +240,29 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) e
 
 	return nil
 }
+
+// messageExists performs a lightweight existence check for a business message,
+// using (message_id, date_time) as the idempotency key. This avoids requiring a
+// unique constraint at the database level while still preventing duplicate effects.
+func (r *Repository) messageExists(ctx context.Context, messageID, dateTime string) (bool, error) {
+	if messageID == "" || dateTime == "" {
+		return false, nil
+	}
+
+	const query = `
+		SELECT 1
+		FROM aviation.telegrams
+		WHERE message_id = $1 AND date_time = $2
+		LIMIT 1
+	`
+
+	var one int
+	if err := r.pool.QueryRow(ctx, query, messageID, dateTime).Scan(&one); err != nil {
+		if err == pgx.ErrNoRows {
+			return false, nil
+		}
+		return false, err
+	}
+
+	return true, nil
+}

internal/observability/logging/logger.go

@@ -0,0 +1,102 @@
+package logging
+
+import "go.uber.org/zap"
+
+// ErrorType represents a coarse-grained categorisation of errors for logging and alerting.
+// Typical values:
+//   - business: validation failures, domain rule violations, payload issues.
+//   - transient: network / DB / NATS glitches that may succeed on retry.
+//   - fatal: programming bugs, schema mismatches, or conditions that require operator action.
+type ErrorType string
+
+const (
+	ErrorTypeBusiness  ErrorType = "business"
+	ErrorTypeTransient ErrorType = "transient"
+	ErrorTypeFatal     ErrorType = "fatal"
+)
+
+// Canonical logging field names for structured logs produced by the CAATSM
+// receiver. Using constants avoids scattering magic strings and keeps log
+// analysis queries stable over time.
+const (
+	FieldService        = "service"
+	FieldTransportMsgID = "transport_msg_id"
+	FieldTelegramMsgID  = "telegram_message_id"
+	FieldCategory       = "category"
+	FieldStream         = "stream"
+	FieldConsumer       = "consumer"
+	FieldSubject        = "subject"
+	FieldNATSSequence   = "nats_sequence"
+	FieldRequestID      = "request_id"
+	FieldTraceID        = "trace_id"
+	FieldErrorType      = "error_type"
+)
+
+// MessageFields captures the common structured logging contract for message-processing logs.
+// All fields are optional; empty values will simply be skipped.
+type MessageFields struct {
+	// Identifiers
+	Service        string
+	TransportMsgID string
+	BusinessMsgID  string
+	Category       string
+
+	// NATS / JetStream context
+	Stream     string
+	Consumer   string
+	Subject    string
+	JSSequence uint64
+
+	// Correlation / tracing
+	RequestID string
+	TraceID   string
+
+	// Error classification
+	ErrorType ErrorType
+}
+
+// WithMessageContext returns a logger pre-populated with the structured fields defined in MessageFields.
+// This is the primary entry point for enforcing the logging contract in the codebase.
+func WithMessageContext(logger *zap.Logger, mf MessageFields) *zap.Logger {
+	if logger == nil {
+		return zap.NewNop()
+	}
+
+	fields := make([]zap.Field, 0, 12)
+
+	if mf.Service != "" {
+		fields = append(fields, zap.String(FieldService, mf.Service))
+	}
+	if mf.TransportMsgID != "" {
+		fields = append(fields, zap.String(FieldTransportMsgID, mf.TransportMsgID))
+	}
+	if mf.BusinessMsgID != "" {
+		fields = append(fields, zap.String(FieldTelegramMsgID, mf.BusinessMsgID))
+	}
+	if mf.Category != "" {
+		fields = append(fields, zap.String(FieldCategory, mf.Category))
+	}
+	if mf.Stream != "" {
+		fields = append(fields, zap.String(FieldStream, mf.Stream))
+	}
+	if mf.Consumer != "" {
+		fields = append(fields, zap.String(FieldConsumer, mf.Consumer))
+	}
+	if mf.Subject != "" {
+		fields = append(fields, zap.String(FieldSubject, mf.Subject))
+	}
+	if mf.JSSequence > 0 {
+		fields = append(fields, zap.Uint64(FieldNATSSequence, mf.JSSequence))
+	}
+	if mf.RequestID != "" {
+		fields = append(fields, zap.String(FieldRequestID, mf.RequestID))
+	}
+	if mf.TraceID != "" {
+		fields = append(fields, zap.String(FieldTraceID, mf.TraceID))
+	}
+	if mf.ErrorType != "" {
+		fields = append(fields, zap.String(FieldErrorType, string(mf.ErrorType)))
+	}
+
+	return logger.With(fields...)
+}

internal/observability/metrics/metrics.go

@@ -11,31 +11,130 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
+// Metric and label key/value contracts for the CAATSM receiver. Centralising these
+// names avoids scattering magic strings across the codebase and keeps PromQL and
+// documentation aligned with the implementation.
+const (
+	// Metric names.
+	MetricProcessedTotal        = "caatsm_processed_total"
+	MetricFailuresTotal         = "caatsm_failures_total"
+	MetricParseLatencySeconds   = "caatsm_parse_latency_seconds"
+	MetricMessagesTotal         = "caatsm_messages_total"
+	MetricHandleLatencySeconds  = "caatsm_handle_latency_seconds"
+	MetricRetriesTotal          = "caatsm_retries_total"
+	MetricJSAPICallsTotal       = "caatsm_js_api_calls_total"
+	MetricDBQueriesTotal        = "caatsm_db_queries_total"
+	MetricDBQueryLatencySeconds = "caatsm_db_query_latency_seconds"
+
+	// Common label keys.
+	LabelStatus    = "status"
+	LabelCategory  = "category"
+	LabelStage     = "stage"
+	LabelStream    = "stream"
+	LabelConsumer  = "consumer"
+	LabelResult    = "result"
+	LabelReason    = "reason"
+	LabelOperation = "operation"
+
+	// Standard result label values for caatsm_messages_total.
+	ResultOK            = "ok"
+	ResultFail          = "fail"
+	ResultPermanentFail = "permanent_fail"
+
+	// Standard result values for DB operations.
+	DBResultOK    = "ok"
+	DBResultError = "error"
+
+	// Standard retry reasons.
+	RetryReasonProcessorError = "processor_error"
+)
+
 var (
 	once sync.Once
+
 	registry *prometheus.Registry
+
+	// Legacy metrics (kept for backward compatibility).
 	processedCounter *prometheus.CounterVec
 	failureCounter   *prometheus.CounterVec
 	parseLatency     *prometheus.HistogramVec
+
+	// Message handling metrics (per stream / consumer).
+	messagesTotal   *prometheus.CounterVec
+	handleLatency   *prometheus.HistogramVec
+	retriesTotal    *prometheus.CounterVec
+	jsAPICallsTotal *prometheus.CounterVec
+
+	// Database metrics.
+	dbQueriesTotal *prometheus.CounterVec
+	dbQueryLatency *prometheus.HistogramVec
 )
 
 func initCollectors() {
 	registry = prometheus.NewRegistry()
+
+	// Legacy metrics.
 	processedCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
-		Name: "caatsm_processed_total",
+		Name: MetricProcessedTotal,
 		Help: "Count of telegrams processed by status and category.",
-	}, []string{"status", "category"})
+	}, []string{LabelStatus, LabelCategory})
+
 	failureCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
-		Name: "caatsm_failures_total",
+		Name: MetricFailuresTotal,
 		Help: "Count of processor failures by stage (parser, repository, publisher).",
-	}, []string{"stage"})
+	}, []string{LabelStage})
+
 	parseLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Name:    "caatsm_parse_latency_seconds",
+		Name:    MetricParseLatencySeconds,
 		Help:    "Latency between reception and parse completion.",
 		Buckets: prometheus.DefBuckets,
-	}, []string{"status", "category"})
+	}, []string{LabelStatus, LabelCategory})
 
-	registry.MustRegister(processedCounter, failureCounter, parseLatency)
+	// New message handling metrics.
+	messagesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: MetricMessagesTotal,
+		Help: "Total number of messages handled by the receiver, labelled by stream, consumer and result.",
+	}, []string{LabelStream, LabelConsumer, LabelResult})
+
+	handleLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    MetricHandleLatencySeconds,
+		Help:    "Latency of end-to-end message handling in seconds, from NATS receive to handler completion.",
+		Buckets: prometheus.DefBuckets,
+	}, []string{LabelStream, LabelConsumer})
+
+	retriesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: MetricRetriesTotal,
+		Help: "Total number of message retries (negative acknowledgements), labelled by stream, consumer and reason.",
+	}, []string{LabelStream, LabelConsumer, LabelReason})
+
+	jsAPICallsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: MetricJSAPICallsTotal,
+		Help: "Count of JetStream API calls made by the receiver.",
+	}, []string{LabelOperation})
+
+	// Database metrics.
+	dbQueriesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: MetricDBQueriesTotal,
+		Help: "Total number of database operations, labelled by operation and result.",
+	}, []string{LabelOperation, LabelResult})
+
+	dbQueryLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    MetricDBQueryLatencySeconds,
+		Help:    "Latency of database operations in seconds, labelled by operation.",
+		Buckets: prometheus.DefBuckets,
+	}, []string{LabelOperation})
+
+	registry.MustRegister(
+		processedCounter,
+		failureCounter,
+		parseLatency,
+		messagesTotal,
+		handleLatency,
+		retriesTotal,
+		jsAPICallsTotal,
+		dbQueriesTotal,
+		dbQueryLatency,
+	)
 }
 
 func ensureCollectors() {
@@ -62,6 +161,42 @@ func RecordFailure(stage string) {
 	failureCounter.WithLabelValues(labelValue(stage)).Inc()
 }
 
+// RecordMessageHandled records end-to-end message handling metrics (per stream / consumer).
+// Result is expected to be values such as "ok", "fail", or "retry".
+func RecordMessageHandled(stream, consumer, result string, elapsed time.Duration) {
+	ensureCollectors()
+	if elapsed < 0 {
+		elapsed = 0
+	}
+	messagesTotal.WithLabelValues(labelValue(stream), labelValue(consumer), labelValue(result)).Inc()
+	handleLatency.WithLabelValues(labelValue(stream), labelValue(consumer)).Observe(elapsed.Seconds())
+}
+
+// RecordRetry increments the retry counter for a message that is being negatively acknowledged.
+// Reason can capture the high-level cause, e.g. "processor_error" or "nats_timeout".
+func RecordRetry(stream, consumer, reason string) {
+	ensureCollectors()
+	retriesTotal.WithLabelValues(labelValue(stream), labelValue(consumer), labelValue(reason)).Inc()
+}
+
+// RecordDBQuery records metrics for a single database operation.
+// Operation examples: "insert_one", "insert_batch", "insert_raw".
+// Result is usually "ok" or "error".
+func RecordDBQuery(operation, result string, elapsed time.Duration) {
+	ensureCollectors()
+	if elapsed < 0 {
+		elapsed = 0
+	}
+	dbQueriesTotal.WithLabelValues(labelValue(operation), labelValue(result)).Inc()
+	dbQueryLatency.WithLabelValues(labelValue(operation)).Observe(elapsed.Seconds())
+}
+
+// RecordJSAPICall increments the JetStream API call counter for the given operation.
+func RecordJSAPICall(operation string) {
+	ensureCollectors()
+	jsAPICallsTotal.WithLabelValues(labelValue(operation)).Inc()
+}
+
 func labelValue(value string) string {
 	value = strings.TrimSpace(value)
 	if value == "" {