ruststream_sqlx/outbox/
database.rs

1//! The databases the outbox's default statements run on: each builds a record's statements in the
2//! built-in dialects its connections speak, and a connection picks its own.
3
4#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
5use std::collections::BTreeSet;
6use std::future::Future;
7#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
8use std::sync::{Mutex, PoisonError};
9
10#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
11use ruststream_sqlx_dialect::{self as dialect, OutboxDialect};
12use ruststream_sqlx_dialect::{StatementError, TableSpec};
13
14#[cfg(feature = "any")]
15use sqlx::Any;
16#[cfg(feature = "any")]
17use sqlx::AnyConnection;
18#[cfg(any(
19    feature = "postgres",
20    feature = "mysql",
21    feature = "sqlite",
22    feature = "any"
23))]
24use sqlx::Arguments as _;
25use sqlx::{Database, Error, FromRow};
26#[cfg(feature = "mysql")]
27use sqlx::{MySql, MySqlConnection};
28#[cfg(feature = "postgres")]
29use sqlx::{PgConnection, Postgres};
30#[cfg(feature = "sqlite")]
31use sqlx::{Sqlite, SqliteConnection};
32
33/// The texts of the default statements of one record on one dialect, interned for the life of the
34/// process. Machinery.
35#[doc(hidden)]
36#[derive(Debug, Clone, Copy)]
37pub struct Statements {
38    /// The fetch of a record by its id.
39    pub(super) fetch: &'static str,
40    /// The mark of a processed record, by its id: `Ack` and `Discard`.
41    pub(super) mark: &'static str,
42    /// The recovery of the unprocessed records of a name.
43    pub(super) recover: &'static str,
44}
45
46#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
47impl Statements {
48    /// The statements `dialect` writes for `spec`.
49    fn of(dialect: &impl OutboxDialect, spec: &TableSpec<'_>) -> Result<Self, StatementError> {
50        Ok(Self {
51            fetch: intern(dialect.outbox_fetch(spec)?.sql()),
52            mark: intern(dialect.outbox_mark(spec)?.sql()),
53            recover: intern(dialect.outbox_recover(spec)?.sql()),
54        })
55    }
56}
57
58/// The default statements of one record, built when the record type is registered and kept in its
59/// registry node: one set per built-in dialect the record's database runs, `None` for the others.
60/// Machinery.
61#[doc(hidden)]
62#[derive(Debug, Clone, Copy, Default)]
63pub struct Defaults {
64    #[cfg(feature = "postgres")]
65    postgres: Option<Statements>,
66    #[cfg(feature = "mysql")]
67    mysql: Option<Statements>,
68    #[cfg(feature = "sqlite")]
69    sqlite: Option<Statements>,
70}
71
72/// `text` for the life of the process, written once whatever the number of registries that build
73/// it.
74// Why leaked: sqlx 0.9 runs a `&'static str` as written and copies any shorter-lived text into a
75// new `Arc<str>` on every run, so the per-message path needs `'static` statements. A registry
76// builds them once, at construction; the set keeps each distinct text once, so building a
77// registry again (a test suite, a service rebuilt in one process) leaks nothing new.
78#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
79fn intern(text: &str) -> &'static str {
80    static TEXTS: Mutex<BTreeSet<&'static str>> = Mutex::new(BTreeSet::new());
81    let mut texts = TEXTS.lock().unwrap_or_else(PoisonError::into_inner);
82    if let Some(interned) = texts.get(text) {
83        return interned;
84    }
85    let interned: &'static str = Box::leak(text.into());
86    texts.insert(interned);
87    interned
88}
89
90/// The name `AnyConnection::backend_name` gives each built-in database, as sqlx names it.
91#[cfg(all(feature = "any", feature = "postgres"))]
92const POSTGRES_BACKEND: &str = "PostgreSQL";
93#[cfg(all(feature = "any", feature = "mysql"))]
94const MYSQL_BACKEND: &str = "MySQL";
95#[cfg(all(feature = "any", feature = "sqlite"))]
96const SQLITE_BACKEND: &str = "SQLite";
97
98#[cfg(feature = "any")]
99impl Defaults {
100    /// The statements of the database behind the `AnyPool` backend named `backend`.
101    fn for_backend(&self, backend: &str) -> Option<&Statements> {
102        match backend {
103            #[cfg(feature = "postgres")]
104            POSTGRES_BACKEND => self.postgres.as_ref(),
105            #[cfg(feature = "mysql")]
106            MYSQL_BACKEND => self.mysql.as_ref(),
107            #[cfg(feature = "sqlite")]
108            SQLITE_BACKEND => self.sqlite.as_ref(),
109            _ => None,
110        }
111    }
112}
113
114/// A sqlx database the default events of an outbox record run on.
115///
116/// The registry builds the statements from the record's
117/// [`OutboxTable::TABLE`](super::OutboxTable::TABLE) in each built-in dialect the database's
118/// connections speak, once, when the record type is registered, and a connection picks its own.
119///
120/// [`Postgres`](sqlx::Postgres), [`MySql`](sqlx::MySql) (MySQL and MariaDB) and
121/// [`Sqlite`](sqlx::Sqlite) implement it under their features, and [`Any`](sqlx::Any) under `any`,
122/// where the database the connection reached picks the text. A record's default event on a
123/// backend whose dialect the crate was built without fails with [`sqlx::Error::Configuration`]
124/// naming the record and the event.
125///
126/// # Examples
127///
128/// ```
129/// # #[cfg(feature = "postgres")]
130/// # mod demo {
131/// use ruststream::OutgoingMessage;
132/// use ruststream_sqlx::{Outbox, OutboxDatabase, outbox};
133/// use sqlx::{PgConnection, Postgres};
134///
135/// // outbox: id BIGSERIAL PRIMARY KEY, name TEXT, payload BYTEA; a processed record is deleted
136/// #[derive(Outbox, sqlx::FromRow)]
137/// #[outbox(table = "outbox")]
138/// pub struct OrderEvent {
139///     #[field(id)]
140///     id: i64,
141///     #[field(name)]
142///     name: String,
143///     #[field(payload)]
144///     payload: Vec<u8>,
145/// }
146///
147/// // The record of a published message is the service's own statement.
148/// impl outbox::Publish<Postgres> for OrderEvent {
149///     async fn publish(
150///         conn: &mut PgConnection,
151///         msg: &OutgoingMessage<'_>,
152///     ) -> Result<i64, sqlx::Error> {
153///         sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
154///             .bind(msg.name())
155///             .bind(msg.payload())
156///             .fetch_one(conn)
157///             .await
158///     }
159/// }
160///
161/// // The default events run on any `OutboxDatabase`: here the mark of a processed record.
162/// pub async fn mark<DB: OutboxDatabase>(
163///     conn: &mut DB::Connection,
164///     id: &i64,
165/// ) -> Result<(), sqlx::Error>
166/// where
167///     OrderEvent: outbox::Ack<DB>,
168/// {
169///     <OrderEvent as outbox::Ack<DB>>::ack(conn, id).await
170/// }
171/// # }
172/// # fn main() {}
173/// ```
174pub trait OutboxDatabase: Database {
175    /// The default statements of the table `spec` describes, in each built-in dialect this
176    /// database's connections may speak. Machinery.
177    #[doc(hidden)]
178    fn defaults(spec: &TableSpec<'_>) -> Result<Defaults, StatementError>;
179
180    /// The statements this connection's database runs, or `None` when the crate was built without
181    /// its dialect. Machinery.
182    #[doc(hidden)]
183    fn statements<'d>(conn: &Self::Connection, defaults: &'d Defaults) -> Option<&'d Statements>;
184
185    /// Binds the name a recovery reads the records of. Machinery.
186    #[doc(hidden)]
187    fn bind_name(arguments: &mut Self::Arguments, name: &str) -> Result<(), Error>;
188
189    /// Runs a statement. Machinery.
190    #[doc(hidden)]
191    fn execute<'c>(
192        conn: &'c mut Self::Connection,
193        sql: &'static str,
194        arguments: Self::Arguments,
195    ) -> impl Future<Output = Result<(), Error>> + Send + 'c;
196
197    /// Runs a statement that reads at most one record. Machinery.
198    #[doc(hidden)]
199    fn fetch_optional<'c, Record>(
200        conn: &'c mut Self::Connection,
201        sql: &'static str,
202        arguments: Self::Arguments,
203    ) -> impl Future<Output = Result<Option<Record>, Error>> + Send + 'c
204    where
205        Record: for<'r> FromRow<'r, Self::Row> + Send + Unpin + 'c;
206
207    /// Runs a statement that reads every record it selects. Machinery.
208    #[doc(hidden)]
209    fn fetch_all<'c, Record>(
210        conn: &'c mut Self::Connection,
211        sql: &'static str,
212        arguments: Self::Arguments,
213    ) -> impl Future<Output = Result<Vec<Record>, Error>> + Send + 'c
214    where
215        Record: for<'r> FromRow<'r, Self::Row> + Send + Unpin + 'c;
216}
217
218/// `OutboxDatabase` for one sqlx database: `$build` builds the statements of a table, and the
219/// connection picks the ones `$pick` names.
220#[cfg(any(
221    feature = "postgres",
222    feature = "mysql",
223    feature = "sqlite",
224    feature = "any"
225))]
226macro_rules! outbox_database {
227    (
228        $database:ty, $connection:ty,
229        |$spec:ident| $build:expr,
230        |$conn:ident, $defaults:ident| $pick:expr
231    ) => {
232        impl OutboxDatabase for $database {
233            fn defaults($spec: &TableSpec<'_>) -> Result<Defaults, StatementError> {
234                $build
235            }
236
237            fn statements<'d>(
238                $conn: &$connection,
239                $defaults: &'d Defaults,
240            ) -> Option<&'d Statements> {
241                $pick
242            }
243
244            fn bind_name(arguments: &mut Self::Arguments, name: &str) -> Result<(), Error> {
245                arguments.add(name).map_err(Error::Encode)
246            }
247
248            async fn execute(
249                conn: &mut $connection,
250                sql: &'static str,
251                arguments: Self::Arguments,
252            ) -> Result<(), Error> {
253                sqlx::query_with(sql, arguments).execute(conn).await?;
254                Ok(())
255            }
256
257            async fn fetch_optional<'c, Record>(
258                conn: &'c mut $connection,
259                sql: &'static str,
260                arguments: Self::Arguments,
261            ) -> Result<Option<Record>, Error>
262            where
263                Record: for<'r> FromRow<'r, Self::Row> + Send + Unpin + 'c,
264            {
265                sqlx::query_as_with(sql, arguments)
266                    .fetch_optional(conn)
267                    .await
268            }
269
270            async fn fetch_all<'c, Record>(
271                conn: &'c mut $connection,
272                sql: &'static str,
273                arguments: Self::Arguments,
274            ) -> Result<Vec<Record>, Error>
275            where
276                Record: for<'r> FromRow<'r, Self::Row> + Send + Unpin + 'c,
277            {
278                sqlx::query_as_with(sql, arguments).fetch_all(conn).await
279            }
280        }
281    };
282}
283
284#[cfg(feature = "postgres")]
285outbox_database!(
286    Postgres,
287    PgConnection,
288    |spec| Ok(Defaults {
289        postgres: Some(Statements::of(&dialect::Postgres, spec)?),
290        ..Defaults::default()
291    }),
292    |_conn, defaults| defaults.postgres.as_ref()
293);
294#[cfg(feature = "mysql")]
295outbox_database!(
296    MySql,
297    MySqlConnection,
298    |spec| Ok(Defaults {
299        mysql: Some(Statements::of(&dialect::MySql, spec)?),
300        ..Defaults::default()
301    }),
302    |_conn, defaults| defaults.mysql.as_ref()
303);
304#[cfg(feature = "sqlite")]
305outbox_database!(
306    Sqlite,
307    SqliteConnection,
308    |spec| Ok(Defaults {
309        sqlite: Some(Statements::of(&dialect::Sqlite, spec)?),
310        ..Defaults::default()
311    }),
312    |_conn, defaults| defaults.sqlite.as_ref()
313);
314// `Any` passes a statement to its backend as written, so the backend picks the text.
315#[cfg(feature = "any")]
316outbox_database!(
317    Any,
318    AnyConnection,
319    |spec| Ok(Defaults {
320        #[cfg(feature = "postgres")]
321        postgres: Some(Statements::of(&dialect::Postgres, spec)?),
322        #[cfg(feature = "mysql")]
323        mysql: Some(Statements::of(&dialect::MySql, spec)?),
324        #[cfg(feature = "sqlite")]
325        sqlite: Some(Statements::of(&dialect::Sqlite, spec)?),
326    }),
327    |conn, defaults| defaults.for_backend(conn.backend_name())
328);
329
330/// The error of a default event on a connection whose database the crate was built without.
331pub(super) fn no_outbox_statement(record: &'static str, event: &'static str) -> Error {
332    Error::Configuration(
333        format!("`{record}` has no {event} statement for this connection's database").into(),
334    )
335}
336
337#[cfg(test)]
338mod tests {
339    #[cfg(feature = "postgres")]
340    use std::ptr;
341
342    #[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
343    use ruststream_sqlx_dialect::{Column, Form, OutboxDialect, StatementError, TableSpec};
344
345    use super::no_outbox_statement;
346    #[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
347    use super::{Defaults, OutboxDatabase, dialect};
348
349    #[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
350    const TABLE: TableSpec<'static> = TableSpec::new("outbox", Column::new("id"), Form::RowLock)
351        .group(Column::new("name"))
352        .payload(Column::new("payload"))
353        .processed_at(Column::new("processed_at"))
354        .database_clock();
355
356    #[test]
357    fn the_missing_statement_names_the_record_and_the_event() {
358        assert_eq!(
359            no_outbox_statement("OrderEvent", "fetch").to_string(),
360            "error with configuration: `OrderEvent` has no fetch statement for this connection's \
361             database"
362        );
363    }
364
365    #[cfg(all(
366        feature = "any",
367        feature = "postgres",
368        feature = "mysql",
369        feature = "sqlite"
370    ))]
371    #[test]
372    fn an_any_backend_runs_the_statements_of_its_database() -> Result<(), StatementError> {
373        use sqlx::Any;
374
375        use super::{MYSQL_BACKEND, POSTGRES_BACKEND, SQLITE_BACKEND};
376
377        let defaults = <Any as OutboxDatabase>::defaults(&TABLE)?;
378        // The mark reads each database's clock, so its text differs on every dialect.
379        let mark = |backend| {
380            defaults
381                .for_backend(backend)
382                .map(|statements| statements.mark)
383        };
384        assert_eq!(
385            mark(POSTGRES_BACKEND),
386            Some(dialect::Postgres.outbox_mark(&TABLE)?.sql())
387        );
388        assert_eq!(
389            mark(MYSQL_BACKEND),
390            Some(dialect::MySql.outbox_mark(&TABLE)?.sql())
391        );
392        assert_eq!(
393            mark(SQLITE_BACKEND),
394            Some(dialect::Sqlite.outbox_mark(&TABLE)?.sql())
395        );
396        assert_eq!(mark("MSSQL"), None);
397        Ok(())
398    }
399
400    #[cfg(feature = "postgres")]
401    #[test]
402    fn a_statement_built_twice_is_kept_once() -> Result<(), StatementError> {
403        use sqlx::Postgres;
404
405        let first = <Postgres as OutboxDatabase>::defaults(&TABLE)?;
406        let second = <Postgres as OutboxDatabase>::defaults(&TABLE)?;
407        let mark = |defaults: &Defaults| defaults.postgres.map(|statements| statements.mark);
408        let (Some(first), Some(second)) = (mark(&first), mark(&second)) else {
409            panic!("Postgres builds its own statements");
410        };
411        assert!(
412            ptr::eq(first, second),
413            "the second build reuses the first's text"
414        );
415        Ok(())
416    }
417
418    #[cfg(all(
419        feature = "any",
420        feature = "postgres",
421        feature = "mysql",
422        feature = "sqlite"
423    ))]
424    #[test]
425    fn the_backend_names_are_sqlxs() {
426        use sqlx::Database;
427
428        use super::{MYSQL_BACKEND, POSTGRES_BACKEND, SQLITE_BACKEND};
429
430        assert_eq!(POSTGRES_BACKEND, <sqlx::Postgres as Database>::NAME);
431        assert_eq!(MYSQL_BACKEND, <sqlx::MySql as Database>::NAME);
432        assert_eq!(SQLITE_BACKEND, <sqlx::Sqlite as Database>::NAME);
433    }
434}
ruststream_sqlx/outbox/database.rs

ruststream_sqlx/outbox/
database.rs
