Skip to main content

camel_component_sql/
config.rs

1use std::str::FromStr;
2use std::time::Duration;
3
4use camel_api::component_metadata::{ComponentMetadata, UriOption};
5use camel_component_api::CamelError;
6use camel_component_api::NetworkRetryPolicy;
7use camel_component_api::{UriComponents, UriConfig, parse_uri};
8use tracing::warn;
9
10/// Redaction helper: returns `Some("***")` if the option is `Some`, otherwise `None`.
11fn redacted_opt(opt: &Option<String>) -> Option<&'static str> {
12    if opt.is_some() { Some("***") } else { None }
13}
14
15/// Redacts the user:password portion of a database URL for safe display.
16/// Returns `"scheme://***@host/db"` for URLs with userinfo, or the original URL otherwise.
17pub fn redact_db_url(db_url: &str) -> String {
18    match url::Url::parse(db_url) {
19        Ok(mut parsed) => {
20            if parsed.username().is_empty() && parsed.password().is_none() {
21                return db_url.to_string();
22            }
23            let _ = parsed.set_username("***");
24            let _ = parsed.set_password(Some("***"));
25            parsed.to_string()
26        }
27        Err(_) => {
28            // url::Url rejected the string, so structured redaction is
29            // impossible. Best-effort: if the raw string carries a userinfo
30            // separator (`@` after `://`), redact that segment rather than
31            // risk leaking credentials (rc-7rup). Strings without `@` carry
32            // no userinfo and are returned unchanged.
33            if let Some(scheme_end) = db_url.find("://") {
34                let after_scheme = &db_url[scheme_end + 3..];
35                if let Some(at) = after_scheme.find('@') {
36                    let scheme = &db_url[..scheme_end + 3];
37                    let host_part = &after_scheme[at + 1..];
38                    return format!("{scheme}***@{host_part}");
39                }
40            }
41            db_url.to_string()
42        }
43    }
44}
45
46/// Output type for SQL query results.
47#[derive(Debug, Clone, PartialEq, Default)]
48pub enum SqlOutputType {
49    /// Return all rows as a list.
50    #[default]
51    SelectList,
52    /// Return a single row (first result).
53    SelectOne,
54    /// Stream results as an async iterator.
55    StreamList,
56}
57
58impl FromStr for SqlOutputType {
59    type Err = CamelError;
60
61    fn from_str(s: &str) -> Result<Self, Self::Err> {
62        match s {
63            "SelectList" => Ok(SqlOutputType::SelectList),
64            "SelectOne" => Ok(SqlOutputType::SelectOne),
65            "StreamList" => Ok(SqlOutputType::StreamList),
66            _ => Err(CamelError::InvalidUri(format!(
67                "Unknown output type: {}",
68                s
69            ))),
70        }
71    }
72}
73
74/// Transaction mode for SQL operations.
75///
76/// - `Auto`: Each statement auto-commits (default, current behavior).
77/// - `Managed`: Explicit transaction boundaries (future; currently logs a warning
78///   and falls back to Auto).
79///
80// TODO(SQL-002): managed transaction mode — implement explicit transaction boundaries
81#[derive(Debug, Clone, PartialEq, Default)]
82pub enum TransactionMode {
83    /// Auto-commit each statement (default).
84    #[default]
85    Auto,
86    /// Managed transactions — not yet implemented.
87    Managed,
88}
89
90impl FromStr for TransactionMode {
91    type Err = CamelError;
92
93    fn from_str(s: &str) -> Result<Self, Self::Err> {
94        match s {
95            "Auto" => Ok(TransactionMode::Auto),
96            "Managed" => Ok(TransactionMode::Managed),
97            _ => Err(CamelError::InvalidUri(format!(
98                "Unknown transaction mode: {}. Expected 'Auto' or 'Managed'",
99                s
100            ))),
101        }
102    }
103}
104
105impl std::fmt::Display for TransactionMode {
106    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107        match self {
108            TransactionMode::Auto => write!(f, "Auto"),
109            TransactionMode::Managed => write!(f, "Managed"),
110        }
111    }
112}
113
114/// Processing strategy for SQL consumers.
115#[derive(Debug, Clone, PartialEq, Default)]
116pub enum ProcessingStrategy {
117    /// Process rows directly in the polling task (default).
118    #[default]
119    Direct,
120    /// Schedule processing via a separate task (deferred execution).
121    Scheduled,
122}
123
124impl FromStr for ProcessingStrategy {
125    type Err = CamelError;
126
127    fn from_str(s: &str) -> Result<Self, Self::Err> {
128        match s {
129            "Direct" => Ok(ProcessingStrategy::Direct),
130            "Scheduled" => Ok(ProcessingStrategy::Scheduled),
131            _ => Err(CamelError::InvalidUri(format!(
132                "Unknown processing strategy: {}. Expected 'Direct' or 'Scheduled'",
133                s
134            ))),
135        }
136    }
137}
138
139impl std::fmt::Display for ProcessingStrategy {
140    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
141        match self {
142            ProcessingStrategy::Direct => write!(f, "Direct"),
143            ProcessingStrategy::Scheduled => write!(f, "Scheduled"),
144        }
145    }
146}
147
148/// Poll strategy for SQL consumers.
149#[derive(Debug, Clone, PartialEq, Default)]
150pub enum PollStrategy {
151    /// Poll sequentially with delay between polls (default).
152    #[default]
153    Sequential,
154    /// Poll in bursts — execute multiple queries in rapid succession.
155    Burst,
156}
157
158impl FromStr for PollStrategy {
159    type Err = CamelError;
160
161    fn from_str(s: &str) -> Result<Self, Self::Err> {
162        match s {
163            "Sequential" => Ok(PollStrategy::Sequential),
164            "Burst" => Ok(PollStrategy::Burst),
165            _ => Err(CamelError::InvalidUri(format!(
166                "Unknown poll strategy: {}. Expected 'Sequential' or 'Burst'",
167                s
168            ))),
169        }
170    }
171}
172
173impl std::fmt::Display for PollStrategy {
174    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
175        match self {
176            PollStrategy::Sequential => write!(f, "Sequential"),
177            PollStrategy::Burst => write!(f, "Burst"),
178        }
179    }
180}
181
182/// Global configuration for SQL component.
183///
184/// This struct supports serde deserialization with defaults and builder methods.
185/// It holds pool configuration that can be applied as defaults to endpoints.
186///
187/// **Security note:** `Debug` implementation redacts sensitive fields (SSL key paths).
188#[derive(Clone, PartialEq, serde::Deserialize)]
189#[serde(default)]
190pub struct SqlGlobalConfig {
191    pub max_connections: u32,
192    pub min_connections: u32,
193    pub idle_timeout_secs: u64,
194    pub max_lifetime_secs: u64,
195    // SSL/TLS
196    pub ssl_mode: Option<String>,
197    pub ssl_root_cert: Option<String>,
198    pub ssl_cert: Option<String>,
199    pub ssl_key: Option<String>,
200    /// Retry policy for transient database connection failures.
201    #[serde(default)]
202    pub retry: NetworkRetryPolicy,
203}
204
205impl std::fmt::Debug for SqlGlobalConfig {
206    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
207        f.debug_struct("SqlGlobalConfig")
208            .field("max_connections", &self.max_connections)
209            .field("min_connections", &self.min_connections)
210            .field("idle_timeout_secs", &self.idle_timeout_secs)
211            .field("max_lifetime_secs", &self.max_lifetime_secs)
212            .field("ssl_mode", &self.ssl_mode)
213            .field("ssl_root_cert", &self.ssl_root_cert)
214            .field("ssl_cert", &self.ssl_cert)
215            .field("ssl_key", &redacted_opt(&self.ssl_key))
216            .field("retry", &self.retry)
217            .finish()
218    }
219}
220
221impl Default for SqlGlobalConfig {
222    fn default() -> Self {
223        Self {
224            max_connections: 5,
225            min_connections: 1,
226            idle_timeout_secs: 300,
227            max_lifetime_secs: 1800,
228            ssl_mode: None,
229            ssl_root_cert: None,
230            ssl_cert: None,
231            ssl_key: None,
232            retry: NetworkRetryPolicy::default(),
233        }
234    }
235}
236
237impl SqlGlobalConfig {
238    pub fn new() -> Self {
239        Self::default()
240    }
241
242    pub fn with_max_connections(mut self, value: u32) -> Self {
243        self.max_connections = value;
244        self
245    }
246
247    pub fn with_min_connections(mut self, value: u32) -> Self {
248        self.min_connections = value;
249        self
250    }
251
252    pub fn with_idle_timeout_secs(mut self, value: u64) -> Self {
253        self.idle_timeout_secs = value;
254        self
255    }
256
257    pub fn with_max_lifetime_secs(mut self, value: u64) -> Self {
258        self.max_lifetime_secs = value;
259        self
260    }
261
262    pub fn with_ssl_mode(mut self, value: impl Into<String>) -> Self {
263        self.ssl_mode = Some(value.into());
264        self
265    }
266
267    pub fn with_ssl_root_cert(mut self, value: impl Into<String>) -> Self {
268        self.ssl_root_cert = Some(value.into());
269        self
270    }
271
272    pub fn with_ssl_cert(mut self, value: impl Into<String>) -> Self {
273        self.ssl_cert = Some(value.into());
274        self
275    }
276
277    pub fn with_ssl_key(mut self, value: impl Into<String>) -> Self {
278        self.ssl_key = Some(value.into());
279        self
280    }
281
282    pub fn with_retry(mut self, value: NetworkRetryPolicy) -> Self {
283        self.retry = value;
284        self
285    }
286}
287
288/// Configuration for SQL component endpoints.
289///
290/// URI format: `sql:<query>?db_url=<url>&param1=val1&param2=val2`
291///
292/// The query can be inline SQL or a file reference with `file:` prefix:
293/// - `sql:SELECT * FROM users?db_url=...` - inline SQL
294/// - `sql:file:/path/to/query.sql?db_url=...` - read SQL from file
295///
296/// **Note on file-based queries (SQL-014):** When the query path starts with `file:`,
297/// the file is NOT read synchronously during `from_uri()`. Instead, the file path is
298/// stored in `source_path` and the query is resolved asynchronously via `resolve_file_query()`
299/// during async initialization (producer pool init or consumer start). This avoids
300/// blocking I/O in the synchronous URI parsing path.
301///
302/// **Security note:** `Debug` implementation redacts the `db_url` (which may contain credentials)
303/// and `ssl_key` path. Use `redact_db_url()` for safe logging of database URLs.
304#[derive(Clone)]
305pub struct SqlEndpointConfig {
306    // Connection
307    /// Database connection URL (optional when datasource_name is set).
308    pub db_url: String,
309    /// Named datasource reference (from CamelConfig.datasources).
310    pub datasource_name: Option<String>,
311    /// Maximum connections in the pool. None = use global default.
312    pub max_connections: Option<u32>,
313    /// Minimum connections in the pool. None = use global default.
314    pub min_connections: Option<u32>,
315    /// Idle timeout in seconds. None = use global default.
316    pub idle_timeout_secs: Option<u64>,
317    /// Maximum connection lifetime in seconds. None = use global default.
318    pub max_lifetime_secs: Option<u64>,
319
320    // Query
321    /// The SQL query (from URI path or file).
322    pub query: String,
323    /// Path to the file containing the SQL query (when using `file:` prefix).
324    pub source_path: Option<String>,
325    /// Output type for query results. Default: SelectList.
326    pub output_type: SqlOutputType,
327    /// Placeholder character for parameters. Default: '#'.
328    pub placeholder: char,
329    /// If true, process parameter placeholders in queries. Default: true.
330    pub use_placeholder: bool,
331    /// If true, don't execute the query (dry run). Default: false.
332    pub noop: bool,
333    /// Separator for IN clause expansion. Default: ", ".
334    pub in_separator: String,
335
336    // SQL-005: always populate statement even if body is null/empty
337    /// If true, always bind parameters even if the exchange body is null/empty
338    /// (uses empty defaults). Default: false.
339    pub always_populate_statement: bool,
340
341    // SQL-011: allow named parameters
342    /// If true, recognize `:name` style placeholders and map them from exchange
343    /// headers or body fields. Default: true.
344    pub allow_named_parameters: bool,
345
346    // SQL-016: fetch size hint
347    /// Fetch size hint for query results. None = driver default.
348    pub fetch_size: Option<u32>,
349
350    // SQL-002: transaction mode
351    /// Transaction mode for SQL operations. Default: Auto.
352    pub transaction_mode: TransactionMode,
353
354    // Consumer (polling)
355    /// Delay between polls in milliseconds. Default: 500.
356    pub delay_ms: u64,
357    /// Initial delay before first poll in milliseconds. Default: 1000.
358    pub initial_delay_ms: u64,
359    /// Maximum messages per poll.
360    pub max_messages_per_poll: Option<i32>,
361    /// SQL to execute after consuming each message.
362    pub on_consume: Option<String>,
363    /// SQL to execute if consumption fails.
364    pub on_consume_failed: Option<String>,
365    /// SQL to execute after consuming a batch.
366    pub on_consume_batch_complete: Option<String>,
367    /// Route empty result sets. Default: false.
368    pub route_empty_result_set: bool,
369    /// Use iterator for results. Default: true.
370    pub use_iterator: bool,
371    /// Expected number of rows affected.
372    pub expected_update_count: Option<i64>,
373    /// Break batch on consume failure. Default: false.
374    pub break_batch_on_consume_fail: bool,
375    /// Bridge poll errors into route error handling. Default: false.
376    pub bridge_error_handler: bool,
377
378    // SQL-015: repeat count for consumer polling
379    /// Maximum number of polls before the consumer stops. Omit (None) for infinite
380    /// polling. `0` = never poll (the consumer exits before its first poll); useful
381    /// for disabling a route via config without removing it. See also camel-timer's
382    /// `repeatCount`, which implements the same `0 = never` semantic.
383    pub repeat_count: Option<u32>,
384
385    // SQL-020: break on empty poll
386    /// When true, the consumer stops after a poll that returns zero rows
387    /// (SelectList mode only; ignored with a warning in StreamList mode).
388    /// Default: false.
389    pub break_on_empty: bool,
390
391    // SQL-017: processing strategy
392    /// Processing strategy for consumer. Default: Direct.
393    pub processing_strategy: ProcessingStrategy,
394
395    // SQL-018: poll strategy
396    /// Poll strategy for consumer. Default: Sequential.
397    pub poll_strategy: PollStrategy,
398
399    // Producer
400    /// Enable batch mode. Default: false.
401    pub batch: bool,
402    /// Use message body for SQL. Default: false.
403    pub use_message_body_for_sql: bool,
404    /// Allow queries to be sourced from exchange headers (`CamelSql.Query`) or body.
405    /// Default `false` — dynamic queries are SQLi risk. Set `true` for backward compat.
406    pub allow_dynamic_query: bool,
407
408    // SSL/TLS
409    /// SSL mode for the connection. None = use global default.
410    pub ssl_mode: Option<String>,
411    /// Path to SSL root certificate. None = use global default.
412    pub ssl_root_cert: Option<String>,
413    /// Path to SSL client certificate. None = use global default.
414    pub ssl_cert: Option<String>,
415    /// Path to SSL client key. None = use global default.
416    pub ssl_key: Option<String>,
417
418    /// Retry policy for transient database connection failures.
419    pub retry: NetworkRetryPolicy,
420
421    /// Whether `retry` was explicitly set via URI params. Used by
422    /// [`apply_defaults`] to decide whether URI values win over
423    /// the global config. Internal tracking flag, not serialized.
424    retry_set_from_uri: bool,
425}
426
427impl std::fmt::Debug for SqlEndpointConfig {
428    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
429        f.debug_struct("SqlEndpointConfig")
430            .field("db_url", &redact_db_url(&self.db_url))
431            .field("datasource_name", &self.datasource_name)
432            .field("max_connections", &self.max_connections)
433            .field("min_connections", &self.min_connections)
434            .field("idle_timeout_secs", &self.idle_timeout_secs)
435            .field("max_lifetime_secs", &self.max_lifetime_secs)
436            .field("query", &self.query)
437            .field("source_path", &self.source_path)
438            .field("output_type", &self.output_type)
439            .field("placeholder", &self.placeholder)
440            .field("use_placeholder", &self.use_placeholder)
441            .field("noop", &self.noop)
442            .field("in_separator", &self.in_separator)
443            .field("always_populate_statement", &self.always_populate_statement)
444            .field("allow_named_parameters", &self.allow_named_parameters)
445            .field("fetch_size", &self.fetch_size)
446            .field("transaction_mode", &self.transaction_mode)
447            .field("delay_ms", &self.delay_ms)
448            .field("initial_delay_ms", &self.initial_delay_ms)
449            .field("max_messages_per_poll", &self.max_messages_per_poll)
450            .field("on_consume", &self.on_consume)
451            .field("on_consume_failed", &self.on_consume_failed)
452            .field("on_consume_batch_complete", &self.on_consume_batch_complete)
453            .field("route_empty_result_set", &self.route_empty_result_set)
454            .field("use_iterator", &self.use_iterator)
455            .field("expected_update_count", &self.expected_update_count)
456            .field(
457                "break_batch_on_consume_fail",
458                &self.break_batch_on_consume_fail,
459            )
460            .field("bridge_error_handler", &self.bridge_error_handler)
461            .field("repeat_count", &self.repeat_count)
462            .field("break_on_empty", &self.break_on_empty)
463            .field("processing_strategy", &self.processing_strategy)
464            .field("poll_strategy", &self.poll_strategy)
465            .field("batch", &self.batch)
466            .field("use_message_body_for_sql", &self.use_message_body_for_sql)
467            .field("allow_dynamic_query", &self.allow_dynamic_query)
468            .field("ssl_mode", &self.ssl_mode)
469            .field("ssl_root_cert", &self.ssl_root_cert)
470            .field("ssl_cert", &self.ssl_cert)
471            .field("ssl_key", &redacted_opt(&self.ssl_key))
472            .field("retry", &self.retry)
473            .finish()
474    }
475}
476
477/// Private container for macro-derived `uri_options()` and `metadata()`.
478///
479/// Mirrors `SqlEndpointConfig`'s URI-parsed params exactly. `SqlEndpointConfig`
480/// holds extra non-URI fields (`retry`, `retry_set_from_uri`) and custom
481/// `from_uri` logic; metadata delegation targets this inner type.
482#[derive(Debug, Clone, UriConfig)]
483#[allow(dead_code)]
484#[uri_scheme = "sql"]
485#[uri_config(
486    skip_impl,
487    metadata(
488        scheme = "sql",
489        description = "Execute SQL against a configured datasource",
490        producer,
491        consumer
492    ),
493    crate = "camel_component_api"
494)]
495struct SqlUriConfig {
496    // ── Connection ──
497    #[uri_param(name = "db_url", desc = "Database connection URL", secret)]
498    pub db_url: Option<String>,
499    #[uri_param(
500        name = "datasource",
501        desc = "Named datasource reference (from CamelConfig.datasources)"
502    )]
503    pub datasource: Option<String>,
504    #[uri_param(name = "maxConnections", desc = "Maximum connections in the pool")]
505    pub max_connections: Option<u64>,
506    #[uri_param(name = "minConnections", desc = "Minimum connections in the pool")]
507    pub min_connections: Option<u64>,
508    #[uri_param(name = "idleTimeoutSecs", desc = "Idle timeout in seconds")]
509    pub idle_timeout_secs: Option<u64>,
510    #[uri_param(
511        name = "maxLifetimeSecs",
512        desc = "Maximum connection lifetime in seconds"
513    )]
514    pub max_lifetime_secs: Option<u64>,
515
516    // ── Query ──
517    #[uri_param(
518        name = "outputType",
519        kind = "enum:SelectList,SelectOne,StreamList",
520        default = "SelectList",
521        desc = "Output type for query results"
522    )]
523    pub output_type: Option<String>,
524    #[uri_param(
525        name = "placeholder",
526        default = "#",
527        desc = "Placeholder character for parameters"
528    )]
529    pub placeholder: Option<String>,
530    #[uri_param(
531        name = "usePlaceholder",
532        default = "true",
533        desc = "Process parameter placeholders in queries"
534    )]
535    pub use_placeholder: Option<bool>,
536    #[uri_param(
537        name = "noop",
538        default = "false",
539        desc = "Dry-run mode (don't execute the query)"
540    )]
541    pub noop: Option<bool>,
542    #[uri_param(
543        name = "inSeparator",
544        default = ", ",
545        desc = "Separator for IN clause expansion"
546    )]
547    pub in_separator: Option<String>,
548    #[uri_param(
549        name = "alwaysPopulateStatement",
550        default = "false",
551        desc = "Always bind parameters even if the exchange body is null or empty"
552    )]
553    pub always_populate_statement: Option<bool>,
554    #[uri_param(
555        name = "allowNamedParameters",
556        default = "true",
557        desc = "Recognize :name style placeholders from headers or body"
558    )]
559    pub allow_named_parameters: Option<bool>,
560    #[uri_param(name = "fetchSize", desc = "Fetch size hint for query results")]
561    pub fetch_size: Option<u64>,
562    #[uri_param(
563        name = "transactionMode",
564        kind = "enum:Auto,Managed",
565        default = "Auto",
566        desc = "Transaction mode for SQL operations"
567    )]
568    pub transaction_mode: Option<String>,
569
570    // ── Consumer ──
571    #[uri_param(
572        name = "delay",
573        default = "500",
574        desc = "Delay between polls in milliseconds"
575    )]
576    pub delay: Option<u64>,
577    #[uri_param(
578        name = "initialDelay",
579        default = "1000",
580        desc = "Initial delay before first poll in milliseconds"
581    )]
582    pub initial_delay: Option<u64>,
583    #[uri_param(name = "maxMessagesPerPoll", desc = "Maximum messages per poll")]
584    pub max_messages_per_poll: Option<u64>,
585    #[uri_param(
586        name = "onConsume",
587        desc = "SQL to execute after consuming each message"
588    )]
589    pub on_consume: Option<String>,
590    #[uri_param(name = "onConsumeFailed", desc = "SQL to execute if consumption fails")]
591    pub on_consume_failed: Option<String>,
592    #[uri_param(
593        name = "onConsumeBatchComplete",
594        desc = "SQL to execute after consuming a batch"
595    )]
596    pub on_consume_batch_complete: Option<String>,
597    #[uri_param(
598        name = "routeEmptyResultSet",
599        default = "false",
600        desc = "Route empty result sets"
601    )]
602    pub route_empty_result_set: Option<bool>,
603    #[uri_param(
604        name = "useIterator",
605        default = "true",
606        desc = "Use iterator for results"
607    )]
608    pub use_iterator: Option<bool>,
609    #[uri_param(
610        name = "expectedUpdateCount",
611        desc = "Expected number of rows affected"
612    )]
613    pub expected_update_count: Option<u64>,
614    #[uri_param(
615        name = "breakBatchOnConsumeFail",
616        default = "false",
617        desc = "Break batch on consume failure"
618    )]
619    pub break_batch_on_consume_fail: Option<bool>,
620    #[uri_param(
621        name = "bridgeErrorHandler",
622        default = "false",
623        desc = "Bridge poll errors into route error handling"
624    )]
625    pub bridge_error_handler: Option<bool>,
626    #[uri_param(
627        name = "repeatCount",
628        desc = "Maximum number of polls before the consumer stops"
629    )]
630    pub repeat_count: Option<u64>,
631    #[uri_param(
632        name = "breakOnEmpty",
633        default = "false",
634        desc = "Stop the consumer after a poll that returns zero rows"
635    )]
636    pub break_on_empty: Option<bool>,
637    #[uri_param(
638        name = "processingStrategy",
639        kind = "enum:Direct,Scheduled",
640        default = "Direct",
641        desc = "Processing strategy for the consumer"
642    )]
643    pub processing_strategy: Option<String>,
644    #[uri_param(
645        name = "pollStrategy",
646        kind = "enum:Sequential,Burst",
647        default = "Sequential",
648        desc = "Poll strategy for the consumer"
649    )]
650    pub poll_strategy: Option<String>,
651
652    // ── Producer ──
653    #[uri_param(name = "batch", default = "false", desc = "Enable batch mode")]
654    pub batch: Option<bool>,
655    #[uri_param(
656        name = "useMessageBodyForSql",
657        default = "false",
658        desc = "Use message body for SQL"
659    )]
660    pub use_message_body_for_sql: Option<bool>,
661    #[uri_param(
662        name = "allowDynamicQuery",
663        default = "false",
664        desc = "Allow queries to be sourced from exchange headers or body"
665    )]
666    pub allow_dynamic_query: Option<bool>,
667
668    // ── SSL ──
669    #[uri_param(name = "sslMode", desc = "SSL mode for the connection")]
670    pub ssl_mode: Option<String>,
671    #[uri_param(name = "sslRootCert", desc = "Path to SSL root certificate")]
672    pub ssl_root_cert: Option<String>,
673    #[uri_param(name = "sslCert", desc = "Path to SSL client certificate")]
674    pub ssl_cert: Option<String>,
675    #[uri_param(name = "sslKey", desc = "Path to SSL client key", secret)]
676    pub ssl_key: Option<String>,
677}
678
679impl SqlEndpointConfig {
680    /// Component metadata for the sql scheme, derived from
681    /// `#[uri_param]` annotations on `SqlUriConfig`.
682    pub fn metadata() -> ComponentMetadata {
683        SqlUriConfig::metadata()
684    }
685
686    /// Generated URI option definitions for the sql scheme,
687    /// derived from `#[uri_param]` annotations on `SqlUriConfig`.
688    pub fn uri_options() -> Vec<UriOption> {
689        SqlUriConfig::uri_options()
690    }
691
692    /// Apply defaults from global config, filling None fields without overriding.
693    pub fn apply_defaults(&mut self, defaults: &SqlGlobalConfig) {
694        if self.max_connections.is_none() {
695            self.max_connections = Some(defaults.max_connections);
696        }
697        if self.min_connections.is_none() {
698            self.min_connections = Some(defaults.min_connections);
699        }
700        if self.idle_timeout_secs.is_none() {
701            self.idle_timeout_secs = Some(defaults.idle_timeout_secs);
702        }
703        if self.max_lifetime_secs.is_none() {
704            self.max_lifetime_secs = Some(defaults.max_lifetime_secs);
705        }
706        if self.ssl_mode.is_none() {
707            self.ssl_mode = defaults.ssl_mode.clone();
708        }
709        if self.ssl_root_cert.is_none() {
710            self.ssl_root_cert = defaults.ssl_root_cert.clone();
711        }
712        if self.ssl_cert.is_none() {
713            self.ssl_cert = defaults.ssl_cert.clone();
714        }
715        if self.ssl_key.is_none() {
716            self.ssl_key = defaults.ssl_key.clone();
717        }
718        // retry: URI wins when set_from_uri, else global fills the gap
719        if !self.retry_set_from_uri {
720            self.retry = defaults.retry.clone();
721        }
722    }
723
724    /// Resolve any remaining None fields with built-in defaults.
725    pub fn resolve_defaults(&mut self) {
726        let defaults = SqlGlobalConfig::default();
727        self.apply_defaults(&defaults);
728    }
729
730    /// Asynchronously read the SQL query from the file referenced by `source_path`.
731    ///
732    /// This is the async replacement for the blocking `std::fs::read_to_string` that
733    /// was previously called in `from_uri()`. Must be invoked during async init
734    /// (producer pool init or consumer start) — never in a synchronous context.
735    ///
736    /// After this call, `self.query` contains the file content (trimmed) and
737    /// `self.source_path` is cleared to prevent re-reading.
738    pub async fn resolve_file_query(&mut self) -> Result<(), CamelError> {
739        if let Some(file_path) = self.source_path.take() {
740            let contents = tokio::fs::read_to_string(&file_path).await.map_err(|e| {
741                CamelError::Config(format!("Failed to read SQL file '{}': {}", file_path, e))
742            })?;
743            self.query = contents.trim().to_string();
744            // Keep source_path as Some so tests can still verify the original path
745            self.source_path = Some(file_path);
746        }
747        Ok(())
748    }
749}
750
751struct SslParamMapping {
752    pg_key: &'static str,
753    mysql_key: &'static str,
754}
755
756const SSL_MAPPINGS: &[(&str, SslParamMapping)] = &[
757    (
758        "sslMode",
759        SslParamMapping {
760            pg_key: "sslmode",
761            mysql_key: "ssl-mode",
762        },
763    ),
764    (
765        "sslRootCert",
766        SslParamMapping {
767            pg_key: "sslrootcert",
768            mysql_key: "ssl-ca",
769        },
770    ),
771    (
772        "sslCert",
773        SslParamMapping {
774            pg_key: "sslcert",
775            mysql_key: "ssl-cert",
776        },
777    ),
778    (
779        "sslKey",
780        SslParamMapping {
781            pg_key: "sslkey",
782            mysql_key: "ssl-key",
783        },
784    ),
785];
786
787pub fn enrich_db_url_with_ssl(
788    db_url: &str,
789    config: &SqlEndpointConfig,
790) -> Result<String, CamelError> {
791    enrich_db_url_with_ssl_params(
792        db_url,
793        config.ssl_mode.as_deref(),
794        config.ssl_root_cert.as_deref(),
795        config.ssl_cert.as_deref(),
796        config.ssl_key.as_deref(),
797    )
798}
799
800pub(crate) fn enrich_db_url_with_ssl_params(
801    db_url: &str,
802    ssl_mode: Option<&str>,
803    ssl_root_cert: Option<&str>,
804    ssl_cert: Option<&str>,
805    ssl_key: Option<&str>,
806) -> Result<String, CamelError> {
807    let mut parsed = url::Url::parse(db_url).map_err(|e| {
808        CamelError::InvalidUri(format!(
809            "Cannot parse database URL for SSL enrichment: {}",
810            e
811        ))
812    })?;
813
814    let scheme = parsed.scheme();
815    if scheme.starts_with("sqlite") {
816        if ssl_mode.is_some() || ssl_root_cert.is_some() || ssl_cert.is_some() || ssl_key.is_some()
817        {
818            warn!(
819                "SSL options configured for SQLite database URL, but SQLite does not support SSL/TLS; ignoring sslMode/sslRootCert/sslCert/sslKey"
820            );
821        }
822        return Ok(db_url.to_string());
823    }
824
825    if scheme != "postgres" && scheme != "postgresql" && scheme != "mysql" {
826        return Ok(db_url.to_string());
827    }
828    let is_mysql = scheme == "mysql";
829
830    // Compute effective ssl_mode with per-driver default when none is specified.
831    let effective_ssl_mode: &str = ssl_mode.unwrap_or(if is_mysql { "prefer" } else { "require" });
832
833    let ssl_params: Vec<(&str, &str)> = [
834        Some(("sslMode", effective_ssl_mode)),
835        ssl_root_cert.map(|v| ("sslRootCert", v)),
836        ssl_cert.map(|v| ("sslCert", v)),
837        ssl_key.map(|v| ("sslKey", v)),
838    ]
839    .into_iter()
840    .flatten()
841    .collect();
842
843    let mut query_pairs = parsed.query_pairs().collect::<Vec<_>>();
844    for (camel_name, value) in &ssl_params {
845        if let Some((_, mapping)) = SSL_MAPPINGS.iter().find(|(name, _)| *name == *camel_name) {
846            let driver_key = if is_mysql {
847                mapping.mysql_key
848            } else {
849                mapping.pg_key
850            };
851
852            if let Some(pos) = query_pairs.iter().position(|(k, _)| k == driver_key) {
853                query_pairs[pos].1 = (*value).into();
854            } else {
855                query_pairs.push((driver_key.into(), (*value).into()));
856            }
857        }
858    }
859
860    // Add TCP connect timeout (10s) if not already set.
861    // This is the sqlx URL-level connect_timeout, NOT the pool acquire_timeout.
862    if !query_pairs.iter().any(|(k, _)| k == "connect_timeout") {
863        query_pairs.push(("connect_timeout".into(), "10".into()));
864    }
865
866    {
867        let mut serializer = url::form_urlencoded::Serializer::new(String::new());
868        for (k, v) in &query_pairs {
869            serializer.append_pair(k, v);
870        }
871        parsed.set_query(Some(&serializer.finish()));
872    }
873
874    Ok(parsed.to_string())
875}
876
877impl UriConfig for SqlEndpointConfig {
878    fn scheme() -> &'static str {
879        "sql"
880    }
881
882    fn from_uri(uri: &str) -> Result<Self, CamelError> {
883        let parts = parse_uri(uri)?;
884        Self::from_components(parts)
885    }
886
887    fn from_components(parts: UriComponents) -> Result<Self, CamelError> {
888        // Validate scheme
889        if parts.scheme != Self::scheme() {
890            return Err(CamelError::InvalidUri(format!(
891                "expected scheme '{}' but got '{}'",
892                Self::scheme(),
893                parts.scheme
894            )));
895        }
896
897        let params = &parts.params;
898
899        // Handle file: prefix for query
900        // SQL-014: defer file reading to async init path to avoid blocking I/O
901        // in the synchronous URI parsing path. Store the path; resolve_file_query()
902        // must be called during async initialization (producer pool init or consumer start).
903        let (query, source_path) = if parts.path.starts_with("file:") {
904            let file_path = parts.path.trim_start_matches("file:").to_string();
905            (String::new(), Some(file_path))
906        } else {
907            (parts.path.clone(), None)
908        };
909
910        // Optional parameter: db_url (required when datasource is not set)
911        let db_url = params.get("db_url").cloned().unwrap_or_default();
912
913        // Named datasource reference (from CamelConfig.datasources)
914        let datasource_name = params.get("datasource").cloned();
915
916        // Connection parameters - None when not set by URI param
917        let max_connections = params.get("maxConnections").and_then(|v| v.parse().ok());
918        let min_connections = params.get("minConnections").and_then(|v| v.parse().ok());
919        let idle_timeout_secs = params.get("idleTimeoutSecs").and_then(|v| v.parse().ok());
920        let max_lifetime_secs = params.get("maxLifetimeSecs").and_then(|v| v.parse().ok());
921
922        // Query parameters
923        let output_type = params
924            .get("outputType")
925            .map(|s| s.parse())
926            .transpose()?
927            .unwrap_or_default();
928        let placeholder = params
929            .get("placeholder")
930            .filter(|v| !v.is_empty())
931            .map(|v| {
932                if v.chars().count() != 1 {
933                    return Err(CamelError::InvalidUri(format!(
934                        "placeholder must be exactly one character, got '{}'",
935                        v
936                    )));
937                }
938                if !v.is_ascii() {
939                    return Err(CamelError::InvalidUri(
940                        "placeholder must be a single ASCII character".to_string(),
941                    ));
942                }
943                Ok(v.chars().next().unwrap()) // allow-unwrap
944            })
945            .transpose()?
946            .unwrap_or('#');
947        /// Parse a boolean URI parameter strictly.
948        ///
949        /// Accepts only `"true"` or `"false"` (case-insensitive). Any other value
950        /// returns `CamelError::InvalidUri` to prevent silent misconfiguration.
951        fn parse_bool_param(name: &str, value: &str) -> Result<bool, CamelError> {
952            if value.eq_ignore_ascii_case("true") {
953                Ok(true)
954            } else if value.eq_ignore_ascii_case("false") {
955                Ok(false)
956            } else {
957                Err(CamelError::InvalidUri(format!(
958                    "{} must be 'true' or 'false', got '{}'",
959                    name, value
960                )))
961            }
962        }
963
964        let use_placeholder = params
965            .get("usePlaceholder")
966            .map(|v| parse_bool_param("usePlaceholder", v))
967            .transpose()?
968            .unwrap_or(true);
969        let noop = params
970            .get("noop")
971            .map(|v| parse_bool_param("noop", v))
972            .transpose()?
973            .unwrap_or(false);
974        let in_separator = params
975            .get("inSeparator")
976            .map(|v| v.to_string())
977            .unwrap_or_else(|| ", ".to_string());
978        if in_separator.is_empty() {
979            return Err(CamelError::InvalidUri(
980                "inSeparator must not be empty".to_string(),
981            ));
982        }
983
984        // SQL-005: alwaysPopulateStatement
985        let always_populate_statement = params
986            .get("alwaysPopulateStatement")
987            .map(|v| parse_bool_param("alwaysPopulateStatement", v))
988            .transpose()?
989            .unwrap_or(false);
990
991        // SQL-011: allowNamedParameters
992        let allow_named_parameters = params
993            .get("allowNamedParameters")
994            .map(|v| parse_bool_param("allowNamedParameters", v))
995            .transpose()?
996            .unwrap_or(true);
997
998        // SQL-016: fetchSize
999        let fetch_size = params.get("fetchSize").and_then(|v| v.parse().ok());
1000
1001        // SQL-002: transactionMode
1002        let transaction_mode = params
1003            .get("transactionMode")
1004            .map(|s| s.parse())
1005            .transpose()?
1006            .unwrap_or_default();
1007
1008        // Consumer parameters
1009        let delay_ms = params
1010            .get("delay")
1011            .and_then(|v| v.parse().ok())
1012            .unwrap_or(500);
1013        let initial_delay_ms = params
1014            .get("initialDelay")
1015            .and_then(|v| v.parse().ok())
1016            .unwrap_or(1000);
1017        let max_messages_per_poll = params
1018            .get("maxMessagesPerPoll")
1019            .and_then(|v| v.parse().ok());
1020        let on_consume = params.get("onConsume").cloned();
1021        let on_consume_failed = params.get("onConsumeFailed").cloned();
1022        let on_consume_batch_complete = params.get("onConsumeBatchComplete").cloned();
1023        let route_empty_result_set = params
1024            .get("routeEmptyResultSet")
1025            .map(|v| parse_bool_param("routeEmptyResultSet", v))
1026            .transpose()?
1027            .unwrap_or(false);
1028        let use_iterator = params
1029            .get("useIterator")
1030            .map(|v| parse_bool_param("useIterator", v))
1031            .transpose()?
1032            .unwrap_or(true);
1033        let expected_update_count = params
1034            .get("expectedUpdateCount")
1035            .and_then(|v| v.parse().ok());
1036        let break_batch_on_consume_fail = params
1037            .get("breakBatchOnConsumeFail")
1038            .map(|v| parse_bool_param("breakBatchOnConsumeFail", v))
1039            .transpose()?
1040            .unwrap_or(false);
1041        let bridge_error_handler = params
1042            .get("bridgeErrorHandler")
1043            .map(|v| parse_bool_param("bridgeErrorHandler", v))
1044            .transpose()?
1045            .unwrap_or(false);
1046
1047        // SQL-015: repeatCount
1048        let repeat_count = params.get("repeatCount").and_then(|v| v.parse().ok());
1049
1050        // SQL-020: breakOnEmpty
1051        let break_on_empty = params
1052            .get("breakOnEmpty")
1053            .map(|v| parse_bool_param("breakOnEmpty", v))
1054            .transpose()?
1055            .unwrap_or(false);
1056
1057        // SQL-017: processingStrategy
1058        let processing_strategy = params
1059            .get("processingStrategy")
1060            .map(|s| s.parse())
1061            .transpose()?
1062            .unwrap_or_default();
1063
1064        // SQL-018: pollStrategy
1065        let poll_strategy = params
1066            .get("pollStrategy")
1067            .map(|s| s.parse())
1068            .transpose()?
1069            .unwrap_or_default();
1070
1071        // Producer parameters
1072        let batch = params
1073            .get("batch")
1074            .map(|v| parse_bool_param("batch", v))
1075            .transpose()?
1076            .unwrap_or(false);
1077        let use_message_body_for_sql = params
1078            .get("useMessageBodyForSql")
1079            .map(|v| parse_bool_param("useMessageBodyForSql", v))
1080            .transpose()?
1081            .unwrap_or(false);
1082        let allow_dynamic_query = params
1083            .get("allowDynamicQuery")
1084            .map(|v| parse_bool_param("allowDynamicQuery", v))
1085            .transpose()?
1086            .unwrap_or(false);
1087        let ssl_mode = params.get("sslMode").cloned();
1088        let ssl_root_cert = params.get("sslRootCert").cloned();
1089        let ssl_cert = params.get("sslCert").cloned();
1090        let ssl_key = params.get("sslKey").cloned();
1091
1092        // Parse retry policy from URI params
1093        let mut retry = NetworkRetryPolicy::default();
1094        let mut retry_set_from_uri = false;
1095        if let Some(raw) = params.get("retryEnabled") {
1096            retry.enabled = raw.parse::<bool>().map_err(|_| {
1097                CamelError::InvalidUri(format!("retryEnabled must be a boolean, got '{raw}'"))
1098            })?;
1099            retry_set_from_uri = true;
1100        }
1101        if let Some(raw) = params.get("retryMaxAttempts") {
1102            retry.max_attempts = raw.parse::<u32>().map_err(|_| {
1103                CamelError::InvalidUri(format!("retryMaxAttempts must be a u32, got '{raw}'"))
1104            })?;
1105            retry_set_from_uri = true;
1106        }
1107        if let Some(raw) = params.get("retryInitialDelayMs") {
1108            retry.initial_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
1109                CamelError::InvalidUri(format!("retryInitialDelayMs must be a u64, got '{raw}'"))
1110            })?);
1111            retry_set_from_uri = true;
1112        }
1113        if let Some(raw) = params.get("retryMultiplier") {
1114            retry.multiplier = raw.parse::<f64>().map_err(|_| {
1115                CamelError::InvalidUri(format!("retryMultiplier must be a f64, got '{raw}'"))
1116            })?;
1117            retry_set_from_uri = true;
1118        }
1119        if let Some(raw) = params.get("retryMaxDelayMs") {
1120            retry.max_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
1121                CamelError::InvalidUri(format!("retryMaxDelayMs must be a u64, got '{raw}'"))
1122            })?);
1123            retry_set_from_uri = true;
1124        }
1125        if let Some(raw) = params.get("retryJitter") {
1126            retry.jitter_factor = raw.parse::<f64>().map_err(|_| {
1127                CamelError::InvalidUri(format!("retryJitter must be a f64, got '{raw}'"))
1128            })?;
1129            retry_set_from_uri = true;
1130        }
1131
1132        if datasource_name.is_none() && db_url.is_empty() {
1133            return Err(CamelError::Config(
1134                "either 'datasource' or 'db_url' parameter is required".to_string(),
1135            ));
1136        }
1137
1138        if datasource_name.is_some() && !db_url.is_empty() {
1139            return Err(CamelError::InvalidUri(
1140                "'db_url' not allowed with named datasource — use 'datasource' alone".to_string(),
1141            ));
1142        }
1143
1144        if datasource_name.is_some() {
1145            let overrides: Vec<&str> = {
1146                let mut v = Vec::new();
1147                if max_connections.is_some() {
1148                    v.push("maxConnections");
1149                }
1150                if min_connections.is_some() {
1151                    v.push("minConnections");
1152                }
1153                if idle_timeout_secs.is_some() {
1154                    v.push("idleTimeoutSecs");
1155                }
1156                if max_lifetime_secs.is_some() {
1157                    v.push("maxLifetimeSecs");
1158                }
1159                if ssl_mode.is_some() {
1160                    v.push("sslMode");
1161                }
1162                if ssl_root_cert.is_some() {
1163                    v.push("sslRootCert");
1164                }
1165                if ssl_cert.is_some() {
1166                    v.push("sslCert");
1167                }
1168                if ssl_key.is_some() {
1169                    v.push("sslKey");
1170                }
1171                v
1172            };
1173            if !overrides.is_empty() {
1174                return Err(CamelError::InvalidUri(format!(
1175                    "pool-affecting params not allowed with named datasource: {}",
1176                    overrides.join(", ")
1177                )));
1178            }
1179        }
1180
1181        Ok(Self {
1182            db_url,
1183            datasource_name,
1184            max_connections,
1185            min_connections,
1186            idle_timeout_secs,
1187            max_lifetime_secs,
1188            query,
1189            source_path,
1190            output_type,
1191            placeholder,
1192            use_placeholder,
1193            noop,
1194            in_separator,
1195            always_populate_statement,
1196            allow_named_parameters,
1197            fetch_size,
1198            transaction_mode,
1199            delay_ms,
1200            initial_delay_ms,
1201            max_messages_per_poll,
1202            on_consume,
1203            on_consume_failed,
1204            on_consume_batch_complete,
1205            route_empty_result_set,
1206            use_iterator,
1207            expected_update_count,
1208            break_batch_on_consume_fail,
1209            bridge_error_handler,
1210            repeat_count,
1211            break_on_empty,
1212            processing_strategy,
1213            poll_strategy,
1214            batch,
1215            use_message_body_for_sql,
1216            allow_dynamic_query,
1217            ssl_mode,
1218            ssl_root_cert,
1219            ssl_cert,
1220            ssl_key,
1221            retry,
1222            retry_set_from_uri,
1223        })
1224    }
1225}
1226
1227#[cfg(test)]
1228mod tests {
1229    use super::*;
1230    use camel_component_api::NetworkRetryPolicy;
1231
1232    #[test]
1233    fn config_defaults() {
1234        let mut c =
1235            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1236        c.resolve_defaults();
1237        assert_eq!(c.query, "select 1");
1238        assert_eq!(c.db_url, "postgres://localhost/test");
1239        assert_eq!(c.max_connections, Some(5));
1240        assert_eq!(c.min_connections, Some(1));
1241        assert_eq!(c.idle_timeout_secs, Some(300));
1242        assert_eq!(c.max_lifetime_secs, Some(1800));
1243        assert_eq!(c.output_type, SqlOutputType::SelectList);
1244        assert_eq!(c.placeholder, '#');
1245        assert!(!c.noop);
1246        assert_eq!(c.in_separator, ", ");
1247        assert_eq!(c.delay_ms, 500);
1248        assert_eq!(c.initial_delay_ms, 1000);
1249        assert!(c.max_messages_per_poll.is_none());
1250        assert!(c.on_consume.is_none());
1251        assert!(c.on_consume_failed.is_none());
1252        assert!(c.on_consume_batch_complete.is_none());
1253        assert!(!c.route_empty_result_set);
1254        assert!(c.use_iterator);
1255        assert!(c.expected_update_count.is_none());
1256        assert!(!c.break_batch_on_consume_fail);
1257        assert!(!c.batch);
1258        assert!(!c.use_message_body_for_sql);
1259        assert!(!c.allow_dynamic_query);
1260        assert!(c.ssl_mode.is_none());
1261        assert!(c.ssl_root_cert.is_none());
1262        assert!(c.ssl_cert.is_none());
1263        assert!(c.ssl_key.is_none());
1264        // SQL-005/SQL-011/SQL-016/SQL-002/SQL-015/SQL-017/SQL-018 defaults
1265        assert!(!c.always_populate_statement);
1266        assert!(c.allow_named_parameters);
1267        assert!(c.fetch_size.is_none());
1268        assert_eq!(c.transaction_mode, TransactionMode::Auto);
1269        assert!(c.repeat_count.is_none());
1270        assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
1271        assert_eq!(c.poll_strategy, PollStrategy::Sequential);
1272    }
1273
1274    #[test]
1275    fn ssl_none_by_default() {
1276        let c =
1277            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1278        assert!(c.ssl_mode.is_none());
1279        assert!(c.ssl_root_cert.is_none());
1280        assert!(c.ssl_cert.is_none());
1281        assert!(c.ssl_key.is_none());
1282    }
1283
1284    #[test]
1285    fn ssl_mode_from_uri() {
1286        let c = SqlEndpointConfig::from_uri(
1287            "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1288        )
1289        .unwrap();
1290        assert_eq!(c.ssl_mode, Some("require".to_string()));
1291        assert!(c.ssl_root_cert.is_none());
1292    }
1293
1294    #[test]
1295    fn ssl_all_params_from_uri() {
1296        let c = SqlEndpointConfig::from_uri(
1297            "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1298        )
1299        .unwrap();
1300        assert_eq!(c.ssl_mode, Some("require".to_string()));
1301        assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1302        assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1303        assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1304    }
1305
1306    #[test]
1307    fn ssl_global_applied_to_endpoint() {
1308        let mut c =
1309            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1310        let global = SqlGlobalConfig::default()
1311            .with_ssl_mode("require")
1312            .with_ssl_root_cert("/etc/ssl/ca.pem");
1313        c.apply_defaults(&global);
1314        assert_eq!(c.ssl_mode, Some("require".to_string()));
1315        assert_eq!(c.ssl_root_cert, Some("/etc/ssl/ca.pem".to_string()));
1316        assert!(c.ssl_cert.is_none());
1317        assert!(c.ssl_key.is_none());
1318    }
1319
1320    #[test]
1321    fn ssl_uri_overrides_global() {
1322        let mut c = SqlEndpointConfig::from_uri(
1323            "sql:select 1?db_url=postgres://localhost/test&sslMode=verify-full",
1324        )
1325        .unwrap();
1326        let global = SqlGlobalConfig::default().with_ssl_mode("require");
1327        c.apply_defaults(&global);
1328        assert_eq!(c.ssl_mode, Some("verify-full".to_string()));
1329    }
1330
1331    #[test]
1332    fn config_wrong_scheme() {
1333        assert!(SqlEndpointConfig::from_uri("redis://localhost:6379").is_err());
1334    }
1335
1336    #[test]
1337    fn config_missing_db_url() {
1338        assert!(SqlEndpointConfig::from_uri("sql:select 1").is_err());
1339    }
1340
1341    #[test]
1342    fn config_output_type_select_one() {
1343        let c = SqlEndpointConfig::from_uri(
1344            "sql:select 1?db_url=postgres://localhost/test&outputType=SelectOne",
1345        )
1346        .unwrap();
1347        assert_eq!(c.output_type, SqlOutputType::SelectOne);
1348    }
1349
1350    #[test]
1351    fn config_output_type_stream_list() {
1352        let c = SqlEndpointConfig::from_uri(
1353            "sql:select 1?db_url=postgres://localhost/test&outputType=StreamList",
1354        )
1355        .unwrap();
1356        assert_eq!(c.output_type, SqlOutputType::StreamList);
1357    }
1358
1359    #[test]
1360    fn in_separator_default() {
1361        let c =
1362            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1363        assert_eq!(c.in_separator, ", ");
1364    }
1365
1366    #[test]
1367    fn in_separator_from_uri() {
1368        let c = SqlEndpointConfig::from_uri(
1369            "sql:select 1?db_url=postgres://localhost/test&inSeparator=;",
1370        )
1371        .unwrap();
1372        assert_eq!(c.in_separator, ";");
1373    }
1374
1375    #[test]
1376    fn in_separator_empty_rejected() {
1377        let result = SqlEndpointConfig::from_uri(
1378            "sql:select 1?db_url=postgres://localhost/test&inSeparator=",
1379        );
1380        assert!(result.is_err());
1381        let msg = format!("{:?}", result.unwrap_err());
1382        assert!(msg.contains("inSeparator") || msg.contains("empty"));
1383    }
1384
1385    #[test]
1386    fn config_consumer_options() {
1387        let c = SqlEndpointConfig::from_uri(
1388            "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"
1389        ).unwrap();
1390        assert_eq!(c.delay_ms, 2000);
1391        assert_eq!(c.initial_delay_ms, 500);
1392        assert_eq!(c.max_messages_per_poll, Some(10));
1393        assert_eq!(
1394            c.on_consume,
1395            Some("update t set done=true where id=:#id".to_string())
1396        );
1397        assert_eq!(
1398            c.on_consume_failed,
1399            Some("update t set failed=true where id=:#id".to_string())
1400        );
1401        assert_eq!(
1402            c.on_consume_batch_complete,
1403            Some("delete from t where done=true".to_string())
1404        );
1405        assert!(c.route_empty_result_set);
1406        assert!(!c.use_iterator);
1407        assert_eq!(c.expected_update_count, Some(1));
1408        assert!(c.break_batch_on_consume_fail);
1409        assert!(!c.bridge_error_handler);
1410    }
1411
1412    #[test]
1413    fn config_producer_options() {
1414        let c = SqlEndpointConfig::from_uri(
1415            "sql:insert into t values (#)?db_url=postgres://localhost/test&batch=true&useMessageBodyForSql=true&noop=true"
1416        ).unwrap();
1417        assert!(c.batch);
1418        assert!(c.use_message_body_for_sql);
1419        assert!(c.noop);
1420    }
1421
1422    #[test]
1423    fn config_pool_options() {
1424        let c = SqlEndpointConfig::from_uri(
1425            "sql:select 1?db_url=postgres://localhost/test&maxConnections=20&minConnections=3&idleTimeoutSecs=600&maxLifetimeSecs=3600"
1426        ).unwrap();
1427        assert_eq!(c.max_connections, Some(20));
1428        assert_eq!(c.min_connections, Some(3));
1429        assert_eq!(c.idle_timeout_secs, Some(600));
1430        assert_eq!(c.max_lifetime_secs, Some(3600));
1431    }
1432
1433    #[test]
1434    fn config_query_with_special_chars() {
1435        let c = SqlEndpointConfig::from_uri(
1436            "sql:select * from users where name = :#name and age > #?db_url=postgres://localhost/test",
1437        )
1438        .unwrap();
1439        assert_eq!(
1440            c.query,
1441            "select * from users where name = :#name and age > #"
1442        );
1443    }
1444
1445    #[test]
1446    fn output_type_from_str() {
1447        assert_eq!(
1448            "SelectList".parse::<SqlOutputType>().unwrap(),
1449            SqlOutputType::SelectList
1450        );
1451        assert_eq!(
1452            "SelectOne".parse::<SqlOutputType>().unwrap(),
1453            SqlOutputType::SelectOne
1454        );
1455        assert_eq!(
1456            "StreamList".parse::<SqlOutputType>().unwrap(),
1457            SqlOutputType::StreamList
1458        );
1459        assert!("Invalid".parse::<SqlOutputType>().is_err());
1460    }
1461
1462    // SQL-014: file-not-found is now detected during async resolve_file_query(), not from_uri
1463    #[tokio::test]
1464    async fn config_file_not_found() {
1465        let mut config = SqlEndpointConfig::from_uri(
1466            "sql:file:/nonexistent/path/query.sql?db_url=postgres://localhost/test",
1467        )
1468        .expect("from_uri should defer file reading");
1469        // from_uri no longer reads the file — source_path is set, query is empty
1470        assert_eq!(
1471            config.source_path,
1472            Some("/nonexistent/path/query.sql".to_string())
1473        );
1474        assert!(config.query.is_empty());
1475
1476        // Error occurs during async resolution
1477        let result = config.resolve_file_query().await;
1478        assert!(result.is_err());
1479        let msg = format!("{:?}", result.unwrap_err());
1480        assert!(msg.contains("Failed to read SQL file") || msg.contains("nonexistent"));
1481    }
1482
1483    // SQL-014: file query is now resolved asynchronously
1484    #[tokio::test]
1485    async fn config_file_query() {
1486        use std::io::Write;
1487        let unique_name = format!(
1488            "test_sql_query_{}.sql",
1489            std::time::SystemTime::now()
1490                .duration_since(std::time::UNIX_EPOCH)
1491                .unwrap_or_default()
1492                .as_nanos()
1493        );
1494        let mut tmp = std::env::temp_dir();
1495        tmp.push(unique_name);
1496        {
1497            let mut f = std::fs::File::create(&tmp).unwrap();
1498            writeln!(f, "SELECT * FROM users").unwrap();
1499        }
1500        let uri = format!(
1501            "sql:file:{}?db_url=postgres://localhost/test",
1502            tmp.display()
1503        );
1504        let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
1505        // query is empty until async resolution
1506        assert!(c.query.is_empty());
1507        assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
1508
1509        // Resolve asynchronously
1510        c.resolve_file_query()
1511            .await
1512            .expect("file query should resolve");
1513        assert_eq!(c.query, "SELECT * FROM users");
1514        std::fs::remove_file(&tmp).ok();
1515    }
1516
1517    // New tests for config contract
1518    #[test]
1519    fn pool_fields_none_when_not_set() {
1520        let c =
1521            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1522        assert_eq!(c.max_connections, None);
1523        assert_eq!(c.min_connections, None);
1524        assert_eq!(c.idle_timeout_secs, None);
1525        assert_eq!(c.max_lifetime_secs, None);
1526    }
1527
1528    #[test]
1529    fn apply_defaults_fills_none() {
1530        let mut c =
1531            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1532        let global = SqlGlobalConfig {
1533            max_connections: 10,
1534            min_connections: 2,
1535            idle_timeout_secs: 600,
1536            max_lifetime_secs: 3600,
1537            ssl_mode: None,
1538            ssl_root_cert: None,
1539            ssl_cert: None,
1540            ssl_key: None,
1541            retry: NetworkRetryPolicy::default(),
1542        };
1543        c.apply_defaults(&global);
1544        assert_eq!(c.max_connections, Some(10));
1545        assert_eq!(c.min_connections, Some(2));
1546        assert_eq!(c.idle_timeout_secs, Some(600));
1547        assert_eq!(c.max_lifetime_secs, Some(3600));
1548        assert!(c.ssl_mode.is_none());
1549        assert!(c.ssl_root_cert.is_none());
1550        assert!(c.ssl_cert.is_none());
1551        assert!(c.ssl_key.is_none());
1552    }
1553
1554    #[test]
1555    fn apply_defaults_does_not_override() {
1556        let mut c = SqlEndpointConfig::from_uri(
1557            "sql:select 1?db_url=postgres://localhost/test&maxConnections=99&minConnections=5",
1558        )
1559        .unwrap();
1560        let global = SqlGlobalConfig {
1561            max_connections: 10,
1562            min_connections: 2,
1563            idle_timeout_secs: 600,
1564            max_lifetime_secs: 3600,
1565            ssl_mode: None,
1566            ssl_root_cert: None,
1567            ssl_cert: None,
1568            ssl_key: None,
1569            retry: NetworkRetryPolicy::default(),
1570        };
1571        c.apply_defaults(&global);
1572        // URI-set values should NOT be overridden
1573        assert_eq!(c.max_connections, Some(99));
1574        assert_eq!(c.min_connections, Some(5));
1575        // None fields should be filled from global
1576        assert_eq!(c.idle_timeout_secs, Some(600));
1577        assert_eq!(c.max_lifetime_secs, Some(3600));
1578    }
1579
1580    #[test]
1581    fn resolve_defaults_fills_remaining() {
1582        let mut c = SqlEndpointConfig::from_uri(
1583            "sql:select 1?db_url=postgres://localhost/test&maxConnections=7",
1584        )
1585        .unwrap();
1586        c.resolve_defaults();
1587        assert_eq!(c.max_connections, Some(7)); // from URI
1588        assert_eq!(c.min_connections, Some(1)); // from defaults
1589        assert_eq!(c.idle_timeout_secs, Some(300)); // from defaults
1590        assert_eq!(c.max_lifetime_secs, Some(1800)); // from defaults
1591    }
1592
1593    #[test]
1594    fn global_config_builder() {
1595        let c = SqlGlobalConfig::default()
1596            .with_max_connections(20)
1597            .with_min_connections(3)
1598            .with_idle_timeout_secs(600)
1599            .with_max_lifetime_secs(3600)
1600            .with_ssl_mode("require")
1601            .with_ssl_root_cert("/ca.pem")
1602            .with_ssl_cert("/cert.pem")
1603            .with_ssl_key("/key.pem");
1604        assert_eq!(c.max_connections, 20);
1605        assert_eq!(c.min_connections, 3);
1606        assert_eq!(c.idle_timeout_secs, 600);
1607        assert_eq!(c.max_lifetime_secs, 3600);
1608        assert_eq!(c.ssl_mode, Some("require".to_string()));
1609        assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1610        assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1611        assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1612    }
1613
1614    #[test]
1615    fn enrich_postgres_ssl_mode() {
1616        let mut c = SqlEndpointConfig::from_uri(
1617            "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1618        )
1619        .unwrap();
1620        c.resolve_defaults();
1621        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1622        assert!(url.contains("sslmode=require"), "got: {}", url);
1623    }
1624
1625    #[test]
1626    fn enrich_postgres_all_ssl() {
1627        let mut c = SqlEndpointConfig::from_uri(
1628            "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1629        )
1630        .unwrap();
1631        c.resolve_defaults();
1632        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1633        assert!(url.contains("sslmode=require"), "got: {}", url);
1634        assert!(url.contains("sslrootcert="), "got: {}", url);
1635        assert!(url.contains("sslcert="), "got: {}", url);
1636        assert!(url.contains("sslkey="), "got: {}", url);
1637    }
1638
1639    #[test]
1640    fn enrich_mysql_ssl() {
1641        let mut c = SqlEndpointConfig::from_uri(
1642            "sql:select 1?db_url=mysql://localhost/test&sslMode=require",
1643        )
1644        .unwrap();
1645        c.resolve_defaults();
1646        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1647        assert!(url.contains("ssl-mode=require"), "got: {}", url);
1648    }
1649
1650    #[test]
1651    fn enrich_existing_query_params() {
1652        let mut c = SqlEndpointConfig::from_uri(
1653            "sql:select 1?db_url=postgres://localhost/test?existing=1&sslMode=require",
1654        )
1655        .unwrap();
1656        c.resolve_defaults();
1657        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1658        assert!(url.contains("existing=1"), "got: {}", url);
1659        assert!(url.contains("sslmode=require"), "got: {}", url);
1660    }
1661
1662    #[test]
1663    fn enrich_override_existing() {
1664        let mut c = SqlEndpointConfig::from_uri(
1665            "sql:select 1?db_url=postgres://localhost/test?sslmode=allow&sslMode=require",
1666        )
1667        .unwrap();
1668        c.resolve_defaults();
1669        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1670        assert!(url.contains("sslmode=require"), "got: {}", url);
1671        assert!(!url.contains("sslmode=allow"), "got: {}", url);
1672    }
1673
1674    #[test]
1675    fn enrich_applies_postgres_defaults() {
1676        let mut c =
1677            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1678        c.resolve_defaults();
1679        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1680        // Default ssl_mode and connect_timeout are applied at enrichment time
1681        assert!(
1682            url.contains("sslmode=require"),
1683            "expected sslmode=require, got: {}",
1684            url
1685        );
1686        assert!(
1687            url.contains("connect_timeout=10"),
1688            "expected connect_timeout=10, got: {}",
1689            url
1690        );
1691    }
1692
1693    #[test]
1694    fn enrich_mysql_defaults_to_prefer() {
1695        let mut c =
1696            SqlEndpointConfig::from_uri("sql:select 1?db_url=mysql://localhost/test").unwrap();
1697        c.resolve_defaults();
1698        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1699        assert!(
1700            url.contains("ssl-mode=prefer"),
1701            "expected ssl-mode=prefer, got: {}",
1702            url
1703        );
1704        assert!(
1705            url.contains("connect_timeout=10"),
1706            "expected connect_timeout=10, got: {}",
1707            url
1708        );
1709    }
1710
1711    #[test]
1712    fn enrich_preserves_explicit_connect_timeout() {
1713        let mut c = SqlEndpointConfig::from_uri(
1714            "sql:select 1?db_url=postgres://localhost/test?connect_timeout=5",
1715        )
1716        .unwrap();
1717        c.resolve_defaults();
1718        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1719        assert!(
1720            url.contains("connect_timeout=5"),
1721            "expected explicit connect_timeout=5 preserved, got: {}",
1722            url
1723        );
1724        assert!(
1725            url.contains("sslmode=require"),
1726            "expected sslmode=require, got: {}",
1727            url
1728        );
1729    }
1730
1731    #[test]
1732    fn enrich_url_encodes_paths() {
1733        let mut c = SqlEndpointConfig::from_uri(
1734            "sql:select 1?db_url=postgres://localhost/test&sslRootCert=/path/to/my%20cert.pem",
1735        )
1736        .unwrap();
1737        c.resolve_defaults();
1738        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1739        assert!(url.contains("sslrootcert="), "got: {}", url);
1740    }
1741
1742    #[test]
1743    fn enrich_unsupported_scheme_returns_unchanged() {
1744        let mut c = SqlEndpointConfig::from_uri(
1745            "sql:select 1?db_url=sqlite://localhost/test.db&sslMode=require",
1746        )
1747        .unwrap();
1748        c.resolve_defaults();
1749        let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1750        assert_eq!(url, "sqlite://localhost/test.db");
1751    }
1752
1753    #[test]
1754    fn enrich_invalid_url_returns_error() {
1755        let mut c = SqlEndpointConfig::from_uri(
1756            "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1757        )
1758        .unwrap();
1759        c.resolve_defaults();
1760        let result = enrich_db_url_with_ssl("://not-a-valid-url", &c);
1761        assert!(result.is_err());
1762    }
1763
1764    // --- Phase B hardening tests ---
1765
1766    // SQL-010: Debug output redacts credentials
1767    #[test]
1768    fn debug_redacts_db_url_with_password() {
1769        let c = SqlEndpointConfig::from_uri(
1770            "sql:select 1?db_url=postgres://user:secret123@localhost/test",
1771        )
1772        .unwrap();
1773        let debug_output = format!("{:?}", c);
1774        assert!(
1775            !debug_output.contains("secret123"),
1776            "Debug output must not contain password: {}",
1777            debug_output
1778        );
1779        assert!(
1780            debug_output.contains("***"),
1781            "Debug output must contain redacted marker: {}",
1782            debug_output
1783        );
1784    }
1785
1786    #[test]
1787    fn debug_redacts_ssl_key() {
1788        let c = SqlEndpointConfig::from_uri(
1789            "sql:select 1?db_url=postgres://localhost/test&sslKey=/secret/key.pem",
1790        )
1791        .unwrap();
1792        let debug_output = format!("{:?}", c);
1793        assert!(
1794            !debug_output.contains("/secret/key.pem"),
1795            "Debug output must not contain ssl_key path: {}",
1796            debug_output
1797        );
1798    }
1799
1800    #[test]
1801    fn debug_global_config_redacts_ssl_key() {
1802        let c = SqlGlobalConfig::default().with_ssl_key("/secret/key.pem");
1803        let debug_output = format!("{:?}", c);
1804        assert!(
1805            !debug_output.contains("/secret/key.pem"),
1806            "Debug output must not contain ssl_key path: {}",
1807            debug_output
1808        );
1809        assert!(
1810            debug_output.contains("***"),
1811            "Debug output must contain redacted marker: {}",
1812            debug_output
1813        );
1814    }
1815
1816    #[test]
1817    fn redact_db_url_with_credentials() {
1818        assert_eq!(
1819            redact_db_url("postgres://user:pass@host/db"),
1820            "postgres://***:***@host/db"
1821        );
1822    }
1823
1824    #[test]
1825    fn redact_db_url_without_credentials() {
1826        assert_eq!(redact_db_url("sqlite::memory:"), "sqlite::memory:");
1827    }
1828
1829    #[test]
1830    fn redact_db_url_invalid_returns_original() {
1831        assert_eq!(redact_db_url("not-a-url"), "not-a-url");
1832    }
1833
1834    #[test]
1835    fn redact_db_url_invalid_with_credentials_redacts() {
1836        // Port out of range (99999) makes url::Url::parse fail, but the raw
1837        // string carries userinfo. The Err branch must still redact it rather
1838        // than return the credential-bearing string verbatim (rc-7rup).
1839        let redacted = redact_db_url("postgres://user:secret@host:99999/db");
1840        assert!(
1841            !redacted.contains("secret"),
1842            "password must be redacted in unparseable URL: {redacted}"
1843        );
1844        assert!(
1845            !redacted.contains("user:secret"),
1846            "userinfo must be redacted: {redacted}"
1847        );
1848        assert!(
1849            !redacted.contains("user"),
1850            "username must be redacted: {redacted}"
1851        );
1852    }
1853
1854    #[test]
1855    fn redact_db_url_at_before_scheme_returns_original() {
1856        // `@` preceding `://` is not userinfo in a valid position; the
1857        // best-effort redactor leaves it unchanged (no false redaction).
1858        assert_eq!(redact_db_url("foo@bar://baz"), "foo@bar://baz");
1859    }
1860
1861    // SQL-004: usePlaceholder parsing
1862    #[test]
1863    fn use_placeholder_defaults_to_true() {
1864        let c =
1865            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1866        assert!(c.use_placeholder);
1867    }
1868
1869    #[test]
1870    fn use_placeholder_false_from_uri() {
1871        let c = SqlEndpointConfig::from_uri(
1872            "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=false",
1873        )
1874        .unwrap();
1875        assert!(!c.use_placeholder);
1876    }
1877
1878    #[test]
1879    fn use_placeholder_true_from_uri() {
1880        let c = SqlEndpointConfig::from_uri(
1881            "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=true",
1882        )
1883        .unwrap();
1884        assert!(c.use_placeholder);
1885    }
1886
1887    // SQL-004: strict boolean parsing — invalid values rejected
1888    #[test]
1889    fn use_placeholder_rejects_invalid_value() {
1890        let result = SqlEndpointConfig::from_uri(
1891            "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=1",
1892        );
1893        assert!(result.is_err());
1894        let msg = format!("{:?}", result.unwrap_err());
1895        assert!(msg.contains("usePlaceholder") && msg.contains("true") && msg.contains("false"));
1896    }
1897
1898    #[test]
1899    fn use_placeholder_rejects_typo_tru() {
1900        let result = SqlEndpointConfig::from_uri(
1901            "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=tru",
1902        );
1903        assert!(result.is_err());
1904    }
1905
1906    #[test]
1907    fn use_placeholder_rejects_yes() {
1908        let result = SqlEndpointConfig::from_uri(
1909            "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=yes",
1910        );
1911        assert!(result.is_err());
1912    }
1913
1914    #[test]
1915    fn noop_rejects_invalid_value() {
1916        let result =
1917            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&noop=1");
1918        assert!(result.is_err());
1919        let msg = format!("{:?}", result.unwrap_err());
1920        assert!(msg.contains("noop"));
1921    }
1922
1923    #[test]
1924    fn batch_rejects_invalid_value() {
1925        let result =
1926            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&batch=yes");
1927        assert!(result.is_err());
1928        let msg = format!("{:?}", result.unwrap_err());
1929        assert!(msg.contains("batch"));
1930    }
1931
1932    #[test]
1933    fn route_empty_result_set_rejects_invalid_value() {
1934        let result = SqlEndpointConfig::from_uri(
1935            "sql:select 1?db_url=postgres://localhost/test&routeEmptyResultSet=on",
1936        );
1937        assert!(result.is_err());
1938    }
1939
1940    #[test]
1941    fn use_iterator_rejects_invalid_value() {
1942        let result = SqlEndpointConfig::from_uri(
1943            "sql:select 1?db_url=postgres://localhost/test&useIterator=1",
1944        );
1945        assert!(result.is_err());
1946    }
1947
1948    #[test]
1949    fn break_batch_on_consume_fail_rejects_invalid_value() {
1950        let result = SqlEndpointConfig::from_uri(
1951            "sql:select 1?db_url=postgres://localhost/test&breakBatchOnConsumeFail=yes",
1952        );
1953        assert!(result.is_err());
1954    }
1955
1956    #[test]
1957    fn use_message_body_for_sql_rejects_invalid_value() {
1958        let result = SqlEndpointConfig::from_uri(
1959            "sql:select 1?db_url=postgres://localhost/test&useMessageBodyForSql=1",
1960        );
1961        assert!(result.is_err());
1962    }
1963
1964    // Case-insensitive true/false still works
1965    #[test]
1966    fn boolean_params_case_insensitive() {
1967        let c = SqlEndpointConfig::from_uri(
1968            "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=TRUE&noop=FALSE&batch=True&useIterator=False&bridgeErrorHandler=TRUE",
1969        )
1970        .unwrap();
1971        assert!(c.use_placeholder);
1972        assert!(!c.noop);
1973        assert!(c.batch);
1974        assert!(!c.use_iterator);
1975        assert!(c.bridge_error_handler);
1976    }
1977
1978    // SQL-022: multi-char placeholder rejected
1979    #[test]
1980    fn multi_char_placeholder_rejected() {
1981        let result = SqlEndpointConfig::from_uri(
1982            "sql:select 1?db_url=postgres://localhost/test&placeholder=##",
1983        );
1984        assert!(result.is_err());
1985        let msg = format!("{:?}", result.unwrap_err());
1986        assert!(msg.contains("placeholder") && msg.contains("one character"));
1987    }
1988
1989    #[test]
1990    fn non_ascii_placeholder_rejected() {
1991        let result = SqlEndpointConfig::from_uri(
1992            "sql:select 1?db_url=postgres://localhost/test&placeholder=%C2%A2",
1993        );
1994        assert!(result.is_err());
1995    }
1996
1997    #[test]
1998    fn single_char_placeholder_accepted() {
1999        let c = SqlEndpointConfig::from_uri(
2000            "sql:select 1?db_url=postgres://localhost/test&placeholder=$",
2001        )
2002        .unwrap();
2003        assert_eq!(c.placeholder, '$');
2004    }
2005
2006    #[test]
2007    fn empty_placeholder_falls_back_to_default() {
2008        // Empty string is filtered out by the original logic — falls back to '#'
2009        let c = SqlEndpointConfig::from_uri(
2010            "sql:select 1?db_url=postgres://localhost/test&placeholder=",
2011        )
2012        .unwrap();
2013        assert_eq!(c.placeholder, '#');
2014    }
2015
2016    // SQL-014: file-based SQL config test (verifies async resolution and caching)
2017    #[tokio::test]
2018    async fn file_query_cached_in_config() {
2019        use std::io::Write;
2020        let unique_name = format!(
2021            "test_sql_cached_{}.sql",
2022            std::time::SystemTime::now()
2023                .duration_since(std::time::UNIX_EPOCH)
2024                .unwrap_or_default()
2025                .as_nanos()
2026        );
2027        let mut tmp = std::env::temp_dir();
2028        tmp.push(unique_name);
2029        {
2030            let mut f = std::fs::File::create(&tmp).unwrap();
2031            writeln!(f, "SELECT * FROM cached_test").unwrap();
2032        }
2033        let uri = format!(
2034            "sql:file:{}?db_url=postgres://localhost/test",
2035            tmp.display()
2036        );
2037        let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
2038        // Query is empty before async resolution
2039        assert!(c.query.is_empty());
2040        assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
2041
2042        // Resolve asynchronously — query is cached in config
2043        c.resolve_file_query()
2044            .await
2045            .expect("resolve should succeed");
2046        assert_eq!(c.query, "SELECT * FROM cached_test");
2047
2048        // Delete the file — config still has the query
2049        std::fs::remove_file(&tmp).ok();
2050        assert_eq!(c.query, "SELECT * FROM cached_test");
2051    }
2052
2053    // --- H-03 audit sweep tests ---
2054
2055    // SQL-005: alwaysPopulateStatement
2056    #[test]
2057    fn always_populate_statement_defaults_to_false() {
2058        let c =
2059            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2060        assert!(!c.always_populate_statement);
2061    }
2062
2063    #[test]
2064    fn always_populate_statement_from_uri() {
2065        let c = SqlEndpointConfig::from_uri(
2066            "sql:select 1?db_url=postgres://localhost/test&alwaysPopulateStatement=true",
2067        )
2068        .unwrap();
2069        assert!(c.always_populate_statement);
2070    }
2071
2072    // SQL-011: allowNamedParameters
2073    #[test]
2074    fn allow_named_parameters_defaults_to_true() {
2075        let c =
2076            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2077        assert!(c.allow_named_parameters);
2078    }
2079
2080    #[test]
2081    fn allow_named_parameters_false_from_uri() {
2082        let c = SqlEndpointConfig::from_uri(
2083            "sql:select 1?db_url=postgres://localhost/test&allowNamedParameters=false",
2084        )
2085        .unwrap();
2086        assert!(!c.allow_named_parameters);
2087    }
2088
2089    // SQL-016: fetchSize
2090    #[test]
2091    fn fetch_size_defaults_to_none() {
2092        let c =
2093            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2094        assert!(c.fetch_size.is_none());
2095    }
2096
2097    #[test]
2098    fn fetch_size_from_uri() {
2099        let c = SqlEndpointConfig::from_uri(
2100            "sql:select 1?db_url=postgres://localhost/test&fetchSize=1000",
2101        )
2102        .unwrap();
2103        assert_eq!(c.fetch_size, Some(1000));
2104    }
2105
2106    // SQL-002: transactionMode
2107    #[test]
2108    fn transaction_mode_defaults_to_auto() {
2109        let c =
2110            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2111        assert_eq!(c.transaction_mode, TransactionMode::Auto);
2112    }
2113
2114    #[test]
2115    fn transaction_mode_managed_from_uri() {
2116        let c = SqlEndpointConfig::from_uri(
2117            "sql:select 1?db_url=postgres://localhost/test&transactionMode=Managed",
2118        )
2119        .unwrap();
2120        assert_eq!(c.transaction_mode, TransactionMode::Managed);
2121    }
2122
2123    #[test]
2124    fn transaction_mode_invalid_rejected() {
2125        let result = SqlEndpointConfig::from_uri(
2126            "sql:select 1?db_url=postgres://localhost/test&transactionMode=Invalid",
2127        );
2128        assert!(result.is_err());
2129    }
2130
2131    // SQL-015: repeatCount
2132    #[test]
2133    fn repeat_count_defaults_to_none() {
2134        let c =
2135            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2136        assert!(c.repeat_count.is_none());
2137    }
2138
2139    #[test]
2140    fn repeat_count_from_uri() {
2141        let c = SqlEndpointConfig::from_uri(
2142            "sql:select 1?db_url=postgres://localhost/test&repeatCount=10",
2143        )
2144        .unwrap();
2145        assert_eq!(c.repeat_count, Some(10));
2146    }
2147
2148    // SQL-020: breakOnEmpty
2149    #[test]
2150    fn break_on_empty_defaults_to_false() {
2151        let c =
2152            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2153        assert!(!c.break_on_empty);
2154    }
2155
2156    #[test]
2157    fn break_on_empty_true_from_uri() {
2158        let c = SqlEndpointConfig::from_uri(
2159            "sql:select 1?db_url=postgres://localhost/test&breakOnEmpty=true",
2160        )
2161        .unwrap();
2162        assert!(c.break_on_empty);
2163    }
2164
2165    #[test]
2166    fn break_on_empty_explicit_false_from_uri() {
2167        let c = SqlEndpointConfig::from_uri(
2168            "sql:select 1?db_url=postgres://localhost/test&breakOnEmpty=false",
2169        )
2170        .unwrap();
2171        assert!(!c.break_on_empty);
2172    }
2173
2174    // SQL-017: processingStrategy
2175    #[test]
2176    fn processing_strategy_defaults_to_direct() {
2177        let c =
2178            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2179        assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
2180    }
2181
2182    #[test]
2183    fn processing_strategy_scheduled_from_uri() {
2184        let c = SqlEndpointConfig::from_uri(
2185            "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Scheduled",
2186        )
2187        .unwrap();
2188        assert_eq!(c.processing_strategy, ProcessingStrategy::Scheduled);
2189    }
2190
2191    #[test]
2192    fn processing_strategy_invalid_rejected() {
2193        let result = SqlEndpointConfig::from_uri(
2194            "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Invalid",
2195        );
2196        assert!(result.is_err());
2197    }
2198
2199    // SQL-018: pollStrategy
2200    #[test]
2201    fn poll_strategy_defaults_to_sequential() {
2202        let c =
2203            SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2204        assert_eq!(c.poll_strategy, PollStrategy::Sequential);
2205    }
2206
2207    #[test]
2208    fn poll_strategy_burst_from_uri() {
2209        let c = SqlEndpointConfig::from_uri(
2210            "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Burst",
2211        )
2212        .unwrap();
2213        assert_eq!(c.poll_strategy, PollStrategy::Burst);
2214    }
2215
2216    #[test]
2217    fn poll_strategy_invalid_rejected() {
2218        let result = SqlEndpointConfig::from_uri(
2219            "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Invalid",
2220        );
2221        assert!(result.is_err());
2222    }
2223
2224    // ── RetryPolicy (rc-ddl) ──────────────────────────────────────────────
2225
2226    #[test]
2227    fn sql_endpoint_config_has_retry_policy() {
2228        let cfg = SqlEndpointConfig::from_uri(
2229            "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=3&retryInitialDelayMs=500",
2230        )
2231        .expect("parse");
2232        assert_eq!(cfg.retry.max_attempts, 3);
2233        assert_eq!(
2234            cfg.retry.initial_delay,
2235            std::time::Duration::from_millis(500)
2236        );
2237        assert!(cfg.retry.enabled);
2238    }
2239
2240    #[test]
2241    fn sql_endpoint_config_retry_defaults_when_unspecified() {
2242        let cfg =
2243            SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
2244        // When URI has no retry params, retry defaults to NetworkRetryPolicy::default()
2245        assert!(cfg.retry.enabled);
2246        assert_eq!(cfg.retry.max_attempts, 10); // default
2247    }
2248
2249    #[test]
2250    fn sql_global_config_has_retry_default() {
2251        let cfg = SqlGlobalConfig::default();
2252        assert!(cfg.retry.enabled);
2253    }
2254
2255    #[test]
2256    fn retry_policy_parse_full_uri_params() {
2257        let cfg = SqlEndpointConfig::from_uri(
2258            "sql:select 1?db_url=sqlite::memory:&retryEnabled=false&retryMaxAttempts=7&retryInitialDelayMs=1000&retryMultiplier=3.0&retryMaxDelayMs=60000&retryJitter=0.5",
2259        )
2260        .expect("parse");
2261        assert!(!cfg.retry.enabled);
2262        assert_eq!(cfg.retry.max_attempts, 7);
2263        assert_eq!(
2264            cfg.retry.initial_delay,
2265            std::time::Duration::from_millis(1000)
2266        );
2267        assert!((cfg.retry.multiplier - 3.0).abs() < f64::EPSILON);
2268        assert_eq!(cfg.retry.max_delay, std::time::Duration::from_millis(60000));
2269        assert!((cfg.retry.jitter_factor - 0.5).abs() < f64::EPSILON);
2270    }
2271
2272    #[test]
2273    fn retry_policy_from_uri_survives_apply_defaults_with_global() {
2274        let mut ep = SqlEndpointConfig::from_uri(
2275            "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=10&retryInitialDelayMs=500",
2276        )
2277        .expect("parse");
2278        let global = SqlGlobalConfig::default(); // global has default retry (max_attempts=10)
2279        ep.apply_defaults(&global);
2280        // URI values survive when retry_set_from_uri is true
2281        assert_eq!(ep.retry.max_attempts, 10);
2282        assert_eq!(
2283            ep.retry.initial_delay,
2284            std::time::Duration::from_millis(500)
2285        );
2286    }
2287
2288    #[test]
2289    fn retry_policy_falls_back_to_global_when_uri_has_no_retry_params() {
2290        let mut ep =
2291            SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
2292        let mut global = SqlGlobalConfig::default();
2293        global.retry.max_attempts = 7;
2294        ep.apply_defaults(&global);
2295        // When URI has no retry params, global fills the gap
2296        assert_eq!(ep.retry.max_attempts, 7);
2297    }
2298
2299    #[test]
2300    fn from_uri_with_datasource_name() {
2301        let cfg = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders").unwrap();
2302        assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2303        assert!(cfg.db_url.is_empty());
2304    }
2305
2306    #[test]
2307    fn from_uri_with_datasource_and_behavior_override() {
2308        let cfg =
2309            SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&outputType=SelectOne")
2310                .unwrap();
2311        assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2312    }
2313
2314    #[test]
2315    fn from_uri_datasource_rejects_pool_override() {
2316        let result =
2317            SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&maxConnections=50");
2318        assert!(result.is_err());
2319        let msg = result.unwrap_err().to_string();
2320        assert!(msg.contains("pool-affecting"));
2321    }
2322
2323    #[test]
2324    fn from_uri_neither_datasource_nor_db_url_is_error() {
2325        let result = SqlEndpointConfig::from_uri("sql:SELECT 1");
2326        assert!(result.is_err());
2327    }
2328
2329    #[test]
2330    fn from_uri_db_url_inline_still_works() {
2331        let cfg =
2332            SqlEndpointConfig::from_uri("sql:SELECT 1?db_url=postgres://localhost/test").unwrap();
2333        assert!(cfg.datasource_name.is_none());
2334        assert_eq!(cfg.db_url, "postgres://localhost/test");
2335    }
2336
2337    #[test]
2338    fn from_uri_datasource_rejects_ssl_mode() {
2339        let result = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslMode=require");
2340        assert!(result.is_err());
2341        let msg = result.unwrap_err().to_string();
2342        assert!(msg.contains("pool-affecting"));
2343    }
2344
2345    #[test]
2346    fn from_uri_datasource_rejects_ssl_root_cert() {
2347        let result =
2348            SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslRootCert=/ca.pem");
2349        assert!(result.is_err());
2350    }
2351
2352    #[test]
2353    fn from_uri_datasource_rejects_db_url() {
2354        let result = SqlEndpointConfig::from_uri(
2355            "sql:SELECT 1?datasource=orders&db_url=postgres://evil:5432/pwned",
2356        );
2357        assert!(result.is_err());
2358        let msg = result.unwrap_err().to_string();
2359        assert!(msg.contains("db_url") && msg.contains("datasource"));
2360    }
2361}