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}