Skip to main content

mako_engine/
dead_letter.rs

1//! Dead-letter sink for unroutable or unprocessable inbound messages.
2//!
3//! BDEW AS4 requires that every received message is either processed or
4//! explicitly refused (CONTRL negative acknowledgement). Messages that
5//! cannot be routed to a workflow — because the PID is unknown, the
6//! conversation is not in-flight, or the format version has no adapter —
7//! must not be silently dropped.
8//!
9//! # Design
10//!
11//! [`DeadLetterSink`] is a synchronous trait that receives a structured
12//! [`DeadLetterReason`] for every rejected message. The synchronous contract
13//! keeps dispatch-path hot code fast; implementations that need async work
14//! (e.g. persisting to a durable DLQ or sending a CONTRL) can use
15//! `tokio::spawn` internally.
16//!
17//! # Implementations
18//!
19//! | Type | Behaviour |
20//! |------|-----------|
21//! | [`LogDeadLetterSink`] | Emits structured `tracing::warn!`; suitable for all deployments |
22//! | [`NoopDeadLetterSink`] | Silently discards; **only for testing** |
23//!
24//! # Wiring
25//!
26//! Pass an implementation to [`EngineBuilder::with_dead_letter_sink`].
27//! The default is [`LogDeadLetterSink`] so unroutable messages are always
28//! visible in the log output without any configuration.
29//!
30//! ```rust
31//! use mako_engine::dead_letter::{AuditContext, DeadLetterReason, DeadLetterSink, LogDeadLetterSink};
32//! use mako_engine::ids::Pid;
33//!
34//! let sink = LogDeadLetterSink;
35//! sink.reject(&DeadLetterReason::UnknownPid { pid: Pid::new(99999), context: AuditContext::now() });
36//! ```
37//!
38//! [`EngineBuilder::with_dead_letter_sink`]: crate::builder::EngineBuilder::with_dead_letter_sink
39
40use std::sync::Arc;
41
42use time_tz::{OffsetDateTimeExt as _, timezones};
43
44// ── AuditContext ──────────────────────────────────────────────────────────────
45
46/// Structured audit context attached to every dead-letter event.
47///
48/// All fields are `Option` because they are only partially known at rejection
49/// time (e.g. `pid` is not available for a parse failure before the PID is
50/// decoded). Callers fill in as many fields as they have.
51///
52/// These fields map to § 147 AO / GoBD audit-log requirements for AS4 message
53/// rejection events:
54///
55/// | Field | § 147 AO / GoBD requirement |
56/// |---|---|
57/// | `message_type` | Nachrichtentyp (UTILMD, MSCONS, APERAK, …) |
58/// | `release_code` | Releasekennung (S2.1, G1.1, 2.4c, …) |
59/// | `pid` | Prüfidentifikator |
60/// | `sender_eic` | GLN des Absenders |
61/// | `receiver_eic` | GLN des Empfängers |
62/// | `message_ref` | UNH-Referenz |
63/// | `process_id` | Geschäftsvorfallkennung |
64/// | `tenant_id` | Mandant |
65/// | `correlation_id` | AS4 `ConversationId` or similar |
66/// | `timestamp` | Zeitstempel des Eingangs (German local time) |
67#[derive(Debug, Clone)]
68pub struct AuditContext {
69    /// EDIFACT message type (e.g. `"UTILMD"`, `"MSCONS"`, `"APERAK"`).
70    pub message_type: Option<String>,
71    /// BDEW release code (e.g. `"S2.1"`, `"G1.1"`, `"2.4c"`).
72    pub release_code: Option<String>,
73    /// BDEW Prüfidentifikator numeric code.
74    pub pid: Option<crate::ids::Pid>,
75    /// GLN of the AS4 sender.
76    pub sender_eic: Option<String>,
77    /// GLN of the AS4 receiver.
78    pub receiver_eic: Option<String>,
79    /// UNH message reference (interchange + message ref).
80    pub message_ref: Option<String>,
81    /// Internal process / workflow stream ID.
82    pub process_id: Option<String>,
83    /// Tenant identifier (Mandant).
84    pub tenant_id: Option<String>,
85    /// AS4 ConversationId or engine correlation key.
86    pub correlation_id: Option<String>,
87    /// Timestamp of message receipt, in German local time (CET/CEST).
88    pub timestamp: time::OffsetDateTime,
89}
90
91impl AuditContext {
92    /// Create an `AuditContext` with only a timestamp, all other fields `None`.
93    ///
94    /// The timestamp is set to the current wall-clock time in **German local time**
95    /// (CET = UTC+1 in winter, CEST = UTC+2 in summer), satisfying the § 147 AO / GoBD
96    /// requirement for German-timezone audit records.
97    ///
98    /// Use builder-style setters to fill in known fields:
99    /// ```rust
100    /// use mako_engine::dead_letter::AuditContext;
101    /// use mako_engine::ids::Pid;
102    ///
103    /// let ctx = AuditContext::now()
104    ///     .with_message_type("UTILMD")
105    ///     .with_pid(Pid::new(55001))
106    ///     .with_sender_eic("4012345000023");
107    /// ```
108    #[must_use]
109    pub fn now() -> Self {
110        let berlin = timezones::db::europe::BERLIN;
111        Self {
112            message_type: None,
113            release_code: None,
114            pid: None,
115            sender_eic: None,
116            receiver_eic: None,
117            message_ref: None,
118            process_id: None,
119            tenant_id: None,
120            correlation_id: None,
121            // Use Berlin local time so audit records align with the German
122            // regulatory clock — an off-by-one-hour error at DST transitions
123            // is a reportable BNetzA regulatory violation.
124            timestamp: time::OffsetDateTime::now_utc().to_timezone(berlin),
125        }
126    }
127
128    /// Populate an `AuditContext` from an interchange header and known optional fields.
129    ///
130    /// Fills in `sender_eic`, `receiver_eic`, and `message_ref` (interchange control
131    /// reference) from the parsed UNB header.  All remaining fields (pid, process_id,
132    /// tenant_id, correlation_id) are `None` and should be set via builder setters
133    /// when available.
134    ///
135    /// Satisfies the § 147 AO / GoBD requirement that every dead-letter record carries at
136    /// minimum the sender GLN, receiver GLN, and interchange reference.
137    #[must_use]
138    pub fn from_interchange(sender_id: &str, receiver_id: &str, control_ref: &str) -> Self {
139        Self::now()
140            .with_sender_eic(sender_id)
141            .with_receiver_eic(receiver_id)
142            .with_message_ref(control_ref)
143    }
144
145    /// Set the message type.
146    #[must_use]
147    pub fn with_message_type(mut self, mt: impl Into<String>) -> Self {
148        self.message_type = Some(mt.into());
149        self
150    }
151
152    /// Set the BDEW release code.
153    #[must_use]
154    pub fn with_release_code(mut self, rc: impl Into<String>) -> Self {
155        self.release_code = Some(rc.into());
156        self
157    }
158
159    /// Set the Prüfidentifikator.
160    #[must_use]
161    pub fn with_pid(mut self, pid: crate::ids::Pid) -> Self {
162        self.pid = Some(pid);
163        self
164    }
165
166    /// Set the sender GLN.
167    #[must_use]
168    pub fn with_sender_eic(mut self, eic: impl Into<String>) -> Self {
169        self.sender_eic = Some(eic.into());
170        self
171    }
172
173    /// Set the receiver GLN.
174    #[must_use]
175    pub fn with_receiver_eic(mut self, eic: impl Into<String>) -> Self {
176        self.receiver_eic = Some(eic.into());
177        self
178    }
179
180    /// Set the UNH message reference.
181    #[must_use]
182    pub fn with_message_ref(mut self, r: impl Into<String>) -> Self {
183        self.message_ref = Some(r.into());
184        self
185    }
186
187    /// Set the internal process / stream ID.
188    #[must_use]
189    pub fn with_process_id(mut self, id: impl Into<String>) -> Self {
190        self.process_id = Some(id.into());
191        self
192    }
193
194    /// Set the tenant identifier.
195    #[must_use]
196    pub fn with_tenant_id(mut self, id: impl Into<String>) -> Self {
197        self.tenant_id = Some(id.into());
198        self
199    }
200
201    /// Set the AS4 correlation / conversation ID.
202    #[must_use]
203    pub fn with_correlation_id(mut self, id: impl Into<String>) -> Self {
204        self.correlation_id = Some(id.into());
205        self
206    }
207}
208
209impl Default for AuditContext {
210    fn default() -> Self {
211        Self::now()
212    }
213}
214
215// ── DeadLetterReason ──────────────────────────────────────────────────────────
216
217/// Structured reason why an inbound message was rejected.
218///
219/// The variant gives the dispatch path enough information to emit an
220/// actionable CONTRL or log entry. Each variant carries an [`AuditContext`]
221/// with the § 147 AO / GoBD fields required for regulatory audit logging.
222///
223/// Adding new variants is a non-breaking change thanks to `#[non_exhaustive]`.
224#[derive(Debug, Clone)]
225#[non_exhaustive]
226pub enum DeadLetterReason {
227    /// No workflow is registered for this PID in the [`PidRouter`].
228    ///
229    /// The PID is either from a future BDEW release not yet deployed or a
230    /// malformed message. Respond with a CONTRL negative acknowledgement.
231    ///
232    /// [`PidRouter`]: crate::pid_router::PidRouter
233    UnknownPid {
234        /// The numeric Prüfidentifikator that had no registered workflow.
235        pid: crate::ids::Pid,
236        /// § 147 AO / GoBD structured audit context.
237        context: AuditContext,
238    },
239
240    /// No in-flight process matched the inbound `conversation_id`.
241    ///
242    /// This typically means the process completed, was never started, or
243    /// the [`ProcessRegistry`] was lost on restart (see.
244    ///
245    /// [`ProcessRegistry`]: crate::registry::ProcessRegistry
246    UnknownConversation {
247        /// The `conversation_id` from the inbound EDIFACT interchange.
248        conversation_id: String,
249        /// § 147 AO / GoBD structured audit context.
250        context: AuditContext,
251    },
252
253    /// The message's format version has no registered [`MessageAdapter`].
254    ///
255    /// Either the adapter registry is incomplete (see or the sender
256    /// is using a deprecated / future format version.
257    ///
258    /// [`MessageAdapter`]: crate::message_adapter::MessageAdapter
259    VersionMismatch {
260        /// The format version string the adapter registry expected.
261        expected: String,
262        /// The format version string carried in the inbound message.
263        received: String,
264        /// § 147 AO / GoBD structured audit context.
265        context: AuditContext,
266    },
267
268    /// A message with this inbox key was already accepted (AS4 duplicate).
269    ///
270    /// The AS4 sender retries for up to 72 hours. The [`InboxStore`]
271    /// detected the duplicate and the message must not be processed again.
272    ///
273    /// [`InboxStore`]: crate::inbox::InboxStore
274    DuplicateMessage {
275        /// The inbox deduplication key (typically the AS4 `MessageId`).
276        inbox_key: String,
277        /// § 147 AO / GoBD structured audit context.
278        context: AuditContext,
279    },
280
281    /// A workflow or adapter returned a processing error.
282    ///
283    /// The message was routed correctly but could not be processed. Use
284    /// this variant when the failure is definitive (not retriable).
285    ProcessingError {
286        /// Short, human-readable description of the failure.
287        message: String,
288        /// § 147 AO / GoBD structured audit context.
289        context: AuditContext,
290    },
291
292    /// An interchange flagged with UNB DE0035 = 1 (test indicator) was received
293    /// on a production endpoint.
294    ///
295    /// Per Allgemeine Festlegungen V6.1d §3, test interchanges **must not** be
296    /// processed as production. The interchange is rejected at the ingest boundary
297    /// without being forwarded to any workflow.
298    TestMessage {
299        /// § 147 AO / GoBD structured audit context (contains sender, receiver, control_ref).
300        context: AuditContext,
301    },
302
303    /// A business message arrived without the `NAD` party the answer is
304    /// addressed with, or the one the Sparte is resolved from.
305    ///
306    /// BDEW Allgemeine Festlegungen V6.1d §2.13 identifies the *fachliche*
307    /// sender and receiver on message level in `NAD+MS` / `NAD+MR` DE 3035 and
308    /// states the approach applies "für alle EDI@Energy EDIFACT Nachrichten und
309    /// -dateien einheitlich". Both are load-bearing here: the sender is who the
310    /// answer and the APERAK go back to, and the receiver is which of the
311    /// operator's own MP-IDs — and therefore which Sparte and which Marktrolle —
312    /// the message is addressed to.
313    ///
314    /// Substituting an empty MP-ID produces a process whose answer is addressed
315    /// to nobody and whose Sparte silently defaults, so the message is refused
316    /// at the boundary instead. CONTRL carries no `NAD` at all — it is a
317    /// UN/EDIFACT syntax acknowledgement rather than an EDI@Energy business
318    /// message — and is exempt.
319    MissingInterchangeParty {
320        /// `"MS"` or `"MR"` — which of the two was absent or empty.
321        qualifier: &'static str,
322        /// § 147 AO / GoBD structured audit context.
323        context: AuditContext,
324    },
325
326    /// The PID resolved to a workflow, but the ingest dispatcher has no arm
327    /// for it — so `makod` claimed to route the message and then dropped it.
328    ///
329    /// Always a coverage bug on this side, never a defect in the sender's
330    /// message: [`PidRouter`] answered, the transport acknowledged, and the
331    /// business payload went nowhere. It is recorded rather than logged because
332    /// an acknowledged inbound message that produced no process is exactly what
333    /// § 147 AO / GoBD require a trace of.
334    ///
335    /// [`PidRouter`]: crate::pid_router::PidRouter
336    NotDispatchable {
337        /// The workflow name [`PidRouter`] resolved the PID to.
338        ///
339        /// [`PidRouter`]: crate::pid_router::PidRouter
340        workflow_name: String,
341        /// The Prüfidentifikator carried by the message.
342        pid: crate::ids::Pid,
343        /// Machine-readable skip reason from the dispatcher.
344        reason: String,
345        /// § 147 AO / GoBD structured audit context.
346        context: AuditContext,
347    },
348
349    /// The outbox delivery worker gave up after exhausting all retry attempts.
350    ///
351    /// The message was re-queued `max_attempts` times and never successfully
352    /// delivered to the AS4 endpoint (or ERP webhook). The message is removed
353    /// from the outbox and recorded here for regulatory audit.
354    OutboxExhausted {
355        /// The outbox message ID of the undeliverable message.
356        message_id: crate::ids::OutboxMessageId,
357        /// The message type (e.g. `"APERAK"`, `"CONTRL"`).
358        message_type: String,
359        /// The intended recipient GLN.
360        recipient: String,
361        /// The last error returned by the AS4 sender.
362        last_error: String,
363        /// How many delivery attempts were made.
364        attempts: u32,
365    },
366}
367
368impl DeadLetterReason {
369    /// Short label identifying the rejection category.
370    ///
371    /// Suitable for structured log fields and metric labels.
372    #[must_use]
373    pub fn label(&self) -> &'static str {
374        match self {
375            Self::UnknownPid { .. } => "unknown_pid",
376            Self::UnknownConversation { .. } => "unknown_conversation",
377            Self::VersionMismatch { .. } => "version_mismatch",
378            Self::DuplicateMessage { .. } => "duplicate_message",
379            Self::ProcessingError { .. } => "processing_error",
380            Self::TestMessage { .. } => "test_message",
381            Self::MissingInterchangeParty { .. } => "missing_interchange_party",
382            Self::NotDispatchable { .. } => "not_dispatchable",
383            Self::OutboxExhausted { .. } => "outbox_exhausted",
384        }
385    }
386
387    /// Return the [`AuditContext`] embedded in this reason, if present.
388    ///
389    /// `OutboxExhausted` does not carry an `AuditContext` because it refers
390    /// to an outbound message (not an inbound AS4 message).
391    #[must_use]
392    pub fn audit_context(&self) -> Option<&AuditContext> {
393        match self {
394            Self::UnknownPid { context, .. }
395            | Self::UnknownConversation { context, .. }
396            | Self::VersionMismatch { context, .. }
397            | Self::DuplicateMessage { context, .. }
398            | Self::ProcessingError { context, .. }
399            | Self::TestMessage { context, .. }
400            | Self::MissingInterchangeParty { context, .. }
401            | Self::NotDispatchable { context, .. } => Some(context),
402            Self::OutboxExhausted { .. } => None,
403        }
404    }
405}
406
407impl std::fmt::Display for DeadLetterReason {
408    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
409        match self {
410            Self::UnknownPid { pid, .. } => write!(f, "unknown PID {pid}"),
411            Self::UnknownConversation {
412                conversation_id, ..
413            } => {
414                write!(f, "unknown conversation {conversation_id}")
415            }
416            Self::VersionMismatch {
417                expected, received, ..
418            } => write!(
419                f,
420                "version mismatch: expected {expected}, received {received}"
421            ),
422            Self::DuplicateMessage { inbox_key, .. } => write!(f, "duplicate message {inbox_key}"),
423            Self::ProcessingError { message, .. } => write!(f, "processing error: {message}"),
424            Self::TestMessage { context } => write!(
425                f,
426                "test interchange rejected (DE0035=1): sender={}, receiver={}, ref={}",
427                context.sender_eic.as_deref().unwrap_or(""),
428                context.receiver_eic.as_deref().unwrap_or(""),
429                context.message_ref.as_deref().unwrap_or(""),
430            ),
431            Self::MissingInterchangeParty { qualifier, .. } => write!(
432                f,
433                "message carries no NAD+{qualifier} party (Allgemeine Festlegungen §2.13)"
434            ),
435            Self::NotDispatchable {
436                workflow_name,
437                pid,
438                reason,
439                ..
440            } => write!(
441                f,
442                "PID {pid} routed to workflow {workflow_name} but no dispatch arm handled it ({reason})"
443            ),
444            Self::OutboxExhausted {
445                message_id,
446                message_type,
447                recipient,
448                attempts,
449                ..
450            } => write!(
451                f,
452                "outbox exhausted after {attempts} attempts: {message_type} → {recipient} (id={message_id})"
453            ),
454        }
455    }
456}
457
458// ── DeadLetterSink trait ──────────────────────────────────────────────────────
459
460/// Receives messages that cannot be routed or processed.
461///
462/// Implement this trait to:
463/// - Emit CONTRL negative acknowledgements for unroutable messages
464/// - Persist rejections to a durable dead-letter queue for manual review
465/// - Trigger alerts when duplicate-message counts exceed a threshold
466///
467/// The method is **synchronous**. Implementations that require async work
468/// (network calls, database writes) must use `tokio::spawn` internally.
469///
470/// # Default
471///
472/// The default dead-letter sink is [`LogDeadLetterSink`], which emits
473/// `tracing::warn!` events. Override with
474/// [`EngineBuilder::with_dead_letter_sink`] to add CONTRL dispatch or
475/// persistent DLQ storage.
476///
477/// [`EngineBuilder::with_dead_letter_sink`]: crate::builder::EngineBuilder::with_dead_letter_sink
478pub trait DeadLetterSink: Send + Sync + 'static {
479    /// Record a rejected message.
480    ///
481    /// Called by the dispatch path synchronously, before the inbound
482    /// message is acknowledged at the AS4 transport layer. Must not block.
483    fn reject(&self, reason: &DeadLetterReason);
484}
485
486// ── LogDeadLetterSink ─────────────────────────────────────────────────────────
487
488/// A [`DeadLetterSink`] that emits a structured `tracing::warn!` event for
489/// every rejected message.
490///
491/// Suitable for all deployment tiers. In production, combine with a
492/// `tracing` subscriber that forwards `warn`-level events to your alert
493/// pipeline (Loki, CloudWatch, etc.).
494///
495/// This is the **default** dead-letter sink in [`EngineBuilder`].
496///
497/// [`EngineBuilder`]: crate::builder::EngineBuilder
498#[derive(Debug, Clone, Default)]
499pub struct LogDeadLetterSink;
500
501impl DeadLetterSink for LogDeadLetterSink {
502    fn reject(&self, reason: &DeadLetterReason) {
503        // Increment Prometheus counter for every rejection, regardless of
504        // which sink is wired — mirrors SlateDbDeadLetterSink behaviour so
505        // alerting works in non-SlateDB and smoke environments too.
506        crate::metrics::EngineMetrics::global().dead_letter_recorded(reason.label());
507        // Emit all § 147 AO / GoBD structured audit fields when available.
508        if let Some(ctx) = reason.audit_context() {
509            tracing::warn!(
510                reason = reason.label(),
511                message_type = ctx.message_type.as_deref().unwrap_or(""),
512                release_code = ctx.release_code.as_deref().unwrap_or(""),
513                pid = ctx.pid.map_or(0, crate::ids::Pid::as_u32),
514                sender_eic = ctx.sender_eic.as_deref().unwrap_or(""),
515                receiver_eic = ctx.receiver_eic.as_deref().unwrap_or(""),
516                message_ref = ctx.message_ref.as_deref().unwrap_or(""),
517                process_id = ctx.process_id.as_deref().unwrap_or(""),
518                tenant_id = ctx.tenant_id.as_deref().unwrap_or(""),
519                correlation_id = ctx.correlation_id.as_deref().unwrap_or(""),
520                %ctx.timestamp,
521                "dead letter: {reason}",
522            );
523        } else {
524            // OutboxExhausted has no inbound audit context; log its own fields.
525            match reason {
526                DeadLetterReason::OutboxExhausted {
527                    message_id,
528                    message_type,
529                    recipient,
530                    last_error,
531                    attempts,
532                } => {
533                    tracing::error!(
534                        %message_id,
535                        message_type,
536                        recipient,
537                        last_error,
538                        attempts,
539                        reason = reason.label(),
540                        "dead letter: outbox exhausted — message removed after max delivery \
541                         attempts; manual intervention required to deliver this message",
542                    );
543                }
544                _ => {
545                    tracing::warn!(reason = reason.label(), "dead letter: {reason}");
546                }
547            }
548        }
549    }
550}
551
552// ── NoopDeadLetterSink ────────────────────────────────────────────────────────
553
554/// A [`DeadLetterSink`] that silently discards all rejected messages.
555///
556/// **Use only in unit tests** where dead-letter events are not the subject
557/// under test. Using this in production means unroutable messages are lost
558/// without any diagnostic output, violating BDEW AS4 requirements.
559#[derive(Debug, Clone, Default)]
560#[must_use = "NoopDeadLetterSink discards all rejections; use LogDeadLetterSink in production"]
561#[cfg_attr(
562    not(any(test, feature = "testing")),
563    deprecated = "NoopDeadLetterSink must not be used in production builds; use LogDeadLetterSink instead"
564)]
565pub struct NoopDeadLetterSink;
566
567#[cfg(any(test, feature = "testing"))]
568impl DeadLetterSink for NoopDeadLetterSink {
569    fn reject(&self, _reason: &DeadLetterReason) {}
570}
571
572// ── ArcDeadLetterSink ─────────────────────────────────────────────────────────
573
574/// Blanket implementation so `Arc<T>` is a `DeadLetterSink` whenever `T` is.
575///
576/// This allows passing `Arc<LogDeadLetterSink>` or `Arc<dyn DeadLetterSink>`
577/// wherever a `DeadLetterSink` is expected without an extra wrapper.
578impl<T: DeadLetterSink> DeadLetterSink for Arc<T> {
579    fn reject(&self, reason: &DeadLetterReason) {
580        self.as_ref().reject(reason);
581    }
582}
583
584#[cfg(test)]
585mod tests {
586    use super::*;
587
588    #[test]
589    fn dead_letter_reason_labels() {
590        assert_eq!(
591            DeadLetterReason::UnknownPid {
592                pid: crate::ids::Pid::new(55001),
593                context: AuditContext::now()
594            }
595            .label(),
596            "unknown_pid"
597        );
598        assert_eq!(
599            DeadLetterReason::UnknownConversation {
600                conversation_id: "abc".into(),
601                context: AuditContext::now(),
602            }
603            .label(),
604            "unknown_conversation"
605        );
606        assert_eq!(
607            DeadLetterReason::VersionMismatch {
608                expected: "FV2025-10-01".into(),
609                received: "FV2026-10-01".into(),
610                context: AuditContext::now(),
611            }
612            .label(),
613            "version_mismatch"
614        );
615        assert_eq!(
616            DeadLetterReason::DuplicateMessage {
617                inbox_key: "msg-1".into(),
618                context: AuditContext::now(),
619            }
620            .label(),
621            "duplicate_message"
622        );
623        assert_eq!(
624            DeadLetterReason::ProcessingError {
625                message: "invalid state".into(),
626                context: AuditContext::now(),
627            }
628            .label(),
629            "processing_error"
630        );
631    }
632
633    #[test]
634    fn log_sink_does_not_panic() {
635        let sink = LogDeadLetterSink;
636        sink.reject(&DeadLetterReason::UnknownPid {
637            pid: crate::ids::Pid::new(99999),
638            context: AuditContext::now(),
639        });
640        sink.reject(&DeadLetterReason::UnknownConversation {
641            conversation_id: "conv-123".into(),
642            context: AuditContext::now(),
643        });
644        sink.reject(&DeadLetterReason::VersionMismatch {
645            expected: "FV2025-10-01".into(),
646            received: "FV2026-10-01".into(),
647            context: AuditContext::now(),
648        });
649        sink.reject(&DeadLetterReason::DuplicateMessage {
650            inbox_key: "msg-42".into(),
651            context: AuditContext::now(),
652        });
653        sink.reject(&DeadLetterReason::ProcessingError {
654            message: "workflow rejected command".into(),
655            context: AuditContext::now(),
656        });
657    }
658
659    #[test]
660    fn noop_sink_does_not_panic() {
661        let sink = NoopDeadLetterSink;
662        sink.reject(&DeadLetterReason::UnknownPid {
663            pid: crate::ids::Pid::new(55001),
664            context: AuditContext::now(),
665        });
666    }
667
668    #[test]
669    fn arc_blanket_impl_works() {
670        let sink: Arc<LogDeadLetterSink> = Arc::new(LogDeadLetterSink);
671        sink.reject(&DeadLetterReason::UnknownPid {
672            pid: crate::ids::Pid::new(1),
673            context: AuditContext::now(),
674        });
675    }
676
677    #[test]
678    fn dead_letter_reason_display() {
679        assert_eq!(
680            DeadLetterReason::UnknownPid {
681                pid: crate::ids::Pid::new(55001),
682                context: AuditContext::now()
683            }
684            .to_string(),
685            "unknown PID 55001"
686        );
687        assert!(
688            DeadLetterReason::VersionMismatch {
689                expected: "FV2025-10-01".into(),
690                received: "FV2026-10-01".into(),
691                context: AuditContext::now(),
692            }
693            .to_string()
694            .contains("version mismatch")
695        );
696    }
697
698    #[test]
699    fn audit_context_builder() {
700        let ctx = AuditContext::now()
701            .with_message_type("UTILMD")
702            .with_pid(crate::ids::Pid::new(55001))
703            .with_sender_eic("4012345000023")
704            .with_receiver_eic("9900357000004")
705            .with_message_ref("00001")
706            .with_tenant_id("tenant-a")
707            .with_correlation_id("conv-xyz");
708
709        assert_eq!(ctx.message_type.as_deref(), Some("UTILMD"));
710        assert_eq!(ctx.pid, Some(crate::ids::Pid::new(55001)));
711        assert_eq!(ctx.sender_eic.as_deref(), Some("4012345000023"));
712        assert_eq!(ctx.correlation_id.as_deref(), Some("conv-xyz"));
713    }
714
715    #[test]
716    fn audit_context_returned_for_inbound_reasons() {
717        let r = DeadLetterReason::UnknownPid {
718            pid: crate::ids::Pid::new(99),
719            context: AuditContext::now().with_pid(crate::ids::Pid::new(99)),
720        };
721        assert!(r.audit_context().is_some());
722    }
723}