outbox_core/
service.rs

1//! Producer-side entry point: writes new events into the outbox.
2//!
3//! [`OutboxService`] is what application code calls when a domain action
4//! needs to emit an event. It applies the configured
5//! [`IdempotencyStrategy`](crate::config::IdempotencyStrategy) to produce (or
6//! accept) a token, optionally reserves that token through an external
7//! [`IdempotencyStorageProvider`] to reject duplicates, and then persists the
8//! event via an [`OutboxWriter`]. The worker side
9//! ([`OutboxManager`](crate::manager::OutboxManager)) picks it up later.
10
11use crate::error::OutboxError;
12use crate::idempotency::storage::NoIdempotency;
13use crate::model::Event;
14use crate::object::{EventType, IdempotencyToken, Payload};
15use crate::prelude::{IdempotencyStorageProvider, OutboxConfig};
16use crate::storage::OutboxWriter;
17use serde::Serialize;
18use std::fmt::Debug;
19use std::sync::Arc;
20
21/// Producer-side facade for writing outbox events.
22///
23/// The service is generic over:
24///
25/// - `W` — [`OutboxWriter`] implementation (persists the event row)
26/// - `S` — [`IdempotencyStorageProvider`] implementation used to reserve
27///   tokens; set to [`NoIdempotency`] when no external reservation is needed
28/// - `P` — the user's domain event payload type (`Debug + Clone + Serialize`)
29///
30/// Construct with [`new`](Self::new) when events should be written without an
31/// external idempotency check, or with
32/// [`with_idempotency`](Self::with_idempotency) to wire a reservation backend.
33pub struct OutboxService<W, S, P>
34where
35    P: Debug + Clone + Serialize + Send + Sync,
36{
37    writer: Arc<W>,
38    config: Arc<OutboxConfig<P>>,
39    idempotency_storage: Option<Arc<S>>,
40}
41
42impl<W, P> OutboxService<W, NoIdempotency, P>
43where
44    W: OutboxWriter<P> + Send + Sync + 'static,
45    P: Debug + Clone + Serialize + Send + Sync,
46{
47    /// Creates a service without any external idempotency reservation.
48    ///
49    /// Tokens are still produced according to `config.idempotency_strategy`
50    /// and written alongside the event, so downstream consumers can deduplicate
51    /// on their side, but no pre-insert uniqueness check is performed here.
52    /// Use [`with_idempotency`](OutboxService::with_idempotency) to attach a
53    /// reservation store (for example Redis) when at-producer deduplication
54    /// is required.
55    pub fn new(writer: Arc<W>, config: Arc<OutboxConfig<P>>) -> Self {
56        Self {
57            writer,
58            config,
59            idempotency_storage: None,
60        }
61    }
62}
63
64impl<W, S, P> OutboxService<W, S, P>
65where
66    W: OutboxWriter<P> + Send + Sync + 'static,
67    S: IdempotencyStorageProvider + Send + Sync + 'static,
68    P: Debug + Clone + Serialize + Send + Sync,
69{
70    /// Creates a service wired to an external idempotency reservation store.
71    ///
72    /// Before inserting the event, the service calls
73    /// [`IdempotencyStorageProvider::try_reserve`] with the token produced by
74    /// the configured strategy. If the reservation returns `false`, the insert
75    /// is skipped and [`OutboxError::DuplicateEvent`] is propagated to the
76    /// caller.
77    pub fn with_idempotency(
78        writer: Arc<W>,
79        config: Arc<OutboxConfig<P>>,
80        idempotency_storage: Arc<S>,
81    ) -> Self {
82        Self {
83            writer,
84            idempotency_storage: Some(idempotency_storage),
85            config,
86        }
87    }
88    /// Adds a new event to the outbox storage with idempotency checks.
89    ///
90    /// The token is derived from
91    /// [`IdempotencyStrategy`](crate::config::IdempotencyStrategy) on the
92    /// configured [`OutboxConfig`]:
93    ///
94    /// - `Provided` — uses `provided_token` as-is (`None` skips reservation).
95    /// - `Uuid` — generates a fresh UUID v7; `provided_token` is ignored.
96    /// - `Custom(callback)` — passes the event about to be written to the
97    ///   callback and uses the returned `String` as the token.
98    /// - `None` — no token is produced and reservation is skipped.
99    ///
100    /// If an idempotency provider is configured and a token was produced, it
101    /// will first attempt to reserve the token to prevent duplicate processing.
102    ///
103    /// # Errors
104    ///
105    /// Returns [`OutboxError::DuplicateEvent`] if the event token has already
106    /// been used. Returns any [`OutboxError`] variant propagated from the
107    /// reservation call or from the writer's `insert_event`.
108    ///
109    /// # Transactional usage
110    ///
111    /// The `executor` argument is forwarded to the configured
112    /// [`OutboxWriter`](crate::storage::OutboxWriter) — for sqlx-based backends
113    /// this is typically `&mut *tx` borrowed from a held `sqlx::Transaction`.
114    /// Threading the transaction through the call is what makes the outbox
115    /// write **atomic with the caller's business write**: both rows commit or
116    /// roll back together.
117    ///
118    /// # Example
119    ///
120    /// ```ignore
121    /// use std::sync::Arc;
122    /// use outbox_core::prelude::*;
123    ///
124    /// # async fn emit(
125    /// #     service: OutboxService<impl OutboxWriter<MyEvent> + Send + Sync + 'static,
126    /// #                            NoIdempotency,
127    /// #                            MyEvent>,
128    /// #     payload: MyEvent,
129    /// #     mut tx: sqlx::Transaction<'_, sqlx::Postgres>,
130    /// # ) -> Result<(), OutboxError>
131    /// # where MyEvent: std::fmt::Debug + Clone + serde::Serialize + Send + Sync,
132    /// # {
133    /// service.add_event("order.created", payload, None, &mut *tx).await?;
134    /// # Ok(()) }
135    /// ```
136    pub async fn add_event(
137        &self,
138        event_type: &str,
139        payload: P,
140        provided_token: Option<String>,
141        executor: W::Executor<'_>,
142    ) -> Result<(), OutboxError>
143    where
144        P: Debug + Clone + Serialize + Send + Sync,
145    {
146        let mut event = Event::new(EventType::new(event_type), Payload::new(payload), None);
147
148        let i_token = self
149            .config
150            .idempotency_strategy
151            .invoke(provided_token, &event)
152            .map(IdempotencyToken::new);
153
154        if let Some(i_provider) = &self.idempotency_storage
155            && let Some(ref token) = i_token
156            && !i_provider.try_reserve(token).await?
157        {
158            return Err(OutboxError::DuplicateEvent);
159        }
160
161        event.idempotency_token = i_token;
162        self.writer.insert_event(event, executor).await
163    }
164}
165
166#[cfg(test)]
167#[allow(clippy::unwrap_used)]
168mod tests {
169    use super::*;
170    use crate::config::IdempotencyStrategy;
171    use crate::idempotency::storage::MockIdempotencyStorageProvider;
172    use async_trait::async_trait;
173    use rstest::rstest;
174    use serde::Deserialize;
175    use std::sync::Mutex;
176
177    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
178    struct TestPayload {
179        kind: String,
180    }
181
182    fn payload() -> TestPayload {
183        TestPayload { kind: "k".into() }
184    }
185
186    fn config_with(strategy: IdempotencyStrategy<TestPayload>) -> Arc<OutboxConfig<TestPayload>> {
187        Arc::new(OutboxConfig {
188            batch_size: 100,
189            retention_days: 7,
190            gc_interval_secs: 3600,
191            poll_interval_secs: 10,
192            lock_timeout_mins: 5,
193            idempotency_strategy: strategy,
194            dlq_threshold: 10,
195            dlq_interval_secs: 1,
196        })
197    }
198
199    /// Test double for `OutboxWriter`: captures every event handed to
200    /// `insert_event` and (optionally) returns a pre-configured error instead
201    /// of succeeding. Uses `()` as the executor handle — tests don't need a
202    /// real transaction.
203    #[derive(Default)]
204    struct CapturingWriter {
205        captured: Mutex<Vec<Event<TestPayload>>>,
206        error: Option<OutboxError>,
207    }
208
209    impl CapturingWriter {
210        fn ok() -> Self {
211            Self::default()
212        }
213
214        fn failing(error: OutboxError) -> Self {
215            Self {
216                captured: Mutex::new(Vec::new()),
217                error: Some(error),
218            }
219        }
220
221        fn events(&self) -> Vec<Event<TestPayload>> {
222            self.captured.lock().unwrap().clone()
223        }
224    }
225
226    #[async_trait]
227    impl OutboxWriter<TestPayload> for CapturingWriter {
228        type Executor<'a> = ();
229
230        async fn insert_event<'a>(
231            &self,
232            event: Event<TestPayload>,
233            _: (),
234        ) -> Result<(), OutboxError>
235        where
236            'a: 'async_trait,
237        {
238            if let Some(err) = &self.error {
239                return Err(err.clone());
240            }
241            self.captured.lock().unwrap().push(event);
242            Ok(())
243        }
244    }
245
246    #[rstest]
247    #[tokio::test]
248    async fn none_strategy_without_idempotency_storage_inserts_event_without_token() {
249        let writer = Arc::new(CapturingWriter::ok());
250        let service = OutboxService::new(writer.clone(), config_with(IdempotencyStrategy::None));
251        let result = service.add_event("t", payload(), None, ()).await;
252        assert!(result.is_ok());
253
254        let events = writer.events();
255        assert_eq!(events.len(), 1);
256        assert!(events[0].idempotency_token.is_none());
257        assert_eq!(events[0].event_type.as_str(), "t");
258    }
259
260    #[rstest]
261    #[tokio::test]
262    async fn uuid_strategy_without_idempotency_storage_inserts_event_with_generated_token() {
263        let writer = Arc::new(CapturingWriter::ok());
264        let service = OutboxService::new(writer.clone(), config_with(IdempotencyStrategy::Uuid));
265        let result = service.add_event("t", payload(), None, ()).await;
266        assert!(result.is_ok());
267
268        let events = writer.events();
269        assert_eq!(events.len(), 1);
270        let token = events[0]
271            .idempotency_token
272            .as_ref()
273            .expect("Uuid strategy must produce a token");
274        assert!(!token.as_str().is_empty());
275    }
276
277    #[rstest]
278    #[tokio::test]
279    async fn uuid_strategy_with_storage_reserves_same_token_as_inserted() {
280        let writer = Arc::new(CapturingWriter::ok());
281        let mut idem = MockIdempotencyStorageProvider::new();
282
283        // Capture the reserved token so we can compare it to what reached the writer.
284        let reserved: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
285        let reserved_r = reserved.clone();
286
287        idem.expect_try_reserve().times(1).returning(move |tok| {
288            *reserved_r.lock().unwrap() = Some(tok.as_str().to_owned());
289            Ok(true)
290        });
291
292        let service = OutboxService::with_idempotency(
293            writer.clone(),
294            config_with(IdempotencyStrategy::Uuid),
295            Arc::new(idem),
296        );
297        let result = service.add_event("t", payload(), None, ()).await;
298        assert!(result.is_ok());
299
300        let events = writer.events();
301        assert_eq!(events.len(), 1);
302        let inserted = events[0]
303            .idempotency_token
304            .as_ref()
305            .map(|t| t.as_str().to_owned());
306        let captured = reserved.lock().unwrap().clone();
307        assert_eq!(inserted, captured);
308    }
309
310    #[rstest]
311    #[tokio::test]
312    async fn provided_some_passes_user_token_to_reserve_and_insert() {
313        let writer = Arc::new(CapturingWriter::ok());
314        let mut idem = MockIdempotencyStorageProvider::new();
315
316        idem.expect_try_reserve()
317            .withf(|t| t.as_str() == "user-tok")
318            .times(1)
319            .returning(|_| Ok(true));
320
321        let service = OutboxService::with_idempotency(
322            writer.clone(),
323            config_with(IdempotencyStrategy::Provided),
324            Arc::new(idem),
325        );
326        let result = service
327            .add_event("t", payload(), Some("user-tok".to_string()), ())
328            .await;
329        assert!(result.is_ok());
330
331        let events = writer.events();
332        assert_eq!(events.len(), 1);
333        assert_eq!(
334            events[0].idempotency_token.as_ref().map(|t| t.as_str()),
335            Some("user-tok")
336        );
337    }
338
339    #[rstest]
340    #[tokio::test]
341    async fn provided_none_skips_reserve_and_inserts_without_token() {
342        let writer = Arc::new(CapturingWriter::ok());
343        let mut idem = MockIdempotencyStorageProvider::new();
344        idem.expect_try_reserve().times(0);
345
346        let service = OutboxService::with_idempotency(
347            writer.clone(),
348            config_with(IdempotencyStrategy::Provided),
349            Arc::new(idem),
350        );
351        let result = service.add_event("t", payload(), None, ()).await;
352        assert!(result.is_ok());
353
354        let events = writer.events();
355        assert_eq!(events.len(), 1);
356        assert!(events[0].idempotency_token.is_none());
357    }
358
359    #[rstest]
360    #[tokio::test]
361    async fn custom_strategy_uses_extractor_closure_for_token() {
362        fn derive(event: &Event<TestPayload>) -> String {
363            format!("derived:{}", event.payload.as_value().kind)
364        }
365
366        let writer = Arc::new(CapturingWriter::ok());
367        let mut idem = MockIdempotencyStorageProvider::new();
368
369        idem.expect_try_reserve()
370            .withf(|t| t.as_str() == "derived:k")
371            .times(1)
372            .returning(|_| Ok(true));
373
374        let service = OutboxService::with_idempotency(
375            writer.clone(),
376            config_with(IdempotencyStrategy::Custom(Arc::new(derive))),
377            Arc::new(idem),
378        );
379        let result = service.add_event("t", payload(), None, ()).await;
380        assert!(result.is_ok());
381
382        let events = writer.events();
383        assert_eq!(events.len(), 1);
384        assert_eq!(
385            events[0].idempotency_token.as_ref().map(|t| t.as_str()),
386            Some("derived:k")
387        );
388    }
389
390    #[rstest]
391    #[tokio::test]
392    async fn duplicate_when_reserve_returns_false_and_insert_is_not_called() {
393        let writer = Arc::new(CapturingWriter::ok());
394        let mut idem = MockIdempotencyStorageProvider::new();
395
396        idem.expect_try_reserve().times(1).returning(|_| Ok(false));
397
398        let service = OutboxService::with_idempotency(
399            writer.clone(),
400            config_with(IdempotencyStrategy::Provided),
401            Arc::new(idem),
402        );
403        let result = service
404            .add_event("t", payload(), Some("dup".into()), ())
405            .await;
406        assert!(matches!(result, Err(OutboxError::DuplicateEvent)));
407        assert!(writer.events().is_empty());
408    }
409
410    #[rstest]
411    #[tokio::test]
412    async fn reserve_error_propagates_and_insert_is_not_called() {
413        let writer = Arc::new(CapturingWriter::ok());
414        let mut idem = MockIdempotencyStorageProvider::new();
415
416        idem.expect_try_reserve()
417            .times(1)
418            .returning(|_| Err(OutboxError::InfrastructureError("redis down".into())));
419
420        let service = OutboxService::with_idempotency(
421            writer.clone(),
422            config_with(IdempotencyStrategy::Uuid),
423            Arc::new(idem),
424        );
425        let result = service.add_event("t", payload(), None, ()).await;
426        assert!(matches!(result, Err(OutboxError::InfrastructureError(_))));
427        assert!(writer.events().is_empty());
428    }
429
430    #[rstest]
431    #[tokio::test]
432    async fn insert_error_propagates_after_successful_reserve() {
433        let writer = Arc::new(CapturingWriter::failing(OutboxError::DatabaseError(
434            "pk conflict".into(),
435        )));
436        let mut idem = MockIdempotencyStorageProvider::new();
437
438        idem.expect_try_reserve().times(1).returning(|_| Ok(true));
439
440        let service = OutboxService::with_idempotency(
441            writer.clone(),
442            config_with(IdempotencyStrategy::Uuid),
443            Arc::new(idem),
444        );
445        let result = service.add_event("t", payload(), None, ()).await;
446        assert!(matches!(result, Err(OutboxError::DatabaseError(_))));
447        assert!(writer.events().is_empty());
448    }
449}
outbox_core/service.rs

outbox_core/
service.rs
