Skip to main content

email_transport/
options.rs

1//! Per-send options and provider-specific transport option storage.
2
3use core::any::{Any, TypeId};
4#[cfg(feature = "serde")]
5use std::collections::BTreeSet;
6use std::collections::HashMap;
7use std::time::Duration;
8
9use email_message::Envelope;
10#[cfg(feature = "serde")]
11use thiserror::Error;
12
13/// Per-send controls shared by structured and raw transport sends.
14///
15/// With the `serde` feature enabled, this serializes as a sparse object:
16/// absent options are omitted, [`TransportOptions`] uses its provider-keyed
17/// representation, and `timeout` is encoded as `{ "secs": u64, "nanos": u32 }`.
18/// Deserialization is intentionally registry-driven and not implemented on this
19/// type because provider-specific options need a `TransportOptionRegistry`;
20/// use `TransportOptionRegistry::send_options_seed` (or the convenience
21/// `TransportOptionRegistry::deserialize_send_options` wrapper) instead.
22#[cfg_attr(feature = "serde", derive(serde::Serialize))]
23#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
24#[derive(Debug, Default)]
25#[non_exhaustive]
26pub struct SendOptions {
27    /// Optional custom SMTP envelope for structured [`crate::Transport`] sends.
28    ///
29    /// This is only meaningful for structured transports that advertise
30    /// [`crate::Capabilities::custom_envelope`]. Other structured transports may ignore
31    /// it. [`crate::RawTransport`] methods take an explicit [`Envelope`] argument, and
32    /// that argument is authoritative; raw transports ignore this field.
33    #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
34    pub envelope: Option<Envelope>,
35    /// Typed, in-process provider-specific controls.
36    ///
37    /// With the `serde` feature enabled, [`TransportOptions`] serializes as a
38    /// provider-keyed JSON object and can be hydrated through
39    /// `TransportOptionRegistry`.
40    #[cfg_attr(
41        feature = "serde",
42        serde(skip_serializing_if = "TransportOptions::is_empty")
43    )]
44    #[cfg_attr(
45        feature = "schemars",
46        schemars(default, skip_serializing_if = "TransportOptions::is_empty")
47    )]
48    pub transport_options: TransportOptions,
49    /// Upper bound on provider-call duration for this send attempt. Transports
50    /// advertising `Capabilities::timeout` must honor it; others should ignore it.
51    #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
52    pub timeout: Option<Duration>,
53    /// Provider-level idempotency token for this attempt. Transports advertising
54    /// `Capabilities::idempotency_key` must forward it; others should ignore it.
55    ///
56    /// Validated at construction (rejects empty, NUL, CR/LF, non-tab control
57    /// characters, and values longer than 1 KiB), see [`IdempotencyKey`].
58    #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
59    pub idempotency_key: Option<IdempotencyKey>,
60    /// Opaque correlation identifier carried end-to-end from queue to transport.
61    /// Built-in transports do not automatically expose it to providers; it is
62    /// available to adapter-specific typed options when a provider has a natural
63    /// slot for it.
64    ///
65    /// Validated at construction with the same rules as
66    /// [`IdempotencyKey`], see [`CorrelationId`].
67    #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
68    pub correlation_id: Option<CorrelationId>,
69}
70
71impl SendOptions {
72    /// Create send options with no overrides.
73    #[must_use]
74    pub fn new() -> Self {
75        Self::default()
76    }
77
78    /// Set the structured-send envelope override.
79    #[must_use]
80    pub fn with_envelope(mut self, envelope: Envelope) -> Self {
81        self.envelope = Some(envelope);
82        self
83    }
84
85    /// Replace the provider-specific typed option map.
86    #[must_use]
87    pub fn with_transport_options(mut self, transport_options: TransportOptions) -> Self {
88        self.transport_options = transport_options;
89        self
90    }
91
92    /// Insert or replace the provider option value of type `T`.
93    ///
94    /// Slots are keyed by type, exactly like [`TransportOptions::insert`].
95    /// Unlike [`Self::with_transport_options`], this keeps options of other
96    /// types already present, so calls chain across providers.
97    #[cfg(feature = "serde")]
98    #[must_use]
99    pub fn with_transport_option<T>(mut self, value: T) -> Self
100    where
101        T: TransportOption + serde::Serialize,
102    {
103        self.transport_options.insert(value);
104        self
105    }
106
107    /// Insert or replace the provider option value of type `T`.
108    ///
109    /// Slots are keyed by type, exactly like [`TransportOptions::insert`].
110    /// Unlike [`Self::with_transport_options`], this keeps options of other
111    /// types already present, so calls chain across providers.
112    #[cfg(not(feature = "serde"))]
113    #[must_use]
114    pub fn with_transport_option<T: TransportOption>(mut self, value: T) -> Self {
115        self.transport_options.insert(value);
116        self
117    }
118
119    /// Set the provider-call timeout for this attempt.
120    #[must_use]
121    pub const fn with_timeout(mut self, timeout: Duration) -> Self {
122        self.timeout = Some(timeout);
123        self
124    }
125
126    /// Set the provider idempotency key for this attempt.
127    #[must_use]
128    pub fn with_idempotency_key(mut self, idempotency_key: IdempotencyKey) -> Self {
129        self.idempotency_key = Some(idempotency_key);
130        self
131    }
132
133    /// Set the end-to-end tracing correlation identifier.
134    #[must_use]
135    pub fn with_correlation_id(mut self, correlation_id: CorrelationId) -> Self {
136        self.correlation_id = Some(correlation_id);
137        self
138    }
139
140    /// Return a serializable view that omits the provider idempotency key.
141    ///
142    /// Queue adapters can use this when the idempotency key belongs to the
143    /// queue invocation rather than the eventual provider request.
144    #[cfg(feature = "serde")]
145    #[must_use]
146    pub fn serializable_without_idempotency_key(&self) -> impl serde::Serialize + '_ {
147        SendOptionsWithoutIdempotencyKey(self)
148    }
149}
150
151#[cfg(feature = "serde")]
152struct SendOptionsWithoutIdempotencyKey<'a>(&'a SendOptions);
153
154#[cfg(feature = "serde")]
155impl serde::Serialize for SendOptionsWithoutIdempotencyKey<'_> {
156    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
157    where
158        S: serde::Serializer,
159    {
160        use serde::ser::SerializeStruct as _;
161
162        let SendOptions {
163            envelope,
164            transport_options,
165            timeout,
166            idempotency_key: _,
167            correlation_id,
168        } = self.0;
169        let field_count = usize::from(envelope.is_some())
170            + usize::from(!transport_options.is_empty())
171            + usize::from(timeout.is_some())
172            + usize::from(correlation_id.is_some());
173        let mut state = serializer.serialize_struct("SendOptions", field_count)?;
174
175        if let Some(envelope) = envelope {
176            state.serialize_field("envelope", envelope)?;
177        }
178        if !transport_options.is_empty() {
179            state.serialize_field("transport_options", transport_options)?;
180        }
181        if let Some(timeout) = timeout {
182            state.serialize_field("timeout", timeout)?;
183        }
184        if let Some(correlation_id) = correlation_id {
185            state.serialize_field("correlation_id", correlation_id)?;
186        }
187
188        state.end()
189    }
190}
191
192crate::string_newtype! {
193    /// Provider idempotency token for safe retries.
194    ///
195    /// Validated on construction, empty, NUL, CR/LF, non-tab control
196    /// characters, and values longer than 1 KiB are rejected. Adapters
197    /// that advertise [`crate::Capabilities::idempotency_key`] forward the value
198    /// verbatim to the provider's idempotency header; the validation closes
199    /// a header-injection seam at the type level.
200    ///
201    /// `new_unchecked` is available on this type, it is declared via
202    /// the `@unchecked` matcher arm of [`crate::string_newtype!`] so
203    /// trusted-input construction (internal constants, test fixtures)
204    /// stays available. End-user code should reach for
205    /// [`Self::new`] / [`std::str::FromStr`] for any value that
206    /// originated outside trusted code paths.
207    @unchecked IdempotencyKey
208}
209
210crate::string_newtype! {
211    /// Correlation identifier carried end-to-end from queue to transport.
212    ///
213    /// Available to adapter-specific typed options when a provider has a
214    /// natural slot for it. Validated on construction with the same rules as
215    /// [`IdempotencyKey`].
216    ///
217    /// `new_unchecked` is available, declared via the `@unchecked`
218    /// matcher arm. See [`IdempotencyKey`] for guidance.
219    @unchecked CorrelationId
220}
221
222/// Marker for typed provider-specific send options.
223///
224/// Adapters define their own per-provider option structs and implement this
225/// trait so the typed slot in [`TransportOptions`] can store them keyed by
226/// [`TypeId`] for in-process lookup.
227///
228/// Options also provide [`Self::provider_key`], a stable JSON key for this
229/// option's provider such as `"resend"` or `"postmark"`. The key is
230/// intentionally explicit rather than derived from [`TypeId`] or `type_name`,
231/// both of which are implementation identities rather than wire-format
232/// identifiers.
233///
234/// With the `serde` feature enabled, option types must also implement
235/// `serde::Serialize` for insertion and `serde::Deserialize` for registry
236/// hydration.
237///
238/// # Safety boundary
239///
240/// Provider-specific options may add or relax behavior, such as tags,
241/// metadata, tracking, or templates. They must not constrain delivery, because
242/// a transport that does not recognize the provider-specific option may ignore
243/// it. Safety controls such as sandbox mode, suppression-list toggles, or
244/// "never deliver to real recipients" belong in core [`SendOptions`], where
245/// every transport must honor them or fail the send.
246///
247/// # Send + Sync
248///
249/// The bound is `Send + Sync` on every target, including `wasm32`, even
250/// though most other kernel async surfaces drop `Send` on `wasm32` to
251/// accommodate `!Send` JS handles. `TransportOption` values flow through
252/// the typed slot map and may be inspected from any thread; in practice
253/// adapters store plain newtypes (`Vec<String>`, primitives) that satisfy
254/// the bound trivially. A wasm adapter that wants to stash a raw
255/// `web_sys::JsValue` inside an option would need its own thread-safe
256/// wrapper; the current design assumes that is rare enough not to warrant
257/// a cfg gate.
258pub trait TransportOption: Any + Send + Sync {
259    /// Stable provider key used when serializing this option at queue/wire
260    /// boundaries.
261    ///
262    /// The `Self: Sized` bound keeps [`TransportOption`] dyn-compatible;
263    /// callers that only have a `dyn TransportOption` already receive the
264    /// provider key from the typed slot metadata captured at insertion time.
265    fn provider_key() -> &'static str
266    where
267        Self: Sized;
268}
269
270/// Per-send transport options.
271///
272/// A typed, in-process map keyed by `TypeId` carrying provider-specific strongly
273/// typed values (e.g. `PostmarkTag`, `ResendTags`). Cheap, zero-copy at the
274/// adapter boundary.
275///
276/// Values also carry a stable [`TransportOption::provider_key`]. With the
277/// `serde` feature enabled, this allows the map to serialize into a
278/// provider-keyed object for queue boundaries through any serde format.
279/// Deserialization requires a `TransportOptionRegistry` because Rust cannot
280/// discover concrete `TransportOption` implementors from a string key alone;
281/// drive it through `TransportOptionsSeed` / `SendOptionsSeed` (or the
282/// convenience `TransportOptionRegistry::deserialize_send_options`).
283#[derive(Default)]
284pub struct TransportOptions {
285    inner: HashMap<TypeId, TypedSlot>,
286}
287
288#[cfg(feature = "serde")]
289struct TypedSlot {
290    type_name: &'static str,
291    provider_key: &'static str,
292    value: Box<dyn DynTransportOption>,
293}
294
295#[cfg(not(feature = "serde"))]
296struct TypedSlot {
297    type_name: &'static str,
298    value: Box<dyn Any + Send + Sync>,
299}
300
301/// Erased-serde-aware view of a `TransportOption` value, plus access back to
302/// `Any` for the typed-slot lookup methods.
303#[cfg(feature = "serde")]
304trait DynTransportOption: erased_serde::Serialize + Send + Sync + 'static {
305    fn as_any(&self) -> &dyn Any;
306    fn as_any_mut(&mut self) -> &mut dyn Any;
307    fn into_any(self: Box<Self>) -> Box<dyn Any + Send + Sync>;
308}
309
310#[cfg(feature = "serde")]
311impl<T> DynTransportOption for T
312where
313    T: TransportOption + serde::Serialize,
314{
315    fn as_any(&self) -> &dyn Any {
316        self
317    }
318    fn as_any_mut(&mut self) -> &mut dyn Any {
319        self
320    }
321    fn into_any(self: Box<Self>) -> Box<dyn Any + Send + Sync> {
322        self
323    }
324}
325
326#[cfg(feature = "serde")]
327erased_serde::serialize_trait_object!(DynTransportOption);
328
329impl TransportOptions {
330    /// Insert or replace the option value of type `T`.
331    #[cfg(feature = "serde")]
332    pub fn insert<T>(&mut self, value: T)
333    where
334        T: TransportOption + serde::Serialize,
335    {
336        self.inner.insert(
337            TypeId::of::<T>(),
338            TypedSlot {
339                type_name: std::any::type_name::<T>(),
340                provider_key: T::provider_key(),
341                value: Box::new(value),
342            },
343        );
344    }
345
346    /// Insert or replace the option value of type `T`.
347    #[cfg(not(feature = "serde"))]
348    pub fn insert<T: TransportOption>(&mut self, value: T) {
349        self.inner.insert(
350            TypeId::of::<T>(),
351            TypedSlot {
352                type_name: std::any::type_name::<T>(),
353                value: Box::new(value),
354            },
355        );
356    }
357
358    /// Insert or replace the option value of type `T`, by value.
359    ///
360    /// Builder-style counterpart of [`Self::insert`].
361    #[cfg(feature = "serde")]
362    #[must_use]
363    pub fn with<T>(mut self, value: T) -> Self
364    where
365        T: TransportOption + serde::Serialize,
366    {
367        self.insert(value);
368        self
369    }
370
371    /// Insert or replace the option value of type `T`, by value.
372    ///
373    /// Builder-style counterpart of [`Self::insert`].
374    #[cfg(not(feature = "serde"))]
375    #[must_use]
376    pub fn with<T: TransportOption>(mut self, value: T) -> Self {
377        self.insert(value);
378        self
379    }
380
381    /// Return the stored option of type `T`, if present.
382    #[must_use]
383    pub fn get<T: TransportOption>(&self) -> Option<&T> {
384        let slot = self.inner.get(&TypeId::of::<T>())?;
385        slot.value_any().downcast_ref::<T>()
386    }
387
388    /// Return the stored option of type `T` mutably, if present.
389    pub fn get_mut<T: TransportOption>(&mut self) -> Option<&mut T> {
390        let slot = self.inner.get_mut(&TypeId::of::<T>())?;
391        slot.value_any_mut().downcast_mut::<T>()
392    }
393
394    /// Remove and return the option of type `T`, if present.
395    pub fn remove<T: TransportOption>(&mut self) -> Option<T> {
396        let slot = self.inner.remove(&TypeId::of::<T>())?;
397        slot.into_any().downcast::<T>().ok().map(|v| *v)
398    }
399
400    /// `true` when no typed slots are present.
401    #[must_use]
402    pub fn is_empty(&self) -> bool {
403        self.inner.is_empty()
404    }
405}
406
407#[cfg(feature = "serde")]
408impl TypedSlot {
409    fn value_any(&self) -> &dyn Any {
410        self.value.as_any()
411    }
412
413    fn value_any_mut(&mut self) -> &mut dyn Any {
414        self.value.as_any_mut()
415    }
416
417    fn into_any(self) -> Box<dyn Any + Send + Sync> {
418        self.value.into_any()
419    }
420}
421
422#[cfg(not(feature = "serde"))]
423impl TypedSlot {
424    fn value_any(&self) -> &dyn Any {
425        self.value.as_ref()
426    }
427
428    fn value_any_mut(&mut self) -> &mut dyn Any {
429        self.value.as_mut()
430    }
431
432    fn into_any(self) -> Box<dyn Any + Send + Sync> {
433        self.value
434    }
435}
436
437#[cfg(feature = "serde")]
438impl serde::Serialize for TransportOptions {
439    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
440    where
441        S: serde::Serializer,
442    {
443        use serde::ser::SerializeMap as _;
444
445        let mut slots: Vec<&TypedSlot> = self.inner.values().collect();
446        slots.sort_unstable_by_key(|slot| slot.provider_key);
447
448        let mut seen = BTreeSet::new();
449        let mut map = serializer.serialize_map(Some(slots.len()))?;
450        for slot in slots {
451            if !seen.insert(slot.provider_key) {
452                return Err(serde::ser::Error::custom(format_args!(
453                    "duplicate TransportOption provider key `{}`",
454                    slot.provider_key
455                )));
456            }
457            map.serialize_entry(slot.provider_key, &*slot.value)?;
458        }
459        map.end()
460    }
461}
462
463#[cfg(feature = "schemars")]
464impl schemars::JsonSchema for TransportOptions {
465    fn schema_name() -> std::borrow::Cow<'static, str> {
466        "TransportOptions".into()
467    }
468
469    fn schema_id() -> std::borrow::Cow<'static, str> {
470        concat!(module_path!(), "::TransportOptions").into()
471    }
472
473    fn json_schema(_generator: &mut schemars::SchemaGenerator) -> schemars::Schema {
474        schemars::json_schema!({
475            "type": "object",
476            "additionalProperties": true,
477        })
478    }
479}
480
481/// Registry of queue/wire codecs for concrete [`TransportOption`] types.
482///
483/// Serialization does not need a registry because every typed slot stores its
484/// provider key when inserted into [`TransportOptions`]. Deserialization does
485/// need a registry so a stable provider key can be mapped back to the concrete
486/// Rust type that owns that wire shape; that mapping is exposed through
487/// [`TransportOptionsSeed`] and [`SendOptionsSeed`], which implement
488/// [`serde::de::DeserializeSeed`] so the registry can drive any serde
489/// deserializer (JSON, CBOR, `MessagePack`, postcard, ...) directly into typed
490/// slots without an intermediate `serde_json::Value`.
491#[cfg(feature = "serde")]
492#[derive(Default)]
493pub struct TransportOptionRegistry {
494    decoders: HashMap<&'static str, TransportOptionDecoder>,
495}
496
497#[cfg(feature = "serde")]
498impl std::fmt::Debug for TransportOptionRegistry {
499    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
500        let mut registered: Vec<(&str, &'static str)> = self
501            .decoders
502            .iter()
503            .map(|(provider_key, decoder)| (*provider_key, decoder.type_name))
504            .collect();
505        registered.sort_unstable_by_key(|(provider_key, _)| *provider_key);
506
507        f.debug_struct("TransportOptionRegistry")
508            .field("registered", &registered)
509            .finish()
510    }
511}
512
513#[cfg(feature = "serde")]
514struct TransportOptionDecoder {
515    type_name: &'static str,
516    decode: for<'de> fn(
517        &mut dyn erased_serde::Deserializer<'de>,
518        &mut TransportOptions,
519    ) -> Result<(), erased_serde::Error>,
520}
521
522#[cfg(feature = "serde")]
523impl TransportOptionRegistry {
524    /// Create an empty provider-option registry.
525    #[must_use]
526    pub fn new() -> Self {
527        Self::default()
528    }
529
530    /// Register a concrete provider option type by its
531    /// [`TransportOption::provider_key`].
532    ///
533    /// # Errors
534    ///
535    /// Returns [`TransportOptionRegistryError::DuplicateProviderKey`] if a
536    /// different option type has already claimed the same provider key.
537    pub fn register<T>(&mut self) -> Result<(), TransportOptionRegistryError>
538    where
539        T: TransportOption + serde::Serialize + serde::de::DeserializeOwned,
540    {
541        let provider_key = T::provider_key();
542        if let Some(existing) = self.decoders.get(provider_key) {
543            if existing.type_name == std::any::type_name::<T>() {
544                return Ok(());
545            }
546
547            return Err(TransportOptionRegistryError::DuplicateProviderKey {
548                provider_key,
549                existing_type: existing.type_name,
550                new_type: std::any::type_name::<T>(),
551            });
552        }
553
554        self.decoders.insert(
555            provider_key,
556            TransportOptionDecoder {
557                type_name: std::any::type_name::<T>(),
558                decode: decode_transport_option::<T>,
559            },
560        );
561        Ok(())
562    }
563
564    /// Return whether a provider option decoder is registered for `provider_key`.
565    #[must_use]
566    pub fn contains_provider_key(&self, provider_key: &str) -> bool {
567        self.decoders.contains_key(provider_key)
568    }
569
570    /// Return the registered provider keys in stable lexical order.
571    #[must_use]
572    pub fn provider_keys(&self) -> Vec<&'static str> {
573        let mut provider_keys: Vec<_> = self.decoders.keys().copied().collect();
574        provider_keys.sort_unstable();
575        provider_keys
576    }
577
578    /// Build a [`DeserializeSeed`](serde::de::DeserializeSeed) that hydrates a
579    /// [`TransportOptions`] map from any serde deserializer.
580    ///
581    /// Unknown provider keys are rejected by default. Use
582    /// [`TransportOptionsSeed::ignore_unknown_provider_keys`] to skip them.
583    #[must_use]
584    pub fn transport_options_seed(&self) -> TransportOptionsSeed<'_> {
585        TransportOptionsSeed {
586            registry: self,
587            ignore_unknown: false,
588        }
589    }
590
591    /// Build a [`DeserializeSeed`](serde::de::DeserializeSeed) that hydrates a
592    /// [`SendOptions`] from any serde deserializer.
593    ///
594    /// Unknown top-level fields are ignored for forward compatibility. Unknown
595    /// provider keys inside `transport_options` are rejected by default; use
596    /// [`SendOptionsSeed::ignore_unknown_transport_options`] to skip them.
597    #[must_use]
598    pub fn send_options_seed(&self) -> SendOptionsSeed<'_> {
599        SendOptionsSeed {
600            registry: self,
601            ignore_unknown: false,
602        }
603    }
604
605    /// Deserialize [`SendOptions`] from any serde deserializer.
606    ///
607    /// Convenience wrapper around [`Self::send_options_seed`] for callers that
608    /// only need the default strict behavior.
609    ///
610    /// # Errors
611    ///
612    /// Returns the deserializer's native error when the payload shape is
613    /// malformed, a registered provider option fails to deserialize, or
614    /// `transport_options` contains an unregistered provider key.
615    pub fn deserialize_send_options<'de, D>(&self, deserializer: D) -> Result<SendOptions, D::Error>
616    where
617        D: serde::Deserializer<'de>,
618    {
619        use serde::de::DeserializeSeed as _;
620        self.send_options_seed().deserialize(deserializer)
621    }
622
623    /// Deserialize a single provider option for `provider_key` and overwrite the
624    /// matching typed slot in `options` when the provider key is registered.
625    ///
626    /// This is the per-key dispatch primitive that callers iterating over a
627    /// pre-decoded provider map (for example, a `BTreeMap<String, Value>` wire
628    /// staging type built by a downstream crate that cannot drive
629    /// [`SendOptionsSeed`] directly) reach for. For deserializing a whole
630    /// [`SendOptions`] or [`TransportOptions`] from a serde format, prefer
631    /// [`Self::send_options_seed`] / [`Self::transport_options_seed`]; those
632    /// have a richer strict-vs-ignore policy via builder methods.
633    ///
634    /// Returns `Ok(true)` when a registered option type consumed the value and
635    /// `Ok(false)` for unknown provider keys. Unknown keys are intentionally not
636    /// errors so queue payloads can be forwarded across workers with different
637    /// provider feature sets; the caller is expected to either propagate or
638    /// suppress that signal as the surrounding context requires.
639    ///
640    /// # Errors
641    ///
642    /// Returns the deserializer's native error if `provider_key` is registered
643    /// but the value does not match that option type's serde shape.
644    pub fn hydrate_into<'de, D>(
645        &self,
646        provider_key: &str,
647        deserializer: D,
648        options: &mut TransportOptions,
649    ) -> Result<bool, D::Error>
650    where
651        D: serde::Deserializer<'de>,
652    {
653        let Some(decoder) = self.decoders.get(provider_key) else {
654            return Ok(false);
655        };
656
657        let mut erased = <dyn erased_serde::Deserializer<'de>>::erase(deserializer);
658        (decoder.decode)(&mut erased, options).map_err(serde::de::Error::custom)?;
659        Ok(true)
660    }
661}
662
663#[cfg(feature = "serde")]
664fn decode_transport_option<T>(
665    deserializer: &mut dyn erased_serde::Deserializer<'_>,
666    options: &mut TransportOptions,
667) -> Result<(), erased_serde::Error>
668where
669    T: TransportOption + serde::Serialize + serde::de::DeserializeOwned,
670{
671    let value: T = erased_serde::deserialize(deserializer)?;
672    options.insert(value);
673    Ok(())
674}
675
676/// [`DeserializeSeed`](serde::de::DeserializeSeed) for [`TransportOptions`].
677///
678/// Built through [`TransportOptionRegistry::transport_options_seed`]. This is
679/// the only way to deserialize a [`TransportOptions`] map, since the type
680/// cannot have a plain `Deserialize` impl (the registry has to map provider
681/// keys back to typed slot types at runtime).
682#[cfg(feature = "serde")]
683pub struct TransportOptionsSeed<'a> {
684    registry: &'a TransportOptionRegistry,
685    ignore_unknown: bool,
686}
687
688#[cfg(feature = "serde")]
689impl TransportOptionsSeed<'_> {
690    /// Skip provider keys that the registry has no decoder for instead of
691    /// erroring, so payloads can flow across workers compiled with different
692    /// adapter feature sets.
693    ///
694    /// With the `tracing` feature enabled, each skipped key is emitted as a
695    /// `debug` event carrying a `provider_key` field, so a mistyped key remains
696    /// diagnosable even though it is not an error.
697    #[must_use]
698    pub fn ignore_unknown_provider_keys(mut self) -> Self {
699        self.ignore_unknown = true;
700        self
701    }
702}
703
704#[cfg(feature = "serde")]
705impl<'de> serde::de::DeserializeSeed<'de> for TransportOptionsSeed<'_> {
706    type Value = TransportOptions;
707
708    fn deserialize<D>(self, deserializer: D) -> Result<TransportOptions, D::Error>
709    where
710        D: serde::Deserializer<'de>,
711    {
712        deserializer.deserialize_map(TransportOptionsVisitor {
713            registry: self.registry,
714            ignore_unknown: self.ignore_unknown,
715        })
716    }
717}
718
719#[cfg(feature = "serde")]
720struct TransportOptionsVisitor<'a> {
721    registry: &'a TransportOptionRegistry,
722    ignore_unknown: bool,
723}
724
725#[cfg(feature = "serde")]
726impl<'de> serde::de::Visitor<'de> for TransportOptionsVisitor<'_> {
727    type Value = TransportOptions;
728
729    fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
730        f.write_str("a provider-keyed map of TransportOption values")
731    }
732
733    fn visit_map<A>(self, mut map: A) -> Result<TransportOptions, A::Error>
734    where
735        A: serde::de::MapAccess<'de>,
736    {
737        let mut options = TransportOptions::default();
738        while let Some(key) = map.next_key::<String>()? {
739            if let Some(decoder) = self.registry.decoders.get(key.as_str()) {
740                map.next_value_seed(TransportOptionDecoderSeed {
741                    decode: decoder.decode,
742                    options: &mut options,
743                })?;
744            } else if self.ignore_unknown {
745                // Ignoring is the normal union case, not a fault: a payload may
746                // carry options for providers this worker was not built with.
747                // It is traced because it is also how a mistyped provider key
748                // disappears without a trace otherwise.
749                #[cfg(feature = "tracing")]
750                ::tracing::debug!(
751                    provider_key = key.as_str(),
752                    "ignored unregistered TransportOption provider key"
753                );
754                map.next_value::<serde::de::IgnoredAny>()?;
755            } else {
756                return Err(serde::de::Error::custom(format_args!(
757                    "unknown TransportOption provider key `{key}`"
758                )));
759            }
760        }
761        Ok(options)
762    }
763}
764
765#[cfg(feature = "serde")]
766struct TransportOptionDecoderSeed<'a> {
767    decode: for<'de> fn(
768        &mut dyn erased_serde::Deserializer<'de>,
769        &mut TransportOptions,
770    ) -> Result<(), erased_serde::Error>,
771    options: &'a mut TransportOptions,
772}
773
774#[cfg(feature = "serde")]
775impl<'de> serde::de::DeserializeSeed<'de> for TransportOptionDecoderSeed<'_> {
776    type Value = ();
777
778    fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
779    where
780        D: serde::Deserializer<'de>,
781    {
782        let mut erased = <dyn erased_serde::Deserializer<'de>>::erase(deserializer);
783        (self.decode)(&mut erased, self.options).map_err(serde::de::Error::custom)
784    }
785}
786
787/// [`DeserializeSeed`](serde::de::DeserializeSeed) for [`SendOptions`].
788///
789/// Built through [`TransportOptionRegistry::send_options_seed`].
790///
791/// Use this when [`SendOptions`] is a field of another `DeserializeSeed`
792/// value and the parent visitor needs to thread a registry through. For the
793/// typical "deserialize a `SendOptions` from a queue payload" case, prefer the
794/// thinner [`TransportOptionRegistry::deserialize_send_options`] convenience
795/// wrapper.
796#[cfg(feature = "serde")]
797pub struct SendOptionsSeed<'a> {
798    registry: &'a TransportOptionRegistry,
799    ignore_unknown: bool,
800}
801
802#[cfg(feature = "serde")]
803impl SendOptionsSeed<'_> {
804    /// Skip unknown provider keys inside `transport_options` instead of
805    /// erroring. See [`TransportOptionsSeed::ignore_unknown_provider_keys`].
806    #[must_use]
807    pub fn ignore_unknown_transport_options(mut self) -> Self {
808        self.ignore_unknown = true;
809        self
810    }
811}
812
813#[cfg(feature = "serde")]
814impl<'de> serde::de::DeserializeSeed<'de> for SendOptionsSeed<'_> {
815    type Value = SendOptions;
816
817    fn deserialize<D>(self, deserializer: D) -> Result<SendOptions, D::Error>
818    where
819        D: serde::Deserializer<'de>,
820    {
821        deserializer.deserialize_map(SendOptionsVisitor {
822            registry: self.registry,
823            ignore_unknown: self.ignore_unknown,
824        })
825    }
826}
827
828#[cfg(feature = "serde")]
829struct SendOptionsVisitor<'a> {
830    registry: &'a TransportOptionRegistry,
831    ignore_unknown: bool,
832}
833
834/// Compile-time field identifier for [`SendOptions`].
835///
836/// Adding a field to [`SendOptions`] without adding a variant here means the
837/// new field never reaches the typed value through the seed; the round-trip
838/// test in this module's `tests` will catch that. Adding a variant here without
839/// extending the `match` in [`SendOptionsVisitor::visit_map`] is a compile
840/// error.
841#[cfg(feature = "serde")]
842#[derive(serde::Deserialize)]
843#[serde(field_identifier, rename_all = "snake_case")]
844enum SendOptionsField {
845    Envelope,
846    TransportOptions,
847    Timeout,
848    IdempotencyKey,
849    CorrelationId,
850    #[serde(other)]
851    Other,
852}
853
854#[cfg(feature = "serde")]
855impl<'de> serde::de::Visitor<'de> for SendOptionsVisitor<'_> {
856    type Value = SendOptions;
857
858    fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
859        f.write_str("a SendOptions map")
860    }
861
862    fn visit_map<A>(self, mut map: A) -> Result<SendOptions, A::Error>
863    where
864        A: serde::de::MapAccess<'de>,
865    {
866        let mut options = SendOptions::default();
867        while let Some(field) = map.next_key::<SendOptionsField>()? {
868            match field {
869                SendOptionsField::Envelope => options.envelope = map.next_value()?,
870                SendOptionsField::TransportOptions => {
871                    options.transport_options = map.next_value_seed(TransportOptionsSeed {
872                        registry: self.registry,
873                        ignore_unknown: self.ignore_unknown,
874                    })?;
875                }
876                SendOptionsField::Timeout => options.timeout = map.next_value()?,
877                SendOptionsField::IdempotencyKey => {
878                    options.idempotency_key = map.next_value()?;
879                }
880                SendOptionsField::CorrelationId => {
881                    options.correlation_id = map.next_value()?;
882                }
883                SendOptionsField::Other => {
884                    map.next_value::<serde::de::IgnoredAny>()?;
885                }
886            }
887        }
888        Ok(options)
889    }
890}
891
892#[cfg(feature = "serde")]
893#[derive(Debug, Error)]
894#[non_exhaustive]
895/// Failure to register a provider-option codec.
896pub enum TransportOptionRegistryError {
897    /// Two different option types claimed the same stable provider key.
898    #[error(
899        "duplicate TransportOption provider key `{provider_key}` for `{new_type}`; already registered by `{existing_type}`"
900    )]
901    DuplicateProviderKey {
902        /// Provider key claimed by both option types.
903        provider_key: &'static str,
904        /// Fully qualified name of the previously registered type.
905        existing_type: &'static str,
906        /// Fully qualified name of the type that attempted registration.
907        new_type: &'static str,
908    },
909}
910
911impl std::fmt::Debug for TransportOptions {
912    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
913        let mut typed: Vec<&'static str> = self.inner.values().map(|slot| slot.type_name).collect();
914        typed.sort_unstable();
915
916        f.debug_struct("TransportOptions")
917            .field("typed", &typed)
918            .finish()
919    }
920}
921
922#[cfg(test)]
923mod tests {
924    use crate::{STRING_NEWTYPE_MAX_BYTES, StringNewtypeError};
925
926    #[cfg(feature = "serde")]
927    use email_message::Envelope;
928
929    #[cfg(feature = "serde")]
930    use super::CorrelationId;
931    #[cfg(any(feature = "serde", feature = "schemars"))]
932    use super::SendOptions;
933    use super::{IdempotencyKey, TransportOption, TransportOptions};
934    #[cfg(feature = "serde")]
935    use super::{SendOptionsSeed, TransportOptionRegistry, TransportOptionRegistryError};
936
937    #[derive(Debug, PartialEq, Eq)]
938    #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
939    struct TestOption(String);
940
941    impl TransportOption for TestOption {
942        fn provider_key() -> &'static str {
943            "test"
944        }
945    }
946
947    #[cfg(feature = "serde")]
948    #[derive(Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
949    struct OtherTestOption {
950        value: u32,
951    }
952
953    #[cfg(feature = "serde")]
954    impl TransportOption for OtherTestOption {
955        fn provider_key() -> &'static str {
956            "other"
957        }
958    }
959
960    #[cfg(feature = "serde")]
961    #[derive(Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
962    struct DuplicateKeyOption(String);
963
964    #[cfg(feature = "serde")]
965    impl TransportOption for DuplicateKeyOption {
966        fn provider_key() -> &'static str {
967            "test"
968        }
969    }
970
971    #[test]
972    fn idempotency_key_rejects_crlf_injection() {
973        let err = IdempotencyKey::new("hi\r\nBcc: victim").unwrap_err();
974        assert_eq!(err, StringNewtypeError::Newline);
975    }
976
977    #[test]
978    fn idempotency_key_rejects_empty() {
979        let err = IdempotencyKey::new("").unwrap_err();
980        assert_eq!(err, StringNewtypeError::Empty);
981    }
982
983    #[test]
984    fn idempotency_key_rejects_nul() {
985        let err = IdempotencyKey::new("hi\0bye").unwrap_err();
986        assert_eq!(err, StringNewtypeError::Nul);
987    }
988
989    #[test]
990    fn idempotency_key_rejects_oversize() {
991        let payload = "x".repeat(STRING_NEWTYPE_MAX_BYTES + 1);
992        let err = IdempotencyKey::new(payload).unwrap_err();
993        assert!(matches!(err, StringNewtypeError::TooLong { .. }));
994    }
995
996    #[test]
997    fn idempotency_key_accepts_well_formed() {
998        let key = IdempotencyKey::new("job-12345").unwrap();
999        assert_eq!(key.as_str(), "job-12345");
1000    }
1001
1002    #[test]
1003    fn idempotency_key_rejects_unicode_tag_codepoints() {
1004        // U+E0041 is the tag-form of ASCII 'A'. Invisible to humans
1005        // but readable by some downstream tooling, a known ASCII-
1006        // smuggling vector. The validator must reject it.
1007        let smuggled = format!("trace-{}", '\u{E0041}');
1008        let err = IdempotencyKey::new(smuggled).unwrap_err();
1009        assert_eq!(err, StringNewtypeError::UnicodeTag);
1010    }
1011
1012    #[test]
1013    fn idempotency_key_accepts_legitimate_non_ascii_utf8() {
1014        // Legitimate UTF-8 (e.g. accented letters) stays accepted
1015        // the kernel's gate is for injection prevention, not
1016        // HTTP-header validity. Adapters that forward the value as
1017        // an HTTP header are responsible for additional validation.
1018        let key = IdempotencyKey::new("trace-héllo").unwrap();
1019        assert_eq!(key.as_str(), "trace-héllo");
1020    }
1021
1022    #[test]
1023    #[cfg(feature = "serde")]
1024    fn correlation_id_round_trips_through_serde() {
1025        let id = CorrelationId::new("trace-abc").unwrap();
1026        let json = serde_json::to_string(&id).unwrap();
1027        assert_eq!(json, "\"trace-abc\"");
1028        let back: CorrelationId = serde_json::from_str(&json).unwrap();
1029        assert_eq!(back, id);
1030    }
1031
1032    #[test]
1033    #[cfg(feature = "serde")]
1034    fn correlation_id_serde_rejects_invalid() {
1035        // Deserialize routes through the generated new constructor, so validation runs.
1036        let result: Result<CorrelationId, _> = serde_json::from_str("\"hi\\r\\nbad\"");
1037        assert!(result.is_err());
1038    }
1039
1040    #[test]
1041    #[cfg(feature = "serde")]
1042    fn send_options_serialize_empty_as_empty_object() {
1043        let json = serde_json::to_value(SendOptions::default()).expect("send options serialize");
1044
1045        assert_eq!(json, serde_json::json!({}));
1046    }
1047
1048    #[test]
1049    #[cfg(feature = "serde")]
1050    fn send_options_serialize_sparse_provider_keyed_payload() {
1051        let mut transport_options = TransportOptions::default();
1052        transport_options.insert(TestOption(String::from("value")));
1053
1054        let options = SendOptions::new()
1055            .with_transport_options(transport_options)
1056            .with_timeout(std::time::Duration::new(3, 25))
1057            .with_idempotency_key(IdempotencyKey::new("idem-123").unwrap())
1058            .with_correlation_id(CorrelationId::new("corr-456").unwrap());
1059
1060        let json = serde_json::to_value(options).expect("send options serialize");
1061
1062        assert_eq!(
1063            json,
1064            serde_json::json!({
1065                "transport_options": {"test": "value"},
1066                "timeout": {"secs": 3, "nanos": 25},
1067                "idempotency_key": "idem-123",
1068                "correlation_id": "corr-456"
1069            })
1070        );
1071    }
1072
1073    #[test]
1074    #[cfg(feature = "serde")]
1075    fn send_options_serialize_envelope() {
1076        let envelope = Envelope::new(
1077            Some("sender@example.com".parse().unwrap()),
1078            vec!["recipient@example.com".parse().unwrap()],
1079        );
1080        let options = SendOptions::new().with_envelope(envelope);
1081
1082        let json = serde_json::to_value(options).expect("send options serialize");
1083
1084        assert_eq!(json["envelope"]["mail_from"], "sender@example.com");
1085        assert_eq!(
1086            json["envelope"]["rcpt_to"],
1087            serde_json::json!(["recipient@example.com"])
1088        );
1089    }
1090
1091    #[test]
1092    #[cfg(feature = "schemars")]
1093    fn transport_options_schema_is_provider_keyed_object() {
1094        let schema = schemars::schema_for!(TransportOptions);
1095        let value = schema.as_value();
1096
1097        assert_eq!(
1098            value.get("type").and_then(|value| value.as_str()),
1099            Some("object")
1100        );
1101        assert_eq!(
1102            value
1103                .get("additionalProperties")
1104                .and_then(serde_json::Value::as_bool),
1105            Some(true)
1106        );
1107    }
1108
1109    #[test]
1110    #[cfg(feature = "schemars")]
1111    fn send_options_schema_allows_omitting_transport_options() {
1112        let schema = schemars::schema_for!(SendOptions);
1113        let value = schema.as_value();
1114
1115        assert!(value.pointer("/properties/transport_options").is_some());
1116        if let Some(required) = value.get("required").and_then(|value| value.as_array()) {
1117            assert!(
1118                !required
1119                    .iter()
1120                    .any(|value| value.as_str() == Some("transport_options"))
1121            );
1122        }
1123    }
1124
1125    #[test]
1126    #[cfg(feature = "serde")]
1127    fn send_options_deserialize_hydrates_registered_transport_options() {
1128        let mut registry = TransportOptionRegistry::new();
1129        registry
1130            .register::<TestOption>()
1131            .expect("register succeeds");
1132
1133        let options = registry
1134            .deserialize_send_options(serde_json::json!({
1135                "envelope": {
1136                    "mail_from": "sender@example.com",
1137                    "rcpt_to": ["recipient@example.com"]
1138                },
1139                "transport_options": {"test": "value"},
1140                "timeout": {"secs": 3, "nanos": 25},
1141                "idempotency_key": "idem-123",
1142                "correlation_id": "corr-456",
1143                "future_field": "ignored"
1144            }))
1145            .expect("send options deserialize");
1146
1147        let envelope = options.envelope.as_ref().expect("envelope hydrates");
1148        assert_eq!(
1149            envelope
1150                .mail_from()
1151                .map(email_message::EmailAddress::as_str),
1152            Some("sender@example.com")
1153        );
1154        assert_eq!(
1155            envelope
1156                .rcpt_to()
1157                .iter()
1158                .map(email_message::EmailAddress::as_str)
1159                .collect::<Vec<_>>(),
1160            vec!["recipient@example.com"]
1161        );
1162        assert_eq!(options.timeout, Some(std::time::Duration::new(3, 25)));
1163        assert_eq!(
1164            options.idempotency_key.as_ref().map(IdempotencyKey::as_str),
1165            Some("idem-123")
1166        );
1167        assert_eq!(
1168            options.correlation_id.as_ref().map(CorrelationId::as_str),
1169            Some("corr-456")
1170        );
1171        assert_eq!(
1172            options
1173                .transport_options
1174                .get::<TestOption>()
1175                .map(|value| value.0.as_str()),
1176            Some("value")
1177        );
1178    }
1179
1180    #[test]
1181    #[cfg(feature = "serde")]
1182    fn send_options_deserialize_rejects_unknown_transport_options_by_default() {
1183        let registry = TransportOptionRegistry::new();
1184
1185        let result = registry.deserialize_send_options(serde_json::json!({
1186            "transport_options": {"missing-provider": {"value": 1}}
1187        }));
1188
1189        let error = result.expect_err("unknown provider key should fail");
1190        let rendered = error.to_string();
1191        assert!(
1192            rendered.contains("missing-provider"),
1193            "error should name the unknown provider key, got `{rendered}`"
1194        );
1195    }
1196
1197    /// Drift guard for [`SendOptionsField`]/[`SendOptionsVisitor`].
1198    ///
1199    /// Two layers protect against silent drift when a new field is added to
1200    /// [`SendOptions`]:
1201    ///
1202    /// 1. The destructure below uses no rest pattern, so adding a sixth field
1203    ///    to `SendOptions` is a **compile error** here until the test is
1204    ///    updated.
1205    /// 2. Each named binding is asserted against a non-default value the
1206    ///    `with_*` builder set on `original`. Even if the destructure is
1207    ///    extended without wiring the new field through `SendOptionsField` /
1208    ///    `SendOptionsVisitor`, the assertion for the new field fails at
1209    ///    runtime (`SendOptionsField::Other` would silently route the
1210    ///    serialized key to `IgnoredAny`).
1211    #[test]
1212    #[cfg(feature = "serde")]
1213    fn send_options_seed_round_trip_covers_every_field() {
1214        let mut transport_options = TransportOptions::default();
1215        transport_options.insert(TestOption(String::from("typed")));
1216
1217        let original = SendOptions::new()
1218            .with_envelope(Envelope::new(
1219                Some("sender@example.com".parse().unwrap()),
1220                vec!["recipient@example.com".parse().unwrap()],
1221            ))
1222            .with_transport_options(transport_options)
1223            .with_timeout(std::time::Duration::new(7, 11))
1224            .with_idempotency_key(IdempotencyKey::new("idem-99").unwrap())
1225            .with_correlation_id(CorrelationId::new("corr-77").unwrap());
1226
1227        let json = serde_json::to_value(&original).expect("serialize");
1228
1229        let mut registry = TransportOptionRegistry::new();
1230        registry
1231            .register::<TestOption>()
1232            .expect("register succeeds");
1233
1234        let hydrated = registry
1235            .deserialize_send_options(json)
1236            .expect("deserialize");
1237
1238        // Exhaustive destructure, intentionally without `..`; adding a field
1239        // to `SendOptions` is a compile error here until this test is updated.
1240        let SendOptions {
1241            envelope,
1242            transport_options,
1243            timeout,
1244            idempotency_key,
1245            correlation_id,
1246        } = &hydrated;
1247
1248        assert_eq!(
1249            envelope
1250                .as_ref()
1251                .and_then(|envelope| envelope.mail_from())
1252                .map(email_message::EmailAddress::as_str),
1253            Some("sender@example.com"),
1254            "envelope did not round-trip"
1255        );
1256        assert_eq!(
1257            transport_options
1258                .get::<TestOption>()
1259                .map(|value| value.0.as_str()),
1260            Some("typed"),
1261            "transport_options did not round-trip"
1262        );
1263        assert_eq!(
1264            *timeout,
1265            Some(std::time::Duration::new(7, 11)),
1266            "timeout did not round-trip"
1267        );
1268        assert_eq!(
1269            idempotency_key.as_ref().map(IdempotencyKey::as_str),
1270            Some("idem-99"),
1271            "idempotency_key did not round-trip"
1272        );
1273        assert_eq!(
1274            correlation_id.as_ref().map(CorrelationId::as_str),
1275            Some("corr-77"),
1276            "correlation_id did not round-trip"
1277        );
1278    }
1279
1280    #[test]
1281    #[cfg(feature = "serde")]
1282    fn send_options_seed_can_ignore_unknown_transport_options() {
1283        use serde::de::DeserializeSeed as _;
1284
1285        let mut registry = TransportOptionRegistry::new();
1286        registry
1287            .register::<TestOption>()
1288            .expect("register succeeds");
1289
1290        let payload = serde_json::json!({
1291            "transport_options": {
1292                "test": "value",
1293                "unknown": {"value": 1}
1294            }
1295        });
1296        let options = registry
1297            .send_options_seed()
1298            .ignore_unknown_transport_options()
1299            .deserialize(payload)
1300            .expect("ignore_unknown succeeds");
1301
1302        assert_eq!(
1303            options
1304                .transport_options
1305                .get::<TestOption>()
1306                .map(|value| value.0.as_str()),
1307            Some("value")
1308        );
1309    }
1310
1311    #[test]
1312    #[cfg(feature = "serde")]
1313    fn send_options_deserialize_rejects_malformed_known_transport_options() {
1314        let mut registry = TransportOptionRegistry::new();
1315        registry
1316            .register::<OtherTestOption>()
1317            .expect("register succeeds");
1318
1319        let result = registry.deserialize_send_options(serde_json::json!({
1320            "transport_options": {"other": "not-an-object"}
1321        }));
1322
1323        assert!(result.is_err());
1324    }
1325
1326    #[test]
1327    #[cfg(feature = "serde")]
1328    fn transport_options_seed_drives_a_streaming_deserializer() {
1329        // Format-agnostic seed exercise: drive the seed straight off a
1330        // streaming `serde_json::Deserializer` (not a `serde_json::Value`),
1331        // which is the path any non-JSON serde format would also take.
1332        use serde::de::DeserializeSeed as _;
1333
1334        let mut registry = TransportOptionRegistry::new();
1335        registry
1336            .register::<TestOption>()
1337            .expect("register succeeds");
1338
1339        let mut original = TransportOptions::default();
1340        original.insert(TestOption(String::from("round-trip")));
1341
1342        let bytes = serde_json::to_vec(&original).expect("serialize");
1343        let mut deserializer = serde_json::Deserializer::from_slice(&bytes);
1344        let hydrated = registry
1345            .transport_options_seed()
1346            .deserialize(&mut deserializer)
1347            .expect("hydrate from streaming deserializer");
1348
1349        assert_eq!(
1350            hydrated.get::<TestOption>().map(|value| value.0.as_str()),
1351            Some("round-trip")
1352        );
1353    }
1354
1355    #[test]
1356    #[cfg(feature = "serde")]
1357    fn transport_options_seed_hydrates_multiple_providers_in_one_payload() {
1358        use serde::de::DeserializeSeed as _;
1359
1360        let mut registry = TransportOptionRegistry::new();
1361        registry
1362            .register::<TestOption>()
1363            .expect("register TestOption");
1364        registry
1365            .register::<OtherTestOption>()
1366            .expect("register OtherTestOption");
1367
1368        let payload = serde_json::json!({
1369            "test": "first",
1370            "other": {"value": 42},
1371        });
1372        let hydrated = registry
1373            .transport_options_seed()
1374            .deserialize(payload)
1375            .expect("hydrate succeeds");
1376
1377        assert_eq!(
1378            hydrated.get::<TestOption>().map(|value| value.0.as_str()),
1379            Some("first")
1380        );
1381        assert_eq!(
1382            hydrated.get::<OtherTestOption>().map(|value| value.value),
1383            Some(42)
1384        );
1385    }
1386
1387    #[test]
1388    #[cfg(feature = "serde")]
1389    fn transport_options_seed_can_ignore_unknown_provider_keys() {
1390        use serde::de::DeserializeSeed as _;
1391
1392        let mut registry = TransportOptionRegistry::new();
1393        registry
1394            .register::<TestOption>()
1395            .expect("register succeeds");
1396
1397        let payload = serde_json::json!({
1398            "test": "value",
1399            "missing-provider": {"anything": 1},
1400        });
1401        let hydrated = registry
1402            .transport_options_seed()
1403            .ignore_unknown_provider_keys()
1404            .deserialize(payload)
1405            .expect("ignore_unknown succeeds");
1406
1407        assert_eq!(
1408            hydrated.get::<TestOption>().map(|value| value.0.as_str()),
1409            Some("value")
1410        );
1411    }
1412
1413    #[test]
1414    #[cfg(all(feature = "serde", feature = "tracing"))]
1415    fn transport_options_seed_traces_ignored_provider_keys() {
1416        use std::sync::{Arc, Mutex};
1417
1418        use ::tracing::field::{Field, Visit};
1419        use serde::de::DeserializeSeed as _;
1420        use tracing_subscriber::{Layer, layer::Context, prelude::*, registry::Registry};
1421
1422        #[derive(Clone, Default)]
1423        struct RecordedEvents {
1424            fields: Arc<Mutex<Vec<String>>>,
1425        }
1426
1427        struct EventLayer(RecordedEvents);
1428
1429        impl<S> Layer<S> for EventLayer
1430        where
1431            S: ::tracing::Subscriber,
1432        {
1433            fn on_event(&self, event: &::tracing::Event<'_>, _ctx: Context<'_, S>) {
1434                let mut recorder = FieldRecorder::default();
1435                event.record(&mut recorder);
1436                self.0
1437                    .fields
1438                    .lock()
1439                    .expect("event mutex poisoned")
1440                    .extend(recorder.fields);
1441            }
1442        }
1443
1444        #[derive(Default)]
1445        struct FieldRecorder {
1446            fields: Vec<String>,
1447        }
1448
1449        impl Visit for FieldRecorder {
1450            fn record_str(&mut self, field: &Field, value: &str) {
1451                self.fields.push(format!("{}={value}", field.name()));
1452            }
1453
1454            fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
1455                self.fields.push(format!("{}={value:?}", field.name()));
1456            }
1457        }
1458
1459        let recorded = RecordedEvents::default();
1460        let subscriber = Registry::default().with(EventLayer(recorded.clone()));
1461
1462        let mut registry = TransportOptionRegistry::new();
1463        registry
1464            .register::<TestOption>()
1465            .expect("register succeeds");
1466
1467        ::tracing::subscriber::with_default(subscriber, || {
1468            registry
1469                .transport_options_seed()
1470                .ignore_unknown_provider_keys()
1471                .deserialize(serde_json::json!({"missing-provider": {"anything": 1}}))
1472                .expect("ignore_unknown succeeds");
1473        });
1474
1475        let fields = recorded
1476            .fields
1477            .lock()
1478            .expect("event mutex poisoned")
1479            .clone();
1480        assert!(
1481            fields
1482                .iter()
1483                .any(|field| field == "provider_key=missing-provider"),
1484            "ignored provider key should be traced: {fields:?}"
1485        );
1486    }
1487
1488    #[test]
1489    #[cfg(feature = "serde")]
1490    fn send_options_seed_works_when_nested_inside_another_seed() {
1491        // Exercises the composability case `SendOptionsSeed` exists for: a
1492        // parent custom visitor threads the registry into a nested
1493        // `SendOptionsSeed` via `next_value_seed`, the way a downstream
1494        // `SendEmailRequestSeed` would.
1495        use serde::de::{DeserializeSeed, MapAccess, Visitor};
1496
1497        struct WrapperSeed<'a> {
1498            registry: &'a TransportOptionRegistry,
1499        }
1500        struct WrapperVisitor<'a> {
1501            registry: &'a TransportOptionRegistry,
1502        }
1503
1504        impl<'de> DeserializeSeed<'de> for WrapperSeed<'_> {
1505            type Value = SendOptions;
1506            fn deserialize<D>(self, deserializer: D) -> Result<SendOptions, D::Error>
1507            where
1508                D: serde::Deserializer<'de>,
1509            {
1510                deserializer.deserialize_map(WrapperVisitor {
1511                    registry: self.registry,
1512                })
1513            }
1514        }
1515
1516        impl<'de> Visitor<'de> for WrapperVisitor<'_> {
1517            type Value = SendOptions;
1518            fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1519                f.write_str("a wrapper map containing a `send_options` field")
1520            }
1521            fn visit_map<A>(self, mut map: A) -> Result<SendOptions, A::Error>
1522            where
1523                A: MapAccess<'de>,
1524            {
1525                let mut found = None;
1526                while let Some(key) = map.next_key::<String>()? {
1527                    if key == "send_options" {
1528                        found = Some(map.next_value_seed(SendOptionsSeed {
1529                            registry: self.registry,
1530                            ignore_unknown: false,
1531                        })?);
1532                    } else {
1533                        map.next_value::<serde::de::IgnoredAny>()?;
1534                    }
1535                }
1536                found.ok_or_else(|| serde::de::Error::missing_field("send_options"))
1537            }
1538        }
1539
1540        let mut registry = TransportOptionRegistry::new();
1541        registry
1542            .register::<TestOption>()
1543            .expect("register succeeds");
1544
1545        let payload = serde_json::json!({
1546            "send_options": {
1547                "transport_options": {"test": "nested"},
1548                "timeout": {"secs": 1, "nanos": 0}
1549            }
1550        });
1551        let send_options = WrapperSeed {
1552            registry: &registry,
1553        }
1554        .deserialize(payload)
1555        .expect("nested seed succeeds");
1556
1557        assert_eq!(
1558            send_options
1559                .transport_options
1560                .get::<TestOption>()
1561                .map(|value| value.0.as_str()),
1562            Some("nested")
1563        );
1564        assert_eq!(
1565            send_options.timeout,
1566            Some(std::time::Duration::from_secs(1))
1567        );
1568    }
1569
1570    #[test]
1571    #[cfg(feature = "serde")]
1572    fn empty_send_options_round_trip_through_seed() {
1573        let registry = TransportOptionRegistry::new();
1574        let json = serde_json::to_value(SendOptions::default()).expect("serialize");
1575        let hydrated = registry
1576            .deserialize_send_options(json)
1577            .expect("deserialize empty");
1578
1579        let SendOptions {
1580            envelope,
1581            transport_options,
1582            timeout,
1583            idempotency_key,
1584            correlation_id,
1585        } = &hydrated;
1586
1587        assert!(envelope.is_none());
1588        assert!(transport_options.is_empty());
1589        assert!(timeout.is_none());
1590        assert!(idempotency_key.is_none());
1591        assert!(correlation_id.is_none());
1592    }
1593
1594    #[test]
1595    #[cfg(feature = "serde")]
1596    fn transport_option_registry_debug_lists_registered_provider_keys() {
1597        let mut registry = TransportOptionRegistry::new();
1598        registry
1599            .register::<TestOption>()
1600            .expect("register succeeds");
1601        registry
1602            .register::<OtherTestOption>()
1603            .expect("register succeeds");
1604
1605        let rendered = format!("{registry:?}");
1606        for needle in ["test", "other", "TestOption", "OtherTestOption"] {
1607            assert!(
1608                rendered.contains(needle),
1609                "expected `{needle}` in {rendered}"
1610            );
1611        }
1612    }
1613
1614    #[test]
1615    #[cfg(feature = "serde")]
1616    fn transport_option_registry_lists_provider_keys_in_stable_order() {
1617        let mut registry = TransportOptionRegistry::new();
1618        registry
1619            .register::<TestOption>()
1620            .expect("register succeeds");
1621        registry
1622            .register::<OtherTestOption>()
1623            .expect("register succeeds");
1624
1625        assert_eq!(registry.provider_keys(), vec!["other", "test"]);
1626    }
1627
1628    #[test]
1629    fn transport_options_store_typed_values() {
1630        let mut options = TransportOptions::default();
1631        options.insert(TestOption(String::from("value")));
1632
1633        assert_eq!(
1634            options.get::<TestOption>().map(|value| value.0.as_str()),
1635            Some("value")
1636        );
1637        assert_eq!(
1638            options
1639                .remove::<TestOption>()
1640                .as_ref()
1641                .map(|value| value.0.as_str()),
1642            Some("value")
1643        );
1644        assert!(options.get::<TestOption>().is_none());
1645    }
1646
1647    #[test]
1648    fn transport_options_with_inserts_by_value() {
1649        let options = TransportOptions::default().with(TestOption(String::from("value")));
1650
1651        assert_eq!(
1652            options.get::<TestOption>().map(|value| value.0.as_str()),
1653            Some("value")
1654        );
1655    }
1656
1657    #[test]
1658    #[cfg(feature = "serde")]
1659    fn send_options_with_transport_option_chains_across_providers() {
1660        let options = SendOptions::new()
1661            .with_transport_option(TestOption(String::from("value")))
1662            .with_transport_option(OtherTestOption { value: 42 });
1663
1664        assert_eq!(
1665            options
1666                .transport_options
1667                .get::<TestOption>()
1668                .map(|value| value.0.as_str()),
1669            Some("value")
1670        );
1671        assert_eq!(
1672            options.transport_options.get::<OtherTestOption>(),
1673            Some(&OtherTestOption { value: 42 })
1674        );
1675    }
1676
1677    #[test]
1678    #[cfg(feature = "serde")]
1679    fn send_options_with_transport_option_replaces_same_type() {
1680        let options = SendOptions::new()
1681            .with_transport_option(TestOption(String::from("first")))
1682            .with_transport_option(TestOption(String::from("second")));
1683
1684        assert_eq!(
1685            options
1686                .transport_options
1687                .get::<TestOption>()
1688                .map(|value| value.0.as_str()),
1689            Some("second")
1690        );
1691    }
1692
1693    #[test]
1694    fn transport_options_debug_lists_typed_slot_names() {
1695        let mut options = TransportOptions::default();
1696        options.insert(TestOption(String::from("v")));
1697
1698        let rendered = format!("{options:?}");
1699        assert!(rendered.contains("TestOption"), "got {rendered}");
1700    }
1701
1702    #[test]
1703    #[cfg(feature = "serde")]
1704    fn transport_options_serialize_provider_keyed_map() {
1705        let mut options = TransportOptions::default();
1706        options.insert(TestOption(String::from("value")));
1707        options.insert(OtherTestOption { value: 42 });
1708
1709        let json = serde_json::to_value(&options).expect("transport options serialize");
1710        assert_eq!(json["test"], serde_json::json!("value"));
1711        assert_eq!(json["other"], serde_json::json!({"value": 42}));
1712    }
1713
1714    #[test]
1715    #[cfg(feature = "serde")]
1716    fn transport_option_registry_hydrates_and_overwrites() {
1717        let mut registry = TransportOptionRegistry::new();
1718        registry
1719            .register::<TestOption>()
1720            .expect("register succeeds");
1721
1722        let mut options = TransportOptions::default();
1723        options.insert(TestOption(String::from("typed")));
1724
1725        let hydrated = registry
1726            .hydrate_into("test", serde_json::json!("json"), &mut options)
1727            .expect("hydration succeeds");
1728
1729        assert!(hydrated);
1730        assert_eq!(
1731            options.get::<TestOption>().map(|value| value.0.as_str()),
1732            Some("json")
1733        );
1734    }
1735
1736    #[test]
1737    #[cfg(feature = "serde")]
1738    fn transport_option_registry_ignores_unknown_keys() {
1739        let registry = TransportOptionRegistry::new();
1740        let mut options = TransportOptions::default();
1741
1742        let hydrated = registry
1743            .hydrate_into("unknown", serde_json::json!({"value": 1}), &mut options)
1744            .expect("unknown keys do not error");
1745
1746        assert!(!hydrated);
1747        assert!(options.is_empty());
1748    }
1749
1750    #[test]
1751    #[cfg(feature = "serde")]
1752    fn transport_option_registry_rejects_malformed_known_values() {
1753        let mut registry = TransportOptionRegistry::new();
1754        registry
1755            .register::<OtherTestOption>()
1756            .expect("register succeeds");
1757        let mut options = TransportOptions::default();
1758
1759        let result =
1760            registry.hydrate_into("other", serde_json::json!("not-an-object"), &mut options);
1761
1762        assert!(result.is_err());
1763        assert!(options.is_empty());
1764    }
1765
1766    #[test]
1767    #[cfg(feature = "serde")]
1768    fn transport_option_registry_rejects_duplicate_provider_keys() {
1769        let mut registry = TransportOptionRegistry::new();
1770        registry
1771            .register::<TestOption>()
1772            .expect("register succeeds");
1773
1774        let error = registry
1775            .register::<DuplicateKeyOption>()
1776            .expect_err("duplicate provider key should fail");
1777
1778        match error {
1779            TransportOptionRegistryError::DuplicateProviderKey { provider_key, .. } => {
1780                assert_eq!(provider_key, "test");
1781            }
1782        }
1783    }
1784}