ruststream_sqlx_dialect/
outbox.rs

1//! The statements of a transactional outbox table: the take of a record by its id, the mark of a
2//! processed one, and the recovery of the unprocessed records of a name.
3
4use crate::dialect::Dialect;
5use crate::spec::TableSpec;
6use crate::statement::{Statement, StatementError};
7
8/// The statements a dialect builds for an outbox table, which records what a service publishes
9/// until a consumer has processed it.
10///
11/// An outbox table is described by a [`TableSpec`] whose id identifies a record, whose
12/// [`group`](TableSpec::group) column holds the name the record was published under, and whose
13/// optional [`processed_at`](TableSpec::processed_at) column marks a processed record. A table
14/// without it deletes a processed record instead. The table's [`Form`](crate::Form) does not
15/// apply: a record is taken by its id, never claimed. `#[derive(Outbox)]` in `ruststream-sqlx`
16/// builds these statements at compile time with each built-in dialect, and on the database's own
17/// clock ([`TableSpec::database_clock`]). [`Postgres`](crate::Postgres), [`MySql`](crate::MySql)
18/// and [`Sqlite`](crate::Sqlite) implement it.
19///
20/// # Examples
21///
22/// ```
23/// # #[cfg(feature = "postgres")]
24/// # fn main() -> Result<(), ruststream_sqlx_dialect::StatementError> {
25/// use ruststream_sqlx_dialect::{Column, Form, OutboxDialect, Param, Postgres, TableSpec};
26///
27/// // outbox: id BIGSERIAL PRIMARY KEY, name TEXT, payload BYTEA, processed_at TIMESTAMPTZ
28/// const OUTBOX: TableSpec<'static> = TableSpec::new("outbox", Column::new("id"), Form::RowLock)
29///     .group(Column::new("name"))
30///     .processed_at(Column::new("processed_at"))
31///     .payload(Column::new("payload"))
32///     .database_clock();
33///
34/// // What a consumer of a tracked message runs: the take of its record, then the mark once the
35/// // handler acknowledged it.
36/// let fetch = Postgres.outbox_fetch(&OUTBOX)?;
37/// assert_eq!(
38///     fetch.sql(),
39///     r#"SELECT "id", "name", "processed_at", "payload" FROM "outbox" WHERE "id" = $1 AND "processed_at" IS NULL"#
40/// );
41/// let mark = Postgres.outbox_mark(&OUTBOX)?;
42/// assert_eq!(
43///     mark.sql(),
44///     r#"UPDATE "outbox" SET "processed_at" = statement_timestamp() WHERE "id" = $1"#
45/// );
46///
47/// // What the republish at startup runs for each registered name.
48/// let recover = Postgres.outbox_recover(&OUTBOX)?;
49/// assert_eq!(recover.params(), [Param::Group]);
50/// # Ok(())
51/// # }
52/// # #[cfg(not(feature = "postgres"))]
53/// # fn main() {}
54/// ```
55pub trait OutboxDialect: Dialect {
56    /// The statement that reads the record [`Param::Id`](crate::Param::Id) names while it is
57    /// unprocessed, with every column: what the consumer of a tracked message takes into work.
58    ///
59    /// # Errors
60    ///
61    /// [`StatementError::IdentifierTooLong`] for a name longer than the database keeps.
62    ///
63    /// # Examples
64    ///
65    /// ```
66    /// # #[cfg(feature = "mysql")]
67    /// # fn main() -> Result<(), ruststream_sqlx_dialect::StatementError> {
68    /// use ruststream_sqlx_dialect::{Column, Form, MySql, OutboxDialect, Param, TableSpec};
69    ///
70    /// // A table without `processed_at` deletes a processed record, so every record it holds is
71    /// // unprocessed.
72    /// const OUTBOX: TableSpec<'static> = TableSpec::new("outbox", Column::new("id"), Form::RowLock)
73    ///     .group(Column::new("name"))
74    ///     .payload(Column::new("payload"));
75    ///
76    /// let fetch = MySql.outbox_fetch(&OUTBOX)?;
77    /// assert_eq!(fetch.sql(), "SELECT `id`, `name`, `payload` FROM `outbox` WHERE `id` = ?");
78    /// assert_eq!(fetch.params(), [Param::Id]);
79    /// # Ok(())
80    /// # }
81    /// # #[cfg(not(feature = "mysql"))]
82    /// # fn main() {}
83    /// ```
84    fn outbox_fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
85
86    /// The statement that marks the record [`Param::Id`](crate::Param::Id) names processed: it
87    /// sets `processed_at` to now, or deletes the record in a table without that column.
88    ///
89    /// Where the table runs in the row lock form, this is the dialect's
90    /// [`ack`](Dialect::ack) of it.
91    ///
92    /// # Errors
93    ///
94    /// [`StatementError::IdentifierTooLong`] for a name longer than the database keeps.
95    ///
96    /// # Examples
97    ///
98    /// ```
99    /// # #[cfg(feature = "sqlite")]
100    /// # fn main() -> Result<(), ruststream_sqlx_dialect::StatementError> {
101    /// use ruststream_sqlx_dialect::{Column, Form, OutboxDialect, Sqlite, TableSpec};
102    ///
103    /// const OUTBOX: TableSpec<'static> = TableSpec::new("outbox", Column::new("id"), Form::RowLock)
104    ///     .group(Column::new("name"))
105    ///     .processed_at(Column::new("processed_at"))
106    ///     .payload(Column::new("payload"))
107    ///     .database_clock();
108    ///
109    /// // The mark reads SQLite's own clock, as text that sorts as the times do.
110    /// let mark = Sqlite.outbox_mark(&OUTBOX)?;
111    /// assert_eq!(
112    ///     mark.sql(),
113    ///     "UPDATE `outbox` SET `processed_at` = strftime('%Y-%m-%dT%H:%M:%f+00:00', 'now') \
114    ///      WHERE `id` = ?"
115    /// );
116    /// # Ok(())
117    /// # }
118    /// # #[cfg(not(feature = "sqlite"))]
119    /// # fn main() {}
120    /// ```
121    fn outbox_mark(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
122
123    /// The statement that reads the unprocessed records published under the name
124    /// [`Param::Group`](crate::Param::Group) binds, with every column: what the republish at
125    /// startup sends again.
126    ///
127    /// # Errors
128    ///
129    /// [`StatementError::MissingRole`] for a table without a [`group`](TableSpec::group) column,
130    /// and [`StatementError::IdentifierTooLong`] for a name longer than the database keeps.
131    ///
132    /// # Examples
133    ///
134    /// ```
135    /// # #[cfg(feature = "postgres")]
136    /// # fn main() -> Result<(), ruststream_sqlx_dialect::StatementError> {
137    /// use ruststream_sqlx_dialect::{Column, Form, OutboxDialect, Param, Postgres, TableSpec};
138    ///
139    /// const OUTBOX: TableSpec<'static> = TableSpec::new("outbox", Column::new("id"), Form::RowLock)
140    ///     .within("app")
141    ///     .group(Column::new("name"))
142    ///     .processed_at(Column::new("processed_at"))
143    ///     .payload(Column::new("payload"));
144    ///
145    /// let recover = Postgres.outbox_recover(&OUTBOX)?;
146    /// assert_eq!(
147    ///     recover.sql(),
148    ///     r#"SELECT "id", "name", "processed_at", "payload" FROM "app"."outbox" WHERE "name" = $1 AND "processed_at" IS NULL"#
149    /// );
150    /// assert_eq!(recover.params(), [Param::Group]);
151    /// # Ok(())
152    /// # }
153    /// # #[cfg(not(feature = "postgres"))]
154    /// # fn main() {}
155    /// ```
156    fn outbox_recover(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
157}
ruststream_sqlx_dialect/outbox.rs

ruststream_sqlx_dialect/
outbox.rs
