camel_integration_test/
sql_action.rs

1//! Scenario `sql:` action — datasource state preparation for the
2//! integration tier (bd rc-25lup.1).
3//!
4//! Papal-locked architecture (bd rc-25lup, consult 2026-09-08): this is
5//! the scenario state branch. `sql:` seeds mutable state at rest
6//! through a datasource's pool before route assertions run; reads stay
7//! in the `validate` sql target and the two vocabularies are never
8//! mixed (Citrus precedent: fixture setup and assertions are separate
9//! actions). [`is_read_statement`] + [`validate_sql_action`] enforce
10//! that split at document load time.
11//!
12//! Executor diagnostics pass through [`sanitize_db_error`], which
13//! redacts the datasource URL per ADR-0051 (Credential Redaction at
14//! Diagnostic Boundaries): database URLs carry credential bytes and
15//! must not reach test failure output.
16
17use camel_api::CamelError;
18use camel_config::config::CamelConfig;
19use serde::Deserialize;
20
21/// Key under which a scenario action selects this vocabulary.
22pub const SQL_ACTION_KEY: &str = "sql";
23
24/// Diagnostic key for the boot-time lint that rejects per-connection
25/// sqlite `:memory:` datasource URLs (spa-3, bd rc-25lup.1).
26pub const SQL_MEMORY_NOT_SHARED: &str = "sql-memory-not-shared";
27
28/// Raw serde shape of the scenario `sql:` action as parsed from the
29/// document. Field names stay snake_case despite the `camelCase`
30/// rename (no multi-word fields today) — the attribute is load-bearing
31/// for future fields.
32#[derive(Debug, Clone, Deserialize)]
33#[serde(deny_unknown_fields, rename_all = "camelCase")]
34pub struct RawSqlAction {
35    datasource: String,
36    prepare: Vec<String>,
37}
38
39/// Validated, typed twin of [`RawSqlAction`].
40#[derive(Debug, Clone)]
41pub struct SqlAction {
42    pub datasource: String,
43    pub prepare: Vec<String>,
44}
45
46/// Whether `stmt` is a read (`select`/`with` prefix), tolerating
47/// leading whitespace and a wrapping parenthesis group. Re-trimming
48/// after each dropped `(` is what makes `"  ( select 1 )"` a read.
49pub fn is_read_statement(stmt: &str) -> bool {
50    let mut rest = stmt;
51    loop {
52        rest = rest.trim_start();
53        if let Some(without_paren) = rest.strip_prefix('(') {
54            rest = without_paren;
55        } else {
56            break;
57        }
58    }
59    let lowered = rest.to_lowercase();
60    lowered.starts_with("select") || lowered.starts_with("with")
61}
62
63/// Validates the raw shape into the typed action. Returns owned error
64/// strings; the `document.rs` hook wraps them into `DocError`.
65pub fn validate_sql_action(raw: &RawSqlAction, action_index: usize) -> Result<SqlAction, String> {
66    if raw.prepare.is_empty() {
67        return Err(format!(
68            "sql action {action_index}: prepare list must not be empty"
69        ));
70    }
71    for (i, stmt) in raw.prepare.iter().enumerate() {
72        if is_read_statement(stmt) {
73            return Err(format!(
74                "sql action {action_index}: prepare statement {i} is a read (select/with \
75                 prefix); reads belong to the validate sql target"
76            ));
77        }
78    }
79    Ok(SqlAction {
80        datasource: raw.datasource.clone(),
81        prepare: raw.prepare.clone(),
82    })
83}
84
85/// Replaces every occurrence of `db_url` in `err_text` with
86/// `[REDACTED]` (ADR-0051). An empty `db_url` is a no-op — an empty
87/// pattern would corrupt the text.
88pub fn sanitize_db_error(err_text: &str, db_url: &str) -> String {
89    if db_url.is_empty() {
90        return err_text.to_string();
91    }
92    err_text.replace(db_url, "[REDACTED]")
93}
94
95/// Boot-time lint (ungated): rejects per-connection sqlite `:memory:`
96/// datasource URLs, which give every pooled connection its own private
97/// in-memory database — an INSERT through one connection and a SELECT
98/// through another can hit different databases.
99///
100/// The remediation message names two accepted forms:
101///
102/// - Named shared-memory URI, the scenario-tier convention, e.g.
103///   `sqlite:file:memdb_appdb?mode=memory&cache=shared`. All pool
104///   connections share one named in-memory database; the name is stable
105///   across connections and parses with the sqlx driver. `sqlite:file:`
106///   matches no automatic datasource factory prefix (factories match
107///   `scheme://` or `scheme::`), so the config must also pin
108///   `provider = "sqlx"`.
109/// - Bare shared cache, `sqlite::memory:?cache=shared`. Accepted, but
110///   each parse gets a sqlx-assigned private name, and pool connections
111///   under the Any driver can hold private databases — pin
112///   `max_connections = 1` with this form.
113///
114/// Iterates `config.datasources` in BTreeMap order (the config stores a
115/// `HashMap`, whose iteration order is unspecified) so diagnostics are
116/// deterministic. The error names the datasource only, never its URL
117/// (ADR-0051 credential redaction).
118pub fn ensure_sqlite_memory_shared(config: &CamelConfig) -> Result<(), CamelError> {
119    let mut names: Vec<&String> = config.datasources.keys().collect();
120    names.sort();
121    for name in names {
122        let lowered = config.datasources[name].db_url.to_lowercase();
123        let prefix = [
124            "sqlite::memory:",
125            "sqlite://:memory:",
126            "sqlite://file::memory:",
127        ]
128        .iter()
129        .find(|p| lowered.starts_with(**p))
130        .copied();
131        if let Some(prefix) = prefix {
132            let remainder = &lowered[prefix.len()..];
133            if !remainder.contains("cache=shared") {
134                return Err(CamelError::Config(format!(
135                    "{}: datasource '{}' uses a per-connection sqlite :memory: URL without \
136                     cache=shared; INSERT and SELECT can hit different databases. Prefer the \
137                     named shared-memory URI sqlite:file:memdb_{}?mode=memory&cache=shared; \
138                     sqlite:file: matches no automatic datasource prefix, so also set \
139                     provider = \"sqlx\". The bare sqlite::memory:?cache=shared URL is also \
140                     accepted, but cross-connection sharing is not guaranteed; set \
141                     max_connections = 1 so state stays on one connection",
142                    SQL_MEMORY_NOT_SHARED, name, name
143                )));
144            }
145        }
146    }
147    Ok(())
148}
149
150#[cfg(feature = "sql")]
151use std::sync::Arc;
152
153#[cfg(feature = "sql")]
154use camel_api::datasource::DatasourceCatalog;
155
156/// Executes the prepare statements in order against the named
157/// datasource's pool. Stops at the first failure, reporting the
158/// statement index and a sanitized error; never emits the datasource
159/// URL.
160#[cfg(feature = "sql")]
161pub async fn execute_sql_prepare(
162    catalog: &Arc<dyn DatasourceCatalog>,
163    action: &SqlAction,
164) -> Result<(), String> {
165    let name = &action.datasource;
166    let Some(config) = catalog.get_config(name) else {
167        return Err(format!("sql action: unknown datasource '{name}'"));
168    };
169    let handle = catalog.get_pool(name).await.map_err(|e| {
170        format!(
171            "sql action: datasource '{name}': {}",
172            sanitize_db_error(&e.to_string(), &config.db_url)
173        )
174    })?;
175    let pool = handle.downcast::<sqlx::AnyPool>().map_err(|e| {
176        format!(
177            "sql action: datasource '{name}': {}",
178            sanitize_db_error(&e.to_string(), &config.db_url)
179        )
180    })?;
181    for (i, stmt) in action.prepare.iter().enumerate() {
182        sqlx::query(stmt).execute(&*pool).await.map_err(|e| {
183            format!(
184                "datasource '{name}' statement [{i}]: {}",
185                sanitize_db_error(&e.to_string(), &config.db_url)
186            )
187        })?;
188    }
189    Ok(())
190}
191
192#[cfg(test)]
193mod tests {
194    use std::collections::HashMap;
195
196    use super::*;
197    use camel_api::datasource::DatasourceConfig;
198
199    fn lint_config(db_url: &str) -> CamelConfig {
200        let mut config = CamelConfig::default();
201        config.datasources.insert(
202            "appdb".to_string(),
203            DatasourceConfig {
204                db_url: db_url.to_string(),
205                provider: None,
206                max_connections: None,
207                min_connections: None,
208                idle_timeout_secs: None,
209                max_lifetime_secs: None,
210                ssl_mode: None,
211                ssl_root_cert: None,
212                ssl_cert: None,
213                ssl_key: None,
214                extra: HashMap::new(),
215            },
216        );
217        config
218    }
219
220    fn raw(datasource: &str, prepare: Vec<&str>) -> RawSqlAction {
221        RawSqlAction {
222            datasource: datasource.to_string(),
223            prepare: prepare.into_iter().map(str::to_string).collect(),
224        }
225    }
226
227    #[test]
228    fn read_statement_detection() {
229        assert!(is_read_statement("SELECT 1"));
230        assert!(is_read_statement("  ( select 1 )"));
231        assert!(is_read_statement("With x AS (SELECT 1) SELECT * FROM x"));
232        assert!(!is_read_statement("INSERT INTO t VALUES (1)"));
233        assert!(!is_read_statement("(CREATE TABLE t (x INTEGER))"));
234    }
235
236    #[test]
237    fn validation_rejects_empty_prepare() {
238        let err = validate_sql_action(&raw("appdb", vec![]), 3).unwrap_err();
239        assert!(err.contains("sql action 3"), "got: {err}");
240        assert!(err.contains("prepare list must not be empty"), "got: {err}");
241    }
242
243    #[test]
244    fn validation_rejects_read_prefixes() {
245        let err = validate_sql_action(&raw("appdb", vec!["(SELECT 1)"]), 0).unwrap_err();
246        assert!(err.contains("statement 0"), "got: {err}");
247
248        let err = validate_sql_action(
249            &raw(
250                "appdb",
251                vec![
252                    "CREATE TABLE t (x INTEGER)",
253                    "with cte as (select 1) select * from cte",
254                ],
255            ),
256            0,
257        )
258        .unwrap_err();
259        assert!(err.contains("statement 1"), "got: {err}");
260    }
261
262    #[test]
263    fn validation_accepts_mutations() {
264        let action = validate_sql_action(
265            &raw(
266                "appdb",
267                vec![
268                    "CREATE TABLE t (x INTEGER)",
269                    "INSERT INTO t VALUES (1)",
270                    "DELETE FROM t WHERE x = 1",
271                ],
272            ),
273            0,
274        )
275        .unwrap();
276        assert_eq!(action.datasource, "appdb");
277        assert_eq!(
278            action.prepare,
279            vec![
280                "CREATE TABLE t (x INTEGER)".to_string(),
281                "INSERT INTO t VALUES (1)".to_string(),
282                "DELETE FROM t WHERE x = 1".to_string(),
283            ]
284        );
285    }
286
287    #[test]
288    fn sanitizer_redacts_db_url() {
289        let sanitized = sanitize_db_error(
290            "connect failed sqlite::memory:?cache=shared&x=1",
291            "sqlite::memory:?cache=shared&x=1",
292        );
293        assert!(sanitized.contains("[REDACTED]"), "got: {sanitized}");
294        assert!(!sanitized.contains("cache=shared&x=1"), "got: {sanitized}");
295    }
296
297    #[test]
298    fn sanitizer_empty_url_noop() {
299        assert_eq!(sanitize_db_error("boom", ""), "boom");
300    }
301
302    #[test]
303    fn lint_message_steers_to_named_form() {
304        let err = ensure_sqlite_memory_shared(&lint_config("sqlite::memory:")).unwrap_err();
305        let text = err.to_string();
306        assert!(text.contains(SQL_MEMORY_NOT_SHARED), "got: {text}");
307        assert!(
308            text.contains("sqlite:file:memdb_appdb?mode=memory&cache=shared"),
309            "got: {text}"
310        );
311        assert!(text.contains("provider"), "got: {text}");
312        assert!(text.contains("max_connections = 1"), "got: {text}");
313        assert!(!text.contains("<name>"), "got: {text}");
314        assert!(!text.contains("<scenario>"), "got: {text}");
315    }
316
317    #[test]
318    fn lint_accepts_bare_shared_form() {
319        let config = lint_config("sqlite::memory:?cache=shared");
320        assert!(ensure_sqlite_memory_shared(&config).is_ok());
321    }
322
323    #[test]
324    fn lint_named_file_uri_passes() {
325        let config = lint_config("sqlite:file:memdb_appdb?mode=memory&cache=shared");
326        assert!(ensure_sqlite_memory_shared(&config).is_ok());
327    }
328
329    #[test]
330    fn lint_rejects_file_memory_uri_without_shared_cache() {
331        let config = lint_config("sqlite://file::memory:?mode=memory");
332        let err = ensure_sqlite_memory_shared(&config).unwrap_err();
333        assert!(err.to_string().contains(SQL_MEMORY_NOT_SHARED));
334    }
335
336    #[test]
337    fn lint_accepts_file_memory_uri_with_shared_cache() {
338        let config = lint_config("sqlite://file::memory:?mode=memory&cache=shared");
339        assert!(ensure_sqlite_memory_shared(&config).is_ok());
340    }
341}
camel_integration_test/sql_action.rs

camel_integration_test/
sql_action.rs
