1use std::str::FromStr;
2use std::time::Duration;
3
4use camel_component_api::CamelError;
5use camel_component_api::NetworkRetryPolicy;
6use camel_component_api::{UriComponents, UriConfig, parse_uri};
7use tracing::warn;
8
9fn redacted_opt(opt: &Option<String>) -> Option<&'static str> {
11 if opt.is_some() { Some("***") } else { None }
12}
13
14pub fn redact_db_url(db_url: &str) -> String {
17 match url::Url::parse(db_url) {
18 Ok(mut parsed) => {
19 if parsed.username().is_empty() && parsed.password().is_none() {
20 return db_url.to_string();
21 }
22 let _ = parsed.set_username("***");
23 let _ = parsed.set_password(Some("***"));
24 parsed.to_string()
25 }
26 Err(_) => db_url.to_string(),
27 }
28}
29
30#[derive(Debug, Clone, PartialEq, Default)]
32pub enum SqlOutputType {
33 #[default]
35 SelectList,
36 SelectOne,
38 StreamList,
40}
41
42impl FromStr for SqlOutputType {
43 type Err = CamelError;
44
45 fn from_str(s: &str) -> Result<Self, Self::Err> {
46 match s {
47 "SelectList" => Ok(SqlOutputType::SelectList),
48 "SelectOne" => Ok(SqlOutputType::SelectOne),
49 "StreamList" => Ok(SqlOutputType::StreamList),
50 _ => Err(CamelError::InvalidUri(format!(
51 "Unknown output type: {}",
52 s
53 ))),
54 }
55 }
56}
57
58#[derive(Debug, Clone, PartialEq, Default)]
66pub enum TransactionMode {
67 #[default]
69 Auto,
70 Managed,
72}
73
74impl FromStr for TransactionMode {
75 type Err = CamelError;
76
77 fn from_str(s: &str) -> Result<Self, Self::Err> {
78 match s {
79 "Auto" => Ok(TransactionMode::Auto),
80 "Managed" => Ok(TransactionMode::Managed),
81 _ => Err(CamelError::InvalidUri(format!(
82 "Unknown transaction mode: {}. Expected 'Auto' or 'Managed'",
83 s
84 ))),
85 }
86 }
87}
88
89impl std::fmt::Display for TransactionMode {
90 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91 match self {
92 TransactionMode::Auto => write!(f, "Auto"),
93 TransactionMode::Managed => write!(f, "Managed"),
94 }
95 }
96}
97
98#[derive(Debug, Clone, PartialEq, Default)]
100pub enum ProcessingStrategy {
101 #[default]
103 Direct,
104 Scheduled,
106}
107
108impl FromStr for ProcessingStrategy {
109 type Err = CamelError;
110
111 fn from_str(s: &str) -> Result<Self, Self::Err> {
112 match s {
113 "Direct" => Ok(ProcessingStrategy::Direct),
114 "Scheduled" => Ok(ProcessingStrategy::Scheduled),
115 _ => Err(CamelError::InvalidUri(format!(
116 "Unknown processing strategy: {}. Expected 'Direct' or 'Scheduled'",
117 s
118 ))),
119 }
120 }
121}
122
123impl std::fmt::Display for ProcessingStrategy {
124 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125 match self {
126 ProcessingStrategy::Direct => write!(f, "Direct"),
127 ProcessingStrategy::Scheduled => write!(f, "Scheduled"),
128 }
129 }
130}
131
132#[derive(Debug, Clone, PartialEq, Default)]
134pub enum PollStrategy {
135 #[default]
137 Sequential,
138 Burst,
140}
141
142impl FromStr for PollStrategy {
143 type Err = CamelError;
144
145 fn from_str(s: &str) -> Result<Self, Self::Err> {
146 match s {
147 "Sequential" => Ok(PollStrategy::Sequential),
148 "Burst" => Ok(PollStrategy::Burst),
149 _ => Err(CamelError::InvalidUri(format!(
150 "Unknown poll strategy: {}. Expected 'Sequential' or 'Burst'",
151 s
152 ))),
153 }
154 }
155}
156
157impl std::fmt::Display for PollStrategy {
158 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
159 match self {
160 PollStrategy::Sequential => write!(f, "Sequential"),
161 PollStrategy::Burst => write!(f, "Burst"),
162 }
163 }
164}
165
166#[derive(Clone, PartialEq, serde::Deserialize)]
173#[serde(default)]
174pub struct SqlGlobalConfig {
175 pub max_connections: u32,
176 pub min_connections: u32,
177 pub idle_timeout_secs: u64,
178 pub max_lifetime_secs: u64,
179 pub ssl_mode: Option<String>,
181 pub ssl_root_cert: Option<String>,
182 pub ssl_cert: Option<String>,
183 pub ssl_key: Option<String>,
184 #[serde(default)]
186 pub retry: NetworkRetryPolicy,
187}
188
189impl std::fmt::Debug for SqlGlobalConfig {
190 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
191 f.debug_struct("SqlGlobalConfig")
192 .field("max_connections", &self.max_connections)
193 .field("min_connections", &self.min_connections)
194 .field("idle_timeout_secs", &self.idle_timeout_secs)
195 .field("max_lifetime_secs", &self.max_lifetime_secs)
196 .field("ssl_mode", &self.ssl_mode)
197 .field("ssl_root_cert", &self.ssl_root_cert)
198 .field("ssl_cert", &self.ssl_cert)
199 .field("ssl_key", &redacted_opt(&self.ssl_key))
200 .field("retry", &self.retry)
201 .finish()
202 }
203}
204
205impl Default for SqlGlobalConfig {
206 fn default() -> Self {
207 Self {
208 max_connections: 5,
209 min_connections: 1,
210 idle_timeout_secs: 300,
211 max_lifetime_secs: 1800,
212 ssl_mode: None,
213 ssl_root_cert: None,
214 ssl_cert: None,
215 ssl_key: None,
216 retry: NetworkRetryPolicy::default(),
217 }
218 }
219}
220
221impl SqlGlobalConfig {
222 pub fn new() -> Self {
223 Self::default()
224 }
225
226 pub fn with_max_connections(mut self, value: u32) -> Self {
227 self.max_connections = value;
228 self
229 }
230
231 pub fn with_min_connections(mut self, value: u32) -> Self {
232 self.min_connections = value;
233 self
234 }
235
236 pub fn with_idle_timeout_secs(mut self, value: u64) -> Self {
237 self.idle_timeout_secs = value;
238 self
239 }
240
241 pub fn with_max_lifetime_secs(mut self, value: u64) -> Self {
242 self.max_lifetime_secs = value;
243 self
244 }
245
246 pub fn with_ssl_mode(mut self, value: impl Into<String>) -> Self {
247 self.ssl_mode = Some(value.into());
248 self
249 }
250
251 pub fn with_ssl_root_cert(mut self, value: impl Into<String>) -> Self {
252 self.ssl_root_cert = Some(value.into());
253 self
254 }
255
256 pub fn with_ssl_cert(mut self, value: impl Into<String>) -> Self {
257 self.ssl_cert = Some(value.into());
258 self
259 }
260
261 pub fn with_ssl_key(mut self, value: impl Into<String>) -> Self {
262 self.ssl_key = Some(value.into());
263 self
264 }
265
266 pub fn with_retry(mut self, value: NetworkRetryPolicy) -> Self {
267 self.retry = value;
268 self
269 }
270}
271
272#[derive(Clone)]
289pub struct SqlEndpointConfig {
290 pub db_url: String,
293 pub datasource_name: Option<String>,
295 pub max_connections: Option<u32>,
297 pub min_connections: Option<u32>,
299 pub idle_timeout_secs: Option<u64>,
301 pub max_lifetime_secs: Option<u64>,
303
304 pub query: String,
307 pub source_path: Option<String>,
309 pub output_type: SqlOutputType,
311 pub placeholder: char,
313 pub use_placeholder: bool,
315 pub noop: bool,
317 pub in_separator: String,
319
320 pub always_populate_statement: bool,
324
325 pub allow_named_parameters: bool,
329
330 pub fetch_size: Option<u32>,
333
334 pub transaction_mode: TransactionMode,
337
338 pub delay_ms: u64,
341 pub initial_delay_ms: u64,
343 pub max_messages_per_poll: Option<i32>,
345 pub on_consume: Option<String>,
347 pub on_consume_failed: Option<String>,
349 pub on_consume_batch_complete: Option<String>,
351 pub route_empty_result_set: bool,
353 pub use_iterator: bool,
355 pub expected_update_count: Option<i64>,
357 pub break_batch_on_consume_fail: bool,
359 pub bridge_error_handler: bool,
361
362 pub repeat_count: Option<u32>,
366
367 pub processing_strategy: ProcessingStrategy,
370
371 pub poll_strategy: PollStrategy,
374
375 pub batch: bool,
378 pub use_message_body_for_sql: bool,
380 pub allow_dynamic_query: bool,
383
384 pub ssl_mode: Option<String>,
387 pub ssl_root_cert: Option<String>,
389 pub ssl_cert: Option<String>,
391 pub ssl_key: Option<String>,
393
394 pub retry: NetworkRetryPolicy,
396
397 retry_set_from_uri: bool,
401}
402
403impl std::fmt::Debug for SqlEndpointConfig {
404 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
405 f.debug_struct("SqlEndpointConfig")
406 .field("db_url", &redact_db_url(&self.db_url))
407 .field("datasource_name", &self.datasource_name)
408 .field("max_connections", &self.max_connections)
409 .field("min_connections", &self.min_connections)
410 .field("idle_timeout_secs", &self.idle_timeout_secs)
411 .field("max_lifetime_secs", &self.max_lifetime_secs)
412 .field("query", &self.query)
413 .field("source_path", &self.source_path)
414 .field("output_type", &self.output_type)
415 .field("placeholder", &self.placeholder)
416 .field("use_placeholder", &self.use_placeholder)
417 .field("noop", &self.noop)
418 .field("in_separator", &self.in_separator)
419 .field("always_populate_statement", &self.always_populate_statement)
420 .field("allow_named_parameters", &self.allow_named_parameters)
421 .field("fetch_size", &self.fetch_size)
422 .field("transaction_mode", &self.transaction_mode)
423 .field("delay_ms", &self.delay_ms)
424 .field("initial_delay_ms", &self.initial_delay_ms)
425 .field("max_messages_per_poll", &self.max_messages_per_poll)
426 .field("on_consume", &self.on_consume)
427 .field("on_consume_failed", &self.on_consume_failed)
428 .field("on_consume_batch_complete", &self.on_consume_batch_complete)
429 .field("route_empty_result_set", &self.route_empty_result_set)
430 .field("use_iterator", &self.use_iterator)
431 .field("expected_update_count", &self.expected_update_count)
432 .field(
433 "break_batch_on_consume_fail",
434 &self.break_batch_on_consume_fail,
435 )
436 .field("bridge_error_handler", &self.bridge_error_handler)
437 .field("repeat_count", &self.repeat_count)
438 .field("processing_strategy", &self.processing_strategy)
439 .field("poll_strategy", &self.poll_strategy)
440 .field("batch", &self.batch)
441 .field("use_message_body_for_sql", &self.use_message_body_for_sql)
442 .field("allow_dynamic_query", &self.allow_dynamic_query)
443 .field("ssl_mode", &self.ssl_mode)
444 .field("ssl_root_cert", &self.ssl_root_cert)
445 .field("ssl_cert", &self.ssl_cert)
446 .field("ssl_key", &redacted_opt(&self.ssl_key))
447 .field("retry", &self.retry)
448 .finish()
449 }
450}
451
452impl SqlEndpointConfig {
453 pub fn apply_defaults(&mut self, defaults: &SqlGlobalConfig) {
455 if self.max_connections.is_none() {
456 self.max_connections = Some(defaults.max_connections);
457 }
458 if self.min_connections.is_none() {
459 self.min_connections = Some(defaults.min_connections);
460 }
461 if self.idle_timeout_secs.is_none() {
462 self.idle_timeout_secs = Some(defaults.idle_timeout_secs);
463 }
464 if self.max_lifetime_secs.is_none() {
465 self.max_lifetime_secs = Some(defaults.max_lifetime_secs);
466 }
467 if self.ssl_mode.is_none() {
468 self.ssl_mode = defaults.ssl_mode.clone();
469 }
470 if self.ssl_root_cert.is_none() {
471 self.ssl_root_cert = defaults.ssl_root_cert.clone();
472 }
473 if self.ssl_cert.is_none() {
474 self.ssl_cert = defaults.ssl_cert.clone();
475 }
476 if self.ssl_key.is_none() {
477 self.ssl_key = defaults.ssl_key.clone();
478 }
479 if !self.retry_set_from_uri {
481 self.retry = defaults.retry.clone();
482 }
483 }
484
485 pub fn resolve_defaults(&mut self) {
487 let defaults = SqlGlobalConfig::default();
488 self.apply_defaults(&defaults);
489 }
490
491 pub async fn resolve_file_query(&mut self) -> Result<(), CamelError> {
500 if let Some(file_path) = self.source_path.take() {
501 let contents = tokio::fs::read_to_string(&file_path).await.map_err(|e| {
502 CamelError::Config(format!("Failed to read SQL file '{}': {}", file_path, e))
503 })?;
504 self.query = contents.trim().to_string();
505 self.source_path = Some(file_path);
507 }
508 Ok(())
509 }
510}
511
512struct SslParamMapping {
513 pg_key: &'static str,
514 mysql_key: &'static str,
515}
516
517const SSL_MAPPINGS: &[(&str, SslParamMapping)] = &[
518 (
519 "sslMode",
520 SslParamMapping {
521 pg_key: "sslmode",
522 mysql_key: "ssl-mode",
523 },
524 ),
525 (
526 "sslRootCert",
527 SslParamMapping {
528 pg_key: "sslrootcert",
529 mysql_key: "ssl-ca",
530 },
531 ),
532 (
533 "sslCert",
534 SslParamMapping {
535 pg_key: "sslcert",
536 mysql_key: "ssl-cert",
537 },
538 ),
539 (
540 "sslKey",
541 SslParamMapping {
542 pg_key: "sslkey",
543 mysql_key: "ssl-key",
544 },
545 ),
546];
547
548pub fn enrich_db_url_with_ssl(
549 db_url: &str,
550 config: &SqlEndpointConfig,
551) -> Result<String, CamelError> {
552 enrich_db_url_with_ssl_params(
553 db_url,
554 config.ssl_mode.as_deref(),
555 config.ssl_root_cert.as_deref(),
556 config.ssl_cert.as_deref(),
557 config.ssl_key.as_deref(),
558 )
559}
560
561pub(crate) fn enrich_db_url_with_ssl_params(
562 db_url: &str,
563 ssl_mode: Option<&str>,
564 ssl_root_cert: Option<&str>,
565 ssl_cert: Option<&str>,
566 ssl_key: Option<&str>,
567) -> Result<String, CamelError> {
568 let mut parsed = url::Url::parse(db_url).map_err(|e| {
569 CamelError::InvalidUri(format!(
570 "Cannot parse database URL for SSL enrichment: {}",
571 e
572 ))
573 })?;
574
575 let scheme = parsed.scheme();
576 if scheme.starts_with("sqlite") {
577 if ssl_mode.is_some() || ssl_root_cert.is_some() || ssl_cert.is_some() || ssl_key.is_some()
578 {
579 warn!(
580 "SSL options configured for SQLite database URL, but SQLite does not support SSL/TLS; ignoring sslMode/sslRootCert/sslCert/sslKey"
581 );
582 }
583 return Ok(db_url.to_string());
584 }
585
586 if scheme != "postgres" && scheme != "postgresql" && scheme != "mysql" {
587 return Ok(db_url.to_string());
588 }
589 let is_mysql = scheme == "mysql";
590
591 let effective_ssl_mode: &str = ssl_mode.unwrap_or(if is_mysql { "prefer" } else { "require" });
593
594 let ssl_params: Vec<(&str, &str)> = [
595 Some(("sslMode", effective_ssl_mode)),
596 ssl_root_cert.map(|v| ("sslRootCert", v)),
597 ssl_cert.map(|v| ("sslCert", v)),
598 ssl_key.map(|v| ("sslKey", v)),
599 ]
600 .into_iter()
601 .flatten()
602 .collect();
603
604 let mut query_pairs = parsed.query_pairs().collect::<Vec<_>>();
605 for (camel_name, value) in &ssl_params {
606 if let Some((_, mapping)) = SSL_MAPPINGS.iter().find(|(name, _)| *name == *camel_name) {
607 let driver_key = if is_mysql {
608 mapping.mysql_key
609 } else {
610 mapping.pg_key
611 };
612
613 if let Some(pos) = query_pairs.iter().position(|(k, _)| k == driver_key) {
614 query_pairs[pos].1 = (*value).into();
615 } else {
616 query_pairs.push((driver_key.into(), (*value).into()));
617 }
618 }
619 }
620
621 if !query_pairs.iter().any(|(k, _)| k == "connect_timeout") {
624 query_pairs.push(("connect_timeout".into(), "10".into()));
625 }
626
627 {
628 let mut serializer = url::form_urlencoded::Serializer::new(String::new());
629 for (k, v) in &query_pairs {
630 serializer.append_pair(k, v);
631 }
632 parsed.set_query(Some(&serializer.finish()));
633 }
634
635 Ok(parsed.to_string())
636}
637
638impl UriConfig for SqlEndpointConfig {
639 fn scheme() -> &'static str {
640 "sql"
641 }
642
643 fn from_uri(uri: &str) -> Result<Self, CamelError> {
644 let parts = parse_uri(uri)?;
645 Self::from_components(parts)
646 }
647
648 fn from_components(parts: UriComponents) -> Result<Self, CamelError> {
649 if parts.scheme != Self::scheme() {
651 return Err(CamelError::InvalidUri(format!(
652 "expected scheme '{}' but got '{}'",
653 Self::scheme(),
654 parts.scheme
655 )));
656 }
657
658 let params = &parts.params;
659
660 let (query, source_path) = if parts.path.starts_with("file:") {
665 let file_path = parts.path.trim_start_matches("file:").to_string();
666 (String::new(), Some(file_path))
667 } else {
668 (parts.path.clone(), None)
669 };
670
671 let db_url = params.get("db_url").cloned().unwrap_or_default();
673
674 let datasource_name = params.get("datasource").cloned();
676
677 let max_connections = params.get("maxConnections").and_then(|v| v.parse().ok());
679 let min_connections = params.get("minConnections").and_then(|v| v.parse().ok());
680 let idle_timeout_secs = params.get("idleTimeoutSecs").and_then(|v| v.parse().ok());
681 let max_lifetime_secs = params.get("maxLifetimeSecs").and_then(|v| v.parse().ok());
682
683 let output_type = params
685 .get("outputType")
686 .map(|s| s.parse())
687 .transpose()?
688 .unwrap_or_default();
689 let placeholder = params
690 .get("placeholder")
691 .filter(|v| !v.is_empty())
692 .map(|v| {
693 if v.chars().count() != 1 {
694 return Err(CamelError::InvalidUri(format!(
695 "placeholder must be exactly one character, got '{}'",
696 v
697 )));
698 }
699 if !v.is_ascii() {
700 return Err(CamelError::InvalidUri(
701 "placeholder must be a single ASCII character".to_string(),
702 ));
703 }
704 Ok(v.chars().next().unwrap()) })
706 .transpose()?
707 .unwrap_or('#');
708 fn parse_bool_param(name: &str, value: &str) -> Result<bool, CamelError> {
713 if value.eq_ignore_ascii_case("true") {
714 Ok(true)
715 } else if value.eq_ignore_ascii_case("false") {
716 Ok(false)
717 } else {
718 Err(CamelError::InvalidUri(format!(
719 "{} must be 'true' or 'false', got '{}'",
720 name, value
721 )))
722 }
723 }
724
725 let use_placeholder = params
726 .get("usePlaceholder")
727 .map(|v| parse_bool_param("usePlaceholder", v))
728 .transpose()?
729 .unwrap_or(true);
730 let noop = params
731 .get("noop")
732 .map(|v| parse_bool_param("noop", v))
733 .transpose()?
734 .unwrap_or(false);
735 let in_separator = params
736 .get("inSeparator")
737 .map(|v| v.to_string())
738 .unwrap_or_else(|| ", ".to_string());
739 if in_separator.is_empty() {
740 return Err(CamelError::InvalidUri(
741 "inSeparator must not be empty".to_string(),
742 ));
743 }
744
745 let always_populate_statement = params
747 .get("alwaysPopulateStatement")
748 .map(|v| parse_bool_param("alwaysPopulateStatement", v))
749 .transpose()?
750 .unwrap_or(false);
751
752 let allow_named_parameters = params
754 .get("allowNamedParameters")
755 .map(|v| parse_bool_param("allowNamedParameters", v))
756 .transpose()?
757 .unwrap_or(true);
758
759 let fetch_size = params.get("fetchSize").and_then(|v| v.parse().ok());
761
762 let transaction_mode = params
764 .get("transactionMode")
765 .map(|s| s.parse())
766 .transpose()?
767 .unwrap_or_default();
768
769 let delay_ms = params
771 .get("delay")
772 .and_then(|v| v.parse().ok())
773 .unwrap_or(500);
774 let initial_delay_ms = params
775 .get("initialDelay")
776 .and_then(|v| v.parse().ok())
777 .unwrap_or(1000);
778 let max_messages_per_poll = params
779 .get("maxMessagesPerPoll")
780 .and_then(|v| v.parse().ok());
781 let on_consume = params.get("onConsume").cloned();
782 let on_consume_failed = params.get("onConsumeFailed").cloned();
783 let on_consume_batch_complete = params.get("onConsumeBatchComplete").cloned();
784 let route_empty_result_set = params
785 .get("routeEmptyResultSet")
786 .map(|v| parse_bool_param("routeEmptyResultSet", v))
787 .transpose()?
788 .unwrap_or(false);
789 let use_iterator = params
790 .get("useIterator")
791 .map(|v| parse_bool_param("useIterator", v))
792 .transpose()?
793 .unwrap_or(true);
794 let expected_update_count = params
795 .get("expectedUpdateCount")
796 .and_then(|v| v.parse().ok());
797 let break_batch_on_consume_fail = params
798 .get("breakBatchOnConsumeFail")
799 .map(|v| parse_bool_param("breakBatchOnConsumeFail", v))
800 .transpose()?
801 .unwrap_or(false);
802 let bridge_error_handler = params
803 .get("bridgeErrorHandler")
804 .map(|v| parse_bool_param("bridgeErrorHandler", v))
805 .transpose()?
806 .unwrap_or(false);
807
808 let repeat_count = params.get("repeatCount").and_then(|v| v.parse().ok());
810
811 let processing_strategy = params
813 .get("processingStrategy")
814 .map(|s| s.parse())
815 .transpose()?
816 .unwrap_or_default();
817
818 let poll_strategy = params
820 .get("pollStrategy")
821 .map(|s| s.parse())
822 .transpose()?
823 .unwrap_or_default();
824
825 let batch = params
827 .get("batch")
828 .map(|v| parse_bool_param("batch", v))
829 .transpose()?
830 .unwrap_or(false);
831 let use_message_body_for_sql = params
832 .get("useMessageBodyForSql")
833 .map(|v| parse_bool_param("useMessageBodyForSql", v))
834 .transpose()?
835 .unwrap_or(false);
836 let allow_dynamic_query = params
837 .get("allowDynamicQuery")
838 .map(|v| parse_bool_param("allowDynamicQuery", v))
839 .transpose()?
840 .unwrap_or(false);
841 let ssl_mode = params.get("sslMode").cloned();
842 let ssl_root_cert = params.get("sslRootCert").cloned();
843 let ssl_cert = params.get("sslCert").cloned();
844 let ssl_key = params.get("sslKey").cloned();
845
846 let mut retry = NetworkRetryPolicy::default();
848 let mut retry_set_from_uri = false;
849 if let Some(raw) = params.get("retryEnabled") {
850 retry.enabled = raw.parse::<bool>().map_err(|_| {
851 CamelError::InvalidUri(format!("retryEnabled must be a boolean, got '{raw}'"))
852 })?;
853 retry_set_from_uri = true;
854 }
855 if let Some(raw) = params.get("retryMaxAttempts") {
856 retry.max_attempts = raw.parse::<u32>().map_err(|_| {
857 CamelError::InvalidUri(format!("retryMaxAttempts must be a u32, got '{raw}'"))
858 })?;
859 retry_set_from_uri = true;
860 }
861 if let Some(raw) = params.get("retryInitialDelayMs") {
862 retry.initial_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
863 CamelError::InvalidUri(format!("retryInitialDelayMs must be a u64, got '{raw}'"))
864 })?);
865 retry_set_from_uri = true;
866 }
867 if let Some(raw) = params.get("retryMultiplier") {
868 retry.multiplier = raw.parse::<f64>().map_err(|_| {
869 CamelError::InvalidUri(format!("retryMultiplier must be a f64, got '{raw}'"))
870 })?;
871 retry_set_from_uri = true;
872 }
873 if let Some(raw) = params.get("retryMaxDelayMs") {
874 retry.max_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
875 CamelError::InvalidUri(format!("retryMaxDelayMs must be a u64, got '{raw}'"))
876 })?);
877 retry_set_from_uri = true;
878 }
879 if let Some(raw) = params.get("retryJitter") {
880 retry.jitter_factor = raw.parse::<f64>().map_err(|_| {
881 CamelError::InvalidUri(format!("retryJitter must be a f64, got '{raw}'"))
882 })?;
883 retry_set_from_uri = true;
884 }
885
886 if datasource_name.is_none() && db_url.is_empty() {
887 return Err(CamelError::Config(
888 "either 'datasource' or 'db_url' parameter is required".to_string(),
889 ));
890 }
891
892 if datasource_name.is_some() && !db_url.is_empty() {
893 return Err(CamelError::InvalidUri(
894 "'db_url' not allowed with named datasource — use 'datasource' alone".to_string(),
895 ));
896 }
897
898 if datasource_name.is_some() {
899 let overrides: Vec<&str> = {
900 let mut v = Vec::new();
901 if max_connections.is_some() {
902 v.push("maxConnections");
903 }
904 if min_connections.is_some() {
905 v.push("minConnections");
906 }
907 if idle_timeout_secs.is_some() {
908 v.push("idleTimeoutSecs");
909 }
910 if max_lifetime_secs.is_some() {
911 v.push("maxLifetimeSecs");
912 }
913 if ssl_mode.is_some() {
914 v.push("sslMode");
915 }
916 if ssl_root_cert.is_some() {
917 v.push("sslRootCert");
918 }
919 if ssl_cert.is_some() {
920 v.push("sslCert");
921 }
922 if ssl_key.is_some() {
923 v.push("sslKey");
924 }
925 v
926 };
927 if !overrides.is_empty() {
928 return Err(CamelError::InvalidUri(format!(
929 "pool-affecting params not allowed with named datasource: {}",
930 overrides.join(", ")
931 )));
932 }
933 }
934
935 Ok(Self {
936 db_url,
937 datasource_name,
938 max_connections,
939 min_connections,
940 idle_timeout_secs,
941 max_lifetime_secs,
942 query,
943 source_path,
944 output_type,
945 placeholder,
946 use_placeholder,
947 noop,
948 in_separator,
949 always_populate_statement,
950 allow_named_parameters,
951 fetch_size,
952 transaction_mode,
953 delay_ms,
954 initial_delay_ms,
955 max_messages_per_poll,
956 on_consume,
957 on_consume_failed,
958 on_consume_batch_complete,
959 route_empty_result_set,
960 use_iterator,
961 expected_update_count,
962 break_batch_on_consume_fail,
963 bridge_error_handler,
964 repeat_count,
965 processing_strategy,
966 poll_strategy,
967 batch,
968 use_message_body_for_sql,
969 allow_dynamic_query,
970 ssl_mode,
971 ssl_root_cert,
972 ssl_cert,
973 ssl_key,
974 retry,
975 retry_set_from_uri,
976 })
977 }
978}
979
980#[cfg(test)]
981mod tests {
982 use super::*;
983 use camel_component_api::NetworkRetryPolicy;
984
985 #[test]
986 fn config_defaults() {
987 let mut c =
988 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
989 c.resolve_defaults();
990 assert_eq!(c.query, "select 1");
991 assert_eq!(c.db_url, "postgres://localhost/test");
992 assert_eq!(c.max_connections, Some(5));
993 assert_eq!(c.min_connections, Some(1));
994 assert_eq!(c.idle_timeout_secs, Some(300));
995 assert_eq!(c.max_lifetime_secs, Some(1800));
996 assert_eq!(c.output_type, SqlOutputType::SelectList);
997 assert_eq!(c.placeholder, '#');
998 assert!(!c.noop);
999 assert_eq!(c.in_separator, ", ");
1000 assert_eq!(c.delay_ms, 500);
1001 assert_eq!(c.initial_delay_ms, 1000);
1002 assert!(c.max_messages_per_poll.is_none());
1003 assert!(c.on_consume.is_none());
1004 assert!(c.on_consume_failed.is_none());
1005 assert!(c.on_consume_batch_complete.is_none());
1006 assert!(!c.route_empty_result_set);
1007 assert!(c.use_iterator);
1008 assert!(c.expected_update_count.is_none());
1009 assert!(!c.break_batch_on_consume_fail);
1010 assert!(!c.batch);
1011 assert!(!c.use_message_body_for_sql);
1012 assert!(!c.allow_dynamic_query);
1013 assert!(c.ssl_mode.is_none());
1014 assert!(c.ssl_root_cert.is_none());
1015 assert!(c.ssl_cert.is_none());
1016 assert!(c.ssl_key.is_none());
1017 assert!(!c.always_populate_statement);
1019 assert!(c.allow_named_parameters);
1020 assert!(c.fetch_size.is_none());
1021 assert_eq!(c.transaction_mode, TransactionMode::Auto);
1022 assert!(c.repeat_count.is_none());
1023 assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
1024 assert_eq!(c.poll_strategy, PollStrategy::Sequential);
1025 }
1026
1027 #[test]
1028 fn ssl_none_by_default() {
1029 let c =
1030 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1031 assert!(c.ssl_mode.is_none());
1032 assert!(c.ssl_root_cert.is_none());
1033 assert!(c.ssl_cert.is_none());
1034 assert!(c.ssl_key.is_none());
1035 }
1036
1037 #[test]
1038 fn ssl_mode_from_uri() {
1039 let c = SqlEndpointConfig::from_uri(
1040 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1041 )
1042 .unwrap();
1043 assert_eq!(c.ssl_mode, Some("require".to_string()));
1044 assert!(c.ssl_root_cert.is_none());
1045 }
1046
1047 #[test]
1048 fn ssl_all_params_from_uri() {
1049 let c = SqlEndpointConfig::from_uri(
1050 "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1051 )
1052 .unwrap();
1053 assert_eq!(c.ssl_mode, Some("require".to_string()));
1054 assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1055 assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1056 assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1057 }
1058
1059 #[test]
1060 fn ssl_global_applied_to_endpoint() {
1061 let mut c =
1062 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1063 let global = SqlGlobalConfig::default()
1064 .with_ssl_mode("require")
1065 .with_ssl_root_cert("/etc/ssl/ca.pem");
1066 c.apply_defaults(&global);
1067 assert_eq!(c.ssl_mode, Some("require".to_string()));
1068 assert_eq!(c.ssl_root_cert, Some("/etc/ssl/ca.pem".to_string()));
1069 assert!(c.ssl_cert.is_none());
1070 assert!(c.ssl_key.is_none());
1071 }
1072
1073 #[test]
1074 fn ssl_uri_overrides_global() {
1075 let mut c = SqlEndpointConfig::from_uri(
1076 "sql:select 1?db_url=postgres://localhost/test&sslMode=verify-full",
1077 )
1078 .unwrap();
1079 let global = SqlGlobalConfig::default().with_ssl_mode("require");
1080 c.apply_defaults(&global);
1081 assert_eq!(c.ssl_mode, Some("verify-full".to_string()));
1082 }
1083
1084 #[test]
1085 fn config_wrong_scheme() {
1086 assert!(SqlEndpointConfig::from_uri("redis://localhost:6379").is_err());
1087 }
1088
1089 #[test]
1090 fn config_missing_db_url() {
1091 assert!(SqlEndpointConfig::from_uri("sql:select 1").is_err());
1092 }
1093
1094 #[test]
1095 fn config_output_type_select_one() {
1096 let c = SqlEndpointConfig::from_uri(
1097 "sql:select 1?db_url=postgres://localhost/test&outputType=SelectOne",
1098 )
1099 .unwrap();
1100 assert_eq!(c.output_type, SqlOutputType::SelectOne);
1101 }
1102
1103 #[test]
1104 fn config_output_type_stream_list() {
1105 let c = SqlEndpointConfig::from_uri(
1106 "sql:select 1?db_url=postgres://localhost/test&outputType=StreamList",
1107 )
1108 .unwrap();
1109 assert_eq!(c.output_type, SqlOutputType::StreamList);
1110 }
1111
1112 #[test]
1113 fn in_separator_default() {
1114 let c =
1115 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1116 assert_eq!(c.in_separator, ", ");
1117 }
1118
1119 #[test]
1120 fn in_separator_from_uri() {
1121 let c = SqlEndpointConfig::from_uri(
1122 "sql:select 1?db_url=postgres://localhost/test&inSeparator=;",
1123 )
1124 .unwrap();
1125 assert_eq!(c.in_separator, ";");
1126 }
1127
1128 #[test]
1129 fn in_separator_empty_rejected() {
1130 let result = SqlEndpointConfig::from_uri(
1131 "sql:select 1?db_url=postgres://localhost/test&inSeparator=",
1132 );
1133 assert!(result.is_err());
1134 let msg = format!("{:?}", result.unwrap_err());
1135 assert!(msg.contains("inSeparator") || msg.contains("empty"));
1136 }
1137
1138 #[test]
1139 fn config_consumer_options() {
1140 let c = SqlEndpointConfig::from_uri(
1141 "sql:select * from t?db_url=postgres://localhost/test&delay=2000&initialDelay=500&maxMessagesPerPoll=10&onConsume=update t set done=true where id=:#id&onConsumeFailed=update t set failed=true where id=:#id&onConsumeBatchComplete=delete from t where done=true&routeEmptyResultSet=true&useIterator=false&expectedUpdateCount=1&breakBatchOnConsumeFail=true"
1142 ).unwrap();
1143 assert_eq!(c.delay_ms, 2000);
1144 assert_eq!(c.initial_delay_ms, 500);
1145 assert_eq!(c.max_messages_per_poll, Some(10));
1146 assert_eq!(
1147 c.on_consume,
1148 Some("update t set done=true where id=:#id".to_string())
1149 );
1150 assert_eq!(
1151 c.on_consume_failed,
1152 Some("update t set failed=true where id=:#id".to_string())
1153 );
1154 assert_eq!(
1155 c.on_consume_batch_complete,
1156 Some("delete from t where done=true".to_string())
1157 );
1158 assert!(c.route_empty_result_set);
1159 assert!(!c.use_iterator);
1160 assert_eq!(c.expected_update_count, Some(1));
1161 assert!(c.break_batch_on_consume_fail);
1162 assert!(!c.bridge_error_handler);
1163 }
1164
1165 #[test]
1166 fn config_producer_options() {
1167 let c = SqlEndpointConfig::from_uri(
1168 "sql:insert into t values (#)?db_url=postgres://localhost/test&batch=true&useMessageBodyForSql=true&noop=true"
1169 ).unwrap();
1170 assert!(c.batch);
1171 assert!(c.use_message_body_for_sql);
1172 assert!(c.noop);
1173 }
1174
1175 #[test]
1176 fn config_pool_options() {
1177 let c = SqlEndpointConfig::from_uri(
1178 "sql:select 1?db_url=postgres://localhost/test&maxConnections=20&minConnections=3&idleTimeoutSecs=600&maxLifetimeSecs=3600"
1179 ).unwrap();
1180 assert_eq!(c.max_connections, Some(20));
1181 assert_eq!(c.min_connections, Some(3));
1182 assert_eq!(c.idle_timeout_secs, Some(600));
1183 assert_eq!(c.max_lifetime_secs, Some(3600));
1184 }
1185
1186 #[test]
1187 fn config_query_with_special_chars() {
1188 let c = SqlEndpointConfig::from_uri(
1189 "sql:select * from users where name = :#name and age > #?db_url=postgres://localhost/test",
1190 )
1191 .unwrap();
1192 assert_eq!(
1193 c.query,
1194 "select * from users where name = :#name and age > #"
1195 );
1196 }
1197
1198 #[test]
1199 fn output_type_from_str() {
1200 assert_eq!(
1201 "SelectList".parse::<SqlOutputType>().unwrap(),
1202 SqlOutputType::SelectList
1203 );
1204 assert_eq!(
1205 "SelectOne".parse::<SqlOutputType>().unwrap(),
1206 SqlOutputType::SelectOne
1207 );
1208 assert_eq!(
1209 "StreamList".parse::<SqlOutputType>().unwrap(),
1210 SqlOutputType::StreamList
1211 );
1212 assert!("Invalid".parse::<SqlOutputType>().is_err());
1213 }
1214
1215 #[tokio::test]
1217 async fn config_file_not_found() {
1218 let mut config = SqlEndpointConfig::from_uri(
1219 "sql:file:/nonexistent/path/query.sql?db_url=postgres://localhost/test",
1220 )
1221 .expect("from_uri should defer file reading");
1222 assert_eq!(
1224 config.source_path,
1225 Some("/nonexistent/path/query.sql".to_string())
1226 );
1227 assert!(config.query.is_empty());
1228
1229 let result = config.resolve_file_query().await;
1231 assert!(result.is_err());
1232 let msg = format!("{:?}", result.unwrap_err());
1233 assert!(msg.contains("Failed to read SQL file") || msg.contains("nonexistent"));
1234 }
1235
1236 #[tokio::test]
1238 async fn config_file_query() {
1239 use std::io::Write;
1240 let unique_name = format!(
1241 "test_sql_query_{}.sql",
1242 std::time::SystemTime::now()
1243 .duration_since(std::time::UNIX_EPOCH)
1244 .unwrap_or_default()
1245 .as_nanos()
1246 );
1247 let mut tmp = std::env::temp_dir();
1248 tmp.push(unique_name);
1249 {
1250 let mut f = std::fs::File::create(&tmp).unwrap();
1251 writeln!(f, "SELECT * FROM users").unwrap();
1252 }
1253 let uri = format!(
1254 "sql:file:{}?db_url=postgres://localhost/test",
1255 tmp.display()
1256 );
1257 let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
1258 assert!(c.query.is_empty());
1260 assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
1261
1262 c.resolve_file_query()
1264 .await
1265 .expect("file query should resolve");
1266 assert_eq!(c.query, "SELECT * FROM users");
1267 std::fs::remove_file(&tmp).ok();
1268 }
1269
1270 #[test]
1272 fn pool_fields_none_when_not_set() {
1273 let c =
1274 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1275 assert_eq!(c.max_connections, None);
1276 assert_eq!(c.min_connections, None);
1277 assert_eq!(c.idle_timeout_secs, None);
1278 assert_eq!(c.max_lifetime_secs, None);
1279 }
1280
1281 #[test]
1282 fn apply_defaults_fills_none() {
1283 let mut c =
1284 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1285 let global = SqlGlobalConfig {
1286 max_connections: 10,
1287 min_connections: 2,
1288 idle_timeout_secs: 600,
1289 max_lifetime_secs: 3600,
1290 ssl_mode: None,
1291 ssl_root_cert: None,
1292 ssl_cert: None,
1293 ssl_key: None,
1294 retry: NetworkRetryPolicy::default(),
1295 };
1296 c.apply_defaults(&global);
1297 assert_eq!(c.max_connections, Some(10));
1298 assert_eq!(c.min_connections, Some(2));
1299 assert_eq!(c.idle_timeout_secs, Some(600));
1300 assert_eq!(c.max_lifetime_secs, Some(3600));
1301 assert!(c.ssl_mode.is_none());
1302 assert!(c.ssl_root_cert.is_none());
1303 assert!(c.ssl_cert.is_none());
1304 assert!(c.ssl_key.is_none());
1305 }
1306
1307 #[test]
1308 fn apply_defaults_does_not_override() {
1309 let mut c = SqlEndpointConfig::from_uri(
1310 "sql:select 1?db_url=postgres://localhost/test&maxConnections=99&minConnections=5",
1311 )
1312 .unwrap();
1313 let global = SqlGlobalConfig {
1314 max_connections: 10,
1315 min_connections: 2,
1316 idle_timeout_secs: 600,
1317 max_lifetime_secs: 3600,
1318 ssl_mode: None,
1319 ssl_root_cert: None,
1320 ssl_cert: None,
1321 ssl_key: None,
1322 retry: NetworkRetryPolicy::default(),
1323 };
1324 c.apply_defaults(&global);
1325 assert_eq!(c.max_connections, Some(99));
1327 assert_eq!(c.min_connections, Some(5));
1328 assert_eq!(c.idle_timeout_secs, Some(600));
1330 assert_eq!(c.max_lifetime_secs, Some(3600));
1331 }
1332
1333 #[test]
1334 fn resolve_defaults_fills_remaining() {
1335 let mut c = SqlEndpointConfig::from_uri(
1336 "sql:select 1?db_url=postgres://localhost/test&maxConnections=7",
1337 )
1338 .unwrap();
1339 c.resolve_defaults();
1340 assert_eq!(c.max_connections, Some(7)); assert_eq!(c.min_connections, Some(1)); assert_eq!(c.idle_timeout_secs, Some(300)); assert_eq!(c.max_lifetime_secs, Some(1800)); }
1345
1346 #[test]
1347 fn global_config_builder() {
1348 let c = SqlGlobalConfig::default()
1349 .with_max_connections(20)
1350 .with_min_connections(3)
1351 .with_idle_timeout_secs(600)
1352 .with_max_lifetime_secs(3600)
1353 .with_ssl_mode("require")
1354 .with_ssl_root_cert("/ca.pem")
1355 .with_ssl_cert("/cert.pem")
1356 .with_ssl_key("/key.pem");
1357 assert_eq!(c.max_connections, 20);
1358 assert_eq!(c.min_connections, 3);
1359 assert_eq!(c.idle_timeout_secs, 600);
1360 assert_eq!(c.max_lifetime_secs, 3600);
1361 assert_eq!(c.ssl_mode, Some("require".to_string()));
1362 assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1363 assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1364 assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1365 }
1366
1367 #[test]
1368 fn enrich_postgres_ssl_mode() {
1369 let mut c = SqlEndpointConfig::from_uri(
1370 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1371 )
1372 .unwrap();
1373 c.resolve_defaults();
1374 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1375 assert!(url.contains("sslmode=require"), "got: {}", url);
1376 }
1377
1378 #[test]
1379 fn enrich_postgres_all_ssl() {
1380 let mut c = SqlEndpointConfig::from_uri(
1381 "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1382 )
1383 .unwrap();
1384 c.resolve_defaults();
1385 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1386 assert!(url.contains("sslmode=require"), "got: {}", url);
1387 assert!(url.contains("sslrootcert="), "got: {}", url);
1388 assert!(url.contains("sslcert="), "got: {}", url);
1389 assert!(url.contains("sslkey="), "got: {}", url);
1390 }
1391
1392 #[test]
1393 fn enrich_mysql_ssl() {
1394 let mut c = SqlEndpointConfig::from_uri(
1395 "sql:select 1?db_url=mysql://localhost/test&sslMode=require",
1396 )
1397 .unwrap();
1398 c.resolve_defaults();
1399 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1400 assert!(url.contains("ssl-mode=require"), "got: {}", url);
1401 }
1402
1403 #[test]
1404 fn enrich_existing_query_params() {
1405 let mut c = SqlEndpointConfig::from_uri(
1406 "sql:select 1?db_url=postgres://localhost/test?existing=1&sslMode=require",
1407 )
1408 .unwrap();
1409 c.resolve_defaults();
1410 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1411 assert!(url.contains("existing=1"), "got: {}", url);
1412 assert!(url.contains("sslmode=require"), "got: {}", url);
1413 }
1414
1415 #[test]
1416 fn enrich_override_existing() {
1417 let mut c = SqlEndpointConfig::from_uri(
1418 "sql:select 1?db_url=postgres://localhost/test?sslmode=allow&sslMode=require",
1419 )
1420 .unwrap();
1421 c.resolve_defaults();
1422 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1423 assert!(url.contains("sslmode=require"), "got: {}", url);
1424 assert!(!url.contains("sslmode=allow"), "got: {}", url);
1425 }
1426
1427 #[test]
1428 fn enrich_applies_postgres_defaults() {
1429 let mut c =
1430 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1431 c.resolve_defaults();
1432 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1433 assert!(
1435 url.contains("sslmode=require"),
1436 "expected sslmode=require, got: {}",
1437 url
1438 );
1439 assert!(
1440 url.contains("connect_timeout=10"),
1441 "expected connect_timeout=10, got: {}",
1442 url
1443 );
1444 }
1445
1446 #[test]
1447 fn enrich_mysql_defaults_to_prefer() {
1448 let mut c =
1449 SqlEndpointConfig::from_uri("sql:select 1?db_url=mysql://localhost/test").unwrap();
1450 c.resolve_defaults();
1451 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1452 assert!(
1453 url.contains("ssl-mode=prefer"),
1454 "expected ssl-mode=prefer, got: {}",
1455 url
1456 );
1457 assert!(
1458 url.contains("connect_timeout=10"),
1459 "expected connect_timeout=10, got: {}",
1460 url
1461 );
1462 }
1463
1464 #[test]
1465 fn enrich_preserves_explicit_connect_timeout() {
1466 let mut c = SqlEndpointConfig::from_uri(
1467 "sql:select 1?db_url=postgres://localhost/test?connect_timeout=5",
1468 )
1469 .unwrap();
1470 c.resolve_defaults();
1471 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1472 assert!(
1473 url.contains("connect_timeout=5"),
1474 "expected explicit connect_timeout=5 preserved, got: {}",
1475 url
1476 );
1477 assert!(
1478 url.contains("sslmode=require"),
1479 "expected sslmode=require, got: {}",
1480 url
1481 );
1482 }
1483
1484 #[test]
1485 fn enrich_url_encodes_paths() {
1486 let mut c = SqlEndpointConfig::from_uri(
1487 "sql:select 1?db_url=postgres://localhost/test&sslRootCert=/path/to/my%20cert.pem",
1488 )
1489 .unwrap();
1490 c.resolve_defaults();
1491 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1492 assert!(url.contains("sslrootcert="), "got: {}", url);
1493 }
1494
1495 #[test]
1496 fn enrich_unsupported_scheme_returns_unchanged() {
1497 let mut c = SqlEndpointConfig::from_uri(
1498 "sql:select 1?db_url=sqlite://localhost/test.db&sslMode=require",
1499 )
1500 .unwrap();
1501 c.resolve_defaults();
1502 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1503 assert_eq!(url, "sqlite://localhost/test.db");
1504 }
1505
1506 #[test]
1507 fn enrich_invalid_url_returns_error() {
1508 let mut c = SqlEndpointConfig::from_uri(
1509 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1510 )
1511 .unwrap();
1512 c.resolve_defaults();
1513 let result = enrich_db_url_with_ssl("://not-a-valid-url", &c);
1514 assert!(result.is_err());
1515 }
1516
1517 #[test]
1521 fn debug_redacts_db_url_with_password() {
1522 let c = SqlEndpointConfig::from_uri(
1523 "sql:select 1?db_url=postgres://user:secret123@localhost/test",
1524 )
1525 .unwrap();
1526 let debug_output = format!("{:?}", c);
1527 assert!(
1528 !debug_output.contains("secret123"),
1529 "Debug output must not contain password: {}",
1530 debug_output
1531 );
1532 assert!(
1533 debug_output.contains("***"),
1534 "Debug output must contain redacted marker: {}",
1535 debug_output
1536 );
1537 }
1538
1539 #[test]
1540 fn debug_redacts_ssl_key() {
1541 let c = SqlEndpointConfig::from_uri(
1542 "sql:select 1?db_url=postgres://localhost/test&sslKey=/secret/key.pem",
1543 )
1544 .unwrap();
1545 let debug_output = format!("{:?}", c);
1546 assert!(
1547 !debug_output.contains("/secret/key.pem"),
1548 "Debug output must not contain ssl_key path: {}",
1549 debug_output
1550 );
1551 }
1552
1553 #[test]
1554 fn debug_global_config_redacts_ssl_key() {
1555 let c = SqlGlobalConfig::default().with_ssl_key("/secret/key.pem");
1556 let debug_output = format!("{:?}", c);
1557 assert!(
1558 !debug_output.contains("/secret/key.pem"),
1559 "Debug output must not contain ssl_key path: {}",
1560 debug_output
1561 );
1562 assert!(
1563 debug_output.contains("***"),
1564 "Debug output must contain redacted marker: {}",
1565 debug_output
1566 );
1567 }
1568
1569 #[test]
1570 fn redact_db_url_with_credentials() {
1571 assert_eq!(
1572 redact_db_url("postgres://user:pass@host/db"),
1573 "postgres://***:***@host/db"
1574 );
1575 }
1576
1577 #[test]
1578 fn redact_db_url_without_credentials() {
1579 assert_eq!(redact_db_url("sqlite::memory:"), "sqlite::memory:");
1580 }
1581
1582 #[test]
1583 fn redact_db_url_invalid_returns_original() {
1584 assert_eq!(redact_db_url("not-a-url"), "not-a-url");
1585 }
1586
1587 #[test]
1589 fn use_placeholder_defaults_to_true() {
1590 let c =
1591 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1592 assert!(c.use_placeholder);
1593 }
1594
1595 #[test]
1596 fn use_placeholder_false_from_uri() {
1597 let c = SqlEndpointConfig::from_uri(
1598 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=false",
1599 )
1600 .unwrap();
1601 assert!(!c.use_placeholder);
1602 }
1603
1604 #[test]
1605 fn use_placeholder_true_from_uri() {
1606 let c = SqlEndpointConfig::from_uri(
1607 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=true",
1608 )
1609 .unwrap();
1610 assert!(c.use_placeholder);
1611 }
1612
1613 #[test]
1615 fn use_placeholder_rejects_invalid_value() {
1616 let result = SqlEndpointConfig::from_uri(
1617 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=1",
1618 );
1619 assert!(result.is_err());
1620 let msg = format!("{:?}", result.unwrap_err());
1621 assert!(msg.contains("usePlaceholder") && msg.contains("true") && msg.contains("false"));
1622 }
1623
1624 #[test]
1625 fn use_placeholder_rejects_typo_tru() {
1626 let result = SqlEndpointConfig::from_uri(
1627 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=tru",
1628 );
1629 assert!(result.is_err());
1630 }
1631
1632 #[test]
1633 fn use_placeholder_rejects_yes() {
1634 let result = SqlEndpointConfig::from_uri(
1635 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=yes",
1636 );
1637 assert!(result.is_err());
1638 }
1639
1640 #[test]
1641 fn noop_rejects_invalid_value() {
1642 let result =
1643 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&noop=1");
1644 assert!(result.is_err());
1645 let msg = format!("{:?}", result.unwrap_err());
1646 assert!(msg.contains("noop"));
1647 }
1648
1649 #[test]
1650 fn batch_rejects_invalid_value() {
1651 let result =
1652 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&batch=yes");
1653 assert!(result.is_err());
1654 let msg = format!("{:?}", result.unwrap_err());
1655 assert!(msg.contains("batch"));
1656 }
1657
1658 #[test]
1659 fn route_empty_result_set_rejects_invalid_value() {
1660 let result = SqlEndpointConfig::from_uri(
1661 "sql:select 1?db_url=postgres://localhost/test&routeEmptyResultSet=on",
1662 );
1663 assert!(result.is_err());
1664 }
1665
1666 #[test]
1667 fn use_iterator_rejects_invalid_value() {
1668 let result = SqlEndpointConfig::from_uri(
1669 "sql:select 1?db_url=postgres://localhost/test&useIterator=1",
1670 );
1671 assert!(result.is_err());
1672 }
1673
1674 #[test]
1675 fn break_batch_on_consume_fail_rejects_invalid_value() {
1676 let result = SqlEndpointConfig::from_uri(
1677 "sql:select 1?db_url=postgres://localhost/test&breakBatchOnConsumeFail=yes",
1678 );
1679 assert!(result.is_err());
1680 }
1681
1682 #[test]
1683 fn use_message_body_for_sql_rejects_invalid_value() {
1684 let result = SqlEndpointConfig::from_uri(
1685 "sql:select 1?db_url=postgres://localhost/test&useMessageBodyForSql=1",
1686 );
1687 assert!(result.is_err());
1688 }
1689
1690 #[test]
1692 fn boolean_params_case_insensitive() {
1693 let c = SqlEndpointConfig::from_uri(
1694 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=TRUE&noop=FALSE&batch=True&useIterator=False&bridgeErrorHandler=TRUE",
1695 )
1696 .unwrap();
1697 assert!(c.use_placeholder);
1698 assert!(!c.noop);
1699 assert!(c.batch);
1700 assert!(!c.use_iterator);
1701 assert!(c.bridge_error_handler);
1702 }
1703
1704 #[test]
1706 fn multi_char_placeholder_rejected() {
1707 let result = SqlEndpointConfig::from_uri(
1708 "sql:select 1?db_url=postgres://localhost/test&placeholder=##",
1709 );
1710 assert!(result.is_err());
1711 let msg = format!("{:?}", result.unwrap_err());
1712 assert!(msg.contains("placeholder") && msg.contains("one character"));
1713 }
1714
1715 #[test]
1716 fn non_ascii_placeholder_rejected() {
1717 let result = SqlEndpointConfig::from_uri(
1718 "sql:select 1?db_url=postgres://localhost/test&placeholder=%C2%A2",
1719 );
1720 assert!(result.is_err());
1721 }
1722
1723 #[test]
1724 fn single_char_placeholder_accepted() {
1725 let c = SqlEndpointConfig::from_uri(
1726 "sql:select 1?db_url=postgres://localhost/test&placeholder=$",
1727 )
1728 .unwrap();
1729 assert_eq!(c.placeholder, '$');
1730 }
1731
1732 #[test]
1733 fn empty_placeholder_falls_back_to_default() {
1734 let c = SqlEndpointConfig::from_uri(
1736 "sql:select 1?db_url=postgres://localhost/test&placeholder=",
1737 )
1738 .unwrap();
1739 assert_eq!(c.placeholder, '#');
1740 }
1741
1742 #[tokio::test]
1744 async fn file_query_cached_in_config() {
1745 use std::io::Write;
1746 let unique_name = format!(
1747 "test_sql_cached_{}.sql",
1748 std::time::SystemTime::now()
1749 .duration_since(std::time::UNIX_EPOCH)
1750 .unwrap_or_default()
1751 .as_nanos()
1752 );
1753 let mut tmp = std::env::temp_dir();
1754 tmp.push(unique_name);
1755 {
1756 let mut f = std::fs::File::create(&tmp).unwrap();
1757 writeln!(f, "SELECT * FROM cached_test").unwrap();
1758 }
1759 let uri = format!(
1760 "sql:file:{}?db_url=postgres://localhost/test",
1761 tmp.display()
1762 );
1763 let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
1764 assert!(c.query.is_empty());
1766 assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
1767
1768 c.resolve_file_query()
1770 .await
1771 .expect("resolve should succeed");
1772 assert_eq!(c.query, "SELECT * FROM cached_test");
1773
1774 std::fs::remove_file(&tmp).ok();
1776 assert_eq!(c.query, "SELECT * FROM cached_test");
1777 }
1778
1779 #[test]
1783 fn always_populate_statement_defaults_to_false() {
1784 let c =
1785 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1786 assert!(!c.always_populate_statement);
1787 }
1788
1789 #[test]
1790 fn always_populate_statement_from_uri() {
1791 let c = SqlEndpointConfig::from_uri(
1792 "sql:select 1?db_url=postgres://localhost/test&alwaysPopulateStatement=true",
1793 )
1794 .unwrap();
1795 assert!(c.always_populate_statement);
1796 }
1797
1798 #[test]
1800 fn allow_named_parameters_defaults_to_true() {
1801 let c =
1802 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1803 assert!(c.allow_named_parameters);
1804 }
1805
1806 #[test]
1807 fn allow_named_parameters_false_from_uri() {
1808 let c = SqlEndpointConfig::from_uri(
1809 "sql:select 1?db_url=postgres://localhost/test&allowNamedParameters=false",
1810 )
1811 .unwrap();
1812 assert!(!c.allow_named_parameters);
1813 }
1814
1815 #[test]
1817 fn fetch_size_defaults_to_none() {
1818 let c =
1819 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1820 assert!(c.fetch_size.is_none());
1821 }
1822
1823 #[test]
1824 fn fetch_size_from_uri() {
1825 let c = SqlEndpointConfig::from_uri(
1826 "sql:select 1?db_url=postgres://localhost/test&fetchSize=1000",
1827 )
1828 .unwrap();
1829 assert_eq!(c.fetch_size, Some(1000));
1830 }
1831
1832 #[test]
1834 fn transaction_mode_defaults_to_auto() {
1835 let c =
1836 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1837 assert_eq!(c.transaction_mode, TransactionMode::Auto);
1838 }
1839
1840 #[test]
1841 fn transaction_mode_managed_from_uri() {
1842 let c = SqlEndpointConfig::from_uri(
1843 "sql:select 1?db_url=postgres://localhost/test&transactionMode=Managed",
1844 )
1845 .unwrap();
1846 assert_eq!(c.transaction_mode, TransactionMode::Managed);
1847 }
1848
1849 #[test]
1850 fn transaction_mode_invalid_rejected() {
1851 let result = SqlEndpointConfig::from_uri(
1852 "sql:select 1?db_url=postgres://localhost/test&transactionMode=Invalid",
1853 );
1854 assert!(result.is_err());
1855 }
1856
1857 #[test]
1859 fn repeat_count_defaults_to_none() {
1860 let c =
1861 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1862 assert!(c.repeat_count.is_none());
1863 }
1864
1865 #[test]
1866 fn repeat_count_from_uri() {
1867 let c = SqlEndpointConfig::from_uri(
1868 "sql:select 1?db_url=postgres://localhost/test&repeatCount=10",
1869 )
1870 .unwrap();
1871 assert_eq!(c.repeat_count, Some(10));
1872 }
1873
1874 #[test]
1876 fn processing_strategy_defaults_to_direct() {
1877 let c =
1878 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1879 assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
1880 }
1881
1882 #[test]
1883 fn processing_strategy_scheduled_from_uri() {
1884 let c = SqlEndpointConfig::from_uri(
1885 "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Scheduled",
1886 )
1887 .unwrap();
1888 assert_eq!(c.processing_strategy, ProcessingStrategy::Scheduled);
1889 }
1890
1891 #[test]
1892 fn processing_strategy_invalid_rejected() {
1893 let result = SqlEndpointConfig::from_uri(
1894 "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Invalid",
1895 );
1896 assert!(result.is_err());
1897 }
1898
1899 #[test]
1901 fn poll_strategy_defaults_to_sequential() {
1902 let c =
1903 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1904 assert_eq!(c.poll_strategy, PollStrategy::Sequential);
1905 }
1906
1907 #[test]
1908 fn poll_strategy_burst_from_uri() {
1909 let c = SqlEndpointConfig::from_uri(
1910 "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Burst",
1911 )
1912 .unwrap();
1913 assert_eq!(c.poll_strategy, PollStrategy::Burst);
1914 }
1915
1916 #[test]
1917 fn poll_strategy_invalid_rejected() {
1918 let result = SqlEndpointConfig::from_uri(
1919 "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Invalid",
1920 );
1921 assert!(result.is_err());
1922 }
1923
1924 #[test]
1927 fn sql_endpoint_config_has_retry_policy() {
1928 let cfg = SqlEndpointConfig::from_uri(
1929 "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=3&retryInitialDelayMs=500",
1930 )
1931 .expect("parse");
1932 assert_eq!(cfg.retry.max_attempts, 3);
1933 assert_eq!(
1934 cfg.retry.initial_delay,
1935 std::time::Duration::from_millis(500)
1936 );
1937 assert!(cfg.retry.enabled);
1938 }
1939
1940 #[test]
1941 fn sql_endpoint_config_retry_defaults_when_unspecified() {
1942 let cfg =
1943 SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
1944 assert!(cfg.retry.enabled);
1946 assert_eq!(cfg.retry.max_attempts, 10); }
1948
1949 #[test]
1950 fn sql_global_config_has_retry_default() {
1951 let cfg = SqlGlobalConfig::default();
1952 assert!(cfg.retry.enabled);
1953 }
1954
1955 #[test]
1956 fn retry_policy_parse_full_uri_params() {
1957 let cfg = SqlEndpointConfig::from_uri(
1958 "sql:select 1?db_url=sqlite::memory:&retryEnabled=false&retryMaxAttempts=7&retryInitialDelayMs=1000&retryMultiplier=3.0&retryMaxDelayMs=60000&retryJitter=0.5",
1959 )
1960 .expect("parse");
1961 assert!(!cfg.retry.enabled);
1962 assert_eq!(cfg.retry.max_attempts, 7);
1963 assert_eq!(
1964 cfg.retry.initial_delay,
1965 std::time::Duration::from_millis(1000)
1966 );
1967 assert!((cfg.retry.multiplier - 3.0).abs() < f64::EPSILON);
1968 assert_eq!(cfg.retry.max_delay, std::time::Duration::from_millis(60000));
1969 assert!((cfg.retry.jitter_factor - 0.5).abs() < f64::EPSILON);
1970 }
1971
1972 #[test]
1973 fn retry_policy_from_uri_survives_apply_defaults_with_global() {
1974 let mut ep = SqlEndpointConfig::from_uri(
1975 "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=10&retryInitialDelayMs=500",
1976 )
1977 .expect("parse");
1978 let global = SqlGlobalConfig::default(); ep.apply_defaults(&global);
1980 assert_eq!(ep.retry.max_attempts, 10);
1982 assert_eq!(
1983 ep.retry.initial_delay,
1984 std::time::Duration::from_millis(500)
1985 );
1986 }
1987
1988 #[test]
1989 fn retry_policy_falls_back_to_global_when_uri_has_no_retry_params() {
1990 let mut ep =
1991 SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
1992 let mut global = SqlGlobalConfig::default();
1993 global.retry.max_attempts = 7;
1994 ep.apply_defaults(&global);
1995 assert_eq!(ep.retry.max_attempts, 7);
1997 }
1998
1999 #[test]
2000 fn from_uri_with_datasource_name() {
2001 let cfg = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders").unwrap();
2002 assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2003 assert!(cfg.db_url.is_empty());
2004 }
2005
2006 #[test]
2007 fn from_uri_with_datasource_and_behavior_override() {
2008 let cfg =
2009 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&outputType=SelectOne")
2010 .unwrap();
2011 assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2012 }
2013
2014 #[test]
2015 fn from_uri_datasource_rejects_pool_override() {
2016 let result =
2017 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&maxConnections=50");
2018 assert!(result.is_err());
2019 let msg = result.unwrap_err().to_string();
2020 assert!(msg.contains("pool-affecting"));
2021 }
2022
2023 #[test]
2024 fn from_uri_neither_datasource_nor_db_url_is_error() {
2025 let result = SqlEndpointConfig::from_uri("sql:SELECT 1");
2026 assert!(result.is_err());
2027 }
2028
2029 #[test]
2030 fn from_uri_db_url_inline_still_works() {
2031 let cfg =
2032 SqlEndpointConfig::from_uri("sql:SELECT 1?db_url=postgres://localhost/test").unwrap();
2033 assert!(cfg.datasource_name.is_none());
2034 assert_eq!(cfg.db_url, "postgres://localhost/test");
2035 }
2036
2037 #[test]
2038 fn from_uri_datasource_rejects_ssl_mode() {
2039 let result = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslMode=require");
2040 assert!(result.is_err());
2041 let msg = result.unwrap_err().to_string();
2042 assert!(msg.contains("pool-affecting"));
2043 }
2044
2045 #[test]
2046 fn from_uri_datasource_rejects_ssl_root_cert() {
2047 let result =
2048 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslRootCert=/ca.pem");
2049 assert!(result.is_err());
2050 }
2051
2052 #[test]
2053 fn from_uri_datasource_rejects_db_url() {
2054 let result = SqlEndpointConfig::from_uri(
2055 "sql:SELECT 1?datasource=orders&db_url=postgres://evil:5432/pwned",
2056 );
2057 assert!(result.is_err());
2058 let msg = result.unwrap_err().to_string();
2059 assert!(msg.contains("db_url") && msg.contains("datasource"));
2060 }
2061}