ruststream_sqlx_dialect/table_name.rs
1//! A table named the way a service writes it: `table`, or `schema.table`.
2
3use thiserror::Error;
4
5/// A table outside the queue's own description, such as the table dead-lettered rows move to.
6///
7/// It is read from `table` or `schema.table`, and checked when it is read: no segment is empty,
8/// and there is at most one schema.
9///
10/// # Examples
11///
12/// ```
13/// # use std::num::NonZeroUsize;
14/// use ruststream_sqlx_dialect::{
15/// Dialect, Param, Statement, StatementError, TableName, TableSpec,
16/// };
17///
18/// /// SQL Server, a database without a built-in dialect.
19/// #[derive(Debug)]
20/// pub struct Mssql;
21///
22/// impl Dialect for Mssql {
23/// fn name(&self) -> &'static str {
24/// "mssql"
25/// }
26///
27/// // A dead email of the service's one queue table moves with the time it died.
28/// fn dead_letter_table(
29/// &self,
30/// _spec: &TableSpec<'_>,
31/// target: TableName<'_>,
32/// ) -> Result<Vec<Statement>, StatementError> {
33/// let mut into = String::new();
34/// self.quote_into(target.schema().unwrap_or("dbo"), &mut into);
35/// into.push('.');
36/// self.quote_into(target.table(), &mut into);
37/// Ok(vec![
38/// Statement::new(
39/// format!(
40/// "INSERT INTO {into} ([job_id], [payload], [died_at]) \
41/// SELECT [job_id], [payload], @p1 FROM [email_jobs] WHERE [job_id] = @p2"
42/// ),
43/// [Param::Now, Param::Id],
44/// ),
45/// Statement::new("DELETE FROM [email_jobs] WHERE [job_id] = @p1", [Param::Id]),
46/// ])
47/// }
48///
49/// fn quote_into(&self, ident: &str, out: &mut String) {
50/// out.push('[');
51/// out.push_str(&ident.replace(']', "]]"));
52/// out.push(']');
53/// }
54/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
55/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
56/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
57/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
58/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
59/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
60/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
61/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
62/// }
63/// ```
64#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
65pub struct TableName<'a> {
66 schema: Option<&'a str>,
67 table: &'a str,
68}
69
70impl<'a> TableName<'a> {
71 /// Reads `table` or `schema.table`.
72 ///
73 /// # Errors
74 ///
75 /// [`ParseTableNameError::EmptySegment`] when the name or one of its segments is empty;
76 /// [`ParseTableNameError::TooManySegments`] when the name has more than one dot.
77 ///
78 /// # Examples
79 ///
80 /// ```
81 /// # #[cfg(feature = "postgres")]
82 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
83 /// use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableName, TableSpec};
84 ///
85 /// const EMAILS: TableSpec<'static> =
86 /// TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
87 /// .payload(Column::new("payload"));
88 ///
89 /// // A dead email moves to the table the configuration names, in the `archive` schema.
90 /// let target = TableName::parse("archive.email_jobs_dead")?;
91 /// let moves = Postgres.dead_letter_table(&EMAILS, target)?;
92 /// assert_eq!(
93 /// moves[0].sql(),
94 /// r#"WITH moved AS (DELETE FROM "email_jobs" WHERE "job_id" = $1 RETURNING "job_id", "payload") INSERT INTO "archive"."email_jobs_dead" ("job_id", "payload") SELECT "job_id", "payload" FROM moved"#
95 /// );
96 /// # Ok(())
97 /// # }
98 /// # #[cfg(not(feature = "postgres"))]
99 /// # fn main() {}
100 /// ```
101 pub fn parse(name: &'a str) -> Result<Self, ParseTableNameError> {
102 let parsed = match name.split_once('.') {
103 None => Self {
104 schema: None,
105 table: name,
106 },
107 Some((_, table)) if table.contains('.') => {
108 return Err(ParseTableNameError::TooManySegments {
109 name: name.to_owned(),
110 });
111 }
112 Some((schema, table)) => Self {
113 schema: Some(schema),
114 table,
115 },
116 };
117 if parsed.table.is_empty() || parsed.schema.is_some_and(str::is_empty) {
118 return Err(ParseTableNameError::EmptySegment {
119 name: name.to_owned(),
120 });
121 }
122 Ok(parsed)
123 }
124
125 /// The schema, or `None` for the connection's default.
126 ///
127 /// # Examples
128 ///
129 /// ```
130 /// # use std::num::NonZeroUsize;
131 /// use ruststream_sqlx_dialect::{
132 /// Dialect, Param, Statement, StatementError, TableName, TableSpec,
133 /// };
134 ///
135 /// /// SQL Server, a database without a built-in dialect.
136 /// #[derive(Debug)]
137 /// pub struct Mssql;
138 ///
139 /// impl Dialect for Mssql {
140 /// fn name(&self) -> &'static str {
141 /// "mssql"
142 /// }
143 ///
144 /// fn dead_letter_table(
145 /// &self,
146 /// spec: &TableSpec<'_>,
147 /// target: TableName<'_>,
148 /// ) -> Result<Vec<Statement>, StatementError> {
149 /// // A struct that flattens another hides columns, so its row moves by position.
150 /// let mut columns = String::new();
151 /// if spec.selects_all() {
152 /// columns.push('*');
153 /// } else {
154 /// for column in spec.columns() {
155 /// if !columns.is_empty() {
156 /// columns.push_str(", ");
157 /// }
158 /// self.quote_into(column.name(), &mut columns);
159 /// }
160 /// }
161 /// // A target that names no schema lives in SQL Server's default one, `dbo`.
162 /// let mut into = String::new();
163 /// self.quote_into(target.schema().unwrap_or("dbo"), &mut into);
164 /// into.push('.');
165 /// self.quote_into(target.table(), &mut into);
166 /// let mut from = String::new();
167 /// self.quote_into(spec.table(), &mut from);
168 /// let mut id = String::new();
169 /// self.quote_into(spec.id().name(), &mut id);
170 /// let into_columns = if spec.selects_all() {
171 /// String::new()
172 /// } else {
173 /// format!(" ({columns})")
174 /// };
175 /// Ok(vec![
176 /// Statement::new(
177 /// format!(
178 /// "INSERT INTO {into}{into_columns} SELECT {columns} FROM {from} \
179 /// WHERE {id} = @p1"
180 /// ),
181 /// [Param::Id],
182 /// ),
183 /// Statement::new(format!("DELETE FROM {from} WHERE {id} = @p1"), [Param::Id]),
184 /// ])
185 /// }
186 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
187 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
188 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
189 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
190 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
191 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
192 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
193 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
194 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
195 /// }
196 /// ```
197 #[must_use]
198 pub const fn schema(&self) -> Option<&'a str> {
199 self.schema
200 }
201
202 /// The table's name, without its schema.
203 ///
204 /// # Examples
205 ///
206 /// ```
207 /// # use std::num::NonZeroUsize;
208 /// use ruststream_sqlx_dialect::{
209 /// Dialect, Param, Statement, StatementError, TableName, TableSpec,
210 /// };
211 ///
212 /// /// SQL Server, a database without a built-in dialect.
213 /// #[derive(Debug)]
214 /// pub struct Mssql;
215 ///
216 /// impl Dialect for Mssql {
217 /// fn name(&self) -> &'static str {
218 /// "mssql"
219 /// }
220 ///
221 /// fn dead_letter_table(
222 /// &self,
223 /// spec: &TableSpec<'_>,
224 /// target: TableName<'_>,
225 /// ) -> Result<Vec<Statement>, StatementError> {
226 /// // A struct that flattens another hides columns, so its row moves by position.
227 /// let mut columns = String::new();
228 /// if spec.selects_all() {
229 /// columns.push('*');
230 /// } else {
231 /// for column in spec.columns() {
232 /// if !columns.is_empty() {
233 /// columns.push_str(", ");
234 /// }
235 /// self.quote_into(column.name(), &mut columns);
236 /// }
237 /// }
238 /// // A target that names no schema lives in SQL Server's default one, `dbo`.
239 /// let mut into = String::new();
240 /// self.quote_into(target.schema().unwrap_or("dbo"), &mut into);
241 /// into.push('.');
242 /// self.quote_into(target.table(), &mut into);
243 /// let mut from = String::new();
244 /// self.quote_into(spec.table(), &mut from);
245 /// let mut id = String::new();
246 /// self.quote_into(spec.id().name(), &mut id);
247 /// let into_columns = if spec.selects_all() {
248 /// String::new()
249 /// } else {
250 /// format!(" ({columns})")
251 /// };
252 /// Ok(vec![
253 /// Statement::new(
254 /// format!(
255 /// "INSERT INTO {into}{into_columns} SELECT {columns} FROM {from} \
256 /// WHERE {id} = @p1"
257 /// ),
258 /// [Param::Id],
259 /// ),
260 /// Statement::new(format!("DELETE FROM {from} WHERE {id} = @p1"), [Param::Id]),
261 /// ])
262 /// }
263 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
264 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
265 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
266 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
267 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
268 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
269 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
270 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
271 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
272 /// }
273 /// ```
274 #[must_use]
275 pub const fn table(&self) -> &'a str {
276 self.table
277 }
278}
279
280/// Why a string does not name a table.
281///
282/// # Examples
283///
284/// ```
285/// use ruststream_sqlx_dialect::{ParseTableNameError, TableName};
286///
287/// // A dead-letter table read from configuration that also names a database is refused.
288/// let refused = TableName::parse("db.archive.email_jobs_dead");
289/// assert!(matches!(refused, Err(ParseTableNameError::TooManySegments { .. })));
290/// ```
291#[derive(Debug, Clone, PartialEq, Eq, Error)]
292#[non_exhaustive]
293pub enum ParseTableNameError {
294 /// The name, or one of its segments, is empty.
295 #[error("table name `{name}` has an empty segment: write `table` or `schema.table`")]
296 EmptySegment {
297 /// The name as given.
298 name: String,
299 },
300 /// The name has more than one dot.
301 #[error("table name `{name}` has more than two segments: write `table` or `schema.table`")]
302 TooManySegments {
303 /// The name as given.
304 name: String,
305 },
306}
307
308#[cfg(test)]
309mod tests {
310 use super::{ParseTableNameError, TableName};
311
312 #[test]
313 fn a_name_reads_with_or_without_a_schema() -> Result<(), ParseTableNameError> {
314 let bare = TableName::parse("jobs_dead")?;
315 assert_eq!((bare.schema(), bare.table()), (None, "jobs_dead"));
316 let qualified = TableName::parse("Archive.Jobs Dead")?;
317 assert_eq!(
318 (qualified.schema(), qualified.table()),
319 (Some("Archive"), "Jobs Dead")
320 );
321 Ok(())
322 }
323
324 #[test]
325 fn an_empty_segment_is_refused() {
326 for name in ["", ".jobs_dead", "archive.", "."] {
327 assert_eq!(
328 TableName::parse(name),
329 Err(ParseTableNameError::EmptySegment {
330 name: name.to_owned()
331 })
332 );
333 }
334 }
335
336 #[test]
337 fn more_than_one_schema_is_refused() {
338 for name in ["db.archive.jobs_dead", "archive..jobs_dead"] {
339 assert_eq!(
340 TableName::parse(name),
341 Err(ParseTableNameError::TooManySegments {
342 name: name.to_owned()
343 })
344 );
345 }
346 }
347}