laminar_connectors/nats/
config.rs

1//! NATS source and sink configuration.
2
3use std::str::FromStr;
4use std::time::Duration;
5
6use async_nats::HeaderName;
7
8use crate::config::ConnectorConfig;
9use crate::error::ConnectorError;
10use crate::serde::Format;
11
12/// Keep retained acknowledgement futures below async-nats' internal publish
13/// window and put a fixed ceiling on connector memory.
14pub(super) const MAX_PENDING_PUBLISH_ACKS: usize = 4_096;
15
16/// A connector operation must always have a finite client-side bound. This is
17/// deliberately an implementation ceiling rather than another deployment knob.
18pub(super) const MAX_NATS_IO_TIMEOUT: Duration = Duration::from_secs(300);
19
20const NATS_CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
21const NATS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
22
23/// NATS connector mode.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
25pub enum Mode {
26    /// Plain NATS pub/sub. No durability, no replay, at-most-once.
27    Core,
28    /// `JetStream` with durable pull consumers.
29    #[default]
30    JetStream,
31}
32
33str_enum!(fromstr Mode, lowercase, ConnectorError, "invalid nats mode",
34    Core => "core";
35    JetStream => "jetstream"
36);
37
38/// `JetStream` consumer ack policy.
39#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
40pub enum AckPolicy {
41    /// Each message acked individually.
42    #[default]
43    Explicit,
44    /// No ack required.
45    None,
46}
47
48str_enum!(fromstr AckPolicy, lowercase, ConnectorError, "invalid ack.policy",
49    Explicit => "explicit";
50    None => "none"
51);
52
53/// `JetStream` consumer delivery policy — where to start.
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
55pub enum DeliverPolicy {
56    /// Every message retained by the stream.
57    #[default]
58    All,
59    /// Only messages published after consumer creation.
60    New,
61    /// Start from a user-supplied stream sequence.
62    ByStartSequence,
63    /// Start from a user-supplied timestamp.
64    ByStartTime,
65}
66
67str_enum!(fromstr DeliverPolicy, lowercase, ConnectorError, "invalid deliver.policy",
68    All => "all";
69    New => "new";
70    ByStartSequence => "by_start_sequence";
71    ByStartTime => "by_start_time"
72);
73
74/// Where the sink gets the subject to publish to.
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub enum SubjectSpec {
77    /// A single literal subject used for every row.
78    Literal(String),
79    /// Per-row: read the named column and use its string value.
80    Column(String),
81}
82
83/// NATS user-authentication mode. Custom `Debug` redacts secrets
84/// because the parent config structs derive `Debug`.
85#[derive(Clone, Default, PartialEq, Eq)]
86#[allow(missing_docs)]
87pub enum AuthMode {
88    #[default]
89    None,
90    UserPass {
91        user: String,
92        password: String,
93    },
94    Token(String),
95    CredsFile(String),
96}
97
98impl std::fmt::Debug for AuthMode {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        match self {
101            Self::None => f.write_str("None"),
102            Self::UserPass { user, .. } => f
103                .debug_struct("UserPass")
104                .field("user", user)
105                .field("password", &"<redacted>")
106                .finish(),
107            Self::Token(_) => f.debug_tuple("Token").field(&"<redacted>").finish(),
108            Self::CredsFile(path) => f.debug_tuple("CredsFile").field(path).finish(),
109        }
110    }
111}
112
113/// TLS transport configuration.
114#[derive(Debug, Clone, Default)]
115#[allow(missing_docs)]
116pub struct TlsConfig {
117    pub enabled: bool,
118    pub ca_location: Option<String>,
119    pub cert_location: Option<String>,
120    pub key_location: Option<String>,
121}
122
123/// NATS source configuration.
124#[derive(Debug, Clone)]
125#[allow(missing_docs)] // one-line `///` per field noises up the config struct
126pub struct NatsSourceConfig {
127    pub servers: Vec<String>,
128    pub mode: Mode,
129    pub auth: AuthMode,
130    pub tls: TlsConfig,
131
132    // JetStream
133    pub stream: Option<String>,
134    pub subject: Option<String>,
135    pub subject_filters: Vec<String>,
136    pub consumer: Option<String>,
137    pub deliver_policy: DeliverPolicy,
138    pub start_sequence: Option<u64>,
139    pub start_time: Option<String>, // RFC3339; parsed against the NATS client's time type in open()
140    pub ack_policy: AckPolicy,
141    pub ack_wait: Duration,
142    pub max_deliver: i64,
143    pub max_ack_pending: i64,
144    pub fetch_batch: usize,
145    pub fetch_max_wait: Duration,
146    /// Interval between `consumer.info()` polls that feed the
147    /// `nats_source_consumer_lag` gauge. Zero disables the poll.
148    pub lag_poll_interval: Duration,
149
150    // Core
151    pub queue_group: Option<String>,
152
153    pub format: Format,
154}
155
156impl NatsSourceConfig {
157    /// # Errors
158    ///
159    /// `ConfigurationError` on any parse or validation failure.
160    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
161        let servers = parse_servers(config)?;
162        let mode = parse_or_default::<Mode>(config, "mode")?;
163        let auth = parse_auth(config)?;
164        let tls = parse_tls(config)?;
165
166        let stream = config.get("stream").map(str::to_string);
167        let subject = config.get("subject").map(str::to_string);
168        let subject_filters = config
169            .get("subject.filters")
170            .map(split_csv)
171            .unwrap_or_default();
172        let consumer = config.get("consumer").map(str::to_string);
173        let deliver_policy = parse_or_default::<DeliverPolicy>(config, "deliver.policy")?;
174        let start_sequence = parse_opt_u64(config, "start.sequence")?;
175        let start_time = config.get("start.time").map(str::to_string);
176        let ack_policy = parse_or_default::<AckPolicy>(config, "ack.policy")?;
177        let ack_wait = require_positive_duration(config, "ack.wait.ms", Duration::from_secs(60))?;
178        let max_deliver = require_positive_or_unlimited_i64(config, "max.deliver", 5)?;
179        let max_ack_pending = require_positive_or_unlimited_i64(config, "max.ack.pending", 10_000)?;
180        let fetch_batch = require_positive_usize(config, "fetch.batch", 500)?;
181        let fetch_max_wait =
182            require_positive_duration(config, "fetch.max.wait.ms", Duration::from_millis(500))?;
183        let lag_poll_interval =
184            parse_duration_ms(config, "lag.poll.interval.ms", Duration::from_secs(10))?;
185        let queue_group = config.get("queue.group").map(str::to_string);
186        let format = parse_format(config)?;
187
188        let cfg = Self {
189            servers,
190            mode,
191            auth,
192            tls,
193            stream,
194            subject,
195            subject_filters,
196            consumer,
197            deliver_policy,
198            start_sequence,
199            start_time,
200            ack_policy,
201            ack_wait,
202            max_deliver,
203            max_ack_pending,
204            fetch_batch,
205            fetch_max_wait,
206            lag_poll_interval,
207            queue_group,
208            format,
209        };
210        cfg.validate()?;
211        Ok(cfg)
212    }
213
214    fn validate(&self) -> Result<(), ConnectorError> {
215        match self.mode {
216            Mode::JetStream => {
217                if self.stream.is_none() {
218                    return Err(cfg_err(
219                        "[LDB-5040] jetstream mode requires 'stream' to be set",
220                    ));
221                }
222                if self.consumer.is_none() {
223                    return Err(cfg_err(
224                        "[LDB-5041] jetstream mode requires 'consumer' (durable name) — \
225                         ephemeral consumers are not supported",
226                    ));
227                }
228                if self.subject.is_none() && self.subject_filters.is_empty() {
229                    return Err(cfg_err(
230                        "[LDB-5042] jetstream mode requires 'subject' or 'subject.filters' \
231                         — the consumer must have at least one filter",
232                    ));
233                }
234                if matches!(self.deliver_policy, DeliverPolicy::ByStartSequence)
235                    && self.start_sequence.is_none()
236                {
237                    return Err(cfg_err(
238                        "[LDB-5043] deliver.policy=by_start_sequence requires 'start.sequence'",
239                    ));
240                }
241                if matches!(self.deliver_policy, DeliverPolicy::ByStartTime)
242                    && self.start_time.is_none()
243                {
244                    return Err(cfg_err(
245                        "[LDB-5044] deliver.policy=by_start_time requires 'start.time'",
246                    ));
247                }
248            }
249            Mode::Core => {
250                if self.subject.is_none() {
251                    return Err(cfg_err(
252                        "[LDB-5045] core mode requires 'subject' (JetStream stream config \
253                         does not apply)",
254                    ));
255                }
256            }
257        }
258        Ok(())
259    }
260}
261
262/// NATS sink configuration.
263#[derive(Debug, Clone)]
264#[allow(missing_docs)]
265pub struct NatsSinkConfig {
266    pub servers: Vec<String>,
267    pub mode: Mode,
268    pub auth: AuthMode,
269    pub tls: TlsConfig,
270    pub stream: Option<String>,
271    pub subject: SubjectSpec,
272    pub dedup_id_column: Option<String>,
273    /// Stream's `duplicate_window` must be at least this long when bounded
274    /// `Nats-Msg-Id` deduplication is enabled.
275    pub min_duplicate_window: Duration,
276    pub max_pending: usize,
277    pub ack_timeout: Duration,
278    pub format: Format,
279    pub header_columns: Vec<HeaderName>,
280}
281
282impl NatsSinkConfig {
283    /// # Errors
284    ///
285    /// `ConfigurationError` on any parse or validation failure.
286    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
287        let servers = parse_servers(config)?;
288        let mode = parse_or_default::<Mode>(config, "mode")?;
289        let auth = parse_auth(config)?;
290        let tls = parse_tls(config)?;
291
292        let subject = match (config.get("subject"), config.get("subject.column")) {
293            (Some(s), None) => SubjectSpec::Literal(s.to_string()),
294            (None, Some(col)) => SubjectSpec::Column(col.to_string()),
295            (Some(_), Some(_)) => {
296                return Err(cfg_err(
297                    "[LDB-5050] set 'subject' OR 'subject.column', not both",
298                ));
299            }
300            (None, None) => {
301                return Err(cfg_err(
302                    "[LDB-5051] 'subject' or 'subject.column' is required",
303                ));
304            }
305        };
306
307        let dedup_id_column = config.get("dedup.id.column").map(str::to_string);
308
309        let cfg = Self {
310            servers,
311            mode,
312            auth,
313            tls,
314            stream: config
315                .get("stream")
316                .map(str::trim)
317                .filter(|stream| !stream.is_empty())
318                .map(str::to_string),
319            subject,
320            dedup_id_column,
321            min_duplicate_window: require_positive_duration(
322                config,
323                "min.duplicate.window.ms",
324                Duration::from_secs(120),
325            )?,
326            max_pending: require_positive_usize(config, "max.pending", MAX_PENDING_PUBLISH_ACKS)?,
327            ack_timeout: require_positive_duration(
328                config,
329                "ack.timeout.ms",
330                Duration::from_secs(30),
331            )?,
332            format: parse_format(config)?,
333            header_columns: config
334                .get("header.columns")
335                .map(split_csv)
336                .unwrap_or_default()
337                .into_iter()
338                .map(|name| {
339                    if !name
340                        .bytes()
341                        .all(|byte| (33..=126).contains(&byte) && byte != b':')
342                    {
343                        return Err(cfg_err(&format!(
344                            "[LDB-5066] header.columns entry '{name}' is not a valid ASCII NATS header name"
345                        )));
346                    }
347                    HeaderName::from_str(&name).map_err(|error| {
348                        cfg_err(&format!(
349                            "[LDB-5066] header.columns entry '{name}' is not a valid NATS header name: {error}"
350                        ))
351                    })
352                })
353                .collect::<Result<_, _>>()?,
354        };
355        cfg.validate()?;
356        Ok(cfg)
357    }
358
359    fn validate(&self) -> Result<(), ConnectorError> {
360        if self.max_pending > MAX_PENDING_PUBLISH_ACKS {
361            return Err(cfg_err(&format!(
362                "max.pending must be <= {MAX_PENDING_PUBLISH_ACKS}"
363            )));
364        }
365        if self.ack_timeout > MAX_NATS_IO_TIMEOUT {
366            return Err(cfg_err(&format!(
367                "ack.timeout.ms must be <= {}",
368                MAX_NATS_IO_TIMEOUT.as_millis()
369            )));
370        }
371        if self.mode == Mode::Core && self.dedup_id_column.is_some() {
372            return Err(cfg_err(
373                "[LDB-5053] 'dedup.id.column' requires mode=jetstream because core NATS has no \
374                 server-side Nats-Msg-Id deduplication",
375            ));
376        }
377        if self.dedup_id_column.is_some() && self.stream.is_none() {
378            return Err(cfg_err(
379                "[LDB-5055] 'dedup.id.column' requires 'stream' so the sink can validate its \
380                 duplicate_window at startup",
381            ));
382        }
383        for h in &self.header_columns {
384            if is_reserved_header(h.as_ref()) {
385                return Err(cfg_err(&format!(
386                    "[LDB-5065] header.columns entry '{h}' collides with a reserved NATS \
387                     header (Nats-Msg-Id / Nats-Expected-Stream); rename the column",
388                )));
389            }
390        }
391        match self.mode {
392            Mode::JetStream if self.stream.is_none() => {
393                return Err(cfg_err(
394                    "[LDB-5071] mode=jetstream requires an explicit 'stream' so durable storage, \
395                     replica count, and publish routing can be validated",
396                ));
397            }
398            Mode::Core if self.stream.is_some() => {
399                return Err(cfg_err(
400                    "[LDB-5071] 'stream' is valid only in mode=jetstream; remove it for core NATS",
401                ));
402            }
403            Mode::Core | Mode::JetStream => {}
404        }
405        Ok(())
406    }
407}
408
409/// Header names the sink manages itself; a user header with the same
410/// name would otherwise silently clobber bounded dedup or stream fencing.
411fn is_reserved_header(name: &str) -> bool {
412    const RESERVED: &[&str] = &["Nats-Msg-Id", "Nats-Expected-Stream"];
413    RESERVED.iter().any(|r| r.eq_ignore_ascii_case(name))
414}
415
416// ── helpers ──
417
418fn cfg_err(msg: &str) -> ConnectorError {
419    ConnectorError::ConfigurationError(msg.to_string())
420}
421
422fn parse_servers(config: &ConnectorConfig) -> Result<Vec<String>, ConnectorError> {
423    let raw = config.require("servers")?;
424    let servers: Vec<String> = split_csv(raw);
425    if servers.is_empty() {
426        return Err(cfg_err("[LDB-5030] 'servers' must not be empty"));
427    }
428    Ok(servers)
429}
430
431fn split_csv(raw: &str) -> Vec<String> {
432    raw.split(',')
433        .map(str::trim)
434        .filter(|s| !s.is_empty())
435        .map(str::to_string)
436        .collect()
437}
438
439fn parse_or_default<T: std::str::FromStr + Default>(
440    config: &ConnectorConfig,
441    key: &str,
442) -> Result<T, ConnectorError>
443where
444    T::Err: std::fmt::Display,
445{
446    match config.get(key) {
447        Some(s) => s
448            .parse()
449            .map_err(|e: T::Err| cfg_err(&format!("invalid {key}: {e}"))),
450        None => Ok(T::default()),
451    }
452}
453
454fn parse_format(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
455    match config.get("format") {
456        Some(s) => Format::parse(s).map_err(|e| cfg_err(&e.to_string())),
457        None => Ok(Format::Json),
458    }
459}
460
461fn parse_bool(config: &ConnectorConfig, key: &str, default: bool) -> Result<bool, ConnectorError> {
462    match config.get(key) {
463        Some(s) => s
464            .parse::<bool>()
465            .map_err(|_| cfg_err(&format!("{key} must be 'true' or 'false', got '{s}'"))),
466        None => Ok(default),
467    }
468}
469
470fn parse_opt_u64(config: &ConnectorConfig, key: &str) -> Result<Option<u64>, ConnectorError> {
471    config.get(key).map(|s| parse_int(key, s)).transpose()
472}
473
474fn parse_i64(config: &ConnectorConfig, key: &str, default: i64) -> Result<i64, ConnectorError> {
475    config.get(key).map_or(Ok(default), |s| parse_int(key, s))
476}
477
478fn parse_usize(
479    config: &ConnectorConfig,
480    key: &str,
481    default: usize,
482) -> Result<usize, ConnectorError> {
483    config.get(key).map_or(Ok(default), |s| parse_int(key, s))
484}
485
486fn require_positive_duration(
487    config: &ConnectorConfig,
488    key: &str,
489    default: Duration,
490) -> Result<Duration, ConnectorError> {
491    let v = parse_duration_ms(config, key, default)?;
492    if v.is_zero() {
493        return Err(cfg_err(&format!("{key} must be > 0")));
494    }
495    Ok(v)
496}
497
498fn require_positive_usize(
499    config: &ConnectorConfig,
500    key: &str,
501    default: usize,
502) -> Result<usize, ConnectorError> {
503    let v = parse_usize(config, key, default)?;
504    if v == 0 {
505        return Err(cfg_err(&format!("{key} must be > 0")));
506    }
507    Ok(v)
508}
509
510/// NATS allows `-1` for "unlimited" in `max_deliver` / `max_ack_pending`.
511/// Reject 0 (undefined) and negative values other than -1.
512fn require_positive_or_unlimited_i64(
513    config: &ConnectorConfig,
514    key: &str,
515    default: i64,
516) -> Result<i64, ConnectorError> {
517    let v = parse_i64(config, key, default)?;
518    if v == 0 || v < -1 {
519        return Err(cfg_err(&format!("{key} must be > 0, or -1 for unlimited")));
520    }
521    Ok(v)
522}
523
524fn parse_duration_ms(
525    config: &ConnectorConfig,
526    key: &str,
527    default: Duration,
528) -> Result<Duration, ConnectorError> {
529    config.get(key).map_or(Ok(default), |s| {
530        parse_int::<u64>(key, s).map(Duration::from_millis)
531    })
532}
533
534fn parse_int<T: std::str::FromStr>(key: &str, raw: &str) -> Result<T, ConnectorError> {
535    raw.parse::<T>()
536        .map_err(|_| cfg_err(&format!("{key} must be an integer, got '{raw}'")))
537}
538
539fn parse_auth(config: &ConnectorConfig) -> Result<AuthMode, ConnectorError> {
540    let mode = config.get("auth.mode").unwrap_or("none");
541    match mode {
542        "none" | "" => {
543            if config.get("user").is_some()
544                || config.get("password").is_some()
545                || config.get("token").is_some()
546                || config.get("creds.file").is_some()
547            {
548                return Err(cfg_err(
549                    "[LDB-5063] credentials set but auth.mode=none; \
550                     set auth.mode explicitly or remove the credentials",
551                ));
552            }
553            Ok(AuthMode::None)
554        }
555        "user_pass" => {
556            let user = config
557                .get("user")
558                .ok_or_else(|| cfg_err("[LDB-5060] auth.mode=user_pass requires 'user'"))?
559                .to_string();
560            let password = config
561                .get("password")
562                .ok_or_else(|| cfg_err("[LDB-5060] auth.mode=user_pass requires 'password'"))?
563                .to_string();
564            Ok(AuthMode::UserPass { user, password })
565        }
566        "token" => {
567            let token = config
568                .get("token")
569                .ok_or_else(|| cfg_err("[LDB-5061] auth.mode=token requires 'token'"))?
570                .to_string();
571            Ok(AuthMode::Token(token))
572        }
573        "creds_file" => {
574            let path = config
575                .get("creds.file")
576                .ok_or_else(|| cfg_err("[LDB-5064] auth.mode=creds_file requires 'creds.file'"))?
577                .to_string();
578            Ok(AuthMode::CredsFile(path))
579        }
580        other => Err(cfg_err(&format!(
581            "invalid auth.mode '{other}'; expected none | user_pass | token | creds_file"
582        ))),
583    }
584}
585
586fn parse_tls(config: &ConnectorConfig) -> Result<TlsConfig, ConnectorError> {
587    let cert_location = config.get("tls.cert.location").map(str::to_string);
588    let key_location = config.get("tls.key.location").map(str::to_string);
589    if cert_location.is_some() != key_location.is_some() {
590        return Err(cfg_err(
591            "[LDB-5062] tls.cert.location and tls.key.location must both be set \
592             (mutual-TLS client cert) or both be unset",
593        ));
594    }
595    Ok(TlsConfig {
596        enabled: parse_bool(config, "tls.enabled", false)?,
597        ca_location: config.get("tls.ca.location").map(str::to_string),
598        cert_location,
599        key_location,
600    })
601}
602
603/// Build `ConnectOptions` from parsed auth + TLS. Shared between
604/// source and sink.
605///
606/// # Errors
607///
608/// Returns `ConnectorError` if the creds file can't be read or parsed.
609pub(super) fn build_connect_options(
610    auth: &AuthMode,
611    tls: &TlsConfig,
612) -> Result<async_nats::ConnectOptions, ConnectorError> {
613    let mut opts = async_nats::ConnectOptions::new();
614    match auth {
615        AuthMode::None => {}
616        AuthMode::UserPass { user, password } => {
617            opts = opts.user_and_password(user.clone(), password.clone());
618        }
619        AuthMode::Token(token) => {
620            opts = opts.token(token.clone());
621        }
622        AuthMode::CredsFile(path) => {
623            let contents = std::fs::read_to_string(path)
624                .map_err(|e| cfg_err(&format!("creds.file '{path}': {e}")))?;
625            opts = async_nats::ConnectOptions::with_credentials(&contents)
626                .map_err(|e| cfg_err(&format!("creds.file '{path}' invalid: {e}")))?;
627        }
628    }
629    let tls_touched = tls.enabled || tls.ca_location.is_some() || tls.cert_location.is_some();
630    if tls_touched {
631        opts = opts.require_tls(true);
632    }
633    if let Some(ca) = &tls.ca_location {
634        opts = opts.add_root_certificates(ca.into());
635    }
636    if let (Some(cert), Some(key)) = (&tls.cert_location, &tls.key_location) {
637        opts = opts.add_client_certificate(cert.into(), key.into());
638    }
639    Ok(opts
640        .connection_timeout(NATS_CONNECTION_TIMEOUT)
641        .request_timeout(Some(NATS_REQUEST_TIMEOUT)))
642}
643
644#[cfg(test)]
645#[allow(clippy::disallowed_types)]
646mod tests {
647    use super::*;
648    use std::collections::HashMap;
649
650    fn cfg(pairs: &[(&str, &str)]) -> ConnectorConfig {
651        let mut props = HashMap::new();
652        for (k, v) in pairs {
653            props.insert((*k).to_string(), (*v).to_string());
654        }
655        ConnectorConfig::with_properties("nats", props)
656    }
657
658    // ── source ──
659
660    #[test]
661    fn source_jetstream_requires_stream() {
662        let err = NatsSourceConfig::from_config(&cfg(&[
663            ("servers", "nats://a:4222"),
664            ("consumer", "c"),
665            ("subject", "x.>"),
666        ]))
667        .unwrap_err()
668        .to_string();
669        assert!(err.contains("LDB-5040"), "got: {err}");
670    }
671
672    #[test]
673    fn source_jetstream_requires_consumer() {
674        let err = NatsSourceConfig::from_config(&cfg(&[
675            ("servers", "nats://a:4222"),
676            ("stream", "S"),
677            ("subject", "x.>"),
678        ]))
679        .unwrap_err()
680        .to_string();
681        assert!(err.contains("LDB-5041"), "got: {err}");
682    }
683
684    #[test]
685    fn source_jetstream_requires_subject_or_filters() {
686        let err = NatsSourceConfig::from_config(&cfg(&[
687            ("servers", "nats://a:4222"),
688            ("stream", "S"),
689            ("consumer", "c"),
690        ]))
691        .unwrap_err()
692        .to_string();
693        assert!(err.contains("LDB-5042"), "got: {err}");
694    }
695
696    #[test]
697    fn source_core_requires_subject() {
698        let err =
699            NatsSourceConfig::from_config(&cfg(&[("servers", "nats://a:4222"), ("mode", "core")]))
700                .unwrap_err()
701                .to_string();
702        assert!(err.contains("LDB-5045"), "got: {err}");
703    }
704
705    #[test]
706    fn source_by_start_sequence_requires_start_sequence() {
707        let err = NatsSourceConfig::from_config(&cfg(&[
708            ("servers", "nats://a:4222"),
709            ("stream", "S"),
710            ("consumer", "c"),
711            ("subject", "x.>"),
712            ("deliver.policy", "by_start_sequence"),
713        ]))
714        .unwrap_err()
715        .to_string();
716        assert!(err.contains("LDB-5043"), "got: {err}");
717    }
718
719    #[test]
720    fn source_happy_path_jetstream() {
721        let parsed = NatsSourceConfig::from_config(&cfg(&[
722            ("servers", "nats://a:4222,nats://b:4222"),
723            ("stream", "ORDERS"),
724            ("consumer", "laminar-orders"),
725            ("subject.filters", "orders.us.*,orders.eu.*"),
726            ("ack.wait.ms", "45000"),
727            ("max.deliver", "3"),
728        ]))
729        .unwrap();
730        assert_eq!(parsed.servers.len(), 2);
731        assert_eq!(parsed.mode, Mode::JetStream);
732        assert_eq!(parsed.subject_filters, vec!["orders.us.*", "orders.eu.*"]);
733        assert_eq!(parsed.ack_wait, Duration::from_secs(45));
734        assert_eq!(parsed.max_deliver, 3);
735    }
736
737    #[test]
738    fn source_happy_path_core() {
739        let parsed = NatsSourceConfig::from_config(&cfg(&[
740            ("servers", "nats://a:4222"),
741            ("mode", "core"),
742            ("subject", "events"),
743            ("queue.group", "workers"),
744        ]))
745        .unwrap();
746        assert_eq!(parsed.mode, Mode::Core);
747        assert_eq!(parsed.subject.as_deref(), Some("events"));
748        assert_eq!(parsed.queue_group.as_deref(), Some("workers"));
749    }
750
751    // ── sink ──
752
753    #[test]
754    fn sink_rejects_subject_and_subject_column() {
755        let err = NatsSinkConfig::from_config(&cfg(&[
756            ("servers", "nats://a:4222"),
757            ("subject", "x"),
758            ("subject.column", "c"),
759        ]))
760        .unwrap_err()
761        .to_string();
762        assert!(err.contains("LDB-5050"), "got: {err}");
763    }
764
765    #[test]
766    fn sink_rejects_neither_subject_nor_column() {
767        let err = NatsSinkConfig::from_config(&cfg(&[("servers", "nats://a:4222")]))
768            .unwrap_err()
769            .to_string();
770        assert!(err.contains("LDB-5051"), "got: {err}");
771    }
772
773    #[test]
774    fn sink_rejects_header_column_colliding_with_reserved() {
775        let err = NatsSinkConfig::from_config(&cfg(&[
776            ("servers", "nats://a:4222"),
777            ("subject", "x"),
778            ("header.columns", "trace_id,nats-msg-id"),
779        ]))
780        .unwrap_err()
781        .to_string();
782        assert!(err.contains("LDB-5065"), "got: {err}");
783    }
784
785    #[test]
786    fn sink_rejects_invalid_header_column_name() {
787        for name in ["trace:id", "trŁce"] {
788            let err = NatsSinkConfig::from_config(&cfg(&[
789                ("servers", "nats://a:4222"),
790                ("subject", "x"),
791                ("header.columns", name),
792            ]))
793            .unwrap_err()
794            .to_string();
795            assert!(err.contains("LDB-5066"), "got: {err}");
796        }
797    }
798
799    #[test]
800    fn sink_rejects_core_with_dedup() {
801        let err = NatsSinkConfig::from_config(&cfg(&[
802            ("servers", "nats://a:4222"),
803            ("mode", "core"),
804            ("subject", "x"),
805            ("dedup.id.column", "event_id"),
806        ]))
807        .unwrap_err()
808        .to_string();
809        assert!(err.contains("LDB-5053"), "got: {err}");
810    }
811
812    #[test]
813    fn sink_rejects_dedup_without_stream() {
814        let err = NatsSinkConfig::from_config(&cfg(&[
815            ("servers", "nats://a:4222"),
816            ("subject", "x"),
817            ("dedup.id.column", "event_id"),
818        ]))
819        .unwrap_err()
820        .to_string();
821        assert!(err.contains("LDB-5055"), "got: {err}");
822    }
823
824    #[test]
825    fn jetstream_sink_requires_named_stream() {
826        let err =
827            NatsSinkConfig::from_config(&cfg(&[("servers", "nats://a:4222"), ("subject", "x")]))
828                .unwrap_err()
829                .to_string();
830        assert!(err.contains("LDB-5071"), "got: {err}");
831    }
832
833    #[test]
834    fn jetstream_sink_rejects_blank_stream() {
835        let err = NatsSinkConfig::from_config(&cfg(&[
836            ("servers", "nats://a:4222"),
837            ("stream", "   "),
838            ("subject", "x"),
839        ]))
840        .unwrap_err()
841        .to_string();
842        assert!(err.contains("LDB-5071"), "got: {err}");
843    }
844
845    #[test]
846    fn core_sink_rejects_ignored_stream() {
847        let err = NatsSinkConfig::from_config(&cfg(&[
848            ("servers", "nats://a:4222"),
849            ("mode", "core"),
850            ("stream", "OUT"),
851            ("subject", "x"),
852        ]))
853        .unwrap_err()
854        .to_string();
855        assert!(err.contains("LDB-5071"), "got: {err}");
856    }
857
858    #[test]
859    fn sink_happy_path_bounded_dedup() {
860        let parsed = NatsSinkConfig::from_config(&cfg(&[
861            ("servers", "nats://a:4222"),
862            ("stream", "OUT"),
863            ("subject.column", "out_subject"),
864            ("dedup.id.column", "event_id"),
865            ("header.columns", "trace_id,tenant"),
866        ]))
867        .unwrap();
868        assert_eq!(parsed.stream.as_deref(), Some("OUT"));
869        assert_eq!(parsed.subject, SubjectSpec::Column("out_subject".into()));
870        assert_eq!(parsed.dedup_id_column.as_deref(), Some("event_id"));
871        let header_columns: Vec<&str> = parsed
872            .header_columns
873            .iter()
874            .map(AsRef::<str>::as_ref)
875            .collect();
876        assert_eq!(header_columns, vec!["trace_id", "tenant"]);
877    }
878
879    // ── servers ──
880
881    #[test]
882    fn servers_required() {
883        let err = NatsSourceConfig::from_config(&cfg(&[]))
884            .unwrap_err()
885            .to_string();
886        assert!(err.contains("servers"), "got: {err}");
887    }
888
889    #[test]
890    fn zero_fetch_batch_rejected() {
891        let err = NatsSourceConfig::from_config(&jetstream_happy(&[("fetch.batch", "0")]))
892            .unwrap_err()
893            .to_string();
894        assert!(err.contains("fetch.batch must be > 0"), "got: {err}");
895    }
896
897    #[test]
898    fn zero_ack_wait_rejected() {
899        let err = NatsSourceConfig::from_config(&jetstream_happy(&[("ack.wait.ms", "0")]))
900            .unwrap_err()
901            .to_string();
902        assert!(err.contains("ack.wait.ms must be > 0"), "got: {err}");
903    }
904
905    #[test]
906    fn max_deliver_unlimited_accepted() {
907        let parsed =
908            NatsSourceConfig::from_config(&jetstream_happy(&[("max.deliver", "-1")])).unwrap();
909        assert_eq!(parsed.max_deliver, -1);
910    }
911
912    #[test]
913    fn max_deliver_negative_other_than_minus_one_rejected() {
914        let err = NatsSourceConfig::from_config(&jetstream_happy(&[("max.deliver", "-5")]))
915            .unwrap_err()
916            .to_string();
917        assert!(err.contains("max.deliver"), "got: {err}");
918    }
919
920    #[test]
921    fn sink_zero_max_pending_rejected() {
922        let err = NatsSinkConfig::from_config(&cfg(&[
923            ("servers", "nats://a:4222"),
924            ("subject", "x"),
925            ("max.pending", "0"),
926        ]))
927        .unwrap_err()
928        .to_string();
929        assert!(err.contains("max.pending must be > 0"), "got: {err}");
930    }
931
932    #[test]
933    fn sink_max_pending_is_capped_below_the_client_publish_window() {
934        let err = NatsSinkConfig::from_config(&cfg(&[
935            ("servers", "nats://a:4222"),
936            ("subject", "x"),
937            ("max.pending", "4097"),
938        ]))
939        .unwrap_err()
940        .to_string();
941        assert!(err.contains("max.pending must be <= 4096"), "got: {err}");
942    }
943
944    #[test]
945    fn sink_ack_timeout_has_a_fixed_client_side_ceiling() {
946        let err = NatsSinkConfig::from_config(&cfg(&[
947            ("servers", "nats://a:4222"),
948            ("subject", "x"),
949            ("ack.timeout.ms", "300001"),
950        ]))
951        .unwrap_err()
952        .to_string();
953        assert!(
954            err.contains("ack.timeout.ms must be <= 300000"),
955            "got: {err}"
956        );
957    }
958
959    #[test]
960    fn servers_empty_csv_rejected() {
961        let err = NatsSourceConfig::from_config(&cfg(&[("servers", ",,")]))
962            .unwrap_err()
963            .to_string();
964        assert!(err.contains("LDB-5030"), "got: {err}");
965    }
966
967    // ── auth ──
968
969    fn jetstream_happy(pairs: &[(&str, &str)]) -> ConnectorConfig {
970        let mut base = vec![
971            ("servers", "nats://a:4222"),
972            ("stream", "S"),
973            ("consumer", "c"),
974            ("subject", "x.>"),
975        ];
976        base.extend_from_slice(pairs);
977        cfg(&base)
978    }
979
980    #[test]
981    fn auth_user_pass_requires_user() {
982        let err = NatsSourceConfig::from_config(&jetstream_happy(&[
983            ("auth.mode", "user_pass"),
984            ("password", "secret"),
985        ]))
986        .unwrap_err()
987        .to_string();
988        assert!(err.contains("LDB-5060"), "got: {err}");
989    }
990
991    #[test]
992    fn auth_user_pass_requires_password() {
993        let err = NatsSourceConfig::from_config(&jetstream_happy(&[
994            ("auth.mode", "user_pass"),
995            ("user", "alice"),
996        ]))
997        .unwrap_err()
998        .to_string();
999        assert!(err.contains("LDB-5060"), "got: {err}");
1000    }
1001
1002    #[test]
1003    fn auth_user_pass_happy_path() {
1004        let parsed = NatsSourceConfig::from_config(&jetstream_happy(&[
1005            ("auth.mode", "user_pass"),
1006            ("user", "alice"),
1007            ("password", "wonderland"),
1008        ]))
1009        .unwrap();
1010        assert_eq!(
1011            parsed.auth,
1012            AuthMode::UserPass {
1013                user: "alice".into(),
1014                password: "wonderland".into(),
1015            }
1016        );
1017    }
1018
1019    #[test]
1020    fn auth_token_requires_token() {
1021        let err = NatsSourceConfig::from_config(&jetstream_happy(&[("auth.mode", "token")]))
1022            .unwrap_err()
1023            .to_string();
1024        assert!(err.contains("LDB-5061"), "got: {err}");
1025    }
1026
1027    #[test]
1028    fn auth_token_happy_path() {
1029        let parsed = NatsSourceConfig::from_config(&jetstream_happy(&[
1030            ("auth.mode", "token"),
1031            ("token", "abc123"),
1032        ]))
1033        .unwrap();
1034        assert_eq!(parsed.auth, AuthMode::Token("abc123".into()));
1035    }
1036
1037    #[test]
1038    fn auth_none_with_stray_credentials_rejected() {
1039        let err = NatsSourceConfig::from_config(&jetstream_happy(&[("user", "alice")]))
1040            .unwrap_err()
1041            .to_string();
1042        assert!(err.contains("LDB-5063"), "got: {err}");
1043    }
1044
1045    #[test]
1046    fn auth_creds_file_requires_path() {
1047        let err = NatsSourceConfig::from_config(&jetstream_happy(&[("auth.mode", "creds_file")]))
1048            .unwrap_err()
1049            .to_string();
1050        assert!(err.contains("LDB-5064"), "got: {err}");
1051    }
1052
1053    #[test]
1054    fn auth_creds_file_happy_path() {
1055        let parsed = NatsSourceConfig::from_config(&jetstream_happy(&[
1056            ("auth.mode", "creds_file"),
1057            ("creds.file", "/secrets/user.creds"),
1058        ]))
1059        .unwrap();
1060        assert_eq!(
1061            parsed.auth,
1062            AuthMode::CredsFile("/secrets/user.creds".into())
1063        );
1064    }
1065
1066    #[test]
1067    fn auth_unknown_mode_rejected() {
1068        let err = NatsSourceConfig::from_config(&jetstream_happy(&[("auth.mode", "banana")]))
1069            .unwrap_err()
1070            .to_string();
1071        assert!(err.contains("invalid auth.mode"), "got: {err}");
1072    }
1073
1074    // ── tls ──
1075
1076    #[test]
1077    fn tls_cert_without_key_rejected() {
1078        let err = NatsSourceConfig::from_config(&jetstream_happy(&[(
1079            "tls.cert.location",
1080            "/certs/client.pem",
1081        )]))
1082        .unwrap_err()
1083        .to_string();
1084        assert!(err.contains("LDB-5062"), "got: {err}");
1085    }
1086
1087    #[test]
1088    fn tls_key_without_cert_rejected() {
1089        let err = NatsSourceConfig::from_config(&jetstream_happy(&[(
1090            "tls.key.location",
1091            "/certs/client.key",
1092        )]))
1093        .unwrap_err()
1094        .to_string();
1095        assert!(err.contains("LDB-5062"), "got: {err}");
1096    }
1097
1098    #[test]
1099    fn tls_happy_path() {
1100        let parsed = NatsSourceConfig::from_config(&jetstream_happy(&[
1101            ("tls.enabled", "true"),
1102            ("tls.ca.location", "/certs/ca.pem"),
1103            ("tls.cert.location", "/certs/client.pem"),
1104            ("tls.key.location", "/certs/client.key"),
1105        ]))
1106        .unwrap();
1107        assert!(parsed.tls.enabled);
1108        assert_eq!(parsed.tls.ca_location.as_deref(), Some("/certs/ca.pem"));
1109        assert_eq!(
1110            parsed.tls.cert_location.as_deref(),
1111            Some("/certs/client.pem")
1112        );
1113    }
1114
1115    #[test]
1116    fn auth_and_tls_on_sink() {
1117        let parsed = NatsSinkConfig::from_config(&cfg(&[
1118            ("servers", "nats://a:4222"),
1119            ("stream", "OUT"),
1120            ("subject", "x"),
1121            ("auth.mode", "user_pass"),
1122            ("user", "alice"),
1123            ("password", "wonderland"),
1124            ("tls.enabled", "true"),
1125        ]))
1126        .unwrap();
1127        assert!(matches!(parsed.auth, AuthMode::UserPass { .. }));
1128        assert!(parsed.tls.enabled);
1129    }
1130}
laminar_connectors/nats/config.rs

laminar_connectors/nats/
config.rs
