reliar_store_postgres/outbox/outbox_store_enqueue.rs
1//! [`PostgresOutboxStore`]'s [`OutboxEnqueue`] implementation.
2
3use bytes::Bytes;
4use reliar_core::{ContentType, Message, MessageId, Serializer};
5use reliar_outbox::OutboxEnqueue;
6use sqlx::{Postgres, Transaction};
7use tracing::Instrument as _;
8
9use super::enqueue::{InsertOutboxRowParams, insert_outbox_row};
10use super::error::{EnqueueError, map_enqueue_error};
11
12use super::PostgresOutboxStore;
13
14/// Generic over the envelope body `T` — `T: Message` is never needed here, only
15/// `envelope.message_type` (the promoted `message_type`/`message_version` columns), which every
16/// `Envelope<T>` carries regardless of `T`. Builds [`InsertOutboxRowParams`] (encoding policy —
17/// the `MetadataRest`/`headers` JSON shape and the "empty remainder is SQL NULL" decision are the
18/// store layer's, not the concern function's) and calls the one-statement concern function
19/// directly on the caller's transaction.
20async fn insert_row<T>(
21 tx: &mut Transaction<'_, Postgres>,
22 envelope: &reliar_core::Envelope<T>,
23 payload: &Bytes,
24 content_type: &ContentType,
25) -> Result<(), sqlx::Error> {
26 let corr = &envelope.metadata.correlation;
27 let sent_at_ms = envelope
28 .metadata
29 .delivery
30 .sent_at
31 .map(crate::records::encode_epoch_millis);
32 let rest = crate::records::MetadataRest {
33 trace: crate::records::TraceRest {
34 traceparent: envelope.metadata.trace.traceparent.clone(),
35 tracestate: envelope.metadata.trace.tracestate.clone(),
36 },
37 routing: crate::records::RoutingRest {
38 source: envelope
39 .metadata
40 .routing
41 .source
42 .as_ref()
43 .map(|v| v.as_str().to_owned()),
44 destination: envelope
45 .metadata
46 .routing
47 .destination
48 .as_ref()
49 .map(|v| v.as_str().to_owned()),
50 reply_to: envelope
51 .metadata
52 .routing
53 .reply_to
54 .as_ref()
55 .map(|v| v.as_str().to_owned()),
56 },
57 delivery: crate::records::DeliveryRest {
58 sent_at_ms,
59 deduplication_id: envelope.metadata.delivery.deduplication_id.clone(),
60 },
61 };
62 // An empty remainder is written as SQL NULL, not '{}', so pending rows stay small.
63 let metadata_json = if rest.trace.traceparent.is_none()
64 && rest.trace.tracestate.is_none()
65 && rest.routing.source.is_none()
66 && rest.routing.destination.is_none()
67 && rest.routing.reply_to.is_none()
68 && rest.delivery.sent_at_ms.is_none()
69 && rest.delivery.deduplication_id.is_none()
70 {
71 None
72 } else {
73 // `MetadataRest`'s fields are now all plain owned `String`/`i64`/`Option` values (no
74 // RFC3339 formatting) — `serde_json::to_value` is total over this
75 // shape. The fallback is unreachable in practice; kept non-panicking rather than
76 // `.expect()`'d away, since a panic on the enqueue path is never acceptable. `.ok()` rather than a
77 // `Value::Null` fallback: on the unreachable error branch this writes SQL `NULL` — the
78 // same "no remainder" shape as the empty-check above — rather than a JSON `null` a reader
79 // would then have to treat as yet another poison case.
80 serde_json::to_value(&rest).ok()
81 };
82
83 let headers_json = envelope.headers().filter(|h| !h.is_empty()).map(|h| {
84 let map: serde_json::Map<String, serde_json::Value> = h
85 .iter()
86 .map(|(k, v)| (k.to_owned(), serde_json::Value::String(v.to_owned())))
87 .collect();
88
89 serde_json::Value::Object(map)
90 });
91
92 insert_outbox_row(
93 &mut *tx,
94 InsertOutboxRowParams {
95 message_id: envelope.id.as_uuid(),
96 message_type: envelope.message_type.name(),
97 message_version: i32::from(envelope.message_type.version()),
98 correlation_id: corr
99 .correlation_id
100 .as_ref()
101 .map(reliar_core::CorrelationId::as_str),
102 conversation_id: corr.conversation_id.as_uuid(),
103 causation_id: corr.causation_id.map(|id| id.as_uuid()),
104 request_id: corr.request_id.map(|id| id.as_uuid()),
105 content_type: content_type.as_str(),
106 payload: &payload[..],
107 tenant_id: envelope.metadata.tenant_id.as_deref(),
108 expires_at: envelope.metadata.delivery.expires_at,
109 // No writer sets `ordering_key` in this release — `Ordering::PerKey` is a
110 // configuration error — but the column and its bind stay so a later `PerKey` writer
111 // needs no migration.
112 ordering_key: None,
113 metadata: metadata_json,
114 headers: headers_json,
115 },
116 )
117 .await
118}
119
120/// [`PostgresOutboxStore`]'s [`OutboxEnqueue`] implementation: reuses `insert_row`. Implements
121/// only [`OutboxEnqueue::enqueue_envelope`] — the provided `enqueue` (bare `T: Message`) calls
122/// back into it, so the `reliar.outbox.enqueue` span fires exactly once per row for either
123/// spelling (ADR 0037 amendment A).
124///
125/// **`Ser: 'static`** — needed because the method reaches `self.serializer` (held as `Arc<Ser>`)
126/// across the `.await` in `insert_row`; without it the future fails to type-check (a
127/// borrowed type must outlive the generic parameters it references).
128/// Every concrete serializer (`JsonSerializer` or any owned one) is `'static`, so no real host is
129/// excluded.
130///
131/// **A single lifetime, `'c`, quantified by the impl.** With `&mut Tx` in the trait's own method
132/// signature the reborrow lifetime is quantified by the method itself, so the higher-ranked
133/// "implementation is not general enough" trap an earlier `OutboxEnqueueIn<&'a mut
134/// Transaction<'c, _>>` shape had — where an explicit, implied-looking `where 'c: 'a` bound broke
135/// every `tokio::spawn`/Axum call site — cannot arise here; there is no second lifetime to
136/// accidentally bound. That regression guard lives as
137/// `outbox_enqueue::enqueue_is_send_through_tokio_spawn` in this crate's Postgres suite.
138///
139/// Renamed from `OutboxStaging`/`stage` in 0.4.0; the store's own typed `enqueue`/`enqueue_with`
140/// were folded into this impl since no inherent method may share a name with a trait method — a
141/// caller now needs `use reliar_outbox::OutboxEnqueue;` in scope to call
142/// `store.enqueue(..)`/`store.enqueue_envelope(..)`. The trait's serialized twin,
143/// `enqueue_serialized`, was cut before shipping.
144impl<'c, Ser> OutboxEnqueue<Transaction<'c, Postgres>> for PostgresOutboxStore<Ser>
145where
146 Ser: Serializer + Send + Sync + 'static,
147{
148 type Error = EnqueueError<Ser::Error>;
149
150 /// Serializes `envelope.body` with this store's configured `Serializer` and writes the
151 /// serializer's own `content_type`. Plain `INSERT`, **no `ON CONFLICT`**: a reused
152 /// `MessageId` aborts the caller's transaction rather than silently losing a message.
153 ///
154 /// # Errors
155 ///
156 /// [`EnqueueError::Serialize`] if the configured `Serializer` rejects the body,
157 /// [`EnqueueError::Duplicate`] for a reused [`MessageId`] (`ix_outbox_message_id` violation,
158 /// ADR 0044 §1), or [`EnqueueError::Database`] for any other `sqlx` failure.
159 ///
160 /// [`EnqueueError::Duplicate`]/[`EnqueueError::Database`] leave `tx` aborted: the failed
161 /// `INSERT` puts the PostgreSQL transaction in the aborted state, so PostgreSQL rejects every
162 /// subsequent statement on it, and every earlier write in that transaction is rolled back at
163 /// commit. [`EnqueueError::Serialize`] is returned before any statement runs, so `tx` is
164 /// untouched and stays usable.
165 // Block form, not `async fn` (conventions §3(b)): the trait bounds neither `T: Send` nor
166 // `Tx: Send`, and an `async fn`'s parameters live in its own generator's *unstarted* state
167 // regardless of when the body consumes them, so it would need both. Serializing and dropping
168 // the typed body (`map_body(|_| ())`) happen synchronously, before any future is constructed,
169 // so the `async` block below is built only after `T` is gone — its captured state (`tx`,
170 // `envelope: Envelope<()>`, `payload`) is `Send` independently of `T`.
171 fn enqueue_envelope<T: Message + Sync>(
172 &self,
173 tx: &mut Transaction<'c, Postgres>,
174 typed_envelope: reliar_core::Envelope<T>,
175 ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send {
176 let span = tracing::debug_span!(
177 "reliar.outbox.enqueue",
178 message.id = %typed_envelope.id,
179 message.type = %typed_envelope.message_type,
180 );
181 // Entered, not `.instrument()`ed: serialization is synchronous and must run before the
182 // async block below exists, but a serializer's own `tracing` events still belong inside
183 // this span. The guard is dropped before the block is built.
184 let payload = {
185 let _guard = span.enter();
186
187 self.serializer
188 .serialize(&typed_envelope.body)
189 .map_err(|source| EnqueueError::Serialize { source })
190 };
191 // `insert_row` never reads `envelope.body` anyway.
192 let envelope = typed_envelope.map_body(|_| ());
193
194 async move {
195 let payload = payload?;
196
197 insert_row(tx, &envelope, &payload, self.content_type())
198 .await
199 .map_err(|source| map_enqueue_error(envelope.id, source))?;
200
201 Ok(envelope.id)
202 }
203 .instrument(span)
204 }
205}