faucet_source_mongodb_cdc/
config.rs

1//! Configuration for [`MongoCdcSource`](crate::MongoCdcSource).
2
3use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError};
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7use std::fmt;
8use std::time::Duration;
9
10fn default_idle_timeout() -> Duration {
11    Duration::from_secs(30)
12}
13fn default_max_await_time_ms() -> u64 {
14    1000
15}
16fn default_batch_size() -> usize {
17    DEFAULT_BATCH_SIZE
18}
19
20/// Which change stream to open.
21#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
22#[serde(tag = "type", rename_all = "snake_case")]
23pub enum Scope {
24    /// Watch one collection.
25    Collection {
26        database: String,
27        collection: String,
28    },
29    /// Watch every collection in one database.
30    Database { database: String },
31    /// Watch the whole deployment.
32    #[default]
33    Cluster,
34}
35
36/// Where to start the change stream on a fresh run (no persisted bookmark).
37#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
38#[serde(tag = "type", rename_all = "snake_case")]
39pub enum StartFrom {
40    /// Start at the current cluster time (default — do not replay history).
41    #[default]
42    Now,
43    /// Start at the earliest retained oplog entry (errors if it has rolled).
44    Earliest,
45    /// Resume after a specific opaque token. Overrides a persisted bookmark.
46    ResumeToken { token: Value },
47    /// Start at a wall-clock second. Overrides a persisted bookmark.
48    Timestamp { timestamp_secs: u32 },
49}
50
51/// Post-image inclusion mode (maps to the driver `FullDocumentType`).
52#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
53#[serde(rename_all = "snake_case")]
54pub enum FullDocument {
55    /// Do not request the post-image (default). `after` is null on update/replace.
56    #[default]
57    Off,
58    /// Include the post-image when the server can supply it.
59    WhenAvailable,
60    /// Require the post-image (errors if unavailable).
61    Required,
62    /// Look up the current full document at emit time (read-skew caveat).
63    UpdateLookup,
64}
65
66/// Pre-image inclusion mode (maps to the driver `FullDocumentBeforeChangeType`).
67#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
68#[serde(rename_all = "snake_case")]
69pub enum FullDocumentBeforeChange {
70    /// Do not request the pre-image (default).
71    #[default]
72    Off,
73    /// Include the pre-image when available (MongoDB 6.0+, pre-images enabled).
74    WhenAvailable,
75    /// Require the pre-image (errors if unavailable).
76    Required,
77}
78
79/// Configuration for [`MongoCdcSource`](crate::MongoCdcSource).
80#[derive(Clone, Serialize, Deserialize, JsonSchema)]
81pub struct MongoCdcSourceConfig {
82    /// MongoDB connection URI. Must point at a replica set or sharded cluster.
83    pub connection_uri: String,
84    /// Which change stream to open.
85    #[serde(default)]
86    pub scope: Scope,
87    /// Allowlist of operation types (`insert`/`update`/`replace`/`delete`/...).
88    /// Empty = all. Prepended server-side as a `$match` on `operationType`.
89    #[serde(default)]
90    pub operation_types: Vec<String>,
91    /// Post-image inclusion mode.
92    #[serde(default)]
93    pub full_document: FullDocument,
94    /// Pre-image inclusion mode (MongoDB 6.0+).
95    #[serde(default)]
96    pub full_document_before_change: FullDocumentBeforeChange,
97    /// Start position on a fresh run (ignored once a bookmark exists, except an
98    /// explicit `resume_token`/`timestamp` which overrides the bookmark).
99    #[serde(default)]
100    pub start_from: StartFrom,
101    /// Optional extra aggregation-pipeline stages, each a JSON object, appended
102    /// after the operation-type `$match`.
103    #[serde(default)]
104    pub aggregation_pipeline: Vec<Value>,
105    /// Terminator: flush the buffer and end the fetch cycle after this much
106    /// quiet. Default 30s.
107    #[serde(
108        default = "default_idle_timeout",
109        with = "faucet_core::config::duration_secs"
110    )]
111    #[schemars(with = "u64")]
112    pub idle_timeout: Duration,
113    /// Server-side blocking `getMore` wait. Must be `< idle_timeout`. Default 1000.
114    #[serde(default = "default_max_await_time_ms")]
115    pub max_await_time_ms: u64,
116    /// Records per emitted `StreamPage`. `0` = drain until idle into one page.
117    #[serde(default = "default_batch_size")]
118    pub batch_size: usize,
119    /// Max change records buffered in memory before a page is emitted, after
120    /// which the run aborts with a typed [`FaucetError::Source`]. `None` =
121    /// unbounded. The OOM safety valve: with `batch_size: 0` (or a library
122    /// caller using `fetch_all`) the change-event buffer would otherwise grow
123    /// without bound for the whole fetch cycle and risk an OOM-kill on a
124    /// high-throughput stream. Mirrors the same field on the sibling
125    /// `postgres-cdc` / `mysql-cdc` sources.
126    #[serde(default)]
127    pub max_staged_records: Option<usize>,
128}
129
130impl MongoCdcSourceConfig {
131    /// Validate fail-fast invariants. Called from `MongoCdcSource::new`.
132    pub fn validate(&self) -> Result<(), FaucetError> {
133        if self.connection_uri.trim().is_empty() {
134            return Err(FaucetError::Config(
135                "mongodb-cdc: connection_uri must not be empty".into(),
136            ));
137        }
138        match &self.scope {
139            Scope::Collection {
140                database,
141                collection,
142            } => {
143                if database.trim().is_empty() || collection.trim().is_empty() {
144                    return Err(FaucetError::Config(
145                        "mongodb-cdc: scope.database and scope.collection must not be empty".into(),
146                    ));
147                }
148            }
149            Scope::Database { database } => {
150                if database.trim().is_empty() {
151                    return Err(FaucetError::Config(
152                        "mongodb-cdc: scope.database must not be empty".into(),
153                    ));
154                }
155            }
156            Scope::Cluster => {}
157        }
158        let idle_ms = self.idle_timeout.as_millis();
159        if idle_ms == 0 {
160            return Err(FaucetError::Config(
161                "mongodb-cdc: idle_timeout must be > 0".into(),
162            ));
163        }
164        if u128::from(self.max_await_time_ms) >= idle_ms {
165            return Err(FaucetError::Config(format!(
166                "mongodb-cdc: max_await_time_ms ({}) must be strictly less than \
167                 idle_timeout ({}ms)",
168                self.max_await_time_ms, idle_ms
169            )));
170        }
171        faucet_core::validate_batch_size(self.batch_size)?;
172        // Validate the derived state key is store-safe.
173        let key = crate::state::state_key(&self.scope);
174        faucet_core::state::validate_state_key(&key)?;
175        Ok(())
176    }
177}
178
179impl fmt::Debug for MongoCdcSourceConfig {
180    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
181        f.debug_struct("MongoCdcSourceConfig")
182            .field("connection_uri", &"***")
183            .field("scope", &self.scope)
184            .field("operation_types", &self.operation_types)
185            .field("full_document", &self.full_document)
186            .field(
187                "full_document_before_change",
188                &self.full_document_before_change,
189            )
190            .field("start_from", &self.start_from)
191            .field("aggregation_pipeline", &self.aggregation_pipeline)
192            .field("idle_timeout", &self.idle_timeout)
193            .field("max_await_time_ms", &self.max_await_time_ms)
194            .field("batch_size", &self.batch_size)
195            .field("max_staged_records", &self.max_staged_records)
196            .finish()
197    }
198}
199
200#[cfg(test)]
201mod tests {
202    use super::*;
203    use serde_json::json;
204
205    fn minimal() -> MongoCdcSourceConfig {
206        serde_json::from_value(json!({
207            "connection_uri": "mongodb://localhost:27017/?replicaSet=rs0",
208            "scope": { "type": "collection", "database": "app", "collection": "users" }
209        }))
210        .unwrap()
211    }
212
213    #[test]
214    fn defaults_via_serde() {
215        let c = minimal();
216        assert_eq!(c.idle_timeout.as_secs(), 30);
217        assert_eq!(c.max_await_time_ms, 1000);
218        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
219        assert_eq!(c.start_from, StartFrom::Now);
220        assert_eq!(c.full_document, FullDocument::Off);
221        assert!(c.operation_types.is_empty());
222        // Defaults to unbounded, matching postgres-cdc / mysql-cdc.
223        assert_eq!(c.max_staged_records, None);
224    }
225
226    #[test]
227    fn max_staged_records_round_trips() {
228        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
229            "connection_uri": "mongodb://h/?replicaSet=rs0",
230            "scope": { "type": "cluster" },
231            "max_staged_records": 100000
232        }))
233        .unwrap();
234        assert_eq!(c.max_staged_records, Some(100_000));
235        assert!(c.validate().is_ok());
236    }
237
238    #[test]
239    fn scope_tagged_enum_round_trips() {
240        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
241            "connection_uri": "mongodb://h/?replicaSet=rs0",
242            "scope": { "type": "cluster" }
243        }))
244        .unwrap();
245        assert_eq!(c.scope, Scope::Cluster);
246    }
247
248    #[test]
249    fn validate_rejects_empty_uri() {
250        let mut c = minimal();
251        c.connection_uri = "  ".into();
252        assert!(c.validate().is_err());
253    }
254
255    #[test]
256    fn validate_rejects_max_await_ge_idle() {
257        let mut c = minimal();
258        c.idle_timeout = Duration::from_millis(500);
259        c.max_await_time_ms = 500;
260        assert!(c.validate().is_err());
261    }
262
263    #[test]
264    fn validate_rejects_empty_collection() {
265        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
266            "connection_uri": "mongodb://h/?replicaSet=rs0",
267            "scope": { "type": "collection", "database": "app", "collection": "" }
268        }))
269        .unwrap();
270        assert!(c.validate().is_err());
271    }
272
273    #[test]
274    fn validate_rejects_empty_database_scope() {
275        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
276            "connection_uri": "mongodb://h/?replicaSet=rs0",
277            "scope": { "type": "database", "database": "" }
278        }))
279        .unwrap();
280        assert!(c.validate().is_err());
281    }
282
283    #[test]
284    fn validate_accepts_batch_size_zero_sentinel() {
285        let mut c = minimal();
286        c.batch_size = 0;
287        assert!(c.validate().is_ok());
288    }
289
290    #[test]
291    fn validate_accepts_minimal() {
292        assert!(minimal().validate().is_ok());
293    }
294
295    #[test]
296    fn debug_redacts_connection_uri() {
297        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
298            "connection_uri": "mongodb://user:secret@h:27017/?replicaSet=rs0",
299            "scope": { "type": "cluster" }
300        }))
301        .unwrap();
302        let dbg = format!("{c:?}");
303        assert!(dbg.contains("***"));
304        assert!(!dbg.contains("secret"));
305    }
306
307    #[test]
308    fn schema_includes_connection_uri_required() {
309        let schema = schemars::schema_for!(MongoCdcSourceConfig);
310        let v = serde_json::to_value(&schema).unwrap();
311        let req = v["required"].as_array().unwrap();
312        let names: Vec<_> = req.iter().filter_map(|x| x.as_str()).collect();
313        assert!(names.contains(&"connection_uri"));
314    }
315}
faucet_source_mongodb_cdc/config.rs

faucet_source_mongodb_cdc/
config.rs
