laminar_connectors/mongodb/config/
mod.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;
laminar_connectors/mongodb/config/mod.rs

laminar_connectors/mongodb/config/
mod.rs
