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