ruststream_sqlx/inbox/database/
built_in.rs

1//! The dialect built into the crate for each database: the broker's dialect unless the service
2//! passes its own.
3
4use std::fmt;
5use std::num::NonZeroUsize;
6
7#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
8use ruststream_sqlx_dialect as dialect;
9#[cfg(any(feature = "postgres", feature = "mysql", feature = "any"))]
10use ruststream_sqlx_dialect::RowLock;
11use ruststream_sqlx_dialect::{
12    Advisory, ClaimShape, Dialect, Lease, Opening, Statement, StatementError, TableName, TableSpec,
13};
14#[cfg(any(
15    feature = "postgres",
16    feature = "mysql",
17    feature = "sqlite",
18    feature = "any"
19))]
20use ruststream_sqlx_dialect::{Opens, level};
21#[cfg(feature = "any")]
22use sqlx::{Any, AnyConnection};
23#[cfg(feature = "mysql")]
24use sqlx::{MySql, MySqlConnection};
25#[cfg(feature = "postgres")]
26use sqlx::{PgConnection, Postgres};
27#[cfg(feature = "sqlite")]
28use sqlx::{Sqlite, SqliteConnection};
29
30use super::QueueDatabase;
31#[cfg(feature = "any")]
32use super::any::AnyDialect;
33
34/// The dialect built into the crate for the database `DB`: what [`SqlxBroker::new`] builds its
35/// statements with.
36///
37/// It builds them with the dialect of the crate's [`dialect`](crate::dialect) module that serves
38/// the database: [`Postgres`](crate::dialect::Postgres) on Postgres,
39/// [`MySql`](crate::dialect::MySql) on MySQL and MariaDB, [`Sqlite`](crate::dialect::Sqlite) on
40/// SQLite. On an `AnyPool` it is the dialect of the database the pool reaches, among those whose
41/// features are on, picked when the broker connects.
42///
43/// It serves the forms its database serves, through the traits of the dialect module: every
44/// `BuiltIn` implements [`Lease`] and [`Advisory`], and `BuiltIn<Postgres>`, `BuiltIn<MySql>` and
45/// `BuiltIn<Any>` implement [`RowLock`]. A SQLite table in the row lock form therefore does not
46/// compile. An `AnyPool` names its database only when the broker connects, so on a SQLite backend
47/// the subscription to a row lock table stops when it starts, with
48/// [`StatementError::UnsupportedForm`].
49///
50/// It opens transactions at the isolation levels and SQLite modes its database keeps, one
51/// [`Opens`](crate::dialect::Opens) per level: `BuiltIn<Postgres>` READ COMMITTED, REPEATABLE
52/// READ and SERIALIZABLE, `BuiltIn<MySql>` those and READ UNCOMMITTED, `BuiltIn<Sqlite>` the three
53/// modes. A table that names another does not compile. `BuiltIn<Any>` takes all seven, and the
54/// picked backend's dialect refuses the ones its database lacks when the subscription starts,
55/// with [`StatementError::UnsupportedOpening`].
56///
57/// [`SqlxBroker::new`]: crate::SqlxBroker::new
58///
59/// # Examples
60///
61/// ```
62/// # #[cfg(feature = "postgres")]
63/// # mod demo {
64/// use ruststream_sqlx::BuiltIn;
65/// use ruststream_sqlx::prelude::*;
66/// use serde::Deserialize;
67/// use sqlx::{PgPool, Postgres};
68///
69/// #[derive(Inbox, sqlx::FromRow)]
70/// #[inbox(table = "email_jobs")]
71/// pub struct SendEmail {
72///     #[field(id, generated)]
73///     job_id: i64,
74///     #[field(group)]
75///     name: String,
76///     #[field(payload)]
77///     payload: Vec<u8>,
78/// }
79///
80/// #[derive(Deserialize)]
81/// struct Email {
82///     to: String,
83/// }
84///
85/// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
86/// async fn send(email: &Email) -> HandlerOutcome {
87///     tracing::info!(to = %email.to, "sending");
88///     HandlerOutcome::ack()
89/// }
90///
91/// // `SqlxBroker<Postgres>` names the same type: the built-in dialect is the default.
92/// pub fn broker(pool: PgPool) -> SqlxBroker<Postgres, BuiltIn<Postgres>> {
93///     SqlxBroker::new(pool)
94/// }
95///
96/// pub fn app(pool: PgPool) -> RustStream {
97///     RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker(pool), |b| {
98///         b.include(send);
99///     })
100/// }
101/// # }
102/// # fn main() {}
103/// ```
104///
105/// On SQLite the table takes the lease form, and the same struct serves a `PgPool` and a
106/// `MySqlPool` unchanged:
107///
108/// ```no_run
109/// # #[cfg(all(feature = "sqlite", feature = "chrono"))]
110/// # mod demo {
111/// use chrono::{DateTime, Utc};
112/// use ruststream_sqlx::prelude::*;
113/// use serde::Deserialize;
114/// use sqlx::SqlitePool;
115///
116/// // thumbnail_jobs: id INTEGER PRIMARY KEY, attempt INTEGER NOT NULL DEFAULT 1,
117/// // locked_until TEXT, payload BLOB NOT NULL
118/// #[derive(Inbox, sqlx::FromRow)]
119/// #[inbox(table = "thumbnail_jobs")]
120/// pub struct MakeThumbnail {
121///     #[field(id, generated)]
122///     id: i64,
123///     #[field(attempt, generated)]
124///     attempt: i16,
125///     #[field(locked_until)]
126///     locked_until: Option<DateTime<Utc>>,
127///     #[field(payload)]
128///     payload: Vec<u8>,
129/// }
130///
131/// #[derive(Deserialize)]
132/// pub struct Image {
133///     path: String,
134/// }
135///
136/// # async fn resize(image: &Image) -> bool { !image.path.is_empty() }
137/// #[subscriber(InboxQueue::<MakeThumbnail>::new("thumbnails"))]
138/// async fn thumbnail(image: &Image) -> HandlerOutcome {
139///     if resize(image).await {
140///         HandlerOutcome::ack()
141///     } else {
142///         HandlerOutcome::retry()
143///     }
144/// }
145///
146/// // The same struct serves a `PgPool` and a `MySqlPool` too: the broker takes the dialect of its
147/// // pool, which builds its statements from the table's description.
148/// pub fn app(pool: SqlitePool) -> RustStream {
149///     RustStream::new(AppInfo::new("thumbnails", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
150///         b.include(thumbnail);
151///     })
152/// }
153/// # }
154/// # fn main() {}
155/// ```
156pub struct BuiltIn<DB: BuiltInDialect>(DB::Picked);
157
158impl<DB: BuiltInDialect> BuiltIn<DB> {
159    /// The built-in dialect that `picked` builds the statements of.
160    #[cfg(any(
161        feature = "postgres",
162        feature = "mysql",
163        feature = "sqlite",
164        feature = "any"
165    ))]
166    pub(crate) const fn new(picked: DB::Picked) -> Self {
167        Self(picked)
168    }
169}
170
171impl<DB: BuiltInDialect> Clone for BuiltIn<DB> {
172    fn clone(&self) -> Self {
173        *self
174    }
175}
176
177impl<DB: BuiltInDialect> Copy for BuiltIn<DB> {}
178
179impl<DB: BuiltInDialect> fmt::Debug for BuiltIn<DB> {
180    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
181        f.debug_tuple("BuiltIn").field(&self.0.name()).finish()
182    }
183}
184
185impl<DB: BuiltInDialect> Dialect for BuiltIn<DB> {
186    fn name(&self) -> &'static str {
187        self.0.name()
188    }
189
190    fn quote_into(&self, ident: &str, out: &mut String) {
191        self.0.quote_into(ident, out);
192    }
193
194    fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) {
195        self.0.placeholder_into(index, out);
196    }
197
198    fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
199        self.0.fetch(spec)
200    }
201
202    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
203        self.0.ack(spec)
204    }
205
206    fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
207        self.0.retry(spec)
208    }
209
210    fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
211        self.0.retry_after(spec)
212    }
213
214    fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
215        self.0.discard(spec)
216    }
217
218    fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
219        self.0.dead_letter_group(spec)
220    }
221
222    fn dead_letter_table(
223        &self,
224        spec: &TableSpec<'_>,
225        target: TableName<'_>,
226    ) -> Result<Vec<Statement>, StatementError> {
227        self.0.dead_letter_table(spec, target)
228    }
229
230    fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
231        self.0.insert(spec)
232    }
233
234    fn server_version(&self) -> Option<&'static str> {
235        self.0.server_version()
236    }
237
238    fn check_server(&self, spec: &TableSpec<'_>, version: &str) -> Result<(), StatementError> {
239        self.0.check_server(spec, version)
240    }
241
242    fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> {
243        self.0.begin(opening)
244    }
245
246    fn savepoint(&self) -> &'static str {
247        self.0.savepoint()
248    }
249
250    fn rollback_to_savepoint(&self) -> &'static str {
251        self.0.rollback_to_savepoint()
252    }
253
254    fn fifo_guard(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
255        self.0.fifo_guard(spec)
256    }
257
258    fn count_claimable(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
259        self.0.count_claimable(spec)
260    }
261}
262
263impl<DB: BuiltInDialect> Lease for BuiltIn<DB> {
264    fn lease_claim(
265        &self,
266        spec: &TableSpec<'_>,
267        shape: ClaimShape,
268    ) -> Result<Statement, StatementError> {
269        self.0.lease_claim(spec, shape)
270    }
271
272    fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
273        self.0.extend(spec)
274    }
275
276    fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
277        self.0.stamp(spec)
278    }
279
280    fn claim_writes_lease(&self) -> bool {
281        self.0.claim_writes_lease()
282    }
283
284    fn claim_counts_attempt(&self, spec: &TableSpec<'_>) -> bool {
285        self.0.claim_counts_attempt(spec)
286    }
287
288    fn begin_lease_claim(&self) -> Option<&'static str> {
289        self.0.begin_lease_claim()
290    }
291}
292
293impl<DB: BuiltInDialect> Advisory for BuiltIn<DB> {
294    fn advisory_claim(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
295        self.0.advisory_claim(spec)
296    }
297
298    fn lock(&self) -> Option<Statement> {
299        self.0.lock()
300    }
301
302    fn unlock(&self) -> Option<Statement> {
303        self.0.unlock()
304    }
305
306    fn take(
307        &self,
308        spec: &TableSpec<'_>,
309        shape: ClaimShape,
310    ) -> Result<Vec<Statement>, StatementError> {
311        self.0.take(spec, shape)
312    }
313}
314
315#[cfg(feature = "postgres")]
316impl RowLock for BuiltIn<Postgres> {
317    fn lock_claim(
318        &self,
319        spec: &TableSpec<'_>,
320        shape: ClaimShape,
321    ) -> Result<Statement, StatementError> {
322        self.0.lock_claim(spec, shape)
323    }
324}
325
326#[cfg(feature = "mysql")]
327impl RowLock for BuiltIn<MySql> {
328    fn lock_claim(
329        &self,
330        spec: &TableSpec<'_>,
331        shape: ClaimShape,
332    ) -> Result<Statement, StatementError> {
333        self.0.lock_claim(spec, shape)
334    }
335}
336
337#[cfg(feature = "any")]
338impl RowLock for BuiltIn<Any> {
339    fn lock_claim(
340        &self,
341        spec: &TableSpec<'_>,
342        shape: ClaimShape,
343    ) -> Result<Statement, StatementError> {
344        self.0.lock_claim(spec, shape)
345    }
346}
347
348#[cfg(feature = "postgres")]
349impl Opens<level::ReadCommitted> for BuiltIn<Postgres> {}
350
351#[cfg(feature = "postgres")]
352impl Opens<level::RepeatableRead> for BuiltIn<Postgres> {}
353
354#[cfg(feature = "postgres")]
355impl Opens<level::Serializable> for BuiltIn<Postgres> {}
356
357#[cfg(feature = "mysql")]
358impl Opens<level::ReadUncommitted> for BuiltIn<MySql> {}
359
360#[cfg(feature = "mysql")]
361impl Opens<level::ReadCommitted> for BuiltIn<MySql> {}
362
363#[cfg(feature = "mysql")]
364impl Opens<level::RepeatableRead> for BuiltIn<MySql> {}
365
366#[cfg(feature = "mysql")]
367impl Opens<level::Serializable> for BuiltIn<MySql> {}
368
369#[cfg(feature = "sqlite")]
370impl Opens<level::Deferred> for BuiltIn<Sqlite> {}
371
372#[cfg(feature = "sqlite")]
373impl Opens<level::Immediate> for BuiltIn<Sqlite> {}
374
375#[cfg(feature = "sqlite")]
376impl Opens<level::Exclusive> for BuiltIn<Sqlite> {}
377
378// Why every level and mode compiles on `Any`: an `AnyPool` names its database only when the
379// broker connects, so the type cannot say which of them the database opens. The picked backend's
380// dialect refuses the ones it lacks when the subscription to such a table starts.
381#[cfg(feature = "any")]
382impl Opens<level::ReadUncommitted> for BuiltIn<Any> {}
383
384#[cfg(feature = "any")]
385impl Opens<level::ReadCommitted> for BuiltIn<Any> {}
386
387#[cfg(feature = "any")]
388impl Opens<level::RepeatableRead> for BuiltIn<Any> {}
389
390#[cfg(feature = "any")]
391impl Opens<level::Serializable> for BuiltIn<Any> {}
392
393#[cfg(feature = "any")]
394impl Opens<level::Deferred> for BuiltIn<Any> {}
395
396#[cfg(feature = "any")]
397impl Opens<level::Immediate> for BuiltIn<Any> {}
398
399#[cfg(feature = "any")]
400impl Opens<level::Exclusive> for BuiltIn<Any> {}
401
402/// A database whose dialect is built into this crate, so `SqlxBroker::new` needs no dialect of
403/// the service's own.
404///
405/// Postgres (feature `postgres`), MySQL with MariaDB (feature `mysql`) and SQLite (feature
406/// `sqlite`) have one, [`BuiltIn<DB>`](BuiltIn). `Any` (feature `any`) takes the dialect of the
407/// database its pool reaches, among those whose features are on, and the broker picks it when it
408/// connects. A row on an `AnyPool` holds only the types `sqlx::Any` carries, and no time is among
409/// them, so a table with `locked_until`, `retry_after` or `processed_at` is out of its reach.
410///
411/// # Examples
412///
413/// ```
414/// # #[cfg(all(feature = "any", feature = "postgres", feature = "mysql"))]
415/// # mod demo {
416/// use ruststream_sqlx::prelude::*;
417/// use serde::Deserialize;
418/// use sqlx::AnyPool;
419///
420/// // The columns `sqlx::Any` carries: integers, text and bytes, no time.
421/// #[derive(Inbox, sqlx::FromRow)]
422/// #[inbox(table = "email_jobs")]
423/// pub struct SendEmail {
424///     #[field(id, generated)]
425///     job_id: i64,
426///     #[field(group)]
427///     name: String,
428///     #[field(attempt, generated)]
429///     attempt: i32,
430///     #[field(payload)]
431///     payload: Vec<u8>,
432/// }
433///
434/// #[derive(Deserialize)]
435/// struct Email {
436///     to: String,
437/// }
438///
439/// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
440/// async fn send(email: &Email) -> HandlerOutcome {
441///     tracing::info!(to = %email.to, "sending");
442///     HandlerOutcome::ack()
443/// }
444///
445/// // One build for every deployment: `Any` is a `BuiltInDialect`, and the broker takes the
446/// // dialect of the database the URL names, Postgres or MySQL, when it connects. `main` installs
447/// // sqlx's drivers (`sqlx::any::install_default_drivers`) before it builds the pool.
448/// pub fn app(pool: AnyPool) -> RustStream {
449///     RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
450///         b.include(send);
451///     })
452/// }
453/// # }
454/// # fn main() {}
455/// ```
456#[diagnostic::on_unimplemented(
457    message = "`{Self}` has no dialect built into the crate",
458    label = "no built-in dialect for this database",
459    note = "build the broker with a dialect of the service's own: \
460            `SqlxBroker::with_dialect(pool, dialect)`, a `SqlxBroker<{Self}, YourDialect>`"
461)]
462pub trait BuiltInDialect: QueueDatabase {
463    /// The dialect of the crate's `dialect` module that builds the database's statements.
464    /// Machinery; [`BuiltIn`] holds it.
465    #[doc(hidden)]
466    type Picked: Lease + Advisory + Copy + 'static;
467
468    /// The dialect that builds the statements of the database `conn` reaches, or `None` when no
469    /// built-in dialect serves it: an `AnyPool`'s backend whose feature is off.
470    ///
471    /// Postgres, MySQL and SQLite answer without reading `conn`; the broker asks with the
472    /// connection it checks when it connects.
473    fn dialect(conn: &Self::Connection) -> Option<BuiltIn<Self>>;
474
475    /// The name of the database `conn` reaches, which an error names when no built-in dialect
476    /// serves it. Machinery; the broker calls it.
477    #[doc(hidden)]
478    fn backend(conn: &Self::Connection) -> &str {
479        let _ = conn;
480        Self::NAME
481    }
482}
483
484#[cfg(feature = "postgres")]
485impl BuiltInDialect for Postgres {
486    type Picked = dialect::Postgres;
487
488    fn dialect(_: &PgConnection) -> Option<BuiltIn<Self>> {
489        Some(BuiltIn::new(dialect::Postgres))
490    }
491}
492
493#[cfg(feature = "mysql")]
494impl BuiltInDialect for MySql {
495    type Picked = dialect::MySql;
496
497    fn dialect(_: &MySqlConnection) -> Option<BuiltIn<Self>> {
498        Some(BuiltIn::new(dialect::MySql))
499    }
500}
501
502#[cfg(feature = "sqlite")]
503impl BuiltInDialect for Sqlite {
504    type Picked = dialect::Sqlite;
505
506    fn dialect(_: &SqliteConnection) -> Option<BuiltIn<Self>> {
507        Some(BuiltIn::new(dialect::Sqlite))
508    }
509}
510
511#[cfg(feature = "any")]
512impl BuiltInDialect for Any {
513    type Picked = AnyDialect;
514
515    fn dialect(conn: &AnyConnection) -> Option<BuiltIn<Self>> {
516        AnyDialect::of(conn.backend_name()).map(BuiltIn::new)
517    }
518
519    fn backend(conn: &AnyConnection) -> &str {
520        conn.backend_name()
521    }
522}
523
524/// The name an `AnyConnection` reports for a Postgres backend: its sqlx driver's
525/// `Database::NAME`.
526#[cfg(feature = "any")]
527pub(crate) const POSTGRES_BACKEND: &str = "PostgreSQL";
528
529/// The name an `AnyConnection` reports for a MySQL or MariaDB backend.
530#[cfg(feature = "any")]
531pub(crate) const MYSQL_BACKEND: &str = "MySQL";
532
533/// The name an `AnyConnection` reports for a SQLite backend.
534#[cfg(feature = "any")]
535pub(crate) const SQLITE_BACKEND: &str = "SQLite";
536
537#[cfg(all(test, feature = "any"))]
538mod tests {
539    use sqlx::Database;
540
541    use super::{MYSQL_BACKEND, POSTGRES_BACKEND, SQLITE_BACKEND};
542
543    #[test]
544    fn the_backend_names_are_those_sqlx_reports() {
545        assert_eq!(POSTGRES_BACKEND, <sqlx::Postgres as Database>::NAME);
546        assert_eq!(MYSQL_BACKEND, <sqlx::MySql as Database>::NAME);
547        assert_eq!(SQLITE_BACKEND, <sqlx::Sqlite as Database>::NAME);
548    }
549}
ruststream_sqlx/inbox/database/built_in.rs

ruststream_sqlx/inbox/database/
built_in.rs
