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. The fix is the shared
99/// cache (`?cache=shared`), which makes all connections in the process
100/// share one in-memory database.
101///
102/// Iterates `config.datasources` in BTreeMap order (the config stores a
103/// `HashMap`, whose iteration order is unspecified) so diagnostics are
104/// deterministic. The error names the datasource only, never its URL
105/// (ADR-0051 credential redaction).
106pub fn ensure_sqlite_memory_shared(config: &CamelConfig) -> Result<(), CamelError> {
107    let mut names: Vec<&String> = config.datasources.keys().collect();
108    names.sort();
109    for name in names {
110        let lowered = config.datasources[name].db_url.to_lowercase();
111        let prefix = ["sqlite::memory:", "sqlite://:memory:"]
112            .iter()
113            .find(|p| lowered.starts_with(**p))
114            .copied();
115        if let Some(prefix) = prefix {
116            let remainder = &lowered[prefix.len()..];
117            if !remainder.contains("cache=shared") {
118                return Err(CamelError::Config(format!(
119                    "{}: datasource '{}' uses a per-connection sqlite :memory: URL without \
120                     cache=shared; INSERT and SELECT can hit different databases. Use \
121                     sqlite::memory:?cache=shared",
122                    SQL_MEMORY_NOT_SHARED, name
123                )));
124            }
125        }
126    }
127    Ok(())
128}
129
130#[cfg(feature = "sql")]
131use std::sync::Arc;
132
133#[cfg(feature = "sql")]
134use camel_api::datasource::DatasourceCatalog;
135
136/// Executes the prepare statements in order against the named
137/// datasource's pool. Stops at the first failure, reporting the
138/// statement index and a sanitized error; never emits the datasource
139/// URL.
140#[cfg(feature = "sql")]
141pub async fn execute_sql_prepare(
142    catalog: &Arc<dyn DatasourceCatalog>,
143    action: &SqlAction,
144) -> Result<(), String> {
145    let name = &action.datasource;
146    let Some(config) = catalog.get_config(name) else {
147        return Err(format!("sql action: unknown datasource '{name}'"));
148    };
149    let handle = catalog.get_pool(name).await.map_err(|e| {
150        format!(
151            "sql action: datasource '{name}': {}",
152            sanitize_db_error(&e.to_string(), &config.db_url)
153        )
154    })?;
155    let pool = handle.downcast::<sqlx::AnyPool>().map_err(|e| {
156        format!(
157            "sql action: datasource '{name}': {}",
158            sanitize_db_error(&e.to_string(), &config.db_url)
159        )
160    })?;
161    for (i, stmt) in action.prepare.iter().enumerate() {
162        sqlx::query(stmt).execute(&*pool).await.map_err(|e| {
163            format!(
164                "datasource '{name}' statement [{i}]: {}",
165                sanitize_db_error(&e.to_string(), &config.db_url)
166            )
167        })?;
168    }
169    Ok(())
170}
171
172#[cfg(test)]
173mod tests {
174    use super::*;
175
176    fn raw(datasource: &str, prepare: Vec<&str>) -> RawSqlAction {
177        RawSqlAction {
178            datasource: datasource.to_string(),
179            prepare: prepare.into_iter().map(str::to_string).collect(),
180        }
181    }
182
183    #[test]
184    fn read_statement_detection() {
185        assert!(is_read_statement("SELECT 1"));
186        assert!(is_read_statement("  ( select 1 )"));
187        assert!(is_read_statement("With x AS (SELECT 1) SELECT * FROM x"));
188        assert!(!is_read_statement("INSERT INTO t VALUES (1)"));
189        assert!(!is_read_statement("(CREATE TABLE t (x INTEGER))"));
190    }
191
192    #[test]
193    fn validation_rejects_empty_prepare() {
194        let err = validate_sql_action(&raw("appdb", vec![]), 3).unwrap_err();
195        assert!(err.contains("sql action 3"), "got: {err}");
196        assert!(err.contains("prepare list must not be empty"), "got: {err}");
197    }
198
199    #[test]
200    fn validation_rejects_read_prefixes() {
201        let err = validate_sql_action(&raw("appdb", vec!["(SELECT 1)"]), 0).unwrap_err();
202        assert!(err.contains("statement 0"), "got: {err}");
203
204        let err = validate_sql_action(
205            &raw(
206                "appdb",
207                vec![
208                    "CREATE TABLE t (x INTEGER)",
209                    "with cte as (select 1) select * from cte",
210                ],
211            ),
212            0,
213        )
214        .unwrap_err();
215        assert!(err.contains("statement 1"), "got: {err}");
216    }
217
218    #[test]
219    fn validation_accepts_mutations() {
220        let action = validate_sql_action(
221            &raw(
222                "appdb",
223                vec![
224                    "CREATE TABLE t (x INTEGER)",
225                    "INSERT INTO t VALUES (1)",
226                    "DELETE FROM t WHERE x = 1",
227                ],
228            ),
229            0,
230        )
231        .unwrap();
232        assert_eq!(action.datasource, "appdb");
233        assert_eq!(
234            action.prepare,
235            vec![
236                "CREATE TABLE t (x INTEGER)".to_string(),
237                "INSERT INTO t VALUES (1)".to_string(),
238                "DELETE FROM t WHERE x = 1".to_string(),
239            ]
240        );
241    }
242
243    #[test]
244    fn sanitizer_redacts_db_url() {
245        let sanitized = sanitize_db_error(
246            "connect failed sqlite::memory:?cache=shared&x=1",
247            "sqlite::memory:?cache=shared&x=1",
248        );
249        assert!(sanitized.contains("[REDACTED]"), "got: {sanitized}");
250        assert!(!sanitized.contains("cache=shared&x=1"), "got: {sanitized}");
251    }
252
253    #[test]
254    fn sanitizer_empty_url_noop() {
255        assert_eq!(sanitize_db_error("boom", ""), "boom");
256    }
257}
camel_integration_test/sql_action.rs

camel_integration_test/
sql_action.rs
