reliar_outbox/enqueue.rs
1//! The enqueue capability an application calls directly, in its own transaction (ADR 0036
2//! amendment B): there is no facade type between the caller and the store.
3
4use reliar_core::{Classify, Envelope, Message, MessageId};
5
6/// Enqueuing a typed message in the caller's own transaction. A provider (`PostgresOutboxStore`)
7/// implements this directly, alongside [`crate::OutboxStore`]; there is no separate handle type
8/// to construct.
9///
10/// `Tx` is the provider's transaction type: `sqlx::Transaction<'_, Postgres>` for
11/// `reliar-store-postgres`. It is a **type parameter** precisely so this crate names no storage
12/// type, and one implementor may support several.
13///
14/// Deliberately **not** a method on [`crate::OutboxStore`]: enqueuing takes a transaction handle
15/// the claim side never sees, `OutboxStore` is already published, and a GAT `type Tx<'a>` would
16/// have to spell `&'a mut Transaction<'c, _>` and reintroduce an invariance problem.
17///
18/// **Two methods, one call each** (ADR 0037 amendment A): a required
19/// [`Self::enqueue_envelope`] that every implementor writes, and a provided [`Self::enqueue`] —
20/// the fire-and-forget spelling — built on top of it. An earlier shape tried an `impl
21/// Into<Envelope<T>>` parameter so one method covered both a bare `T: Message` and an
22/// already-built `Envelope<T>`; that was rejected in favor of a second, explicit method instead
23/// — no inference trick, no `Envelope<T>: !Message` invariant to guard.
24///
25/// **One serialized twin, deliberately absent** (ADR 0036 amendment B.10): a
26/// `enqueue_serialized(&mut tx, &SerializedEnvelope)` existed briefly and was cut for having no
27/// production caller; re-adding it is additive.
28///
29/// Renamed from `OutboxStaging` in 0.4.0; `stage` became `enqueue`, and the facade
30/// `OutboxPublisher` that briefly wrapped it (0.4.0, never released) was withdrawn before
31/// shipping in favor of calling this trait directly.
32///
33/// A provider implements the trait, then a caller enqueues in its own transaction — shown here
34/// against the `test-support` in-memory fake:
35#[cfg_attr(not(feature = "test-support"), doc = "```ignore")]
36#[cfg_attr(feature = "test-support", doc = "```")]
37/// # use reliar_core::Message;
38/// # use reliar_outbox::{InMemoryOutboxStore, InMemoryTransaction, OutboxEnqueue};
39/// # #[derive(serde::Serialize, serde::Deserialize)]
40/// # struct OrderCreated;
41/// # impl Message for OrderCreated {
42/// # const TYPE: &'static str = "orders.created";
43/// # const VERSION: u16 = 1;
44/// # }
45/// # #[tokio::main(flavor = "current_thread")]
46/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
47/// let store = InMemoryOutboxStore::default();
48/// let mut tx = InMemoryTransaction;
49/// let id = store.enqueue(&mut tx, OrderCreated).await?;
50/// assert!(store.record(id).is_some());
51/// # Ok(())
52/// # }
53/// ```
54pub trait OutboxEnqueue<Tx>: Send + Sync {
55 /// What enqueuing fails with.
56 type Error: std::error::Error + Send + Sync + 'static + Classify;
57
58 /// Serializes `envelope`'s body with **this implementor's own** configured `Serializer` and
59 /// enqueues it in `tx`. Writes the serializer's own `content_type`.
60 ///
61 /// The **propagating** spelling: use it when an id must carry over from an inbound request
62 /// — conversation, correlation, causation, tenant, trace, headers — via
63 /// [`Envelope::builder`](reliar_core::Envelope::builder), shown here against the
64 /// `test-support` in-memory fake:
65 ///
66 #[cfg_attr(not(feature = "test-support"), doc = "```ignore")]
67 #[cfg_attr(feature = "test-support", doc = "```")]
68 /// # use reliar_core::{ConversationId, Envelope, Message};
69 /// # use reliar_outbox::{InMemoryOutboxStore, InMemoryTransaction, OutboxEnqueue};
70 /// # use uuid::Uuid;
71 /// # #[derive(serde::Serialize, serde::Deserialize)]
72 /// # struct OrderCreated;
73 /// # impl Message for OrderCreated {
74 /// # const TYPE: &'static str = "orders.created";
75 /// # const VERSION: u16 = 1;
76 /// # }
77 /// # #[tokio::main(flavor = "current_thread")]
78 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
79 /// # let store = InMemoryOutboxStore::default();
80 /// # let mut tx = InMemoryTransaction;
81 /// # let evt = OrderCreated;
82 /// # let cx = ConversationId::from_uuid(Uuid::now_v7());
83 /// store.enqueue_envelope(&mut tx, Envelope::builder(evt).conversation(cx).build()).await?;
84 /// # Ok(())
85 /// # }
86 /// ```
87 ///
88 /// Returns the id written, for the caller's own use — e.g. as a *next* message's
89 /// `causation_id` in the same transaction. The envelope already carries its `id`; this is not
90 /// how a caller learns it, only a convenience.
91 ///
92 /// **Implementors:** the trait bounds neither `T` nor `Tx` on `Send`, so a plain `async fn`
93 /// that carries `envelope: Envelope<T>` (or `tx`) across an `.await` will not satisfy this
94 /// method's `+ Send` return bound. Serialize (or otherwise consume) `T` synchronously, before
95 /// the async block is built — see `PostgresOutboxStore::enqueue_envelope`'s `//` comment in
96 /// `reliar-store-postgres` for the reference shape and the full argument.
97 ///
98 /// The implementation SHALL issue no network I/O other than the statement itself, and SHALL
99 /// NOT commit, roll back or otherwise consume `tx` — the caller owns it.
100 ///
101 /// # Errors
102 ///
103 /// Provider-defined. An `Err` **MAY** leave `tx` unusable, and whether it does is the
104 /// provider's contract — every implementor documents which. The portable rule a caller can
105 /// rely on is therefore: treat any enqueue error as *abort this transaction* — issue no
106 /// further statement on `tx`, roll it back, and consider every earlier write in it lost. With
107 /// `reliar-store-postgres` the transaction **is** aborted: PostgreSQL rejects every
108 /// subsequent statement on it, so no earlier write in that transaction can still be committed.
109 fn enqueue_envelope<T: Message + Sync>(
110 &self,
111 tx: &mut Tx,
112 envelope: Envelope<T>,
113 ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send;
114
115 /// The **fire-and-forget** spelling: builds `body` into an envelope with default metadata
116 /// and a freshly rooted conversation — exactly `Envelope::builder(body).build()` — then
117 /// enqueues it via [`Self::enqueue_envelope`], shown here against the `test-support`
118 /// in-memory fake:
119 ///
120 #[cfg_attr(not(feature = "test-support"), doc = "```ignore")]
121 #[cfg_attr(feature = "test-support", doc = "```")]
122 /// # use reliar_core::Message;
123 /// # use reliar_outbox::{InMemoryOutboxStore, InMemoryTransaction, OutboxEnqueue};
124 /// # #[derive(serde::Serialize, serde::Deserialize)]
125 /// # struct OrderCreated {
126 /// # order_id: u64,
127 /// # }
128 /// # impl Message for OrderCreated {
129 /// # const TYPE: &'static str = "orders.created";
130 /// # const VERSION: u16 = 1;
131 /// # }
132 /// # #[tokio::main(flavor = "current_thread")]
133 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
134 /// # let store = InMemoryOutboxStore::default();
135 /// # let mut tx = InMemoryTransaction;
136 /// # let order_id = 42u64;
137 /// store.enqueue(&mut tx, OrderCreated { order_id }).await?;
138 /// # Ok(())
139 /// # }
140 /// ```
141 ///
142 /// Use [`Self::enqueue_envelope`] instead when an id must propagate from an inbound request;
143 /// this spelling never has anything to propagate from.
144 ///
145 /// Implementors **SHALL NOT** override this method — it is a fixed, provided spelling of
146 /// [`Self::enqueue_envelope`], not an extension point. `Envelope::builder(body).build()` mints
147 /// the envelope and its id eagerly, at call time, not on the returned future's first poll.
148 ///
149 /// # Errors
150 ///
151 /// Same as [`Self::enqueue_envelope`].
152 // Direct delegation, not `async fn` and no `async move` block either — returns
153 // `enqueue_envelope`'s own future unchanged. See `PostgresOutboxStore::enqueue_envelope`'s
154 // comment in `reliar-store-postgres/src/store.rs` for the full `Send`/capture argument.
155 fn enqueue<T: Message + Sync>(
156 &self,
157 tx: &mut Tx,
158 body: T,
159 ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send {
160 self.enqueue_envelope(tx, Envelope::builder(body).build())
161 }
162}