camel_api/
datasource.rs

1use std::any::Any;
2use std::collections::HashMap;
3use std::fmt;
4use std::future::Future;
5use std::pin::Pin;
6use std::sync::Arc;
7
8use serde::{Deserialize, Serialize};
9
10use crate::error::CamelError;
11use crate::lifecycle::HealthStatus;
12
13#[derive(Clone, Deserialize, Serialize, PartialEq)]
14#[serde(deny_unknown_fields)]
15pub struct DatasourceConfig {
16    pub db_url: String,
17    #[serde(default)]
18    pub provider: Option<String>,
19    #[serde(default)]
20    pub max_connections: Option<u32>,
21    #[serde(default)]
22    pub min_connections: Option<u32>,
23    #[serde(default)]
24    pub idle_timeout_secs: Option<u64>,
25    #[serde(default)]
26    pub max_lifetime_secs: Option<u64>,
27    #[serde(default)]
28    pub ssl_mode: Option<String>,
29    #[serde(default)]
30    pub ssl_root_cert: Option<String>,
31    #[serde(default)]
32    pub ssl_cert: Option<String>,
33    #[serde(default)]
34    pub ssl_key: Option<String>,
35    /// Generic key-value pairs for database-specific configuration.
36    /// Components read their bespoke fields from here.
37    /// SQL ignores this; SurrealDB reads namespace/database/username/password.
38    #[serde(default)]
39    pub extra: HashMap<String, toml::Value>,
40}
41
42impl fmt::Debug for DatasourceConfig {
43    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
44        f.debug_struct("DatasourceConfig")
45            .field("db_url", &"[REDACTED]")
46            .field("provider", &self.provider)
47            .field("max_connections", &self.max_connections)
48            .field("min_connections", &self.min_connections)
49            .field("idle_timeout_secs", &self.idle_timeout_secs)
50            .field("max_lifetime_secs", &self.max_lifetime_secs)
51            .field("ssl_mode", &self.ssl_mode)
52            .field("ssl_root_cert", &self.ssl_root_cert)
53            .field("ssl_cert", &self.ssl_cert)
54            .field("ssl_key", &self.ssl_key.as_ref().map(|_| "[REDACTED]"))
55            .field("extra", &"[REDACTED]")
56            .finish()
57    }
58}
59
60impl DatasourceConfig {
61    pub fn validate(&self) -> Result<(), CamelError> {
62        if self.db_url.trim().is_empty() {
63            return Err(CamelError::Config(
64                "datasource db_url cannot be empty".into(),
65            ));
66        }
67        Ok(())
68    }
69}
70
71#[derive(Clone)]
72pub struct DatasourceHandle {
73    pub name: String,
74    pub provider: String,
75    inner: Arc<dyn Any + Send + Sync>,
76}
77
78impl DatasourceHandle {
79    pub fn new(name: String, provider: String, inner: Arc<dyn Any + Send + Sync>) -> Self {
80        Self {
81            name,
82            provider,
83            inner,
84        }
85    }
86
87    pub fn downcast<T: 'static + Send + Sync>(&self) -> Result<Arc<T>, CamelError> {
88        self.inner.clone().downcast::<T>().map_err(|_| {
89            CamelError::ProcessorError(format!(
90                "datasource '{}' (provider '{}'): failed to downcast handle",
91                self.name, self.provider
92            ))
93        })
94    }
95}
96
97impl fmt::Debug for DatasourceHandle {
98    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
99        f.debug_struct("DatasourceHandle")
100            .field("name", &self.name)
101            .field("provider", &self.provider)
102            .finish()
103    }
104}
105
106#[doc(hidden)]
107pub struct ResourceRef {
108    pub kind: String,
109    pub name: String,
110}
111
112impl fmt::Debug for ResourceRef {
113    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
114        f.debug_struct("ResourceRef")
115            .field("kind", &self.kind)
116            .field("name", &self.name)
117            .finish()
118    }
119}
120
121pub type CreatePoolResult = Result<Arc<dyn Any + Send + Sync>, CamelError>;
122pub type CreatePoolFuture<'a> = Pin<Box<dyn Future<Output = CreatePoolResult> + Send + 'a>>;
123pub type CheckFuture<'a> = Pin<Box<dyn Future<Output = HealthStatus> + Send + 'a>>;
124pub type CloseFuture<'a> = Pin<Box<dyn Future<Output = Result<(), CamelError>> + Send + 'a>>;
125
126pub trait PoolFactory: Send + Sync + 'static {
127    fn create<'a>(&'a self, config: &'a DatasourceConfig) -> CreatePoolFuture<'a>;
128
129    fn check<'a>(&'a self, handle: &'a DatasourceHandle) -> CheckFuture<'a>;
130
131    /// Close a handle this factory created. The default is a no-op so
132    /// providers without an explicit close keep compiling; providers that
133    /// own pools (sqlx) override it so teardown drains their connections.
134    /// Close MAY run more than once per handle and MUST stay safe to
135    /// re-run (idempotent by contract).
136    fn close<'a>(&'a self, _handle: &'a DatasourceHandle) -> CloseFuture<'a> {
137        Box::pin(async { Ok(()) })
138    }
139
140    fn supported_schemes(&self) -> &[&str];
141
142    fn matches(&self, config: &DatasourceConfig) -> bool {
143        self.supported_schemes().iter().any(|s| {
144            config.db_url.starts_with(&format!("{}://", s))
145                || config.db_url.starts_with(&format!("{}::", s))
146        })
147    }
148
149    fn name(&self) -> &'static str;
150}
151
152pub type GetPoolFuture<'a> =
153    Pin<Box<dyn Future<Output = Result<DatasourceHandle, CamelError>> + Send + 'a>>;
154pub type CloseAllFuture<'a> = Pin<Box<dyn Future<Output = Result<(), CamelError>> + Send + 'a>>;
155
156pub trait DatasourceCatalog: Send + Sync {
157    fn get_config(&self, name: &str) -> Option<DatasourceConfig>;
158    fn get_pool<'a>(&'a self, name: &'a str) -> GetPoolFuture<'a>;
159    fn register_factory(&self, kind: &str, factory: Arc<dyn PoolFactory>)
160    -> Result<(), CamelError>;
161
162    /// Close every initialized pool. The default is a no-op; the runtime
163    /// catalog overrides it so a boot teardown deterministically drains its
164    /// datasource connections (bd rc-25lup.4). Close MAY run again after a
165    /// completed run; implementors keep it safe to re-run.
166    fn close_all(&self) -> CloseAllFuture<'_> {
167        Box::pin(async { Ok(()) })
168    }
169}
170
171#[cfg(test)]
172mod tests {
173    use super::*;
174
175    #[test]
176    fn datasource_config_validate_rejects_empty_db_url() {
177        let config = DatasourceConfig {
178            db_url: "".to_string(),
179            provider: None,
180            max_connections: None,
181            min_connections: None,
182            idle_timeout_secs: None,
183            max_lifetime_secs: None,
184            ssl_mode: None,
185            ssl_root_cert: None,
186            ssl_cert: None,
187            ssl_key: None,
188            extra: HashMap::new(),
189        };
190        let result = config.validate();
191        assert!(result.is_err());
192        assert!(result.unwrap_err().to_string().contains("empty"));
193    }
194
195    #[test]
196    fn datasource_config_validate_accepts_valid() {
197        let config = DatasourceConfig {
198            db_url: "postgresql://localhost:5432/mydb".to_string(),
199            provider: None,
200            max_connections: Some(10),
201            min_connections: Some(2),
202            idle_timeout_secs: Some(300),
203            max_lifetime_secs: Some(1800),
204            ssl_mode: None,
205            ssl_root_cert: None,
206            ssl_cert: None,
207            ssl_key: None,
208            extra: HashMap::new(),
209        };
210        assert!(config.validate().is_ok());
211    }
212
213    #[test]
214    fn datasource_config_debug_redacts_db_url() {
215        let config = DatasourceConfig {
216            db_url: "postgresql://user:pass@localhost:5432/mydb".to_string(),
217            provider: None,
218            max_connections: None,
219            min_connections: None,
220            idle_timeout_secs: None,
221            max_lifetime_secs: None,
222            ssl_mode: None,
223            ssl_root_cert: None,
224            ssl_cert: None,
225            ssl_key: None,
226            extra: HashMap::new(),
227        };
228        let debug_str = format!("{:?}", config);
229        assert!(
230            debug_str.contains("[REDACTED]"),
231            "Debug output should redact db_url: {}",
232            debug_str
233        );
234        assert!(
235            !debug_str.contains("user:pass"),
236            "Debug output should not contain credentials: {}",
237            debug_str
238        );
239    }
240
241    #[test]
242    fn datasource_handle_downcast_fails_on_wrong_type() {
243        let handle = DatasourceHandle::new("test".to_string(), "mock".to_string(), Arc::new(42u32));
244        let result: Result<Arc<String>, CamelError> = handle.downcast();
245        assert!(result.is_err());
246        let err = result.unwrap_err();
247        assert!(err.to_string().contains("failed to downcast"));
248    }
249
250    #[test]
251    fn pool_factory_matches_by_scheme() {
252        struct PostgresFactory;
253        impl PoolFactory for PostgresFactory {
254            fn create<'a>(&'a self, _config: &'a DatasourceConfig) -> CreatePoolFuture<'a> {
255                Box::pin(async { Ok(Arc::new("pool") as Arc<dyn Any + Send + Sync>) })
256            }
257            fn check<'a>(&'a self, _handle: &'a DatasourceHandle) -> CheckFuture<'a> {
258                Box::pin(async { HealthStatus::Healthy })
259            }
260            fn supported_schemes(&self) -> &[&str] {
261                &["postgresql", "postgres"]
262            }
263            fn name(&self) -> &'static str {
264                "postgres"
265            }
266        }
267
268        let factory = PostgresFactory;
269        let pg_config = DatasourceConfig {
270            db_url: "postgresql://localhost/mydb".to_string(),
271            provider: None,
272            max_connections: None,
273            min_connections: None,
274            idle_timeout_secs: None,
275            max_lifetime_secs: None,
276            ssl_mode: None,
277            ssl_root_cert: None,
278            ssl_cert: None,
279            ssl_key: None,
280            extra: HashMap::new(),
281        };
282        assert!(factory.matches(&pg_config));
283
284        let mysql_config = DatasourceConfig {
285            db_url: "mysql://localhost/mydb".to_string(),
286            provider: None,
287            max_connections: None,
288            min_connections: None,
289            idle_timeout_secs: None,
290            max_lifetime_secs: None,
291            ssl_mode: None,
292            ssl_root_cert: None,
293            ssl_cert: None,
294            ssl_key: None,
295            extra: HashMap::new(),
296        };
297        assert!(!factory.matches(&mysql_config));
298    }
299
300    #[test]
301    fn datasource_config_extra_defaults_empty() {
302        let config = DatasourceConfig {
303            db_url: "ws://localhost:8000".to_string(),
304            provider: None,
305            max_connections: None,
306            min_connections: None,
307            idle_timeout_secs: None,
308            max_lifetime_secs: None,
309            ssl_mode: None,
310            ssl_root_cert: None,
311            ssl_cert: None,
312            ssl_key: None,
313            extra: HashMap::new(),
314        };
315        assert!(config.extra.is_empty());
316    }
317
318    #[test]
319    fn datasource_config_extra_deserializes_from_toml() {
320        let toml_str = r#"
321db_url = "ws://localhost:8000"
322provider = "surrealdb"
323
324[extra]
325namespace = "camel"
326database = "runtime"
327"#;
328        let config: DatasourceConfig = toml::from_str(toml_str).unwrap(); // allow-unwrap
329        assert_eq!(config.db_url, "ws://localhost:8000");
330        assert_eq!(config.provider.as_deref(), Some("surrealdb"));
331        assert_eq!(config.extra.len(), 2);
332        assert_eq!(
333            config.extra.get("namespace").and_then(|v| v.as_str()),
334            Some("camel")
335        );
336    }
337
338    #[test]
339    fn datasource_config_extra_backward_compat_without_extra_block() {
340        let toml_str = r#"
341db_url = "ws://localhost:8000"
342"#;
343        let config: DatasourceConfig = toml::from_str(toml_str).unwrap(); // allow-unwrap
344        assert_eq!(config.db_url, "ws://localhost:8000");
345        assert!(
346            config.extra.is_empty(),
347            "TOML without [extra] must deserialize with empty extra (#[serde(default)])"
348        );
349    }
350
351    #[test]
352    fn datasource_config_debug_redacts_extra() {
353        let mut extra = HashMap::new();
354        extra.insert(
355            "password".to_string(),
356            toml::Value::String("secret123".to_string()),
357        );
358        let config = DatasourceConfig {
359            db_url: "ws://localhost:8000".to_string(),
360            provider: None,
361            max_connections: None,
362            min_connections: None,
363            idle_timeout_secs: None,
364            max_lifetime_secs: None,
365            ssl_mode: None,
366            ssl_root_cert: None,
367            ssl_cert: None,
368            ssl_key: None,
369            extra,
370        };
371        let debug_str = format!("{:?}", config);
372        assert!(
373            debug_str.contains("[REDACTED]"),
374            "extra should be redacted in Debug: {}",
375            debug_str
376        );
377        assert!(
378            !debug_str.contains("secret123"),
379            "password must not appear in Debug: {}",
380            debug_str
381        );
382    }
383}
camel_api/datasource.rs

camel_api/
datasource.rs
