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>,
368
369 pub break_on_empty: bool,
374
375 pub processing_strategy: ProcessingStrategy,
378
379 pub poll_strategy: PollStrategy,
382
383 pub batch: bool,
386 pub use_message_body_for_sql: bool,
388 pub allow_dynamic_query: bool,
391
392 pub ssl_mode: Option<String>,
395 pub ssl_root_cert: Option<String>,
397 pub ssl_cert: Option<String>,
399 pub ssl_key: Option<String>,
401
402 pub retry: NetworkRetryPolicy,
404
405 retry_set_from_uri: bool,
409}
410
411impl std::fmt::Debug for SqlEndpointConfig {
412 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
413 f.debug_struct("SqlEndpointConfig")
414 .field("db_url", &redact_db_url(&self.db_url))
415 .field("datasource_name", &self.datasource_name)
416 .field("max_connections", &self.max_connections)
417 .field("min_connections", &self.min_connections)
418 .field("idle_timeout_secs", &self.idle_timeout_secs)
419 .field("max_lifetime_secs", &self.max_lifetime_secs)
420 .field("query", &self.query)
421 .field("source_path", &self.source_path)
422 .field("output_type", &self.output_type)
423 .field("placeholder", &self.placeholder)
424 .field("use_placeholder", &self.use_placeholder)
425 .field("noop", &self.noop)
426 .field("in_separator", &self.in_separator)
427 .field("always_populate_statement", &self.always_populate_statement)
428 .field("allow_named_parameters", &self.allow_named_parameters)
429 .field("fetch_size", &self.fetch_size)
430 .field("transaction_mode", &self.transaction_mode)
431 .field("delay_ms", &self.delay_ms)
432 .field("initial_delay_ms", &self.initial_delay_ms)
433 .field("max_messages_per_poll", &self.max_messages_per_poll)
434 .field("on_consume", &self.on_consume)
435 .field("on_consume_failed", &self.on_consume_failed)
436 .field("on_consume_batch_complete", &self.on_consume_batch_complete)
437 .field("route_empty_result_set", &self.route_empty_result_set)
438 .field("use_iterator", &self.use_iterator)
439 .field("expected_update_count", &self.expected_update_count)
440 .field(
441 "break_batch_on_consume_fail",
442 &self.break_batch_on_consume_fail,
443 )
444 .field("bridge_error_handler", &self.bridge_error_handler)
445 .field("repeat_count", &self.repeat_count)
446 .field("break_on_empty", &self.break_on_empty)
447 .field("processing_strategy", &self.processing_strategy)
448 .field("poll_strategy", &self.poll_strategy)
449 .field("batch", &self.batch)
450 .field("use_message_body_for_sql", &self.use_message_body_for_sql)
451 .field("allow_dynamic_query", &self.allow_dynamic_query)
452 .field("ssl_mode", &self.ssl_mode)
453 .field("ssl_root_cert", &self.ssl_root_cert)
454 .field("ssl_cert", &self.ssl_cert)
455 .field("ssl_key", &redacted_opt(&self.ssl_key))
456 .field("retry", &self.retry)
457 .finish()
458 }
459}
460
461impl SqlEndpointConfig {
462 pub fn apply_defaults(&mut self, defaults: &SqlGlobalConfig) {
464 if self.max_connections.is_none() {
465 self.max_connections = Some(defaults.max_connections);
466 }
467 if self.min_connections.is_none() {
468 self.min_connections = Some(defaults.min_connections);
469 }
470 if self.idle_timeout_secs.is_none() {
471 self.idle_timeout_secs = Some(defaults.idle_timeout_secs);
472 }
473 if self.max_lifetime_secs.is_none() {
474 self.max_lifetime_secs = Some(defaults.max_lifetime_secs);
475 }
476 if self.ssl_mode.is_none() {
477 self.ssl_mode = defaults.ssl_mode.clone();
478 }
479 if self.ssl_root_cert.is_none() {
480 self.ssl_root_cert = defaults.ssl_root_cert.clone();
481 }
482 if self.ssl_cert.is_none() {
483 self.ssl_cert = defaults.ssl_cert.clone();
484 }
485 if self.ssl_key.is_none() {
486 self.ssl_key = defaults.ssl_key.clone();
487 }
488 if !self.retry_set_from_uri {
490 self.retry = defaults.retry.clone();
491 }
492 }
493
494 pub fn resolve_defaults(&mut self) {
496 let defaults = SqlGlobalConfig::default();
497 self.apply_defaults(&defaults);
498 }
499
500 pub async fn resolve_file_query(&mut self) -> Result<(), CamelError> {
509 if let Some(file_path) = self.source_path.take() {
510 let contents = tokio::fs::read_to_string(&file_path).await.map_err(|e| {
511 CamelError::Config(format!("Failed to read SQL file '{}': {}", file_path, e))
512 })?;
513 self.query = contents.trim().to_string();
514 self.source_path = Some(file_path);
516 }
517 Ok(())
518 }
519}
520
521struct SslParamMapping {
522 pg_key: &'static str,
523 mysql_key: &'static str,
524}
525
526const SSL_MAPPINGS: &[(&str, SslParamMapping)] = &[
527 (
528 "sslMode",
529 SslParamMapping {
530 pg_key: "sslmode",
531 mysql_key: "ssl-mode",
532 },
533 ),
534 (
535 "sslRootCert",
536 SslParamMapping {
537 pg_key: "sslrootcert",
538 mysql_key: "ssl-ca",
539 },
540 ),
541 (
542 "sslCert",
543 SslParamMapping {
544 pg_key: "sslcert",
545 mysql_key: "ssl-cert",
546 },
547 ),
548 (
549 "sslKey",
550 SslParamMapping {
551 pg_key: "sslkey",
552 mysql_key: "ssl-key",
553 },
554 ),
555];
556
557pub fn enrich_db_url_with_ssl(
558 db_url: &str,
559 config: &SqlEndpointConfig,
560) -> Result<String, CamelError> {
561 enrich_db_url_with_ssl_params(
562 db_url,
563 config.ssl_mode.as_deref(),
564 config.ssl_root_cert.as_deref(),
565 config.ssl_cert.as_deref(),
566 config.ssl_key.as_deref(),
567 )
568}
569
570pub(crate) fn enrich_db_url_with_ssl_params(
571 db_url: &str,
572 ssl_mode: Option<&str>,
573 ssl_root_cert: Option<&str>,
574 ssl_cert: Option<&str>,
575 ssl_key: Option<&str>,
576) -> Result<String, CamelError> {
577 let mut parsed = url::Url::parse(db_url).map_err(|e| {
578 CamelError::InvalidUri(format!(
579 "Cannot parse database URL for SSL enrichment: {}",
580 e
581 ))
582 })?;
583
584 let scheme = parsed.scheme();
585 if scheme.starts_with("sqlite") {
586 if ssl_mode.is_some() || ssl_root_cert.is_some() || ssl_cert.is_some() || ssl_key.is_some()
587 {
588 warn!(
589 "SSL options configured for SQLite database URL, but SQLite does not support SSL/TLS; ignoring sslMode/sslRootCert/sslCert/sslKey"
590 );
591 }
592 return Ok(db_url.to_string());
593 }
594
595 if scheme != "postgres" && scheme != "postgresql" && scheme != "mysql" {
596 return Ok(db_url.to_string());
597 }
598 let is_mysql = scheme == "mysql";
599
600 let effective_ssl_mode: &str = ssl_mode.unwrap_or(if is_mysql { "prefer" } else { "require" });
602
603 let ssl_params: Vec<(&str, &str)> = [
604 Some(("sslMode", effective_ssl_mode)),
605 ssl_root_cert.map(|v| ("sslRootCert", v)),
606 ssl_cert.map(|v| ("sslCert", v)),
607 ssl_key.map(|v| ("sslKey", v)),
608 ]
609 .into_iter()
610 .flatten()
611 .collect();
612
613 let mut query_pairs = parsed.query_pairs().collect::<Vec<_>>();
614 for (camel_name, value) in &ssl_params {
615 if let Some((_, mapping)) = SSL_MAPPINGS.iter().find(|(name, _)| *name == *camel_name) {
616 let driver_key = if is_mysql {
617 mapping.mysql_key
618 } else {
619 mapping.pg_key
620 };
621
622 if let Some(pos) = query_pairs.iter().position(|(k, _)| k == driver_key) {
623 query_pairs[pos].1 = (*value).into();
624 } else {
625 query_pairs.push((driver_key.into(), (*value).into()));
626 }
627 }
628 }
629
630 if !query_pairs.iter().any(|(k, _)| k == "connect_timeout") {
633 query_pairs.push(("connect_timeout".into(), "10".into()));
634 }
635
636 {
637 let mut serializer = url::form_urlencoded::Serializer::new(String::new());
638 for (k, v) in &query_pairs {
639 serializer.append_pair(k, v);
640 }
641 parsed.set_query(Some(&serializer.finish()));
642 }
643
644 Ok(parsed.to_string())
645}
646
647impl UriConfig for SqlEndpointConfig {
648 fn scheme() -> &'static str {
649 "sql"
650 }
651
652 fn from_uri(uri: &str) -> Result<Self, CamelError> {
653 let parts = parse_uri(uri)?;
654 Self::from_components(parts)
655 }
656
657 fn from_components(parts: UriComponents) -> Result<Self, CamelError> {
658 if parts.scheme != Self::scheme() {
660 return Err(CamelError::InvalidUri(format!(
661 "expected scheme '{}' but got '{}'",
662 Self::scheme(),
663 parts.scheme
664 )));
665 }
666
667 let params = &parts.params;
668
669 let (query, source_path) = if parts.path.starts_with("file:") {
674 let file_path = parts.path.trim_start_matches("file:").to_string();
675 (String::new(), Some(file_path))
676 } else {
677 (parts.path.clone(), None)
678 };
679
680 let db_url = params.get("db_url").cloned().unwrap_or_default();
682
683 let datasource_name = params.get("datasource").cloned();
685
686 let max_connections = params.get("maxConnections").and_then(|v| v.parse().ok());
688 let min_connections = params.get("minConnections").and_then(|v| v.parse().ok());
689 let idle_timeout_secs = params.get("idleTimeoutSecs").and_then(|v| v.parse().ok());
690 let max_lifetime_secs = params.get("maxLifetimeSecs").and_then(|v| v.parse().ok());
691
692 let output_type = params
694 .get("outputType")
695 .map(|s| s.parse())
696 .transpose()?
697 .unwrap_or_default();
698 let placeholder = params
699 .get("placeholder")
700 .filter(|v| !v.is_empty())
701 .map(|v| {
702 if v.chars().count() != 1 {
703 return Err(CamelError::InvalidUri(format!(
704 "placeholder must be exactly one character, got '{}'",
705 v
706 )));
707 }
708 if !v.is_ascii() {
709 return Err(CamelError::InvalidUri(
710 "placeholder must be a single ASCII character".to_string(),
711 ));
712 }
713 Ok(v.chars().next().unwrap()) })
715 .transpose()?
716 .unwrap_or('#');
717 fn parse_bool_param(name: &str, value: &str) -> Result<bool, CamelError> {
722 if value.eq_ignore_ascii_case("true") {
723 Ok(true)
724 } else if value.eq_ignore_ascii_case("false") {
725 Ok(false)
726 } else {
727 Err(CamelError::InvalidUri(format!(
728 "{} must be 'true' or 'false', got '{}'",
729 name, value
730 )))
731 }
732 }
733
734 let use_placeholder = params
735 .get("usePlaceholder")
736 .map(|v| parse_bool_param("usePlaceholder", v))
737 .transpose()?
738 .unwrap_or(true);
739 let noop = params
740 .get("noop")
741 .map(|v| parse_bool_param("noop", v))
742 .transpose()?
743 .unwrap_or(false);
744 let in_separator = params
745 .get("inSeparator")
746 .map(|v| v.to_string())
747 .unwrap_or_else(|| ", ".to_string());
748 if in_separator.is_empty() {
749 return Err(CamelError::InvalidUri(
750 "inSeparator must not be empty".to_string(),
751 ));
752 }
753
754 let always_populate_statement = params
756 .get("alwaysPopulateStatement")
757 .map(|v| parse_bool_param("alwaysPopulateStatement", v))
758 .transpose()?
759 .unwrap_or(false);
760
761 let allow_named_parameters = params
763 .get("allowNamedParameters")
764 .map(|v| parse_bool_param("allowNamedParameters", v))
765 .transpose()?
766 .unwrap_or(true);
767
768 let fetch_size = params.get("fetchSize").and_then(|v| v.parse().ok());
770
771 let transaction_mode = params
773 .get("transactionMode")
774 .map(|s| s.parse())
775 .transpose()?
776 .unwrap_or_default();
777
778 let delay_ms = params
780 .get("delay")
781 .and_then(|v| v.parse().ok())
782 .unwrap_or(500);
783 let initial_delay_ms = params
784 .get("initialDelay")
785 .and_then(|v| v.parse().ok())
786 .unwrap_or(1000);
787 let max_messages_per_poll = params
788 .get("maxMessagesPerPoll")
789 .and_then(|v| v.parse().ok());
790 let on_consume = params.get("onConsume").cloned();
791 let on_consume_failed = params.get("onConsumeFailed").cloned();
792 let on_consume_batch_complete = params.get("onConsumeBatchComplete").cloned();
793 let route_empty_result_set = params
794 .get("routeEmptyResultSet")
795 .map(|v| parse_bool_param("routeEmptyResultSet", v))
796 .transpose()?
797 .unwrap_or(false);
798 let use_iterator = params
799 .get("useIterator")
800 .map(|v| parse_bool_param("useIterator", v))
801 .transpose()?
802 .unwrap_or(true);
803 let expected_update_count = params
804 .get("expectedUpdateCount")
805 .and_then(|v| v.parse().ok());
806 let break_batch_on_consume_fail = params
807 .get("breakBatchOnConsumeFail")
808 .map(|v| parse_bool_param("breakBatchOnConsumeFail", v))
809 .transpose()?
810 .unwrap_or(false);
811 let bridge_error_handler = params
812 .get("bridgeErrorHandler")
813 .map(|v| parse_bool_param("bridgeErrorHandler", v))
814 .transpose()?
815 .unwrap_or(false);
816
817 let repeat_count = params.get("repeatCount").and_then(|v| v.parse().ok());
819
820 let break_on_empty = params
822 .get("breakOnEmpty")
823 .map(|v| parse_bool_param("breakOnEmpty", v))
824 .transpose()?
825 .unwrap_or(false);
826
827 let processing_strategy = params
829 .get("processingStrategy")
830 .map(|s| s.parse())
831 .transpose()?
832 .unwrap_or_default();
833
834 let poll_strategy = params
836 .get("pollStrategy")
837 .map(|s| s.parse())
838 .transpose()?
839 .unwrap_or_default();
840
841 let batch = params
843 .get("batch")
844 .map(|v| parse_bool_param("batch", v))
845 .transpose()?
846 .unwrap_or(false);
847 let use_message_body_for_sql = params
848 .get("useMessageBodyForSql")
849 .map(|v| parse_bool_param("useMessageBodyForSql", v))
850 .transpose()?
851 .unwrap_or(false);
852 let allow_dynamic_query = params
853 .get("allowDynamicQuery")
854 .map(|v| parse_bool_param("allowDynamicQuery", v))
855 .transpose()?
856 .unwrap_or(false);
857 let ssl_mode = params.get("sslMode").cloned();
858 let ssl_root_cert = params.get("sslRootCert").cloned();
859 let ssl_cert = params.get("sslCert").cloned();
860 let ssl_key = params.get("sslKey").cloned();
861
862 let mut retry = NetworkRetryPolicy::default();
864 let mut retry_set_from_uri = false;
865 if let Some(raw) = params.get("retryEnabled") {
866 retry.enabled = raw.parse::<bool>().map_err(|_| {
867 CamelError::InvalidUri(format!("retryEnabled must be a boolean, got '{raw}'"))
868 })?;
869 retry_set_from_uri = true;
870 }
871 if let Some(raw) = params.get("retryMaxAttempts") {
872 retry.max_attempts = raw.parse::<u32>().map_err(|_| {
873 CamelError::InvalidUri(format!("retryMaxAttempts must be a u32, got '{raw}'"))
874 })?;
875 retry_set_from_uri = true;
876 }
877 if let Some(raw) = params.get("retryInitialDelayMs") {
878 retry.initial_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
879 CamelError::InvalidUri(format!("retryInitialDelayMs must be a u64, got '{raw}'"))
880 })?);
881 retry_set_from_uri = true;
882 }
883 if let Some(raw) = params.get("retryMultiplier") {
884 retry.multiplier = raw.parse::<f64>().map_err(|_| {
885 CamelError::InvalidUri(format!("retryMultiplier must be a f64, got '{raw}'"))
886 })?;
887 retry_set_from_uri = true;
888 }
889 if let Some(raw) = params.get("retryMaxDelayMs") {
890 retry.max_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
891 CamelError::InvalidUri(format!("retryMaxDelayMs must be a u64, got '{raw}'"))
892 })?);
893 retry_set_from_uri = true;
894 }
895 if let Some(raw) = params.get("retryJitter") {
896 retry.jitter_factor = raw.parse::<f64>().map_err(|_| {
897 CamelError::InvalidUri(format!("retryJitter must be a f64, got '{raw}'"))
898 })?;
899 retry_set_from_uri = true;
900 }
901
902 if datasource_name.is_none() && db_url.is_empty() {
903 return Err(CamelError::Config(
904 "either 'datasource' or 'db_url' parameter is required".to_string(),
905 ));
906 }
907
908 if datasource_name.is_some() && !db_url.is_empty() {
909 return Err(CamelError::InvalidUri(
910 "'db_url' not allowed with named datasource — use 'datasource' alone".to_string(),
911 ));
912 }
913
914 if datasource_name.is_some() {
915 let overrides: Vec<&str> = {
916 let mut v = Vec::new();
917 if max_connections.is_some() {
918 v.push("maxConnections");
919 }
920 if min_connections.is_some() {
921 v.push("minConnections");
922 }
923 if idle_timeout_secs.is_some() {
924 v.push("idleTimeoutSecs");
925 }
926 if max_lifetime_secs.is_some() {
927 v.push("maxLifetimeSecs");
928 }
929 if ssl_mode.is_some() {
930 v.push("sslMode");
931 }
932 if ssl_root_cert.is_some() {
933 v.push("sslRootCert");
934 }
935 if ssl_cert.is_some() {
936 v.push("sslCert");
937 }
938 if ssl_key.is_some() {
939 v.push("sslKey");
940 }
941 v
942 };
943 if !overrides.is_empty() {
944 return Err(CamelError::InvalidUri(format!(
945 "pool-affecting params not allowed with named datasource: {}",
946 overrides.join(", ")
947 )));
948 }
949 }
950
951 Ok(Self {
952 db_url,
953 datasource_name,
954 max_connections,
955 min_connections,
956 idle_timeout_secs,
957 max_lifetime_secs,
958 query,
959 source_path,
960 output_type,
961 placeholder,
962 use_placeholder,
963 noop,
964 in_separator,
965 always_populate_statement,
966 allow_named_parameters,
967 fetch_size,
968 transaction_mode,
969 delay_ms,
970 initial_delay_ms,
971 max_messages_per_poll,
972 on_consume,
973 on_consume_failed,
974 on_consume_batch_complete,
975 route_empty_result_set,
976 use_iterator,
977 expected_update_count,
978 break_batch_on_consume_fail,
979 bridge_error_handler,
980 repeat_count,
981 break_on_empty,
982 processing_strategy,
983 poll_strategy,
984 batch,
985 use_message_body_for_sql,
986 allow_dynamic_query,
987 ssl_mode,
988 ssl_root_cert,
989 ssl_cert,
990 ssl_key,
991 retry,
992 retry_set_from_uri,
993 })
994 }
995}
996
997#[cfg(test)]
998mod tests {
999 use super::*;
1000 use camel_component_api::NetworkRetryPolicy;
1001
1002 #[test]
1003 fn config_defaults() {
1004 let mut c =
1005 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1006 c.resolve_defaults();
1007 assert_eq!(c.query, "select 1");
1008 assert_eq!(c.db_url, "postgres://localhost/test");
1009 assert_eq!(c.max_connections, Some(5));
1010 assert_eq!(c.min_connections, Some(1));
1011 assert_eq!(c.idle_timeout_secs, Some(300));
1012 assert_eq!(c.max_lifetime_secs, Some(1800));
1013 assert_eq!(c.output_type, SqlOutputType::SelectList);
1014 assert_eq!(c.placeholder, '#');
1015 assert!(!c.noop);
1016 assert_eq!(c.in_separator, ", ");
1017 assert_eq!(c.delay_ms, 500);
1018 assert_eq!(c.initial_delay_ms, 1000);
1019 assert!(c.max_messages_per_poll.is_none());
1020 assert!(c.on_consume.is_none());
1021 assert!(c.on_consume_failed.is_none());
1022 assert!(c.on_consume_batch_complete.is_none());
1023 assert!(!c.route_empty_result_set);
1024 assert!(c.use_iterator);
1025 assert!(c.expected_update_count.is_none());
1026 assert!(!c.break_batch_on_consume_fail);
1027 assert!(!c.batch);
1028 assert!(!c.use_message_body_for_sql);
1029 assert!(!c.allow_dynamic_query);
1030 assert!(c.ssl_mode.is_none());
1031 assert!(c.ssl_root_cert.is_none());
1032 assert!(c.ssl_cert.is_none());
1033 assert!(c.ssl_key.is_none());
1034 assert!(!c.always_populate_statement);
1036 assert!(c.allow_named_parameters);
1037 assert!(c.fetch_size.is_none());
1038 assert_eq!(c.transaction_mode, TransactionMode::Auto);
1039 assert!(c.repeat_count.is_none());
1040 assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
1041 assert_eq!(c.poll_strategy, PollStrategy::Sequential);
1042 }
1043
1044 #[test]
1045 fn ssl_none_by_default() {
1046 let c =
1047 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1048 assert!(c.ssl_mode.is_none());
1049 assert!(c.ssl_root_cert.is_none());
1050 assert!(c.ssl_cert.is_none());
1051 assert!(c.ssl_key.is_none());
1052 }
1053
1054 #[test]
1055 fn ssl_mode_from_uri() {
1056 let c = SqlEndpointConfig::from_uri(
1057 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1058 )
1059 .unwrap();
1060 assert_eq!(c.ssl_mode, Some("require".to_string()));
1061 assert!(c.ssl_root_cert.is_none());
1062 }
1063
1064 #[test]
1065 fn ssl_all_params_from_uri() {
1066 let c = SqlEndpointConfig::from_uri(
1067 "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1068 )
1069 .unwrap();
1070 assert_eq!(c.ssl_mode, Some("require".to_string()));
1071 assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1072 assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1073 assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1074 }
1075
1076 #[test]
1077 fn ssl_global_applied_to_endpoint() {
1078 let mut c =
1079 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1080 let global = SqlGlobalConfig::default()
1081 .with_ssl_mode("require")
1082 .with_ssl_root_cert("/etc/ssl/ca.pem");
1083 c.apply_defaults(&global);
1084 assert_eq!(c.ssl_mode, Some("require".to_string()));
1085 assert_eq!(c.ssl_root_cert, Some("/etc/ssl/ca.pem".to_string()));
1086 assert!(c.ssl_cert.is_none());
1087 assert!(c.ssl_key.is_none());
1088 }
1089
1090 #[test]
1091 fn ssl_uri_overrides_global() {
1092 let mut c = SqlEndpointConfig::from_uri(
1093 "sql:select 1?db_url=postgres://localhost/test&sslMode=verify-full",
1094 )
1095 .unwrap();
1096 let global = SqlGlobalConfig::default().with_ssl_mode("require");
1097 c.apply_defaults(&global);
1098 assert_eq!(c.ssl_mode, Some("verify-full".to_string()));
1099 }
1100
1101 #[test]
1102 fn config_wrong_scheme() {
1103 assert!(SqlEndpointConfig::from_uri("redis://localhost:6379").is_err());
1104 }
1105
1106 #[test]
1107 fn config_missing_db_url() {
1108 assert!(SqlEndpointConfig::from_uri("sql:select 1").is_err());
1109 }
1110
1111 #[test]
1112 fn config_output_type_select_one() {
1113 let c = SqlEndpointConfig::from_uri(
1114 "sql:select 1?db_url=postgres://localhost/test&outputType=SelectOne",
1115 )
1116 .unwrap();
1117 assert_eq!(c.output_type, SqlOutputType::SelectOne);
1118 }
1119
1120 #[test]
1121 fn config_output_type_stream_list() {
1122 let c = SqlEndpointConfig::from_uri(
1123 "sql:select 1?db_url=postgres://localhost/test&outputType=StreamList",
1124 )
1125 .unwrap();
1126 assert_eq!(c.output_type, SqlOutputType::StreamList);
1127 }
1128
1129 #[test]
1130 fn in_separator_default() {
1131 let c =
1132 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1133 assert_eq!(c.in_separator, ", ");
1134 }
1135
1136 #[test]
1137 fn in_separator_from_uri() {
1138 let c = SqlEndpointConfig::from_uri(
1139 "sql:select 1?db_url=postgres://localhost/test&inSeparator=;",
1140 )
1141 .unwrap();
1142 assert_eq!(c.in_separator, ";");
1143 }
1144
1145 #[test]
1146 fn in_separator_empty_rejected() {
1147 let result = SqlEndpointConfig::from_uri(
1148 "sql:select 1?db_url=postgres://localhost/test&inSeparator=",
1149 );
1150 assert!(result.is_err());
1151 let msg = format!("{:?}", result.unwrap_err());
1152 assert!(msg.contains("inSeparator") || msg.contains("empty"));
1153 }
1154
1155 #[test]
1156 fn config_consumer_options() {
1157 let c = SqlEndpointConfig::from_uri(
1158 "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"
1159 ).unwrap();
1160 assert_eq!(c.delay_ms, 2000);
1161 assert_eq!(c.initial_delay_ms, 500);
1162 assert_eq!(c.max_messages_per_poll, Some(10));
1163 assert_eq!(
1164 c.on_consume,
1165 Some("update t set done=true where id=:#id".to_string())
1166 );
1167 assert_eq!(
1168 c.on_consume_failed,
1169 Some("update t set failed=true where id=:#id".to_string())
1170 );
1171 assert_eq!(
1172 c.on_consume_batch_complete,
1173 Some("delete from t where done=true".to_string())
1174 );
1175 assert!(c.route_empty_result_set);
1176 assert!(!c.use_iterator);
1177 assert_eq!(c.expected_update_count, Some(1));
1178 assert!(c.break_batch_on_consume_fail);
1179 assert!(!c.bridge_error_handler);
1180 }
1181
1182 #[test]
1183 fn config_producer_options() {
1184 let c = SqlEndpointConfig::from_uri(
1185 "sql:insert into t values (#)?db_url=postgres://localhost/test&batch=true&useMessageBodyForSql=true&noop=true"
1186 ).unwrap();
1187 assert!(c.batch);
1188 assert!(c.use_message_body_for_sql);
1189 assert!(c.noop);
1190 }
1191
1192 #[test]
1193 fn config_pool_options() {
1194 let c = SqlEndpointConfig::from_uri(
1195 "sql:select 1?db_url=postgres://localhost/test&maxConnections=20&minConnections=3&idleTimeoutSecs=600&maxLifetimeSecs=3600"
1196 ).unwrap();
1197 assert_eq!(c.max_connections, Some(20));
1198 assert_eq!(c.min_connections, Some(3));
1199 assert_eq!(c.idle_timeout_secs, Some(600));
1200 assert_eq!(c.max_lifetime_secs, Some(3600));
1201 }
1202
1203 #[test]
1204 fn config_query_with_special_chars() {
1205 let c = SqlEndpointConfig::from_uri(
1206 "sql:select * from users where name = :#name and age > #?db_url=postgres://localhost/test",
1207 )
1208 .unwrap();
1209 assert_eq!(
1210 c.query,
1211 "select * from users where name = :#name and age > #"
1212 );
1213 }
1214
1215 #[test]
1216 fn output_type_from_str() {
1217 assert_eq!(
1218 "SelectList".parse::<SqlOutputType>().unwrap(),
1219 SqlOutputType::SelectList
1220 );
1221 assert_eq!(
1222 "SelectOne".parse::<SqlOutputType>().unwrap(),
1223 SqlOutputType::SelectOne
1224 );
1225 assert_eq!(
1226 "StreamList".parse::<SqlOutputType>().unwrap(),
1227 SqlOutputType::StreamList
1228 );
1229 assert!("Invalid".parse::<SqlOutputType>().is_err());
1230 }
1231
1232 #[tokio::test]
1234 async fn config_file_not_found() {
1235 let mut config = SqlEndpointConfig::from_uri(
1236 "sql:file:/nonexistent/path/query.sql?db_url=postgres://localhost/test",
1237 )
1238 .expect("from_uri should defer file reading");
1239 assert_eq!(
1241 config.source_path,
1242 Some("/nonexistent/path/query.sql".to_string())
1243 );
1244 assert!(config.query.is_empty());
1245
1246 let result = config.resolve_file_query().await;
1248 assert!(result.is_err());
1249 let msg = format!("{:?}", result.unwrap_err());
1250 assert!(msg.contains("Failed to read SQL file") || msg.contains("nonexistent"));
1251 }
1252
1253 #[tokio::test]
1255 async fn config_file_query() {
1256 use std::io::Write;
1257 let unique_name = format!(
1258 "test_sql_query_{}.sql",
1259 std::time::SystemTime::now()
1260 .duration_since(std::time::UNIX_EPOCH)
1261 .unwrap_or_default()
1262 .as_nanos()
1263 );
1264 let mut tmp = std::env::temp_dir();
1265 tmp.push(unique_name);
1266 {
1267 let mut f = std::fs::File::create(&tmp).unwrap();
1268 writeln!(f, "SELECT * FROM users").unwrap();
1269 }
1270 let uri = format!(
1271 "sql:file:{}?db_url=postgres://localhost/test",
1272 tmp.display()
1273 );
1274 let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
1275 assert!(c.query.is_empty());
1277 assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
1278
1279 c.resolve_file_query()
1281 .await
1282 .expect("file query should resolve");
1283 assert_eq!(c.query, "SELECT * FROM users");
1284 std::fs::remove_file(&tmp).ok();
1285 }
1286
1287 #[test]
1289 fn pool_fields_none_when_not_set() {
1290 let c =
1291 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1292 assert_eq!(c.max_connections, None);
1293 assert_eq!(c.min_connections, None);
1294 assert_eq!(c.idle_timeout_secs, None);
1295 assert_eq!(c.max_lifetime_secs, None);
1296 }
1297
1298 #[test]
1299 fn apply_defaults_fills_none() {
1300 let mut c =
1301 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1302 let global = SqlGlobalConfig {
1303 max_connections: 10,
1304 min_connections: 2,
1305 idle_timeout_secs: 600,
1306 max_lifetime_secs: 3600,
1307 ssl_mode: None,
1308 ssl_root_cert: None,
1309 ssl_cert: None,
1310 ssl_key: None,
1311 retry: NetworkRetryPolicy::default(),
1312 };
1313 c.apply_defaults(&global);
1314 assert_eq!(c.max_connections, Some(10));
1315 assert_eq!(c.min_connections, Some(2));
1316 assert_eq!(c.idle_timeout_secs, Some(600));
1317 assert_eq!(c.max_lifetime_secs, Some(3600));
1318 assert!(c.ssl_mode.is_none());
1319 assert!(c.ssl_root_cert.is_none());
1320 assert!(c.ssl_cert.is_none());
1321 assert!(c.ssl_key.is_none());
1322 }
1323
1324 #[test]
1325 fn apply_defaults_does_not_override() {
1326 let mut c = SqlEndpointConfig::from_uri(
1327 "sql:select 1?db_url=postgres://localhost/test&maxConnections=99&minConnections=5",
1328 )
1329 .unwrap();
1330 let global = SqlGlobalConfig {
1331 max_connections: 10,
1332 min_connections: 2,
1333 idle_timeout_secs: 600,
1334 max_lifetime_secs: 3600,
1335 ssl_mode: None,
1336 ssl_root_cert: None,
1337 ssl_cert: None,
1338 ssl_key: None,
1339 retry: NetworkRetryPolicy::default(),
1340 };
1341 c.apply_defaults(&global);
1342 assert_eq!(c.max_connections, Some(99));
1344 assert_eq!(c.min_connections, Some(5));
1345 assert_eq!(c.idle_timeout_secs, Some(600));
1347 assert_eq!(c.max_lifetime_secs, Some(3600));
1348 }
1349
1350 #[test]
1351 fn resolve_defaults_fills_remaining() {
1352 let mut c = SqlEndpointConfig::from_uri(
1353 "sql:select 1?db_url=postgres://localhost/test&maxConnections=7",
1354 )
1355 .unwrap();
1356 c.resolve_defaults();
1357 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)); }
1362
1363 #[test]
1364 fn global_config_builder() {
1365 let c = SqlGlobalConfig::default()
1366 .with_max_connections(20)
1367 .with_min_connections(3)
1368 .with_idle_timeout_secs(600)
1369 .with_max_lifetime_secs(3600)
1370 .with_ssl_mode("require")
1371 .with_ssl_root_cert("/ca.pem")
1372 .with_ssl_cert("/cert.pem")
1373 .with_ssl_key("/key.pem");
1374 assert_eq!(c.max_connections, 20);
1375 assert_eq!(c.min_connections, 3);
1376 assert_eq!(c.idle_timeout_secs, 600);
1377 assert_eq!(c.max_lifetime_secs, 3600);
1378 assert_eq!(c.ssl_mode, Some("require".to_string()));
1379 assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1380 assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1381 assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1382 }
1383
1384 #[test]
1385 fn enrich_postgres_ssl_mode() {
1386 let mut c = SqlEndpointConfig::from_uri(
1387 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1388 )
1389 .unwrap();
1390 c.resolve_defaults();
1391 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1392 assert!(url.contains("sslmode=require"), "got: {}", url);
1393 }
1394
1395 #[test]
1396 fn enrich_postgres_all_ssl() {
1397 let mut c = SqlEndpointConfig::from_uri(
1398 "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1399 )
1400 .unwrap();
1401 c.resolve_defaults();
1402 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1403 assert!(url.contains("sslmode=require"), "got: {}", url);
1404 assert!(url.contains("sslrootcert="), "got: {}", url);
1405 assert!(url.contains("sslcert="), "got: {}", url);
1406 assert!(url.contains("sslkey="), "got: {}", url);
1407 }
1408
1409 #[test]
1410 fn enrich_mysql_ssl() {
1411 let mut c = SqlEndpointConfig::from_uri(
1412 "sql:select 1?db_url=mysql://localhost/test&sslMode=require",
1413 )
1414 .unwrap();
1415 c.resolve_defaults();
1416 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1417 assert!(url.contains("ssl-mode=require"), "got: {}", url);
1418 }
1419
1420 #[test]
1421 fn enrich_existing_query_params() {
1422 let mut c = SqlEndpointConfig::from_uri(
1423 "sql:select 1?db_url=postgres://localhost/test?existing=1&sslMode=require",
1424 )
1425 .unwrap();
1426 c.resolve_defaults();
1427 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1428 assert!(url.contains("existing=1"), "got: {}", url);
1429 assert!(url.contains("sslmode=require"), "got: {}", url);
1430 }
1431
1432 #[test]
1433 fn enrich_override_existing() {
1434 let mut c = SqlEndpointConfig::from_uri(
1435 "sql:select 1?db_url=postgres://localhost/test?sslmode=allow&sslMode=require",
1436 )
1437 .unwrap();
1438 c.resolve_defaults();
1439 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1440 assert!(url.contains("sslmode=require"), "got: {}", url);
1441 assert!(!url.contains("sslmode=allow"), "got: {}", url);
1442 }
1443
1444 #[test]
1445 fn enrich_applies_postgres_defaults() {
1446 let mut c =
1447 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1448 c.resolve_defaults();
1449 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1450 assert!(
1452 url.contains("sslmode=require"),
1453 "expected sslmode=require, got: {}",
1454 url
1455 );
1456 assert!(
1457 url.contains("connect_timeout=10"),
1458 "expected connect_timeout=10, got: {}",
1459 url
1460 );
1461 }
1462
1463 #[test]
1464 fn enrich_mysql_defaults_to_prefer() {
1465 let mut c =
1466 SqlEndpointConfig::from_uri("sql:select 1?db_url=mysql://localhost/test").unwrap();
1467 c.resolve_defaults();
1468 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1469 assert!(
1470 url.contains("ssl-mode=prefer"),
1471 "expected ssl-mode=prefer, got: {}",
1472 url
1473 );
1474 assert!(
1475 url.contains("connect_timeout=10"),
1476 "expected connect_timeout=10, got: {}",
1477 url
1478 );
1479 }
1480
1481 #[test]
1482 fn enrich_preserves_explicit_connect_timeout() {
1483 let mut c = SqlEndpointConfig::from_uri(
1484 "sql:select 1?db_url=postgres://localhost/test?connect_timeout=5",
1485 )
1486 .unwrap();
1487 c.resolve_defaults();
1488 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1489 assert!(
1490 url.contains("connect_timeout=5"),
1491 "expected explicit connect_timeout=5 preserved, got: {}",
1492 url
1493 );
1494 assert!(
1495 url.contains("sslmode=require"),
1496 "expected sslmode=require, got: {}",
1497 url
1498 );
1499 }
1500
1501 #[test]
1502 fn enrich_url_encodes_paths() {
1503 let mut c = SqlEndpointConfig::from_uri(
1504 "sql:select 1?db_url=postgres://localhost/test&sslRootCert=/path/to/my%20cert.pem",
1505 )
1506 .unwrap();
1507 c.resolve_defaults();
1508 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1509 assert!(url.contains("sslrootcert="), "got: {}", url);
1510 }
1511
1512 #[test]
1513 fn enrich_unsupported_scheme_returns_unchanged() {
1514 let mut c = SqlEndpointConfig::from_uri(
1515 "sql:select 1?db_url=sqlite://localhost/test.db&sslMode=require",
1516 )
1517 .unwrap();
1518 c.resolve_defaults();
1519 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1520 assert_eq!(url, "sqlite://localhost/test.db");
1521 }
1522
1523 #[test]
1524 fn enrich_invalid_url_returns_error() {
1525 let mut c = SqlEndpointConfig::from_uri(
1526 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1527 )
1528 .unwrap();
1529 c.resolve_defaults();
1530 let result = enrich_db_url_with_ssl("://not-a-valid-url", &c);
1531 assert!(result.is_err());
1532 }
1533
1534 #[test]
1538 fn debug_redacts_db_url_with_password() {
1539 let c = SqlEndpointConfig::from_uri(
1540 "sql:select 1?db_url=postgres://user:secret123@localhost/test",
1541 )
1542 .unwrap();
1543 let debug_output = format!("{:?}", c);
1544 assert!(
1545 !debug_output.contains("secret123"),
1546 "Debug output must not contain password: {}",
1547 debug_output
1548 );
1549 assert!(
1550 debug_output.contains("***"),
1551 "Debug output must contain redacted marker: {}",
1552 debug_output
1553 );
1554 }
1555
1556 #[test]
1557 fn debug_redacts_ssl_key() {
1558 let c = SqlEndpointConfig::from_uri(
1559 "sql:select 1?db_url=postgres://localhost/test&sslKey=/secret/key.pem",
1560 )
1561 .unwrap();
1562 let debug_output = format!("{:?}", c);
1563 assert!(
1564 !debug_output.contains("/secret/key.pem"),
1565 "Debug output must not contain ssl_key path: {}",
1566 debug_output
1567 );
1568 }
1569
1570 #[test]
1571 fn debug_global_config_redacts_ssl_key() {
1572 let c = SqlGlobalConfig::default().with_ssl_key("/secret/key.pem");
1573 let debug_output = format!("{:?}", c);
1574 assert!(
1575 !debug_output.contains("/secret/key.pem"),
1576 "Debug output must not contain ssl_key path: {}",
1577 debug_output
1578 );
1579 assert!(
1580 debug_output.contains("***"),
1581 "Debug output must contain redacted marker: {}",
1582 debug_output
1583 );
1584 }
1585
1586 #[test]
1587 fn redact_db_url_with_credentials() {
1588 assert_eq!(
1589 redact_db_url("postgres://user:pass@host/db"),
1590 "postgres://***:***@host/db"
1591 );
1592 }
1593
1594 #[test]
1595 fn redact_db_url_without_credentials() {
1596 assert_eq!(redact_db_url("sqlite::memory:"), "sqlite::memory:");
1597 }
1598
1599 #[test]
1600 fn redact_db_url_invalid_returns_original() {
1601 assert_eq!(redact_db_url("not-a-url"), "not-a-url");
1602 }
1603
1604 #[test]
1606 fn use_placeholder_defaults_to_true() {
1607 let c =
1608 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1609 assert!(c.use_placeholder);
1610 }
1611
1612 #[test]
1613 fn use_placeholder_false_from_uri() {
1614 let c = SqlEndpointConfig::from_uri(
1615 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=false",
1616 )
1617 .unwrap();
1618 assert!(!c.use_placeholder);
1619 }
1620
1621 #[test]
1622 fn use_placeholder_true_from_uri() {
1623 let c = SqlEndpointConfig::from_uri(
1624 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=true",
1625 )
1626 .unwrap();
1627 assert!(c.use_placeholder);
1628 }
1629
1630 #[test]
1632 fn use_placeholder_rejects_invalid_value() {
1633 let result = SqlEndpointConfig::from_uri(
1634 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=1",
1635 );
1636 assert!(result.is_err());
1637 let msg = format!("{:?}", result.unwrap_err());
1638 assert!(msg.contains("usePlaceholder") && msg.contains("true") && msg.contains("false"));
1639 }
1640
1641 #[test]
1642 fn use_placeholder_rejects_typo_tru() {
1643 let result = SqlEndpointConfig::from_uri(
1644 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=tru",
1645 );
1646 assert!(result.is_err());
1647 }
1648
1649 #[test]
1650 fn use_placeholder_rejects_yes() {
1651 let result = SqlEndpointConfig::from_uri(
1652 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=yes",
1653 );
1654 assert!(result.is_err());
1655 }
1656
1657 #[test]
1658 fn noop_rejects_invalid_value() {
1659 let result =
1660 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&noop=1");
1661 assert!(result.is_err());
1662 let msg = format!("{:?}", result.unwrap_err());
1663 assert!(msg.contains("noop"));
1664 }
1665
1666 #[test]
1667 fn batch_rejects_invalid_value() {
1668 let result =
1669 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&batch=yes");
1670 assert!(result.is_err());
1671 let msg = format!("{:?}", result.unwrap_err());
1672 assert!(msg.contains("batch"));
1673 }
1674
1675 #[test]
1676 fn route_empty_result_set_rejects_invalid_value() {
1677 let result = SqlEndpointConfig::from_uri(
1678 "sql:select 1?db_url=postgres://localhost/test&routeEmptyResultSet=on",
1679 );
1680 assert!(result.is_err());
1681 }
1682
1683 #[test]
1684 fn use_iterator_rejects_invalid_value() {
1685 let result = SqlEndpointConfig::from_uri(
1686 "sql:select 1?db_url=postgres://localhost/test&useIterator=1",
1687 );
1688 assert!(result.is_err());
1689 }
1690
1691 #[test]
1692 fn break_batch_on_consume_fail_rejects_invalid_value() {
1693 let result = SqlEndpointConfig::from_uri(
1694 "sql:select 1?db_url=postgres://localhost/test&breakBatchOnConsumeFail=yes",
1695 );
1696 assert!(result.is_err());
1697 }
1698
1699 #[test]
1700 fn use_message_body_for_sql_rejects_invalid_value() {
1701 let result = SqlEndpointConfig::from_uri(
1702 "sql:select 1?db_url=postgres://localhost/test&useMessageBodyForSql=1",
1703 );
1704 assert!(result.is_err());
1705 }
1706
1707 #[test]
1709 fn boolean_params_case_insensitive() {
1710 let c = SqlEndpointConfig::from_uri(
1711 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=TRUE&noop=FALSE&batch=True&useIterator=False&bridgeErrorHandler=TRUE",
1712 )
1713 .unwrap();
1714 assert!(c.use_placeholder);
1715 assert!(!c.noop);
1716 assert!(c.batch);
1717 assert!(!c.use_iterator);
1718 assert!(c.bridge_error_handler);
1719 }
1720
1721 #[test]
1723 fn multi_char_placeholder_rejected() {
1724 let result = SqlEndpointConfig::from_uri(
1725 "sql:select 1?db_url=postgres://localhost/test&placeholder=##",
1726 );
1727 assert!(result.is_err());
1728 let msg = format!("{:?}", result.unwrap_err());
1729 assert!(msg.contains("placeholder") && msg.contains("one character"));
1730 }
1731
1732 #[test]
1733 fn non_ascii_placeholder_rejected() {
1734 let result = SqlEndpointConfig::from_uri(
1735 "sql:select 1?db_url=postgres://localhost/test&placeholder=%C2%A2",
1736 );
1737 assert!(result.is_err());
1738 }
1739
1740 #[test]
1741 fn single_char_placeholder_accepted() {
1742 let c = SqlEndpointConfig::from_uri(
1743 "sql:select 1?db_url=postgres://localhost/test&placeholder=$",
1744 )
1745 .unwrap();
1746 assert_eq!(c.placeholder, '$');
1747 }
1748
1749 #[test]
1750 fn empty_placeholder_falls_back_to_default() {
1751 let c = SqlEndpointConfig::from_uri(
1753 "sql:select 1?db_url=postgres://localhost/test&placeholder=",
1754 )
1755 .unwrap();
1756 assert_eq!(c.placeholder, '#');
1757 }
1758
1759 #[tokio::test]
1761 async fn file_query_cached_in_config() {
1762 use std::io::Write;
1763 let unique_name = format!(
1764 "test_sql_cached_{}.sql",
1765 std::time::SystemTime::now()
1766 .duration_since(std::time::UNIX_EPOCH)
1767 .unwrap_or_default()
1768 .as_nanos()
1769 );
1770 let mut tmp = std::env::temp_dir();
1771 tmp.push(unique_name);
1772 {
1773 let mut f = std::fs::File::create(&tmp).unwrap();
1774 writeln!(f, "SELECT * FROM cached_test").unwrap();
1775 }
1776 let uri = format!(
1777 "sql:file:{}?db_url=postgres://localhost/test",
1778 tmp.display()
1779 );
1780 let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
1781 assert!(c.query.is_empty());
1783 assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
1784
1785 c.resolve_file_query()
1787 .await
1788 .expect("resolve should succeed");
1789 assert_eq!(c.query, "SELECT * FROM cached_test");
1790
1791 std::fs::remove_file(&tmp).ok();
1793 assert_eq!(c.query, "SELECT * FROM cached_test");
1794 }
1795
1796 #[test]
1800 fn always_populate_statement_defaults_to_false() {
1801 let c =
1802 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1803 assert!(!c.always_populate_statement);
1804 }
1805
1806 #[test]
1807 fn always_populate_statement_from_uri() {
1808 let c = SqlEndpointConfig::from_uri(
1809 "sql:select 1?db_url=postgres://localhost/test&alwaysPopulateStatement=true",
1810 )
1811 .unwrap();
1812 assert!(c.always_populate_statement);
1813 }
1814
1815 #[test]
1817 fn allow_named_parameters_defaults_to_true() {
1818 let c =
1819 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1820 assert!(c.allow_named_parameters);
1821 }
1822
1823 #[test]
1824 fn allow_named_parameters_false_from_uri() {
1825 let c = SqlEndpointConfig::from_uri(
1826 "sql:select 1?db_url=postgres://localhost/test&allowNamedParameters=false",
1827 )
1828 .unwrap();
1829 assert!(!c.allow_named_parameters);
1830 }
1831
1832 #[test]
1834 fn fetch_size_defaults_to_none() {
1835 let c =
1836 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1837 assert!(c.fetch_size.is_none());
1838 }
1839
1840 #[test]
1841 fn fetch_size_from_uri() {
1842 let c = SqlEndpointConfig::from_uri(
1843 "sql:select 1?db_url=postgres://localhost/test&fetchSize=1000",
1844 )
1845 .unwrap();
1846 assert_eq!(c.fetch_size, Some(1000));
1847 }
1848
1849 #[test]
1851 fn transaction_mode_defaults_to_auto() {
1852 let c =
1853 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1854 assert_eq!(c.transaction_mode, TransactionMode::Auto);
1855 }
1856
1857 #[test]
1858 fn transaction_mode_managed_from_uri() {
1859 let c = SqlEndpointConfig::from_uri(
1860 "sql:select 1?db_url=postgres://localhost/test&transactionMode=Managed",
1861 )
1862 .unwrap();
1863 assert_eq!(c.transaction_mode, TransactionMode::Managed);
1864 }
1865
1866 #[test]
1867 fn transaction_mode_invalid_rejected() {
1868 let result = SqlEndpointConfig::from_uri(
1869 "sql:select 1?db_url=postgres://localhost/test&transactionMode=Invalid",
1870 );
1871 assert!(result.is_err());
1872 }
1873
1874 #[test]
1876 fn repeat_count_defaults_to_none() {
1877 let c =
1878 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1879 assert!(c.repeat_count.is_none());
1880 }
1881
1882 #[test]
1883 fn repeat_count_from_uri() {
1884 let c = SqlEndpointConfig::from_uri(
1885 "sql:select 1?db_url=postgres://localhost/test&repeatCount=10",
1886 )
1887 .unwrap();
1888 assert_eq!(c.repeat_count, Some(10));
1889 }
1890
1891 #[test]
1893 fn break_on_empty_defaults_to_false() {
1894 let c =
1895 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1896 assert!(!c.break_on_empty);
1897 }
1898
1899 #[test]
1900 fn break_on_empty_true_from_uri() {
1901 let c = SqlEndpointConfig::from_uri(
1902 "sql:select 1?db_url=postgres://localhost/test&breakOnEmpty=true",
1903 )
1904 .unwrap();
1905 assert!(c.break_on_empty);
1906 }
1907
1908 #[test]
1909 fn break_on_empty_explicit_false_from_uri() {
1910 let c = SqlEndpointConfig::from_uri(
1911 "sql:select 1?db_url=postgres://localhost/test&breakOnEmpty=false",
1912 )
1913 .unwrap();
1914 assert!(!c.break_on_empty);
1915 }
1916
1917 #[test]
1919 fn processing_strategy_defaults_to_direct() {
1920 let c =
1921 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1922 assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
1923 }
1924
1925 #[test]
1926 fn processing_strategy_scheduled_from_uri() {
1927 let c = SqlEndpointConfig::from_uri(
1928 "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Scheduled",
1929 )
1930 .unwrap();
1931 assert_eq!(c.processing_strategy, ProcessingStrategy::Scheduled);
1932 }
1933
1934 #[test]
1935 fn processing_strategy_invalid_rejected() {
1936 let result = SqlEndpointConfig::from_uri(
1937 "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Invalid",
1938 );
1939 assert!(result.is_err());
1940 }
1941
1942 #[test]
1944 fn poll_strategy_defaults_to_sequential() {
1945 let c =
1946 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1947 assert_eq!(c.poll_strategy, PollStrategy::Sequential);
1948 }
1949
1950 #[test]
1951 fn poll_strategy_burst_from_uri() {
1952 let c = SqlEndpointConfig::from_uri(
1953 "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Burst",
1954 )
1955 .unwrap();
1956 assert_eq!(c.poll_strategy, PollStrategy::Burst);
1957 }
1958
1959 #[test]
1960 fn poll_strategy_invalid_rejected() {
1961 let result = SqlEndpointConfig::from_uri(
1962 "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Invalid",
1963 );
1964 assert!(result.is_err());
1965 }
1966
1967 #[test]
1970 fn sql_endpoint_config_has_retry_policy() {
1971 let cfg = SqlEndpointConfig::from_uri(
1972 "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=3&retryInitialDelayMs=500",
1973 )
1974 .expect("parse");
1975 assert_eq!(cfg.retry.max_attempts, 3);
1976 assert_eq!(
1977 cfg.retry.initial_delay,
1978 std::time::Duration::from_millis(500)
1979 );
1980 assert!(cfg.retry.enabled);
1981 }
1982
1983 #[test]
1984 fn sql_endpoint_config_retry_defaults_when_unspecified() {
1985 let cfg =
1986 SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
1987 assert!(cfg.retry.enabled);
1989 assert_eq!(cfg.retry.max_attempts, 10); }
1991
1992 #[test]
1993 fn sql_global_config_has_retry_default() {
1994 let cfg = SqlGlobalConfig::default();
1995 assert!(cfg.retry.enabled);
1996 }
1997
1998 #[test]
1999 fn retry_policy_parse_full_uri_params() {
2000 let cfg = SqlEndpointConfig::from_uri(
2001 "sql:select 1?db_url=sqlite::memory:&retryEnabled=false&retryMaxAttempts=7&retryInitialDelayMs=1000&retryMultiplier=3.0&retryMaxDelayMs=60000&retryJitter=0.5",
2002 )
2003 .expect("parse");
2004 assert!(!cfg.retry.enabled);
2005 assert_eq!(cfg.retry.max_attempts, 7);
2006 assert_eq!(
2007 cfg.retry.initial_delay,
2008 std::time::Duration::from_millis(1000)
2009 );
2010 assert!((cfg.retry.multiplier - 3.0).abs() < f64::EPSILON);
2011 assert_eq!(cfg.retry.max_delay, std::time::Duration::from_millis(60000));
2012 assert!((cfg.retry.jitter_factor - 0.5).abs() < f64::EPSILON);
2013 }
2014
2015 #[test]
2016 fn retry_policy_from_uri_survives_apply_defaults_with_global() {
2017 let mut ep = SqlEndpointConfig::from_uri(
2018 "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=10&retryInitialDelayMs=500",
2019 )
2020 .expect("parse");
2021 let global = SqlGlobalConfig::default(); ep.apply_defaults(&global);
2023 assert_eq!(ep.retry.max_attempts, 10);
2025 assert_eq!(
2026 ep.retry.initial_delay,
2027 std::time::Duration::from_millis(500)
2028 );
2029 }
2030
2031 #[test]
2032 fn retry_policy_falls_back_to_global_when_uri_has_no_retry_params() {
2033 let mut ep =
2034 SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
2035 let mut global = SqlGlobalConfig::default();
2036 global.retry.max_attempts = 7;
2037 ep.apply_defaults(&global);
2038 assert_eq!(ep.retry.max_attempts, 7);
2040 }
2041
2042 #[test]
2043 fn from_uri_with_datasource_name() {
2044 let cfg = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders").unwrap();
2045 assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2046 assert!(cfg.db_url.is_empty());
2047 }
2048
2049 #[test]
2050 fn from_uri_with_datasource_and_behavior_override() {
2051 let cfg =
2052 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&outputType=SelectOne")
2053 .unwrap();
2054 assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2055 }
2056
2057 #[test]
2058 fn from_uri_datasource_rejects_pool_override() {
2059 let result =
2060 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&maxConnections=50");
2061 assert!(result.is_err());
2062 let msg = result.unwrap_err().to_string();
2063 assert!(msg.contains("pool-affecting"));
2064 }
2065
2066 #[test]
2067 fn from_uri_neither_datasource_nor_db_url_is_error() {
2068 let result = SqlEndpointConfig::from_uri("sql:SELECT 1");
2069 assert!(result.is_err());
2070 }
2071
2072 #[test]
2073 fn from_uri_db_url_inline_still_works() {
2074 let cfg =
2075 SqlEndpointConfig::from_uri("sql:SELECT 1?db_url=postgres://localhost/test").unwrap();
2076 assert!(cfg.datasource_name.is_none());
2077 assert_eq!(cfg.db_url, "postgres://localhost/test");
2078 }
2079
2080 #[test]
2081 fn from_uri_datasource_rejects_ssl_mode() {
2082 let result = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslMode=require");
2083 assert!(result.is_err());
2084 let msg = result.unwrap_err().to_string();
2085 assert!(msg.contains("pool-affecting"));
2086 }
2087
2088 #[test]
2089 fn from_uri_datasource_rejects_ssl_root_cert() {
2090 let result =
2091 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslRootCert=/ca.pem");
2092 assert!(result.is_err());
2093 }
2094
2095 #[test]
2096 fn from_uri_datasource_rejects_db_url() {
2097 let result = SqlEndpointConfig::from_uri(
2098 "sql:SELECT 1?datasource=orders&db_url=postgres://evil:5432/pwned",
2099 );
2100 assert!(result.is_err());
2101 let msg = result.unwrap_err().to_string();
2102 assert!(msg.contains("db_url") && msg.contains("datasource"));
2103 }
2104}