outbox_core/
builder.rs

1//! Fluent builder for [`OutboxManager`].
2//!
3//! The manager has several collaborators that must all be provided before it
4//! can run: a storage backend, a message transport, a configuration, and a
5//! shutdown channel — plus a DLQ heap when the `dlq` feature is enabled.
6//! [`OutboxManagerBuilder`] assembles these pieces step by step and validates
7//! them once at [`build`](OutboxManagerBuilder::build) time, returning a
8//! structured [`OutboxError::ConfigError`] if anything is missing rather than
9//! panicking.
10//!
11//! # Example
12//!
13//! ```ignore
14//! use std::sync::Arc;
15//! use tokio::sync::watch;
16//! use outbox_core::prelude::*;
17//!
18//! # async fn run(
19//! #     storage: Arc<impl OutboxStorage<MyEvent> + Send + Sync + 'static>,
20//! #     publisher: Arc<impl Transport<MyEvent> + Send + Sync + 'static>,
21//! # ) -> Result<(), OutboxError>
22//! # where MyEvent: std::fmt::Debug + Clone + serde::Serialize + Send + Sync + 'static,
23//! # {
24//! let (_tx, rx) = watch::channel(false);
25//! let manager = OutboxManagerBuilder::new()
26//!     .storage(storage)
27//!     .publisher(publisher)
28//!     .config(Arc::new(OutboxConfig::default()))
29//!     .shutdown_rx(rx)
30//!     .build()?;
31//! manager.run().await
32//! # }
33//! ```
34
35use crate::config::OutboxConfig;
36#[cfg(feature = "dlq")]
37use crate::dlq::storage::DlqHeap;
38use crate::error::OutboxError;
39use crate::manager::OutboxManager;
40use crate::prelude::{OutboxStorage, Transport};
41use serde::Serialize;
42use std::fmt::Debug;
43use std::sync::Arc;
44use tokio::sync::watch::Receiver;
45
46/// Fluent builder that assembles an [`OutboxManager`].
47///
48/// All fields are optional during construction — validation happens once in
49/// [`build`](Self::build). Setter methods consume and return `self`, so a
50/// builder can be constructed in a single chained expression. The builder is
51/// generic over:
52///
53/// - `S` — [`OutboxStorage`] implementation (typically a database adapter)
54/// - `P` — [`Transport`] implementation (message broker publisher)
55/// - `PT` — the user's domain event payload type (`Debug + Clone + Serialize`)
56///
57/// # Required vs optional
58///
59/// | Setter | Required | Notes |
60/// |---|---|---|
61/// | [`storage`](Self::storage) | yes | fails `build()` if missing |
62/// | [`publisher`](Self::publisher) | yes | fails `build()` if missing |
63/// | [`config`](Self::config) | yes | fails `build()` if missing |
64/// | [`shutdown_rx`](Self::shutdown_rx) | yes | fails `build()` if missing |
65/// | [`dlq_heap`](Self::dlq_heap) | yes *(feature `dlq` only)* | fails `build()` if missing when feature is on |
66pub struct OutboxManagerBuilder<S, P, PT>
67where
68    PT: Debug + Clone + Serialize,
69{
70    storage: Option<Arc<S>>,
71    publisher: Option<Arc<P>>,
72    config: Option<Arc<OutboxConfig<PT>>>,
73    shutdown_rx: Option<Receiver<bool>>,
74    #[cfg(feature = "dlq")]
75    dlq_heap: Option<Arc<dyn DlqHeap>>,
76}
77impl<S, P, PT> Default for OutboxManagerBuilder<S, P, PT>
78where
79    PT: Debug + Clone + Serialize,
80{
81    fn default() -> Self {
82        Self {
83            storage: None,
84            publisher: None,
85            config: None,
86            shutdown_rx: None,
87            #[cfg(feature = "dlq")]
88            dlq_heap: None,
89        }
90    }
91}
92
93impl<S, P, PT> OutboxManagerBuilder<S, P, PT>
94where
95    S: OutboxStorage<PT> + Send + Sync + 'static,
96    P: Transport<PT> + Send + Sync + 'static,
97    PT: Debug + Clone + Serialize + Send + Sync + 'static,
98{
99    /// Creates an empty builder with all fields unset.
100    ///
101    /// Equivalent to [`Default::default`]; kept as a discoverable entry point
102    /// so callers do not need to import the `Default` trait.
103    #[must_use]
104    pub fn new() -> Self {
105        Self::default()
106    }
107
108    /// Sets the storage backend used for reading pending events, locking them,
109    /// persisting status transitions, and deleting expired rows.
110    #[must_use]
111    pub fn storage(mut self, s: Arc<S>) -> Self {
112        self.storage = Some(s);
113        self
114    }
115
116    /// Sets the transport (broker publisher) that delivers events to the
117    /// outside world.
118    #[must_use]
119    pub fn publisher(mut self, p: Arc<P>) -> Self {
120        self.publisher = Some(p);
121        self
122    }
123
124    /// Sets the runtime configuration (batch size, poll interval, GC cadence,
125    /// lock timeout, idempotency strategy).
126    #[must_use]
127    pub fn config(mut self, config: Arc<OutboxConfig<PT>>) -> Self {
128        self.config = Some(config);
129        self
130    }
131
132    /// Sets the shutdown channel. The manager stops its worker loop as soon as
133    /// `true` is observed on this receiver.
134    #[must_use]
135    pub fn shutdown_rx(mut self, rx: Receiver<bool>) -> Self {
136        self.shutdown_rx = Some(rx);
137        self
138    }
139
140    /// Sets the Dead-Letter-Queue heap that tracks per-event failure counts.
141    ///
142    /// Required when the crate is built with `--features dlq`.
143    #[cfg(feature = "dlq")]
144    #[must_use]
145    pub fn dlq_heap(mut self, heap: Arc<dyn DlqHeap>) -> Self {
146        self.dlq_heap = Some(heap);
147        self
148    }
149
150    /// Consumes the builder and returns a fully wired [`OutboxManager`].
151    ///
152    /// # Errors
153    ///
154    /// Returns [`OutboxError::ConfigError`] with a message identifying the
155    /// first missing dependency if any required field has not been set.
156    /// The diagnostic mentions one of: `Storage`, `Publisher`, `Config`,
157    /// `Shutdown channel`, or — under feature `dlq` — `DLQ heap`.
158    pub fn build(self) -> Result<OutboxManager<S, P, PT>, OutboxError> {
159        #[cfg(feature = "dlq")]
160        return Ok(OutboxManager::new(
161            self.storage
162                .ok_or_else(|| OutboxError::ConfigError("Storage is missing".to_string()))?,
163            self.publisher
164                .ok_or_else(|| OutboxError::ConfigError("Publisher is missing".to_string()))?,
165            self.config
166                .ok_or_else(|| OutboxError::ConfigError("Config is missing".to_string()))?,
167            self.dlq_heap
168                .ok_or_else(|| OutboxError::ConfigError("DLQ heap is missing".to_string()))?,
169            self.shutdown_rx.ok_or_else(|| {
170                OutboxError::ConfigError("Shutdown channel is missing".to_string())
171            })?,
172        ));
173        #[cfg(not(feature = "dlq"))]
174        return Ok(OutboxManager::new(
175            self.storage
176                .ok_or_else(|| OutboxError::ConfigError("Storage is missing".to_string()))?,
177            self.publisher
178                .ok_or_else(|| OutboxError::ConfigError("Publisher is missing".to_string()))?,
179            self.config
180                .ok_or_else(|| OutboxError::ConfigError("Config is missing".to_string()))?,
181            self.shutdown_rx.ok_or_else(|| {
182                OutboxError::ConfigError("Shutdown channel is missing".to_string())
183            })?,
184        ));
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191    use crate::config::IdempotencyStrategy;
192    use crate::dlq::storage::MockDlqHeap;
193    use crate::publisher::MockTransport;
194    use crate::storage::MockOutboxStorage;
195    use rstest::rstest;
196    use serde::Deserialize;
197    use tokio::sync::watch;
198
199    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
200    enum SomeDomainEvent {
201        SomeEvent(String),
202    }
203
204    #[rstest]
205    fn test_success_build() {
206        let config = OutboxConfig {
207            batch_size: 100,
208            retention_days: 7,
209            gc_interval_secs: 3600,
210            poll_interval_secs: 5,
211            lock_timeout_mins: 5,
212            idempotency_strategy: IdempotencyStrategy::<SomeDomainEvent>::None,
213            ..Default::default()
214        };
215
216        let storage_mock = MockOutboxStorage::<SomeDomainEvent>::new();
217        let transport_mock = MockTransport::<SomeDomainEvent>::new();
218
219        #[cfg(feature = "dlq")]
220        let dlq_heap_mock: MockDlqHeap = MockDlqHeap::new();
221
222        let (_shutdown_tx, shutdown_rx) = watch::channel(false);
223
224        #[cfg(feature = "dlq")]
225        let outbox_manager = OutboxManagerBuilder::new()
226            .storage(Arc::new(storage_mock))
227            .publisher(Arc::new(transport_mock))
228            .config(Arc::new(config))
229            .shutdown_rx(shutdown_rx)
230            .dlq_heap(Arc::new(dlq_heap_mock))
231            .build();
232
233        #[cfg(not(feature = "dlq"))]
234        let outbox_manager = OutboxManagerBuilder::new()
235            .storage(Arc::new(storage_mock))
236            .publisher(Arc::new(transport_mock))
237            .config(Arc::new(config))
238            .shutdown_rx(shutdown_rx)
239            .build();
240
241        assert!(outbox_manager.is_ok());
242    }
243
244    type Builder = OutboxManagerBuilder<
245        MockOutboxStorage<SomeDomainEvent>,
246        MockTransport<SomeDomainEvent>,
247        SomeDomainEvent,
248    >;
249
250    fn default_config() -> Arc<OutboxConfig<SomeDomainEvent>> {
251        Arc::new(OutboxConfig {
252            batch_size: 100,
253            retention_days: 7,
254            gc_interval_secs: 3600,
255            poll_interval_secs: 5,
256            lock_timeout_mins: 5,
257            idempotency_strategy: IdempotencyStrategy::None,
258            ..Default::default()
259        })
260    }
261
262    #[rstest]
263    fn default_builder_fails_on_build_with_config_error() {
264        let result = Builder::default().build();
265        assert!(matches!(result, Err(OutboxError::ConfigError(_))));
266    }
267
268    fn assert_config_error_with(
269        result: Result<
270            OutboxManager<
271                MockOutboxStorage<SomeDomainEvent>,
272                MockTransport<SomeDomainEvent>,
273                SomeDomainEvent,
274            >,
275            OutboxError,
276        >,
277        needle: &str,
278    ) {
279        match result {
280            Err(OutboxError::ConfigError(msg)) => {
281                assert!(msg.contains(needle), "expected '{needle}' in '{msg}'");
282            }
283            Err(other) => panic!("expected ConfigError, got {other:?}"),
284            Ok(_) => panic!("expected ConfigError, got Ok(..)"),
285        }
286    }
287
288    #[rstest]
289    fn new_matches_default() {
290        let r1 = Builder::new().build();
291        let r2 = Builder::default().build();
292        let m1 = match r1 {
293            Err(OutboxError::ConfigError(m)) => m,
294            _ => panic!("new().build() should fail with ConfigError"),
295        };
296        let m2 = match r2 {
297            Err(OutboxError::ConfigError(m)) => m,
298            _ => panic!("default().build() should fail with ConfigError"),
299        };
300        assert_eq!(m1, m2);
301    }
302
303    #[rstest]
304    fn build_fails_without_storage() {
305        let (_tx, rx) = watch::channel(false);
306        let b = Builder::new()
307            .publisher(Arc::new(MockTransport::new()))
308            .config(default_config())
309            .shutdown_rx(rx);
310        #[cfg(feature = "dlq")]
311        let b = b.dlq_heap(Arc::new(MockDlqHeap::new()));
312
313        assert_config_error_with(b.build(), "Storage");
314    }
315
316    #[rstest]
317    fn build_fails_without_publisher() {
318        let (_tx, rx) = watch::channel(false);
319        let b = Builder::new()
320            .storage(Arc::new(MockOutboxStorage::new()))
321            .config(default_config())
322            .shutdown_rx(rx);
323        #[cfg(feature = "dlq")]
324        let b = b.dlq_heap(Arc::new(MockDlqHeap::new()));
325
326        assert_config_error_with(b.build(), "Publisher");
327    }
328
329    #[rstest]
330    fn build_fails_without_config() {
331        let (_tx, rx) = watch::channel(false);
332        let b = Builder::new()
333            .storage(Arc::new(MockOutboxStorage::new()))
334            .publisher(Arc::new(MockTransport::new()))
335            .shutdown_rx(rx);
336        #[cfg(feature = "dlq")]
337        let b = b.dlq_heap(Arc::new(MockDlqHeap::new()));
338
339        assert_config_error_with(b.build(), "Config");
340    }
341
342    #[rstest]
343    fn build_fails_without_shutdown_rx() {
344        let b = Builder::new()
345            .storage(Arc::new(MockOutboxStorage::new()))
346            .publisher(Arc::new(MockTransport::new()))
347            .config(default_config());
348        #[cfg(feature = "dlq")]
349        let b = b.dlq_heap(Arc::new(MockDlqHeap::new()));
350
351        assert_config_error_with(b.build(), "Shutdown");
352    }
353
354    #[cfg(feature = "dlq")]
355    #[rstest]
356    fn build_fails_without_dlq_heap() {
357        let (_tx, rx) = watch::channel(false);
358        let result = Builder::new()
359            .storage(Arc::new(MockOutboxStorage::new()))
360            .publisher(Arc::new(MockTransport::new()))
361            .config(default_config())
362            .shutdown_rx(rx)
363            .build();
364        assert_config_error_with(result, "DLQ");
365    }
366
367    #[rstest]
368    fn build_is_insensitive_to_setter_order() {
369        let (_tx, rx) = watch::channel(false);
370        let b = Builder::new()
371            .shutdown_rx(rx)
372            .config(default_config())
373            .publisher(Arc::new(MockTransport::new()))
374            .storage(Arc::new(MockOutboxStorage::new()));
375        #[cfg(feature = "dlq")]
376        let b = b.dlq_heap(Arc::new(MockDlqHeap::new()));
377
378        assert!(b.build().is_ok());
379    }
380}
outbox_core/builder.rs

outbox_core/
builder.rs
