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}
120
121impl MongoCdcSourceConfig {
122    /// Validate fail-fast invariants. Called from `MongoCdcSource::new`.
123    pub fn validate(&self) -> Result<(), FaucetError> {
124        if self.connection_uri.trim().is_empty() {
125            return Err(FaucetError::Config(
126                "mongodb-cdc: connection_uri must not be empty".into(),
127            ));
128        }
129        match &self.scope {
130            Scope::Collection {
131                database,
132                collection,
133            } => {
134                if database.trim().is_empty() || collection.trim().is_empty() {
135                    return Err(FaucetError::Config(
136                        "mongodb-cdc: scope.database and scope.collection must not be empty".into(),
137                    ));
138                }
139            }
140            Scope::Database { database } => {
141                if database.trim().is_empty() {
142                    return Err(FaucetError::Config(
143                        "mongodb-cdc: scope.database must not be empty".into(),
144                    ));
145                }
146            }
147            Scope::Cluster => {}
148        }
149        let idle_ms = self.idle_timeout.as_millis();
150        if idle_ms == 0 {
151            return Err(FaucetError::Config(
152                "mongodb-cdc: idle_timeout must be > 0".into(),
153            ));
154        }
155        if u128::from(self.max_await_time_ms) >= idle_ms {
156            return Err(FaucetError::Config(format!(
157                "mongodb-cdc: max_await_time_ms ({}) must be strictly less than \
158                 idle_timeout ({}ms)",
159                self.max_await_time_ms, idle_ms
160            )));
161        }
162        faucet_core::validate_batch_size(self.batch_size)?;
163        // Validate the derived state key is store-safe.
164        let key = crate::state::state_key(&self.scope);
165        faucet_core::state::validate_state_key(&key)?;
166        Ok(())
167    }
168}
169
170impl fmt::Debug for MongoCdcSourceConfig {
171    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
172        f.debug_struct("MongoCdcSourceConfig")
173            .field("connection_uri", &"***")
174            .field("scope", &self.scope)
175            .field("operation_types", &self.operation_types)
176            .field("full_document", &self.full_document)
177            .field(
178                "full_document_before_change",
179                &self.full_document_before_change,
180            )
181            .field("start_from", &self.start_from)
182            .field("aggregation_pipeline", &self.aggregation_pipeline)
183            .field("idle_timeout", &self.idle_timeout)
184            .field("max_await_time_ms", &self.max_await_time_ms)
185            .field("batch_size", &self.batch_size)
186            .finish()
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193    use serde_json::json;
194
195    fn minimal() -> MongoCdcSourceConfig {
196        serde_json::from_value(json!({
197            "connection_uri": "mongodb://localhost:27017/?replicaSet=rs0",
198            "scope": { "type": "collection", "database": "app", "collection": "users" }
199        }))
200        .unwrap()
201    }
202
203    #[test]
204    fn defaults_via_serde() {
205        let c = minimal();
206        assert_eq!(c.idle_timeout.as_secs(), 30);
207        assert_eq!(c.max_await_time_ms, 1000);
208        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
209        assert_eq!(c.start_from, StartFrom::Now);
210        assert_eq!(c.full_document, FullDocument::Off);
211        assert!(c.operation_types.is_empty());
212    }
213
214    #[test]
215    fn scope_tagged_enum_round_trips() {
216        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
217            "connection_uri": "mongodb://h/?replicaSet=rs0",
218            "scope": { "type": "cluster" }
219        }))
220        .unwrap();
221        assert_eq!(c.scope, Scope::Cluster);
222    }
223
224    #[test]
225    fn validate_rejects_empty_uri() {
226        let mut c = minimal();
227        c.connection_uri = "  ".into();
228        assert!(c.validate().is_err());
229    }
230
231    #[test]
232    fn validate_rejects_max_await_ge_idle() {
233        let mut c = minimal();
234        c.idle_timeout = Duration::from_millis(500);
235        c.max_await_time_ms = 500;
236        assert!(c.validate().is_err());
237    }
238
239    #[test]
240    fn validate_rejects_empty_collection() {
241        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
242            "connection_uri": "mongodb://h/?replicaSet=rs0",
243            "scope": { "type": "collection", "database": "app", "collection": "" }
244        }))
245        .unwrap();
246        assert!(c.validate().is_err());
247    }
248
249    #[test]
250    fn validate_rejects_empty_database_scope() {
251        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
252            "connection_uri": "mongodb://h/?replicaSet=rs0",
253            "scope": { "type": "database", "database": "" }
254        }))
255        .unwrap();
256        assert!(c.validate().is_err());
257    }
258
259    #[test]
260    fn validate_accepts_batch_size_zero_sentinel() {
261        let mut c = minimal();
262        c.batch_size = 0;
263        assert!(c.validate().is_ok());
264    }
265
266    #[test]
267    fn validate_accepts_minimal() {
268        assert!(minimal().validate().is_ok());
269    }
270
271    #[test]
272    fn debug_redacts_connection_uri() {
273        let c: MongoCdcSourceConfig = serde_json::from_value(json!({
274            "connection_uri": "mongodb://user:secret@h:27017/?replicaSet=rs0",
275            "scope": { "type": "cluster" }
276        }))
277        .unwrap();
278        let dbg = format!("{c:?}");
279        assert!(dbg.contains("***"));
280        assert!(!dbg.contains("secret"));
281    }
282
283    #[test]
284    fn schema_includes_connection_uri_required() {
285        let schema = schemars::schema_for!(MongoCdcSourceConfig);
286        let v = serde_json::to_value(&schema).unwrap();
287        let req = v["required"].as_array().unwrap();
288        let names: Vec<_> = req.iter().filter_map(|x| x.as_str()).collect();
289        assert!(names.contains(&"connection_uri"));
290    }
291}
faucet_source_mongodb_cdc/config.rs

faucet_source_mongodb_cdc/
config.rs
