Expose actual RabbitMQ dead-letter publish outcomes

README.md

@@ -77,6 +77,10 @@ testing can enable confirmed delivery and call `ReadMessagesToDeadletter`. The
 method preserves the original body and acknowledges it only after the broker
 confirms the dead-letter envelope.
 
+RabbitMQ also offers an optional `RabbitOptions.DeadLetterObserver` callback for
+payload-free final DLQ publish outcomes, including failures and confirmation
+status. See [dead-letter publish outcomes](README.rabbitmq.md#dead-letter-publish-outcomes).
+
 ### Pipeline Actions
 
 Message handlers return a `models.PipelineAction`. The queue client maps that

README.rabbitmq.md

@@ -64,6 +64,7 @@ methods or channels is required.
 | `PrefetchCount` | No | Maximum unacknowledged deliveries; defaults to `5` |
 | `MaxRetries` | No | Retry limit before dead-lettering; defaults to `10` |
 | `ConfirmedDelivery` | No | Provisions the confirmed producer used by explicit confirmed methods; defaults to `false` |
+| `DeadLetterObserver` | No | Optional `func(DeadLetterPublishEvent)` observing final DLQ publish outcomes |
 | `Exchange` | No | Compatibility field; the default exchange is used |
 | `Uri` | No | Compatibility field; connection uses host and credentials |
 | `TLS` | No | Enables an `amqps://` connection |
@@ -147,6 +148,48 @@ rabbit.SetReturnHandler(func(returned amqp.Return) {
 })
 ```
 
+### Dead-letter publish outcomes
+
+Set `RabbitOptions.DeadLetterObserver` or use `SetDeadLetterObserver` before
+creating the client:
+
+```go
+options := queue.NewRabbitOptions().
+    SetDeadLetterObserver(func(event queue.DeadLetterPublishEvent) {
+        // Send these fields to your application's structured logger or metrics.
+        // Do not treat Err == nil alone as proof of durable delivery.
+    })
+```
+
+The callback runs once per logical publish to the configured DLQ, **after** its
+final result (including reconnect), not when a handler returns `PipelineError`.
+It covers malformed messages, handler errors, failed forwarding, exhausted
+retries, forced `ReadMessagesToDeadletter` transfers, `AddToDeadletter`, direct
+`Publish`/`PublishConfirmed`, delayed DLQ publishes, and `PublishDeadLetter`.
+Inspection, scanning, replay, discard, and restoration of retained messages do
+not emit new-entry events. Validation failures before publishing do not emit.
+
+`Source`, `Reason`, and `Attempts` come from the envelope. `Destination` is the
+actual configured DLQ, not legacy envelope destination metadata that may identify
+a replay queue. `TraceID` and `MediaFileName` are extracted only from the inner
+JSON's `traceId` and `payload.key` (falling back to top-level `fileName`).
+Malformed inner bodies still produce outcomes, with empty correlation fields.
+No payload or arbitrary attributes are exposed, and the observer adds no logger
+or tracing dependency.
+
+`Confirmed` is true only when a broker-confirmed publish succeeds (`Err == nil`).
+An unconfirmed publish can return nil before a mandatory return arrives; its
+event remains `Confirmed: false`. Failed publishes retain their error even if
+subsequent disaster recovery succeeds. Observation precedes source settlement,
+so success does not imply the source acknowledgement succeeded. Redelivery can
+produce another logical publish and event.
+
+Callbacks run synchronously outside publishing locks, including on the delayed
+publisher goroutine. They may run concurrently: handlers must be concurrency-safe,
+return promptly, and not panic. A slow callback delays settlement; avoid recursive
+DLQ publishing from the callback. Do not mutate options after starting the client.
+When unset, no envelope or correlation decoding is performed.
+
 ## TLS
 
 Prefixing `Host` with `amqps://` automatically enables TLS. TLS can also be

pkg/queue/deadletter_observer.go

@@ -0,0 +1,65 @@
+package queue
+
+import "encoding/json"
+
+// DeadLetterPublishEvent reports one logical RabbitMQ DLQ publish after its final
+// outcome, including any reconnect attempt, but before delivery settlement or
+// disaster recovery. It contains no message body.
+type DeadLetterPublishEvent struct {
+	Source        string
+	Destination   string // Actual configured DLQ, not an envelope's replay destination.
+	Reason        DeadLetterReason
+	Attempts      int
+	TraceID       string
+	MediaFileName string
+	Confirmed     bool // True only for a broker-confirmed publish with Err == nil.
+	Err           error
+}
+
+func (r *RabbitMQ) publishAndObserveDeadLetter(destination string, body []byte, confirmed bool, publish func() error) error {
+	observer := r.options.DeadLetterObserver
+	if observer == nil || destination != r.options.DeadletterQueue {
+		return publish()
+	}
+
+	err := publish()
+	event := DeadLetterPublishEvent{
+		Source:      r.options.ConsumerQueue,
+		Destination: destination,
+		Reason:      DeadLetterReasonUnspecified,
+		Confirmed:   confirmed && err == nil,
+		Err:         err,
+	}
+	// Project only routing metadata and correlation fields, never the complete
+	// application model (which may contain credentials or other sensitive data).
+	var envelope struct {
+		Schema     string `json:"schema"`
+		Payload    []byte `json:"payload"`
+		DeadLetter struct {
+			Source   string           `json:"source"`
+			Reason   DeadLetterReason `json:"reason"`
+			Attempts int              `json:"attempts"`
+		} `json:"deadLetter"`
+	}
+	if json.Unmarshal(body, &envelope) == nil && envelope.Schema == DeadLetterSchemaV1 {
+		event.Source = envelope.DeadLetter.Source
+		event.Reason = envelope.DeadLetter.Reason
+		event.Attempts = envelope.DeadLetter.Attempts
+		var correlation struct {
+			TraceID  string `json:"traceId"`
+			FileName string `json:"fileName"`
+			Payload  struct {
+				Key string `json:"key"`
+			} `json:"payload"`
+		}
+		if json.Unmarshal(envelope.Payload, &correlation) == nil {
+			event.TraceID = correlation.TraceID
+			event.MediaFileName = correlation.Payload.Key
+			if event.MediaFileName == "" {
+				event.MediaFileName = correlation.FileName
+			}
+		}
+	}
+	observer(event)
+	return err
+}