sofie_db/
seam.rs

1use std::fmt;
2use std::num::NonZeroUsize;
3use std::pin::Pin;
4use std::sync::Arc;
5use std::time::Duration;
6
7use sofie_core::{ColumnReader, Error, Result, SqlType, SqlValue};
8
9use crate::{GuardCheck, MessageSql, Migrations};
10
11pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
12
13/// Long enough for a burst to pass, well short of a request's own timeout.
14const DEFAULT_POOL_WAIT: Duration = Duration::from_secs(5);
15
16#[derive(Clone, Copy, Debug, PartialEq, Eq)]
17pub struct PoolSettings {
18    /// Twice the logical CPUs when `None`.
19    pub connections: Option<NonZeroUsize>,
20    pub wait: Duration,
21}
22
23impl Default for PoolSettings {
24    fn default() -> Self {
25        Self {
26            connections: None,
27            wait: DEFAULT_POOL_WAIT,
28        }
29    }
30}
31
32#[derive(Debug)]
33pub enum Step {
34    /// A mismatched `expect_rows` rolls back with its error; a check costs a round trip.
35    Statement {
36        sql: String,
37        params: Vec<SqlValue>,
38        expect_rows: Option<ExpectedRows>,
39    },
40    /// Trusted SQL only (migrations, framework setup), never input.
41    Script(String),
42}
43
44impl Step {
45    /// A mismatched `expect_rows` is a `Conflict`.
46    #[must_use]
47    pub fn statement(
48        sql: impl Into<String>,
49        params: Vec<SqlValue>,
50        expect_rows: Option<u64>,
51    ) -> Self {
52        Self::Statement {
53            sql: sql.into(),
54            params,
55            expect_rows: expect_rows.map(ExpectedRows::or_conflict),
56        }
57    }
58}
59
60#[derive(Clone, Debug, PartialEq, Eq)]
61pub struct ExpectedRows {
62    pub rows: u64,
63    pub otherwise: Unmet,
64}
65
66/// What a statement that touched another number of rows than expected fails with.
67#[derive(Clone, Debug, PartialEq, Eq)]
68pub enum Unmet {
69    Conflict,
70    NotFound,
71    Refused(Arc<GuardCheck>),
72}
73
74impl ExpectedRows {
75    #[must_use]
76    pub const fn or_conflict(rows: u64) -> Self {
77        Self {
78            rows,
79            otherwise: Unmet::Conflict,
80        }
81    }
82
83    #[must_use]
84    pub const fn or_not_found(rows: u64) -> Self {
85        Self {
86            rows,
87            otherwise: Unmet::NotFound,
88        }
89    }
90
91    pub fn check(&self, affected: u64) -> Result<()> {
92        let met = affected == self.rows;
93        match &self.otherwise {
94            Unmet::Refused(check) => check.checked(!met),
95            _ if met => Ok(()),
96            Unmet::Conflict => Err(Error::Conflict),
97            Unmet::NotFound => Err(Error::NotFound),
98        }
99    }
100}
101
102pub trait RowSink: Send {
103    /// Checked once before any row is read; `None` takes any type, fewer columns are refused.
104    fn expects(&self) -> &[Option<SqlType>];
105
106    fn row(&mut self, row: &mut dyn ColumnReader) -> Result<()>;
107}
108
109pub trait Driver: Send + Sync + 'static {
110    fn dialect(&self) -> &dyn Dialect;
111
112    fn query<'a>(
113        &'a self,
114        sql: &'a str,
115        params: &'a [SqlValue],
116        rows: &'a mut dyn RowSink,
117    ) -> BoxFuture<'a, Result<()>>;
118
119    /// Read-only, as nothing else keeps raw SQL from writing.
120    #[cfg(feature = "raw-sql")]
121    fn query_read_only<'a>(
122        &'a self,
123        role: Option<&'a str>,
124        sql: &'a str,
125        params: &'a [SqlValue],
126        rows: &'a mut dyn RowSink,
127    ) -> BoxFuture<'a, Result<()>> {
128        let _ = (role, params, rows);
129        crate::raw::refuse_read_only(sql)
130    }
131
132    /// A unique-index refusal fails as `Internal` with a [`UniqueViolation`] source; a check,
133    /// exclusion or foreign key as `Rule` ([`check_refusal`], [`exclusion_refusal`], …).
134    fn transaction(&self, steps: Vec<Step>) -> BoxFuture<'_, Result<()>>;
135
136    /// Held over a whole `Schema::migrate` until dropped; else each migration's transaction locks.
137    fn lock_migrations(&self) -> Option<BoxFuture<'_, Result<Box<dyn MigrationsLock>>>> {
138        None
139    }
140
141    /// Outside a transaction, under `lock_migrations`; other migrations get `record` back.
142    #[cfg(feature = "concurrent-indexes")]
143    fn migrate_alone<'a>(
144        &'a self,
145        schema: &'a str,
146        sql: &'a str,
147        record: Step,
148    ) -> std::result::Result<BoxFuture<'a, Result<()>>, Step> {
149        let _ = (schema, sql);
150        Err(record)
151    }
152
153    fn catalog(&self) -> Option<&dyn crate::Catalog> {
154        None
155    }
156
157    fn wakeups(&self) -> Option<BoxFuture<'_, Result<Box<dyn Wakeups>>>> {
158        None
159    }
160
161    fn flag_wakeups(&self) -> Option<BoxFuture<'_, Result<Box<dyn Wakeups>>>> {
162        None
163    }
164
165    #[cfg(feature = "shield")]
166    fn shield_wakeups(&self) -> Option<BoxFuture<'_, Result<Box<dyn Wakeups>>>> {
167        None
168    }
169
170    /// The payloads sent to the `sofie_cache` channel, such as by `pg_notify`.
171    #[cfg(feature = "cache")]
172    fn cache_notifications(&self) -> Option<BoxFuture<'_, Result<Box<dyn Notifications>>>> {
173        None
174    }
175
176    /// The payloads `Dialect::live_notify` sent.
177    #[cfg(feature = "live")]
178    fn change_notifications(&self) -> Option<BoxFuture<'_, Result<Box<dyn Notifications>>>> {
179        None
180    }
181}
182
183#[derive(Debug)]
184pub struct UniqueViolation {
185    pub index: String,
186}
187
188impl fmt::Display for UniqueViolation {
189    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
190        write!(formatter, "unique index {} refused the write", self.index)
191    }
192}
193
194impl std::error::Error for UniqueViolation {}
195
196#[must_use]
197pub fn check_refusal(constraint: &str) -> String {
198    format!("a value breaks the rule `{constraint}`")
199}
200
201/// SQLite names no foreign key it refuses, so neither engine's refusal does.
202pub const REFERENCE_REFUSAL: &str =
203    "the write names a record that does not exist, or removes one another record still names";
204
205#[must_use]
206pub fn exclusion_refusal(constraint: &str) -> String {
207    format!("a value overlaps one another record holds, under the rule `{constraint}`")
208}
209
210#[derive(Clone, Copy, Debug, PartialEq, Eq)]
211pub enum Engine {
212    Postgres,
213    Sqlite,
214}
215
216/// Rendered SQL is cached by the dialect's type, so it must not vary per instance.
217pub trait Dialect: Send + Sync + std::any::Any {
218    /// `index` counts from 1.
219    fn push_placeholder(&self, sql: &mut String, index: usize);
220
221    fn push_quoted(&self, sql: &mut String, identifier: &str) {
222        sql.push('"');
223        for character in identifier.chars() {
224            if character == '"' {
225                sql.push('"');
226            }
227            sql.push(character);
228        }
229        sql.push('"');
230    }
231
232    fn placeholder(&self, index: usize) -> String {
233        let mut sql = String::new();
234        self.push_placeholder(&mut sql, index);
235        sql
236    }
237
238    fn quote(&self, identifier: &str) -> String {
239        let mut sql = String::with_capacity(identifier.len() + 2);
240        self.push_quoted(&mut sql, identifier);
241        sql
242    }
243
244    /// Which of a module's migration files apply: SQLite's where it has its own.
245    fn engine(&self) -> Engine {
246        Engine::Postgres
247    }
248
249    /// Locks its one text parameter until the transaction ends; `None` where commits queue anyway.
250    fn key_lock(&self) -> Option<&'static str> {
251        None
252    }
253
254    fn shared_key_lock(&self) -> Option<&'static str> {
255        None
256    }
257
258    /// A module's table: in its schema, or named after it where the database has no schemas.
259    fn qualified(&self, schema: &str, name: &str) -> String {
260        format!("{}.{}", self.quote(schema), self.quote(name))
261    }
262
263    /// A bytes column as the JSON of loaded child rows holds it: `\x` and hex.
264    fn bytes_in_json(&self, column: &str) -> String {
265        column.to_owned()
266    }
267
268    fn nested_rows(&self, columns: &str, from: &str) -> String {
269        panic!("this database cannot load child rows, so `{columns}` of `{from}` cannot be read")
270    }
271
272    fn binds_arrays(&self) -> bool {
273        false
274    }
275
276    fn migrations_setup(&self) -> &'static str;
277
278    fn migrations_lock(&self) -> &'static str;
279
280    fn enter_schema(&self, schema: &str) -> String;
281
282    /// Answers, one per row, the statements `Schema::fresh` drops every table with.
283    fn drop_every_table(&self) -> Option<&'static str> {
284        None
285    }
286
287    /// Without full-text search, the words are found as one piece of text, ignoring case.
288    #[cfg(feature = "search")]
289    fn text_search(&self, column: &str, words: &str) -> String {
290        let (column, words) = (self.lowercase(column), self.lowercase(words));
291        let escaped =
292            format!(r"replace(replace(replace({words}, '\', '\\'), '%', '\%'), '_', '\_')");
293        format!(r"({column} LIKE '%' || {escaped} || '%' ESCAPE '\')")
294    }
295
296    fn lowercase(&self, expression: &str) -> String {
297        format!("lower({expression})")
298    }
299
300    /// After a nullable column's direction: empty values sort as Postgres's do, last ascending.
301    fn empty_order(&self, descending: bool) -> &'static str {
302        let _ = descending;
303        ""
304    }
305
306    /// `None` returns to the app's login.
307    #[cfg(feature = "module-roles")]
308    fn enter_role(&self, module: Option<&str>) -> Option<String> {
309        let _ = module;
310        None
311    }
312
313    fn framework_migrations(&self) -> Migrations;
314
315    fn messages(&self) -> &'static MessageSql;
316
317    /// Committed with a change to the stored flags or settings.
318    fn flags_notify(&self) -> Option<&'static str> {
319        None
320    }
321
322    #[cfg(feature = "shield")]
323    fn shield(&self) -> Option<&'static crate::ShieldSql> {
324        None
325    }
326
327    #[cfg(feature = "remote-contracts")]
328    fn feed(&self) -> Option<&'static crate::FeedSql> {
329        None
330    }
331
332    #[cfg(feature = "product-reports")]
333    fn product_events(&self) -> Option<&'static crate::ProductSql> {
334        None
335    }
336
337    /// Sends its one text parameter to every instance's `Driver::change_notifications`.
338    #[cfg(feature = "live")]
339    fn live_notify(&self) -> Option<&'static str> {
340        None
341    }
342}
343
344pub trait MigrationsLock: Send {}
345
346pub trait Wakeups: Send {
347    fn next(&mut self) -> BoxFuture<'_, Result<()>>;
348}
349
350#[cfg(any(feature = "cache", feature = "live"))]
351pub trait Notifications: Send {
352    /// Every payload buffered so far, waiting for at least one.
353    fn next(&mut self) -> BoxFuture<'_, Result<Vec<String>>>;
354}
sofie_db/seam.rs

sofie_db/
seam.rs
