ruststream_sqlx/outbox.rs
1//! The transactional outbox: what a service publishes is recorded in its own table until a
2//! consumer has processed it, and what was not processed is published again at startup.
3//!
4//! A struct deriving [`Outbox`](derive@crate::Outbox), or implementing [`OutboxTable`] by hand,
5//! describes the table; `outbox!`, [`Outbox::register`] or [`Outbox::track`] registers it under the
6//! names it tracks, and the registry hands out the two middlewares and the republish. The events
7//! below are what the record's statements do; a service implements the ones it names as its own
8//! (`#[outbox(custom(..))]`, or [`OutboxSpec::own`]), and always [`Publish`], which has no
9//! default.
10//!
11//! Each topic has its page on the site:
12//!
13//! - [The outbox](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/): what it tracks and the at-least-once guarantee.
14//! - [Record](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/record/): the outbox table, derived or described by hand.
15//! - [Registry](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/registry/): `outbox!`, [`Outbox::track`] and
16//! [`Outbox::register`].
17//! - [Tracking](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/tracking/): [`Outbox::layer`] and [`Outbox::publish_layer`].
18//! - [republish](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/republish/): [`Outbox::republish`] at startup.
19//! - [wrap](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/wrap/): [`Outbox::wrap`], a publisher outside the pipeline.
20//! - [set_pool](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/set-pool/): [`Outbox::set_pool`] inside `#[ruststream::app]`.
21//! - [Statements](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/statements/): the record's events in the service's own SQL.
22//! - [Interplay](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/interplay/): the outbox beside the inbox.
23//! - [Switch](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/switch/): `RUSTSTREAM_SQLX_OUTBOX=on` in test builds.
24//! - [OutboxError](https://powersemmi.github.io/ruststream-sqlx/latest/components/outbox/outbox-error/): [`OutboxError`].
25//!
26//! # Examples
27//!
28//! A service whose replies to `orders` are tracked over the in-memory broker. Inside
29//! `#[ruststream::app]` the registry starts without a pool, and `on_startup` hands it one:
30//!
31//! ```
32//! # #[cfg(all(feature = "outbox", feature = "postgres"))]
33//! # mod demo {
34//! use std::convert::Infallible;
35//! use std::io;
36//!
37//! use chrono::{DateTime, Utc};
38//! use ruststream::OutgoingMessage;
39//! use ruststream::memory::prelude::*;
40//! use ruststream_sqlx::{Outbox, outbox};
41//! use serde::{Deserialize, Serialize};
42//! use sqlx::postgres::{PgConnection, PgPool, Postgres};
43//!
44//! // outbox: id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, payload BYTEA NOT NULL,
45//! // processed_at TIMESTAMPTZ
46//! #[derive(Outbox, sqlx::FromRow)]
47//! #[outbox(table = "outbox")]
48//! struct OrderOutbox {
49//! #[field(id)]
50//! id: i64,
51//! #[field(name)]
52//! name: String,
53//! #[field(payload)]
54//! payload: Vec<u8>,
55//! #[field(processed_at)]
56//! processed_at: Option<DateTime<Utc>>,
57//! }
58//!
59//! impl outbox::Publish<Postgres> for OrderOutbox {
60//! async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
61//! sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
62//! .bind(msg.name())
63//! .bind(msg.payload())
64//! .fetch_one(conn)
65//! .await
66//! }
67//! }
68//!
69//! #[derive(Deserialize)]
70//! struct PlaceOrder {
71//! id: u64,
72//! }
73//!
74//! #[derive(Serialize, Deserialize, Outgoing)]
75//! #[outgoing(name = "orders")]
76//! struct OrderPlaced {
77//! id: u64,
78//! }
79//!
80//! #[subscriber("checkout", reply)]
81//! async fn place(cmd: &PlaceOrder) -> OrderPlaced {
82//! OrderPlaced { id: cmd.id }
83//! }
84//!
85//! #[subscriber("orders")]
86//! async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
87//! tracing::info!(order = order.id, "fulfilling the order");
88//! HandlerOutcome::ack()
89//! }
90//!
91//! /// What `#[ruststream::app]` runs.
92//! pub fn app() -> impl App {
93//! let tracking = outbox! {
94//! "orders" => OrderOutbox,
95//! };
96//! let registry = tracking.clone();
97//!
98//! RustStream::new(AppInfo::new("orders", "0.1.0"))
99//! .on_startup(async move |()| {
100//! let pool = PgPool::connect("postgres://localhost/orders")
101//! .await
102//! .map_err(io::Error::other)?;
103//! registry.set_pool(pool.clone()).map_err(io::Error::other)?;
104//! Ok::<_, io::Error>(pool)
105//! })
106//! .layer(tracking.layer())
107//! .publish_layer(tracking.publish_layer())
108//! .after_shutdown(async move |pool| {
109//! pool.close().await;
110//! Ok::<_, Infallible>(())
111//! })
112//! .with_broker(MemoryBroker::new(), |b| {
113//! b.include(place).out_reply(Publish);
114//! b.include(fulfil);
115//! b.after_startup(Publish, tracking.republish());
116//! })
117//! }
118//! # }
119//! # fn main() {}
120//! ```
121
122mod database;
123mod dispatch;
124mod error;
125mod events;
126mod layer;
127mod publish;
128mod registry;
129mod republish;
130mod row;
131pub mod spec;
132mod store;
133mod switch;
134mod wrap;
135
136pub use database::OutboxDatabase;
137#[doc(hidden)]
138pub use database::{Defaults, Statements};
139#[doc(hidden)]
140pub use dispatch::{AckBy, DiscardBy, FetchBy, HeadersOf, RecoverBy, RetryBy, Slot};
141pub use error::{OutboxError, PoolAlreadySet, TrackedPublishError};
142pub use events::{Ack, Discard, Fetch, Publish, Recover, Retry, Tracked};
143pub use layer::TrackingLayer;
144pub use publish::TrackingPublishLayer;
145pub use registry::{Checked, Nil, Outbox, Registered, TrackedName};
146#[doc(hidden)]
147pub use registry::{Lacks, RecordList, RecordNames};
148pub use republish::Republishing;
149pub use row::OutboxRow;
150pub use spec::{OutboxSpec, OutboxTable};
151#[doc(hidden)]
152pub use store::Store;
153pub use wrap::TrackedPublisher;
154
155/// The header a tracked message carries its record's id in: written with the id's `Display`,
156/// read back with its `FromStr`.
157///
158/// # Examples
159///
160/// ```
161/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
162/// # mod demo {
163/// # use ruststream::OutgoingMessage;
164/// # use ruststream::memory::prelude::*;
165/// # use ruststream_sqlx::{Outbox, outbox};
166/// # use serde::{Deserialize, Serialize};
167/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
168/// # #[derive(Outbox, sqlx::FromRow)]
169/// # #[outbox(table = "outbox")]
170/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
171/// # impl outbox::Publish<Postgres> for OrderOutbox {
172/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
173/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
174/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
175/// # }
176/// # }
177/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
178/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
179/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
180/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
181/// use ruststream::runtime::{BlanketLayer, Context, Handler};
182/// use ruststream_sqlx::outbox::OUTBOX_ID_HEADER;
183///
184/// /// A layer of the service's own that logs the record id of every tracked delivery.
185/// #[derive(Clone, Copy)]
186/// pub struct RecordIds;
187///
188/// pub struct Logged<H>(H);
189///
190/// impl BlanketLayer for RecordIds {
191/// fn apply<M, C, S, H>(&self, handler: H) -> impl Handler<M, C, S> + 'static
192/// where
193/// M: Send + Sync + 'static,
194/// C: Send + 'static,
195/// S: Send + Sync + 'static,
196/// H: Handler<M, C, S> + 'static,
197/// {
198/// Logged(handler)
199/// }
200/// }
201///
202/// impl<M: Sync, C: Send, S: Send + Sync, H: Handler<M, C, S>> Handler<M, C, S> for Logged<H> {
203/// async fn handle(&self, msg: &M, ctx: &mut Context<'_, C, S>) -> HandlerOutcome {
204/// if let Some(id) = ctx.headers().get(OUTBOX_ID_HEADER) {
205/// let id = String::from_utf8_lossy(id);
206/// tracing::info!(subscription = ctx.name(), %id, "a tracked delivery");
207/// }
208/// self.0.handle(msg, ctx).await
209/// }
210/// }
211///
212/// pub fn app(pool: PgPool) -> impl App {
213/// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
214/// RustStream::new(AppInfo::new("orders", "0.1.0"))
215/// .layer(RecordIds)
216/// .layer(tracking.layer())
217/// .with_broker(MemoryBroker::new(), |b| {
218/// b.include(fulfil);
219/// })
220/// }
221/// # }
222/// # fn main() {}
223/// ```
224pub const OUTBOX_ID_HEADER: &str = "x-ruststream-outbox-id";