laminar_connectors/mongodb/
change_event.rs

1//! `MongoDB` change stream event types.
2//!
3//! Represents the structure of change events emitted by `MongoDB` change
4//! streams. Maps `MongoDB`'s change event document structure to typed Rust
5//! structs for safe downstream processing.
6//!
7//! # Operation Types
8//!
9//! `MongoDB` change streams emit events for the following operations:
10//!
11//! - `insert` — new document created; `full_document` always present
12//! - `update` — document modified; `update_description` contains delta
13//! - `replace` — full document replacement
14//! - `delete` — document removed; only `document_key` present
15//! - `drop` — collection dropped (lifecycle event)
16//! - `rename` — collection renamed (lifecycle event)
17//! - `invalidate` — stream invalidated; `resumeAfter` is no longer legal
18
19use std::collections::HashMap;
20
21/// The type of operation described by a change event.
22#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
23#[serde(rename_all = "camelCase")]
24pub enum OperationType {
25    /// A new document was inserted.
26    Insert,
27    /// An existing document was updated (delta available).
28    Update,
29    /// An existing document was fully replaced.
30    Replace,
31    /// A document was deleted.
32    Delete,
33    /// The collection was dropped.
34    Drop,
35    /// The collection was renamed.
36    Rename,
37    /// The change stream was invalidated and must be restarted.
38    Invalidate,
39    /// The database was dropped.
40    DropDatabase,
41    /// An operation type not yet mapped by this enum.
42    Other(String),
43}
44
45impl OperationType {
46    /// Returns the single-character code for the operation, compatible with
47    /// the CDC envelope format used elsewhere in `LaminarDB`.
48    #[must_use]
49    pub fn as_str(&self) -> &str {
50        match self {
51            Self::Insert => "I",
52            Self::Update => "U",
53            Self::Replace => "R",
54            Self::Delete => "D",
55            Self::Drop => "DROP",
56            Self::Rename => "RENAME",
57            Self::Invalidate => "INVALIDATE",
58            Self::DropDatabase => "DROP_DATABASE",
59            Self::Other(s) => s.as_str(),
60        }
61    }
62}
63
64impl std::fmt::Display for OperationType {
65    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66        f.write_str(self.as_str())
67    }
68}
69
70/// Describes the fields changed during an `update` operation.
71///
72/// Mirrors `MongoDB`'s `updateDescription` subdocument in change events.
73/// Only populated for `update` operations; `insert`, `replace`, and
74/// `delete` events do not include this.
75#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
76pub struct UpdateDescription {
77    /// Fields that were set or modified, as key-value JSON pairs.
78    pub updated_fields: HashMap<String, serde_json::Value>,
79
80    /// Fields that were removed from the document.
81    pub removed_fields: Vec<String>,
82
83    /// Arrays that were truncated (`MongoDB` 6.0+).
84    #[serde(default)]
85    pub truncated_arrays: Vec<TruncatedArray>,
86
87    /// Ambiguous update paths mapped to their exact path components.
88    ///
89    /// Replay must not treat these keys as ordinary dotted paths. The `MongoDB` sink rejects a
90    /// non-empty map until it can express the update faithfully with `$setField` semantics.
91    #[serde(default)]
92    pub disambiguated_paths: HashMap<String, serde_json::Value>,
93}
94
95/// Describes a truncated array in an update operation (`MongoDB` 6.0+).
96#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
97pub struct TruncatedArray {
98    /// Dot-notation path of the truncated array field.
99    pub field: String,
100    /// New size of the array after truncation.
101    pub new_size: u32,
102}
103
104/// The namespace (database + collection) for a change event.
105#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
106pub struct Namespace {
107    /// Database name.
108    pub db: String,
109    /// Collection name.
110    pub coll: String,
111}
112
113impl Namespace {
114    /// Returns the fully qualified namespace as `db.coll`.
115    #[must_use]
116    pub fn full_name(&self) -> String {
117        format!("{}.{}", self.db, self.coll)
118    }
119}
120
121impl std::fmt::Display for Namespace {
122    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
123        write!(f, "{}.{}", self.db, self.coll)
124    }
125}
126
127/// A change event from a `MongoDB` change stream.
128///
129/// This struct captures all fields relevant for CDC processing. The
130/// `resume_token` is the opaque `_id` field from the change event document,
131/// used for resuming the stream.
132#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
133pub struct MongoDbChangeEvent {
134    /// The operation type.
135    pub operation_type: OperationType,
136
137    /// The namespace (database + collection).
138    pub namespace: Namespace,
139
140    /// The document key (`_id` and any shard key fields).
141    /// Serialized as a JSON string for Arrow compatibility.
142    pub document_key: String,
143
144    /// The full document (present for insert, replace, and optionally
145    /// for update events depending on `FullDocumentMode`).
146    pub full_document: Option<String>,
147
148    /// Update delta (only for `update` operations).
149    pub update_description: Option<UpdateDescription>,
150
151    /// Server timestamp of the operation (cluster time).
152    pub cluster_time_secs: u32,
153
154    /// Cluster time increment.
155    pub cluster_time_inc: u32,
156
157    /// The opaque resume token for this event, serialized as JSON.
158    pub resume_token: String,
159
160    /// Wall clock timestamp in milliseconds since Unix epoch.
161    pub wall_time_ms: i64,
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167
168    #[test]
169    fn test_operation_type_as_str() {
170        assert_eq!(OperationType::Insert.as_str(), "I");
171        assert_eq!(OperationType::Update.as_str(), "U");
172        assert_eq!(OperationType::Replace.as_str(), "R");
173        assert_eq!(OperationType::Delete.as_str(), "D");
174        assert_eq!(OperationType::Drop.as_str(), "DROP");
175        assert_eq!(OperationType::Rename.as_str(), "RENAME");
176        assert_eq!(OperationType::Invalidate.as_str(), "INVALIDATE");
177    }
178
179    #[test]
180    fn test_namespace_full_name() {
181        let ns = Namespace {
182            db: "mydb".to_string(),
183            coll: "users".to_string(),
184        };
185        assert_eq!(ns.full_name(), "mydb.users");
186        assert_eq!(ns.to_string(), "mydb.users");
187    }
188
189    #[test]
190    fn test_update_description_default() {
191        let ud = UpdateDescription::default();
192        assert!(ud.updated_fields.is_empty());
193        assert!(ud.removed_fields.is_empty());
194        assert!(ud.truncated_arrays.is_empty());
195        assert!(ud.disambiguated_paths.is_empty());
196    }
197
198    #[test]
199    fn test_change_event_serde_roundtrip() {
200        let event = MongoDbChangeEvent {
201            operation_type: OperationType::Insert,
202            namespace: Namespace {
203                db: "test".to_string(),
204                coll: "docs".to_string(),
205            },
206            document_key: r#"{"_id": "abc123"}"#.to_string(),
207            full_document: Some(r#"{"_id": "abc123", "name": "Alice"}"#.to_string()),
208            update_description: None,
209            cluster_time_secs: 1_700_000_000,
210            cluster_time_inc: 1,
211            resume_token: r#"{"_data": "token123"}"#.to_string(),
212            wall_time_ms: 1_700_000_000_000,
213        };
214
215        let json = serde_json::to_string(&event).unwrap();
216        let deserialized: MongoDbChangeEvent = serde_json::from_str(&json).unwrap();
217        assert_eq!(deserialized.operation_type, OperationType::Insert);
218        assert_eq!(deserialized.namespace.db, "test");
219    }
220}
laminar_connectors/mongodb/change_event.rs

laminar_connectors/mongodb/
change_event.rs
