laminar_connectors/mongodb/
config.rs

1//! `MongoDB` connector configuration.
2//!
3//! Provides [`MongoDbSourceConfig`] for the CDC change stream source and
4//! [`MongoDbSinkConfig`] for the write sink. Both support construction
5//! from a generic [`ConnectorConfig`] key-value map and programmatic builders.
6//!
7//! User-supplied change-stream pipelines are limited to `$match` stages so they
8//! cannot alter `MongoDB`'s resume token.
9
10use std::time::Duration;
11
12use crate::config::ConnectorConfig;
13use crate::error::ConnectorError;
14
15use super::timeseries::{CollectionKind, TimeSeriesConfig, TimeSeriesGranularity};
16use super::write_model::WriteMode;
17
18const REMOVED_SOURCE_CONFIG_KEYS: &[&str] = &[
19    "batch.size",
20    "max.buffered.events",
21    "max.await.time.ms",
22    "resume.token.store",
23    "split.large.events",
24    "max.poll.records",
25];
26
27const SOURCE_CONFIG_KEYS: &[&str] = &[
28    "connection.uri",
29    "database",
30    "collection",
31    "full.document.mode",
32    "pipeline",
33    "max.buffered.bytes",
34    "laminar.source.name",
35    "_arrow_schema",
36    "_primary_key_columns",
37];
38
39pub(super) const DEFAULT_MAX_BUFFERED_BYTES: usize = 64 * 1024 * 1024;
40pub(super) const MAX_PIPELINE_STAGES: usize = 64;
41pub(super) const MAX_PIPELINE_JSON_BYTES: usize = 256 * 1024;
42const MIN_BUFFERED_BYTES: usize = 1024 * 1024;
43const MAX_BUFFERED_BYTES: usize = 4 * 1024 * 1024 * 1024;
44const ESTIMATED_BUFFERED_ITEM_BYTES: usize = 64 * 1024;
45const MAX_READER_CHANNEL_ITEMS: usize = 4096;
46const MAX_CURSOR_BATCH_ITEMS: u32 = 1000;
47
48const SINK_CONFIG_KEYS: &[&str] = &[
49    "connection.uri",
50    "database",
51    "collection",
52    "flush.interval.ms",
53    "write.mode",
54    "write.mode.key_fields",
55    "timeseries.time_field",
56    "timeseries.meta_field",
57    "timeseries.granularity",
58    "timeseries.bucket_max_span_seconds",
59    "timeseries.bucket_rounding_seconds",
60    "timeseries.expire_after_seconds",
61    "delivery.guarantee",
62    "sink.write.timeout.ms",
63    "_arrow_schema",
64];
65
66const REMOVED_SINK_CONFIG_KEYS: &[&str] = &["batch.size", "write_concern.timeout_ms"];
67
68/// Mode for requesting full documents on update events.
69///
70/// Controls the `fullDocument` option on the change stream cursor.
71/// The choice has significant correctness and performance implications.
72#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
73#[serde(rename_all = "snake_case")]
74pub enum FullDocumentMode {
75    /// Default: only the delta (`updateDescription`) for update events.
76    /// `fullDocument` is `None` on updates.
77    #[default]
78    Delta,
79
80    /// `fullDocument: "required"` — the collection must have
81    /// `changeStreamPreAndPostImages` enabled. Admission fails before the
82    /// change stream opens when the option is disabled.
83    #[serde(rename = "required")]
84    RequirePostImage,
85}
86
87str_enum!(FullDocumentMode, lowercase, ConnectorError, "unknown full document mode",
88    Delta => "delta";
89    RequirePostImage => "required"
90);
91
92/// Configuration for the `MongoDB` CDC source connector.
93#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
94#[serde(deny_unknown_fields)]
95pub struct MongoDbSourceConfig {
96    /// `MongoDB` connection URI (e.g., `mongodb://host:27017`).
97    pub connection_uri: String,
98
99    /// Database name.
100    pub database: String,
101
102    /// Fixed collection name.
103    pub collection: String,
104
105    /// Full document retrieval mode for update events.
106    pub full_document_mode: FullDocumentMode,
107
108    /// Additional `$match` stages applied when the change stream opens.
109    #[serde(default)]
110    pub pipeline: Vec<serde_json::Value>,
111
112    /// Maximum retained decoded bytes across the reader channel and poll buffer.
113    #[serde(default = "default_max_buffered_bytes")]
114    pub max_buffered_bytes: usize,
115}
116
117fn default_max_buffered_bytes() -> usize {
118    DEFAULT_MAX_BUFFERED_BYTES
119}
120
121impl Default for MongoDbSourceConfig {
122    fn default() -> Self {
123        Self {
124            connection_uri: "mongodb://localhost:27017".to_string(),
125            database: String::new(),
126            collection: String::new(),
127            full_document_mode: FullDocumentMode::default(),
128            pipeline: Vec::new(),
129            max_buffered_bytes: default_max_buffered_bytes(),
130        }
131    }
132}
133
134impl MongoDbSourceConfig {
135    /// Creates a new source config with required fields.
136    #[must_use]
137    pub fn new(connection_uri: &str, database: &str, collection: &str) -> Self {
138        Self {
139            connection_uri: connection_uri.to_string(),
140            database: database.to_string(),
141            collection: collection.to_string(),
142            ..Self::default()
143        }
144    }
145
146    /// Private queue capacity derived from the one public ownership budget.
147    pub(crate) fn reader_channel_capacity(&self) -> usize {
148        (self.max_buffered_bytes / ESTIMATED_BUFFERED_ITEM_BYTES).clamp(1, MAX_READER_CHANNEL_ITEMS)
149    }
150
151    /// Cursor batch hint derived from the one public ownership budget.
152    pub(crate) fn cursor_batch_size(&self) -> u32 {
153        u32::try_from(self.reader_channel_capacity())
154            .unwrap_or(MAX_CURSOR_BATCH_ITEMS)
155            .min(MAX_CURSOR_BATCH_ITEMS)
156    }
157
158    /// Validates the configuration.
159    ///
160    /// # Errors
161    ///
162    /// Returns `ConnectorError::ConfigurationError` for invalid settings.
163    pub fn validate(&self) -> Result<(), ConnectorError> {
164        crate::config::require_non_empty(&self.connection_uri, "connection_uri")?;
165        crate::config::require_non_empty(&self.database, "database")?;
166        crate::config::require_non_empty(&self.collection, "collection")?;
167        if self.collection == "*" {
168            return Err(ConnectorError::ConfigurationError(
169                "MongoDB CDC collection must name one fixed collection; database-wide wildcard \
170                 streams cannot be bound to an immutable collection UUID"
171                    .into(),
172            ));
173        }
174
175        if !(MIN_BUFFERED_BYTES..=MAX_BUFFERED_BYTES).contains(&self.max_buffered_bytes) {
176            return Err(ConnectorError::ConfigurationError(format!(
177                "max.buffered.bytes must be between {MIN_BUFFERED_BYTES} and \
178                     {MAX_BUFFERED_BYTES}"
179            )));
180        }
181
182        validate_pipeline(&self.pipeline)?;
183
184        Ok(())
185    }
186
187    /// Parses configuration from a generic [`ConnectorConfig`].
188    ///
189    /// # Errors
190    ///
191    /// Returns `ConnectorError` if required keys are missing or invalid.
192    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
193        if let Some(key) = REMOVED_SOURCE_CONFIG_KEYS
194            .iter()
195            .find(|key| config.get(key).is_some())
196        {
197            let reason = if *key == "max.buffered.events" {
198                "retained ownership is bounded by max.buffered.bytes instead"
199            } else {
200                "the connector did not execute it"
201            };
202            return Err(ConnectorError::ConfigurationError(format!(
203                "MongoDB CDC property '{key}' is not supported: {reason}"
204            )));
205        }
206
207        config.reject_unknown_properties(SOURCE_CONFIG_KEYS, "MongoDB CDC")?;
208
209        let mut cfg = Self {
210            connection_uri: config.require("connection.uri")?.to_string(),
211            database: config.require("database")?.to_string(),
212            collection: config.require("collection")?.to_string(),
213            ..Self::default()
214        };
215
216        if let Some(mode) = config.get_parsed::<FullDocumentMode>("full.document.mode")? {
217            cfg.full_document_mode = mode;
218        }
219        if let Some(pipeline) = config.get("pipeline") {
220            cfg.pipeline = parse_pipeline_property(pipeline)?;
221        }
222        if let Some(max) = config.get_parsed::<usize>("max.buffered.bytes")? {
223            cfg.max_buffered_bytes = max;
224        }
225
226        cfg.validate()?;
227        Ok(cfg)
228    }
229}
230
231fn canonicalize_json(value: serde_json::Value) -> serde_json::Value {
232    match value {
233        serde_json::Value::Array(values) => {
234            serde_json::Value::Array(values.into_iter().map(canonicalize_json).collect())
235        }
236        serde_json::Value::Object(object) => {
237            let mut fields: Vec<_> = object.into_iter().collect();
238            fields.sort_unstable_by(|left, right| left.0.cmp(&right.0));
239            serde_json::Value::Object(
240                fields
241                    .into_iter()
242                    .map(|(key, value)| (key, canonicalize_json(value)))
243                    .collect(),
244            )
245        }
246        scalar => scalar,
247    }
248}
249
250pub(crate) fn canonical_pipeline_json(pipeline: &[serde_json::Value]) -> String {
251    let canonical = canonicalize_json(serde_json::Value::Array(pipeline.to_vec()));
252    serde_json::to_string(&canonical).expect("serde_json::Value serialization cannot fail")
253}
254
255fn normalized_pipeline(
256    pipeline: &[serde_json::Value],
257) -> Result<Vec<serde_json::Value>, ConnectorError> {
258    if pipeline.len() > MAX_PIPELINE_STAGES {
259        return Err(ConnectorError::ConfigurationError(format!(
260            "MongoDB CDC pipeline has {} stages; the maximum is {MAX_PIPELINE_STAGES}",
261            pipeline.len()
262        )));
263    }
264
265    let encoded = serde_json::to_vec(pipeline).map_err(|error| {
266        ConnectorError::ConfigurationError(format!(
267            "MongoDB CDC pipeline cannot be encoded as JSON: {error}"
268        ))
269    })?;
270    if encoded.len() > MAX_PIPELINE_JSON_BYTES {
271        return Err(ConnectorError::ConfigurationError(format!(
272            "MongoDB CDC pipeline is {} bytes; the maximum is {MAX_PIPELINE_JSON_BYTES}",
273            encoded.len()
274        )));
275    }
276
277    let canonical: Vec<_> = pipeline.iter().cloned().map(canonicalize_json).collect();
278    for (i, stage) in canonical.iter().enumerate() {
279        let obj = stage.as_object().ok_or_else(|| {
280            ConnectorError::ConfigurationError(format!("pipeline stage {i} must be a JSON object"))
281        })?;
282        if obj.len() != 1 || !obj.contains_key("$match") {
283            return Err(ConnectorError::ConfigurationError(format!(
284                "pipeline stage {i} is unsafe: MongoDB CDC only supports $match stages because \
285                 projection/replacement stages can alter the resume token"
286            )));
287        }
288        if !obj["$match"].is_object() {
289            return Err(ConnectorError::ConfigurationError(format!(
290                "pipeline stage {i} must contain a JSON object as $match"
291            )));
292        }
293        mongodb::bson::to_document(stage).map_err(|error| {
294            ConnectorError::ConfigurationError(format!(
295                "pipeline stage {i} cannot be represented as BSON: {error}"
296            ))
297        })?;
298    }
299
300    let canonical_bytes = canonical_pipeline_json(&canonical).len();
301    if canonical_bytes > MAX_PIPELINE_JSON_BYTES {
302        return Err(ConnectorError::ConfigurationError(format!(
303            "canonical MongoDB CDC pipeline is {canonical_bytes} bytes; the maximum is \
304             {MAX_PIPELINE_JSON_BYTES}"
305        )));
306    }
307    Ok(canonical)
308}
309
310fn parse_pipeline_property(raw: &str) -> Result<Vec<serde_json::Value>, ConnectorError> {
311    if raw.len() > MAX_PIPELINE_JSON_BYTES {
312        return Err(ConnectorError::ConfigurationError(format!(
313            "MongoDB CDC pipeline property is {} bytes; the maximum is \
314             {MAX_PIPELINE_JSON_BYTES}",
315            raw.len()
316        )));
317    }
318    let value: serde_json::Value = serde_json::from_str(raw).map_err(|error| {
319        ConnectorError::ConfigurationError(format!(
320            "MongoDB CDC pipeline must be valid JSON: {error}"
321        ))
322    })?;
323    let serde_json::Value::Array(pipeline) = value else {
324        return Err(ConnectorError::ConfigurationError(
325            "MongoDB CDC pipeline must be a JSON array".into(),
326        ));
327    };
328    normalized_pipeline(&pipeline)
329}
330
331/// Restrict change-stream pipelines to bounded `$match` stages.
332fn validate_pipeline(pipeline: &[serde_json::Value]) -> Result<(), ConnectorError> {
333    normalized_pipeline(pipeline).map(|_| ())
334}
335
336impl MongoDbSourceConfig {
337    pub(crate) fn normalize_pipeline(&mut self) -> Result<(), ConnectorError> {
338        self.pipeline = normalized_pipeline(&self.pipeline)?;
339        Ok(())
340    }
341}
342
343/// Configuration for the `MongoDB` sink connector.
344#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
345#[serde(deny_unknown_fields)]
346pub struct MongoDbSinkConfig {
347    /// `MongoDB` connection URI.
348    pub connection_uri: String,
349
350    /// Target database name.
351    pub database: String,
352
353    /// Target collection name.
354    pub collection: String,
355
356    /// Whether the target is a standard or time series collection.
357    #[serde(default)]
358    pub collection_kind: CollectionKind,
359
360    /// Write operation mode.
361    #[serde(default)]
362    pub write_mode: WriteMode,
363
364    /// Maximum time between flushes in milliseconds.
365    #[serde(default = "default_flush_interval_ms")]
366    pub flush_interval_ms: u64,
367}
368
369fn default_flush_interval_ms() -> u64 {
370    250
371}
372
373impl Default for MongoDbSinkConfig {
374    fn default() -> Self {
375        Self {
376            connection_uri: "mongodb://localhost:27017".to_string(),
377            database: String::new(),
378            collection: String::new(),
379            collection_kind: CollectionKind::default(),
380            write_mode: WriteMode::default(),
381            flush_interval_ms: default_flush_interval_ms(),
382        }
383    }
384}
385
386impl MongoDbSinkConfig {
387    /// Creates a new sink config with required fields.
388    #[must_use]
389    pub fn new(connection_uri: &str, database: &str, collection: &str) -> Self {
390        Self {
391            connection_uri: connection_uri.to_string(),
392            database: database.to_string(),
393            collection: collection.to_string(),
394            ..Self::default()
395        }
396    }
397
398    /// Returns the flush interval as a `Duration`.
399    #[must_use]
400    pub fn flush_interval(&self) -> Duration {
401        Duration::from_millis(self.flush_interval_ms)
402    }
403
404    /// Validates the configuration.
405    ///
406    /// # Errors
407    ///
408    /// Returns `ConnectorError::ConfigurationError` for invalid settings.
409    pub fn validate(&self) -> Result<(), ConnectorError> {
410        crate::config::require_non_empty(&self.connection_uri, "connection_uri")?;
411        crate::config::require_non_empty(&self.database, "database")?;
412        crate::config::require_non_empty(&self.collection, "collection")?;
413        if self.collection == "*" {
414            return Err(ConnectorError::ConfigurationError(
415                "MongoDB sink collection must name one fixed destination".into(),
416            ));
417        }
418
419        if self.flush_interval_ms == 0 {
420            return Err(ConnectorError::ConfigurationError(
421                "flush_interval_ms must be > 0".to_string(),
422            ));
423        }
424        if let WriteMode::Upsert { key_fields } = &self.write_mode {
425            let mut unique = std::collections::HashSet::with_capacity(key_fields.len());
426            for key in key_fields {
427                if key.trim().is_empty() {
428                    return Err(ConnectorError::ConfigurationError(
429                        "write.mode.key_fields must not contain empty fields".to_string(),
430                    ));
431                }
432                if matches!(key.as_str(), "_op" | "__weight") {
433                    return Err(ConnectorError::ConfigurationError(format!(
434                        "write.mode.key_fields cannot use engine metadata field '{key}'"
435                    )));
436                }
437                if key.starts_with('$') || key.contains('.') {
438                    return Err(ConnectorError::ConfigurationError(format!(
439                        "write.mode.key_fields entry '{key}' must be a top-level MongoDB field and cannot start with '$' or contain '.'"
440                    )));
441                }
442                if !unique.insert(key.as_str()) {
443                    return Err(ConnectorError::ConfigurationError(format!(
444                        "write.mode.key_fields contains duplicate field '{key}'"
445                    )));
446                }
447            }
448            if unique.is_empty() {
449                return Err(ConnectorError::ConfigurationError(
450                    "write.mode.key_fields must contain at least one field".to_string(),
451                ));
452            }
453        }
454
455        // Time series collections only support Insert mode.
456        if let CollectionKind::TimeSeries(time_series) = &self.collection_kind {
457            time_series.validate()?;
458            super::write_model::validate_timeseries_write_mode(&self.write_mode)?;
459        }
460
461        Ok(())
462    }
463
464    /// Parses configuration from a generic [`ConnectorConfig`].
465    ///
466    /// # Errors
467    ///
468    /// Returns `ConnectorError` if required keys are missing or invalid.
469    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
470        if let Some(key) = REMOVED_SINK_CONFIG_KEYS
471            .iter()
472            .find(|key| config.get(key).is_some())
473        {
474            let replacement = match *key {
475                "batch.size" => "fixed memory and MongoDB wire limits govern batching",
476                "write_concern.timeout_ms" => {
477                    "sink.write.timeout.ms governs the complete write deadline"
478                }
479                _ => unreachable!("removed sink key must have a migration message"),
480            };
481            return Err(ConnectorError::ConfigurationError(format!(
482                "MongoDB sink property '{key}' is not supported; {replacement}"
483            )));
484        }
485        config.reject_unknown_properties(SINK_CONFIG_KEYS, "MongoDB sink")?;
486
487        let mut cfg = Self {
488            connection_uri: config.require("connection.uri")?.to_string(),
489            database: config.require("database")?.to_string(),
490            collection: config.require("collection")?.to_string(),
491            ..Self::default()
492        };
493
494        if let Some(interval) = config.get_parsed::<u64>("flush.interval.ms")? {
495            cfg.flush_interval_ms = interval;
496        }
497        if let Some(mode) = config.get("write.mode") {
498            cfg.write_mode = match mode {
499                "insert" => WriteMode::Insert,
500                "cdc_replay" | "cdc-replay" => WriteMode::CdcReplay,
501                "upsert" => {
502                    let keys = config
503                        .require("write.mode.key_fields")?
504                        .split(',')
505                        .map(|s| s.trim().to_string())
506                        .collect();
507                    WriteMode::Upsert { key_fields: keys }
508                }
509                other => {
510                    return Err(ConnectorError::ConfigurationError(format!(
511                        "unknown write.mode: {other}"
512                    )));
513                }
514            };
515        }
516
517        if config.get("write.mode.key_fields").is_some()
518            && !matches!(&cfg.write_mode, WriteMode::Upsert { .. })
519        {
520            return Err(ConnectorError::ConfigurationError(
521                "MongoDB sink property 'write.mode.key_fields' is only valid for write.mode=upsert"
522                    .to_string(),
523            ));
524        }
525        if let Some(time_field) = config.get("timeseries.time_field") {
526            if time_field.trim().is_empty() {
527                return Err(ConnectorError::ConfigurationError(
528                    "timeseries.time_field must not be empty".to_string(),
529                ));
530            }
531            let meta_field = config
532                .get("timeseries.meta_field")
533                .map(str::trim)
534                .filter(|value| !value.is_empty())
535                .map(String::from);
536            let granularity = if let Some(gran_str) = config.get("timeseries.granularity") {
537                match gran_str.to_lowercase().as_str() {
538                    "seconds" => TimeSeriesGranularity::Seconds,
539                    "minutes" => TimeSeriesGranularity::Minutes,
540                    "hours" => TimeSeriesGranularity::Hours,
541                    "custom" => {
542                        let span =
543                            config.require_parsed::<u32>("timeseries.bucket_max_span_seconds")?;
544                        let rounding =
545                            config.require_parsed::<u32>("timeseries.bucket_rounding_seconds")?;
546                        TimeSeriesGranularity::custom(span, rounding)?
547                    }
548                    other => {
549                        return Err(ConnectorError::ConfigurationError(format!(
550                            "unknown timeseries granularity: {other}"
551                        )));
552                    }
553                }
554            } else {
555                TimeSeriesGranularity::Seconds
556            };
557
558            let expire_after_seconds =
559                config.get_parsed::<u64>("timeseries.expire_after_seconds")?;
560
561            cfg.collection_kind = CollectionKind::TimeSeries(TimeSeriesConfig {
562                time_field: time_field.to_string(),
563                meta_field,
564                granularity,
565                expire_after_seconds,
566            });
567        } else if let Some(key) = [
568            "timeseries.meta_field",
569            "timeseries.granularity",
570            "timeseries.bucket_max_span_seconds",
571            "timeseries.bucket_rounding_seconds",
572            "timeseries.expire_after_seconds",
573        ]
574        .into_iter()
575        .find(|key| config.get(key).is_some())
576        {
577            return Err(ConnectorError::ConfigurationError(format!(
578                "MongoDB sink property '{key}' requires 'timeseries.time_field'"
579            )));
580        }
581
582        if !matches!(
583            &cfg.collection_kind,
584            CollectionKind::TimeSeries(TimeSeriesConfig {
585                granularity: TimeSeriesGranularity::Custom { .. },
586                ..
587            })
588        ) {
589            if let Some(key) = [
590                "timeseries.bucket_max_span_seconds",
591                "timeseries.bucket_rounding_seconds",
592            ]
593            .into_iter()
594            .find(|key| config.get(key).is_some())
595            {
596                return Err(ConnectorError::ConfigurationError(format!(
597                    "MongoDB sink property '{key}' is only valid for timeseries.granularity=custom"
598                )));
599            }
600        }
601
602        cfg.validate()?;
603        Ok(cfg)
604    }
605}
606
607#[cfg(test)]
608mod tests {
609    use super::*;
610
611    // ── Source config tests ──
612
613    #[test]
614    fn test_source_config_default() {
615        let cfg = MongoDbSourceConfig::default();
616        assert_eq!(cfg.connection_uri, "mongodb://localhost:27017");
617        assert_eq!(cfg.full_document_mode, FullDocumentMode::Delta);
618        assert_eq!(cfg.max_buffered_bytes, 64 * 1024 * 1024);
619    }
620
621    #[test]
622    fn test_source_config_new() {
623        let cfg = MongoDbSourceConfig::new("mongodb://db:27017", "mydb", "users");
624        assert_eq!(cfg.connection_uri, "mongodb://db:27017");
625        assert_eq!(cfg.database, "mydb");
626        assert_eq!(cfg.collection, "users");
627    }
628
629    #[test]
630    fn test_source_config_validate_empty_uri() {
631        let cfg = MongoDbSourceConfig::new("", "db", "coll");
632        let err = cfg.validate().unwrap_err();
633        assert!(err.to_string().contains("connection_uri"));
634    }
635
636    #[test]
637    fn test_source_config_validate_empty_database() {
638        let cfg = MongoDbSourceConfig::new("mongodb://db:27017", "", "coll");
639        let err = cfg.validate().unwrap_err();
640        assert!(err.to_string().contains("database"));
641    }
642
643    #[test]
644    fn source_config_rejects_database_wildcard_watch() {
645        let cfg = MongoDbSourceConfig::new("mongodb://db:27017", "db", "*");
646        let error = cfg.validate().unwrap_err();
647        assert!(error.to_string().contains("fixed collection"), "{error}");
648        assert!(error.to_string().contains("UUID"), "{error}");
649    }
650
651    #[test]
652    fn source_buffer_byte_bound_has_a_finite_operational_range() {
653        let mut cfg = MongoDbSourceConfig::new("mongodb://db:27017", "db", "coll");
654        for invalid in [MIN_BUFFERED_BYTES - 1, MAX_BUFFERED_BYTES + 1] {
655            cfg.max_buffered_bytes = invalid;
656            let error = cfg.validate().unwrap_err();
657            assert!(error.to_string().contains("max.buffered.bytes"), "{error}");
658        }
659        for valid in [MIN_BUFFERED_BYTES, MAX_BUFFERED_BYTES] {
660            cfg.max_buffered_bytes = valid;
661            cfg.validate().unwrap();
662        }
663    }
664
665    #[test]
666    fn removed_source_properties_are_rejected_explicitly() {
667        for key in REMOVED_SOURCE_CONFIG_KEYS {
668            let mut config = ConnectorConfig::new("mongodb-cdc");
669            config.set("connection.uri", "mongodb://host:27017");
670            config.set("database", "testdb");
671            config.set("collection", "events");
672            config.set(*key, "removed-value");
673            let error = MongoDbSourceConfig::from_config(&config).unwrap_err();
674            assert!(error.to_string().contains(key));
675        }
676    }
677
678    #[test]
679    fn test_source_config_from_connector_config() {
680        let mut config = ConnectorConfig::new("mongodb-cdc");
681        config.set("connection.uri", "mongodb://host:27017");
682        config.set("database", "testdb");
683        config.set("collection", "events");
684        config.set("full.document.mode", "required");
685        config.set(
686            "pipeline",
687            r#"[{"$match":{"z":1,"operationType":"insert"}}]"#,
688        );
689        config.set("max.buffered.bytes", "33554432");
690
691        let cfg = MongoDbSourceConfig::from_config(&config).unwrap();
692        assert_eq!(cfg.connection_uri, "mongodb://host:27017");
693        assert_eq!(cfg.database, "testdb");
694        assert_eq!(cfg.collection, "events");
695        assert_eq!(cfg.full_document_mode, FullDocumentMode::RequirePostImage);
696        assert_eq!(
697            canonical_pipeline_json(&cfg.pipeline),
698            r#"[{"$match":{"operationType":"insert","z":1}}]"#
699        );
700        assert_eq!(cfg.max_buffered_bytes, 32 * 1024 * 1024);
701        assert_eq!(cfg.reader_channel_capacity(), 512);
702        assert_eq!(cfg.cursor_batch_size(), 512);
703    }
704
705    #[test]
706    fn cursor_batch_hint_is_derived_from_buffer_budget() {
707        let mut cfg = MongoDbSourceConfig::new("mongodb://db:27017", "db", "coll");
708        cfg.max_buffered_bytes = MIN_BUFFERED_BYTES;
709        assert_eq!(cfg.reader_channel_capacity(), 16);
710        assert_eq!(cfg.cursor_batch_size(), 16);
711
712        cfg.max_buffered_bytes = DEFAULT_MAX_BUFFERED_BYTES;
713        assert_eq!(cfg.reader_channel_capacity(), 1024);
714        assert_eq!(cfg.cursor_batch_size(), 1000);
715
716        cfg.max_buffered_bytes = usize::MAX;
717        assert_eq!(cfg.reader_channel_capacity(), MAX_READER_CHANNEL_ITEMS);
718        assert_eq!(cfg.cursor_batch_size(), 1000);
719    }
720
721    #[test]
722    fn test_source_config_from_config_missing_required() {
723        let config = ConnectorConfig::new("mongodb-cdc");
724        assert!(MongoDbSourceConfig::from_config(&config).is_err());
725    }
726
727    #[test]
728    fn source_config_rejects_unknown_property() {
729        let mut config = ConnectorConfig::new("mongodb-cdc");
730        config.set("connection.uri", "mongodb://host:27017");
731        config.set("database", "testdb");
732        config.set("collection", "events");
733        config.set("max.await.time.mss", "25");
734
735        let error = MongoDbSourceConfig::from_config(&config).unwrap_err();
736        assert!(error.to_string().contains("max.await.time.mss"));
737    }
738
739    // ── Pipeline validation tests ──
740
741    #[test]
742    fn test_pipeline_valid_match() {
743        let pipeline = vec![serde_json::json!({
744            "$match": { "operationType": "insert" }
745        })];
746        validate_pipeline(&pipeline).unwrap();
747    }
748
749    #[test]
750    fn pipeline_stage_must_be_a_document() {
751        let error = validate_pipeline(&[serde_json::json!("$match")]).unwrap_err();
752        assert!(error.to_string().contains("must be a JSON object"));
753    }
754
755    #[test]
756    fn pipeline_stage_must_be_bson_representable() {
757        let pipeline = vec![serde_json::json!({
758            "$match": { "value": u64::MAX }
759        })];
760        let error = validate_pipeline(&pipeline).unwrap_err();
761        assert!(error.to_string().contains("cannot be represented as BSON"));
762    }
763
764    #[test]
765    fn pipeline_only_accepts_match_stages() {
766        for stage in [
767            serde_json::json!({ "$project": { "_id": 1, "name": 1 } }),
768            serde_json::json!({ "$unset": "_id" }),
769            serde_json::json!({ "$set": { "_id": "overwritten" } }),
770            serde_json::json!({ "$replaceRoot": { "newRoot": "$fullDocument" } }),
771            serde_json::json!({ "$match": {}, "$project": { "_id": 1 } }),
772        ] {
773            let error = validate_pipeline(&[stage]).unwrap_err();
774            assert!(error.to_string().contains("unsafe"), "{error}");
775        }
776    }
777
778    #[test]
779    fn pipeline_property_requires_a_bounded_json_array() {
780        let mut config = ConnectorConfig::new("mongodb-cdc");
781        config.set("connection.uri", "mongodb://host:27017");
782        config.set("database", "testdb");
783        config.set("collection", "events");
784
785        config.set("pipeline", r#"{"$match":{}}"#);
786        let error = MongoDbSourceConfig::from_config(&config).unwrap_err();
787        assert!(error.to_string().contains("JSON array"), "{error}");
788
789        let stages = vec![serde_json::json!({ "$match": {} }); MAX_PIPELINE_STAGES + 1];
790        config.set("pipeline", serde_json::to_string(&stages).unwrap());
791        let error = MongoDbSourceConfig::from_config(&config).unwrap_err();
792        assert!(error.to_string().contains("maximum"), "{error}");
793
794        let oversized = format!(
795            r#"[{{"$match":{{"payload":"{}"}}}}]"#,
796            "x".repeat(MAX_PIPELINE_JSON_BYTES)
797        );
798        config.set("pipeline", oversized);
799        let error = MongoDbSourceConfig::from_config(&config).unwrap_err();
800        assert!(error.to_string().contains("maximum"), "{error}");
801    }
802
803    #[test]
804    fn pipeline_is_recursively_canonicalized() {
805        let pipeline =
806            parse_pipeline_property(r#"[{"$match":{"z":1,"a":{"y":2,"b":3}}}]"#).unwrap();
807        assert_eq!(
808            canonical_pipeline_json(&pipeline),
809            r#"[{"$match":{"a":{"b":3,"y":2},"z":1}}]"#
810        );
811    }
812
813    #[test]
814    fn pipeline_match_expression_must_be_a_document() {
815        let error = validate_pipeline(&[serde_json::json!({ "$match": "insert" })]).unwrap_err();
816        assert!(error.to_string().contains("as $match"), "{error}");
817    }
818
819    // ── Full document mode tests ──
820
821    #[test]
822    fn test_full_document_mode_fromstr() {
823        assert_eq!(
824            "delta".parse::<FullDocumentMode>().unwrap(),
825            FullDocumentMode::Delta
826        );
827        assert_eq!(
828            "required".parse::<FullDocumentMode>().unwrap(),
829            FullDocumentMode::RequirePostImage
830        );
831        assert!("update_lookup".parse::<FullDocumentMode>().is_err());
832        assert!("when_available".parse::<FullDocumentMode>().is_err());
833        assert!("bad".parse::<FullDocumentMode>().is_err());
834    }
835
836    #[test]
837    fn test_full_document_mode_display() {
838        assert_eq!(FullDocumentMode::Delta.to_string(), "delta");
839        assert_eq!(FullDocumentMode::RequirePostImage.to_string(), "required");
840    }
841
842    // ── Sink config tests ──
843
844    #[test]
845    fn test_sink_config_default() {
846        let cfg = MongoDbSinkConfig::default();
847        assert_eq!(cfg.flush_interval_ms, 250);
848        assert!(matches!(cfg.collection_kind, CollectionKind::Standard));
849    }
850
851    #[test]
852    fn test_sink_config_new() {
853        let cfg = MongoDbSinkConfig::new("mongodb://db:27017", "mydb", "events");
854        assert_eq!(cfg.connection_uri, "mongodb://db:27017");
855        assert_eq!(cfg.database, "mydb");
856        assert_eq!(cfg.collection, "events");
857    }
858
859    #[test]
860    fn test_sink_config_validate_empty_uri() {
861        let cfg = MongoDbSinkConfig::new("", "db", "coll");
862        let err = cfg.validate().unwrap_err();
863        assert!(err.to_string().contains("connection_uri"));
864    }
865
866    #[test]
867    fn test_sink_config_validate_zero_flush_interval() {
868        let mut cfg = MongoDbSinkConfig::new("mongodb://db:27017", "db", "coll");
869        cfg.flush_interval_ms = 0;
870        let err = cfg.validate().unwrap_err();
871        assert!(err.to_string().contains("flush_interval_ms"));
872    }
873
874    #[test]
875    fn sink_requires_a_fixed_destination_collection() {
876        let cfg = MongoDbSinkConfig::new("mongodb://db:27017", "db", "*");
877        let error = cfg.validate().unwrap_err();
878        assert!(error.to_string().contains("fixed destination"));
879    }
880
881    #[test]
882    fn test_sink_config_timeseries_upsert_rejected() {
883        let mut cfg = MongoDbSinkConfig::new("mongodb://db:27017", "db", "ts");
884        cfg.collection_kind =
885            CollectionKind::TimeSeries(super::super::timeseries::TimeSeriesConfig {
886                time_field: "ts".to_string(),
887                meta_field: None,
888                granularity: super::super::timeseries::TimeSeriesGranularity::Seconds,
889                expire_after_seconds: None,
890            });
891        cfg.write_mode = WriteMode::Upsert {
892            key_fields: vec!["id".to_string()],
893        };
894        let err = cfg.validate().unwrap_err();
895        assert!(err.to_string().contains("time series"));
896    }
897
898    #[test]
899    fn test_sink_config_flush_interval() {
900        let cfg = MongoDbSinkConfig::default();
901        assert_eq!(cfg.flush_interval(), Duration::from_millis(250));
902    }
903
904    #[test]
905    fn test_sink_config_from_connector_config() {
906        let mut config = ConnectorConfig::new("mongodb-sink");
907        config.set("connection.uri", "mongodb://host:27017");
908        config.set("database", "testdb");
909        config.set("collection", "out");
910        config.set("write.mode", "upsert");
911        config.set("write.mode.key_fields", "id");
912
913        let cfg = MongoDbSinkConfig::from_config(&config).unwrap();
914        assert!(matches!(cfg.write_mode, WriteMode::Upsert { .. }));
915    }
916
917    #[test]
918    fn removed_sink_properties_are_rejected() {
919        for key in REMOVED_SINK_CONFIG_KEYS {
920            let mut config = ConnectorConfig::new("mongodb-sink");
921            config.set("connection.uri", "mongodb://host:27017");
922            config.set("database", "testdb");
923            config.set("collection", "out");
924            config.set(*key, "1000");
925
926            let error = MongoDbSinkConfig::from_config(&config).unwrap_err();
927            assert!(error.to_string().contains(key));
928        }
929    }
930
931    #[test]
932    fn sink_upsert_keys_must_be_non_empty_and_unique() {
933        for keys in [
934            vec![],
935            vec![String::new()],
936            vec!["id".to_string(), "id".to_string()],
937            vec!["_op".to_string()],
938            vec!["$expr".to_string()],
939            vec!["customer.id".to_string()],
940        ] {
941            let mut config = MongoDbSinkConfig::new("mongodb://host:27017", "db", "out");
942            config.write_mode = WriteMode::Upsert { key_fields: keys };
943            assert!(config.validate().is_err());
944        }
945    }
946
947    #[test]
948    fn sink_config_rejects_mode_irrelevant_properties() {
949        let cases = [
950            ("insert", "write.mode.key_fields", "id"),
951            ("insert", "ordered", "false"),
952            ("insert", "write.mode.upsert_on_missing", "true"),
953            ("insert", "write_concern.journal", "false"),
954        ];
955
956        for (mode, key, value) in cases {
957            let mut config = ConnectorConfig::new("mongodb-sink");
958            config.set("connection.uri", "mongodb://host:27017");
959            config.set("database", "testdb");
960            config.set("collection", "out");
961            config.set("write.mode", mode);
962            config.set(key, value);
963
964            let error = MongoDbSinkConfig::from_config(&config).unwrap_err();
965            assert!(error.to_string().contains(key), "{error}");
966        }
967
968        let mut config = ConnectorConfig::new("mongodb-sink");
969        config.set("connection.uri", "mongodb://host:27017");
970        config.set("database", "testdb");
971        config.set("collection", "out");
972        config.set("write.mode", "replace");
973        let error = MongoDbSinkConfig::from_config(&config).unwrap_err();
974        assert!(error.to_string().contains("replace"), "{error}");
975    }
976
977    #[test]
978    fn sink_config_rejects_irrelevant_timeseries_properties() {
979        for key in [
980            "timeseries.meta_field",
981            "timeseries.granularity",
982            "timeseries.bucket_max_span_seconds",
983            "timeseries.bucket_rounding_seconds",
984            "timeseries.expire_after_seconds",
985        ] {
986            let mut config = ConnectorConfig::new("mongodb-sink");
987            config.set("connection.uri", "mongodb://host:27017");
988            config.set("database", "testdb");
989            config.set("collection", "out");
990            config.set(key, "60");
991
992            let error = MongoDbSinkConfig::from_config(&config).unwrap_err();
993            assert!(error.to_string().contains(key), "{error}");
994        }
995
996        let mut config = ConnectorConfig::new("mongodb-sink");
997        config.set("connection.uri", "mongodb://host:27017");
998        config.set("database", "testdb");
999        config.set("collection", "out");
1000        config.set("timeseries.time_field", "timestamp");
1001        config.set("timeseries.granularity", "seconds");
1002        config.set("timeseries.bucket_max_span_seconds", "60");
1003
1004        let error = MongoDbSinkConfig::from_config(&config).unwrap_err();
1005        assert!(error
1006            .to_string()
1007            .contains("timeseries.bucket_max_span_seconds"));
1008    }
1009
1010    #[test]
1011    fn sink_config_rejects_unknown_property() {
1012        let mut config = ConnectorConfig::new("mongodb-sink");
1013        config.set("connection.uri", "mongodb://host:27017");
1014        config.set("database", "testdb");
1015        config.set("collection", "out");
1016        config.set("flush.intervall.ms", "10");
1017
1018        let error = MongoDbSinkConfig::from_config(&config).unwrap_err();
1019        assert!(error.to_string().contains("flush.intervall.ms"));
1020    }
1021
1022    #[test]
1023    fn test_sink_config_timeseries_parsing() {
1024        // Test standard granularity with metadata and TTL
1025        let mut config = ConnectorConfig::new("mongodb-sink");
1026        config.set("connection.uri", "mongodb://host:27017");
1027        config.set("database", "testdb");
1028        config.set("collection", "ts_out");
1029        config.set("timeseries.time_field", "timestamp");
1030        config.set("timeseries.meta_field", "sensor_id");
1031        config.set("timeseries.granularity", "minutes");
1032        config.set("timeseries.expire_after_seconds", "86400");
1033
1034        let cfg = MongoDbSinkConfig::from_config(&config).unwrap();
1035        if let CollectionKind::TimeSeries(ts) = cfg.collection_kind {
1036            assert_eq!(ts.time_field, "timestamp");
1037            assert_eq!(ts.meta_field.as_deref(), Some("sensor_id"));
1038            assert_eq!(ts.granularity, TimeSeriesGranularity::Minutes);
1039            assert_eq!(ts.expire_after_seconds, Some(86400));
1040        } else {
1041            panic!("Expected TimeSeries collection kind");
1042        }
1043
1044        // Test custom granularity
1045        let mut config_custom = ConnectorConfig::new("mongodb-sink");
1046        config_custom.set("connection.uri", "mongodb://host:27017");
1047        config_custom.set("database", "testdb");
1048        config_custom.set("collection", "ts_custom");
1049        config_custom.set("timeseries.time_field", "timestamp");
1050        config_custom.set("timeseries.granularity", "custom");
1051        config_custom.set("timeseries.bucket_max_span_seconds", "3600");
1052        config_custom.set("timeseries.bucket_rounding_seconds", "3600");
1053
1054        let cfg_custom = MongoDbSinkConfig::from_config(&config_custom).unwrap();
1055        if let CollectionKind::TimeSeries(ts) = cfg_custom.collection_kind {
1056            assert_eq!(ts.time_field, "timestamp");
1057            assert_eq!(
1058                ts.granularity,
1059                TimeSeriesGranularity::Custom {
1060                    bucket_max_span_seconds: 3600,
1061                    bucket_rounding_seconds: 3600,
1062                }
1063            );
1064        } else {
1065            panic!("Expected TimeSeries collection kind");
1066        }
1067    }
1068
1069    #[test]
1070    fn test_sink_config_timeseries_empty_time_field_rejected() {
1071        let mut config = ConnectorConfig::new("mongodb-sink");
1072        config.set("connection.uri", "mongodb://host:27017");
1073        config.set("database", "testdb");
1074        config.set("collection", "ts");
1075        config.set("timeseries.time_field", "  ");
1076        let err = MongoDbSinkConfig::from_config(&config).unwrap_err();
1077        assert!(err.to_string().contains("time_field"));
1078    }
1079}
laminar_connectors/mongodb/config.rs

laminar_connectors/mongodb/
config.rs
