ruststream_sqlx_dialect/column.rs
1//! One column of a queue table.
2
3/// One column of a queue table: its name in the database, and whether the database fills it in.
4///
5/// # Examples
6///
7/// ```
8/// # use std::num::NonZeroUsize;
9/// use ruststream_sqlx_dialect::{
10/// ClaimShape, Dialect, Form, Lease, Param, Statement, StatementError, TableSpec,
11/// };
12/// # use ruststream_sqlx_dialect::TableName;
13///
14/// /// SQL Server, a database without a built-in dialect, recording when a handler last extended
15/// /// its lease: every lease table of the service has a `seen_at` column.
16/// #[derive(Debug)]
17/// pub struct Mssql;
18///
19/// impl Lease for Mssql {
20/// fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
21/// // The lease form carries the column of its expiry.
22/// let Form::Lease(expiry) = spec.form() else {
23/// return Err(StatementError::FormMismatch {
24/// statement: "extend",
25/// form: spec.form().name(),
26/// });
27/// };
28/// let mut sql = String::from("UPDATE ");
29/// self.quote_into(spec.table(), &mut sql);
30/// sql.push_str(" SET ");
31/// self.quote_into(expiry.name(), &mut sql);
32/// sql.push_str(" = @p1, [seen_at] = @p2 WHERE ");
33/// self.quote_into(spec.id().name(), &mut sql);
34/// sql.push_str(" = @p3 AND ");
35/// self.quote_into(expiry.name(), &mut sql);
36/// sql.push_str(" = @p4");
37/// Ok(Statement::new(sql, [Param::Lease, Param::Now, Param::Id, Param::Held]))
38/// }
39/// # fn lease_claim(&self, spec: &TableSpec<'_>, shape: ClaimShape) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
40/// # fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
41/// }
42/// # impl Dialect for Mssql {
43/// # fn name(&self) -> &'static str { "mssql" }
44/// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
45/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
46/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
47/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
48/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
49/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
50/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
51/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
52/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
53/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
54/// # }
55/// ```
56#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
57pub struct Column<'a> {
58 name: &'a str,
59 generated: bool,
60}
61
62impl<'a> Column<'a> {
63 /// A column the service writes.
64 ///
65 /// `#[derive(Inbox)]` names the column of each field with it.
66 ///
67 /// # Examples
68 ///
69 /// ```
70 /// # #[cfg(feature = "postgres")]
71 /// # fn main() {
72 /// use ruststream_sqlx_dialect::insert::{self, Sql};
73 /// use ruststream_sqlx_dialect::{Column, Form, TableSpec};
74 ///
75 /// // email_jobs: job_id BIGINT PRIMARY KEY, payload BYTEA; the service writes both.
76 /// const EMAILS: TableSpec<'static> =
77 /// TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
78 /// .payload(Column::new("payload"));
79 ///
80 /// const INSERT: Sql<128> = insert::postgres(&EMAILS);
81 /// assert_eq!(INSERT.as_str(), r#"INSERT INTO "email_jobs" ("job_id", "payload") VALUES ($1, $2)"#);
82 /// # }
83 /// # #[cfg(not(feature = "postgres"))]
84 /// # fn main() {}
85 /// ```
86 #[must_use]
87 pub const fn new(name: &'a str) -> Self {
88 Self {
89 name,
90 generated: false,
91 }
92 }
93
94 /// The same column, filled in by the database, so an insert leaves it out.
95 ///
96 /// `#[field(generated)]` on a field sets it, as `#[field(id, generated)]` does on a
97 /// sequence's id.
98 ///
99 /// # Examples
100 ///
101 /// ```
102 /// # #[cfg(feature = "postgres")]
103 /// # fn main() {
104 /// use ruststream_sqlx_dialect::insert::{self, Sql};
105 /// use ruststream_sqlx_dialect::{Column, Form, TableSpec};
106 ///
107 /// // email_jobs: job_id BIGSERIAL PRIMARY KEY, payload BYTEA; the sequence numbers the rows.
108 /// const EMAILS: TableSpec<'static> =
109 /// TableSpec::new("email_jobs", Column::new("job_id").generated(), Form::RowLock)
110 /// .payload(Column::new("payload"));
111 ///
112 /// const INSERT: Sql<128> = insert::postgres(&EMAILS);
113 /// assert_eq!(INSERT.as_str(), r#"INSERT INTO "email_jobs" ("payload") VALUES ($1)"#);
114 /// # }
115 /// # #[cfg(not(feature = "postgres"))]
116 /// # fn main() {}
117 /// ```
118 #[must_use]
119 pub const fn generated(self) -> Self {
120 Self {
121 generated: true,
122 ..self
123 }
124 }
125
126 /// The column's name in the database.
127 ///
128 /// # Examples
129 ///
130 /// ```
131 /// # use std::num::NonZeroUsize;
132 /// use ruststream_sqlx_dialect::{Dialect, Param, Role, Statement, StatementError, TableSpec};
133 /// # use ruststream_sqlx_dialect::TableName;
134 ///
135 /// /// SQL Server, a database without a built-in dialect; the service's tables take their rows
136 /// /// by row lock.
137 /// #[derive(Debug)]
138 /// pub struct Mssql;
139 ///
140 /// impl Dialect for Mssql {
141 /// fn name(&self) -> &'static str {
142 /// "mssql"
143 /// }
144 ///
145 /// // A dead letter starts its new group with the attempt count of a new row.
146 /// fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
147 /// let Some(group) = spec.column(Role::Group) else {
148 /// return Err(StatementError::MissingRole {
149 /// statement: "dead_letter_group",
150 /// role: Role::Group,
151 /// });
152 /// };
153 /// let mut sql = String::from("UPDATE ");
154 /// self.quote_into(spec.table(), &mut sql);
155 /// sql.push_str(" SET ");
156 /// self.quote_into(group.name(), &mut sql);
157 /// sql.push_str(" = @p1");
158 /// if let Some(attempt) = spec.column(Role::Attempt) {
159 /// sql.push_str(", ");
160 /// self.quote_into(attempt.name(), &mut sql);
161 /// sql.push_str(" = DEFAULT");
162 /// }
163 /// sql.push_str(" WHERE ");
164 /// self.quote_into(spec.id().name(), &mut sql);
165 /// sql.push_str(" = @p2");
166 /// Ok(Statement::new(sql, [Param::Destination, Param::Id]))
167 /// }
168 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
169 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
170 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
171 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
172 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
173 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
174 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
175 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
176 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
177 /// }
178 /// ```
179 #[must_use]
180 pub const fn name(&self) -> &'a str {
181 self.name
182 }
183
184 /// Whether the database fills the column in, so an insert leaves it out.
185 ///
186 /// # Examples
187 ///
188 /// ```
189 /// # use ruststream_sqlx_dialect::TableName;
190 /// use std::num::NonZeroUsize;
191 ///
192 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
193 ///
194 /// /// SQL Server, a database without a built-in dialect.
195 /// #[derive(Debug)]
196 /// pub struct Mssql;
197 ///
198 /// impl Dialect for Mssql {
199 /// fn name(&self) -> &'static str {
200 /// "mssql"
201 /// }
202 ///
203 /// fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
204 /// if spec.selects_all() {
205 /// return Err(StatementError::Flattened { statement: "insert" });
206 /// }
207 /// let mut names = String::new();
208 /// let mut values = String::new();
209 /// let mut params = Vec::new();
210 /// // Every column the database does not fill, bound by its position in the
211 /// // description.
212 /// for (index, column) in spec.columns().enumerate() {
213 /// if column.is_generated() {
214 /// continue;
215 /// }
216 /// if !params.is_empty() {
217 /// names.push_str(", ");
218 /// values.push_str(", ");
219 /// }
220 /// self.quote_into(column.name(), &mut names);
221 /// let position = NonZeroUsize::MIN.saturating_add(params.len());
222 /// self.placeholder_into(position, &mut values);
223 /// params.push(Param::Column(index));
224 /// }
225 /// let mut table = String::new();
226 /// self.quote_into(spec.table(), &mut table);
227 /// let sql = format!("INSERT INTO {table} ({names}) VALUES ({values})");
228 /// Ok(Statement::new(sql, params))
229 /// }
230 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
231 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
232 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
233 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
234 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
235 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
236 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
237 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
238 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
239 /// }
240 /// ```
241 #[must_use]
242 pub const fn is_generated(&self) -> bool {
243 self.generated
244 }
245}
246
247#[cfg(test)]
248mod tests {
249 use super::Column;
250
251 #[test]
252 fn a_column_carries_its_name_and_generation() {
253 let id = Column::new("job_id").generated();
254 assert_eq!(id.name(), "job_id");
255 assert!(id.is_generated());
256 let subject = Column::new("subject");
257 assert!(!subject.is_generated());
258 }
259}