laminar_connectors/mongodb/change_event/
mod.rs

1//! `MongoDB` change-event classification and canonical Extended JSON rendering.
2//!
3//! Change events stay as raw BSON from the driver until Arrow assembly decodes them once.
4//! Operation names are the server's own `operationType` strings, so history records never
5//! use the engine's `_op` changelog vocabulary.
6
7use mongodb::bson::{Bson, RawBsonRef, RawDocument};
8
9use crate::error::ConnectorError;
10
11/// Semantic class of a change-stream `operationType`.
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum ChangeOperation {
14    /// A document was inserted.
15    Insert,
16    /// A document was modified; the event carries a delta and optionally a post-image.
17    Update,
18    /// A document was replaced in full.
19    Replace,
20    /// A document was deleted; only its key is present.
21    Delete,
22    /// The change stream was invalidated; only `startAfter` can continue it.
23    Invalidate,
24    /// The watched collection was dropped.
25    Drop,
26    /// The watched collection was renamed.
27    Rename,
28    /// The database containing the collection was dropped.
29    DropDatabase,
30    /// Expanded collection metadata event that leaves documents unchanged.
31    Metadata,
32    /// An operation this connector does not recognize.
33    Unknown,
34}
35
36impl ChangeOperation {
37    /// Classify a server `operationType`.
38    #[must_use]
39    pub fn classify(operation: &str) -> Self {
40        match operation {
41            "insert" => Self::Insert,
42            "update" => Self::Update,
43            "replace" => Self::Replace,
44            "delete" => Self::Delete,
45            "invalidate" => Self::Invalidate,
46            "drop" => Self::Drop,
47            "rename" => Self::Rename,
48            "dropDatabase" => Self::DropDatabase,
49            "create"
50            | "createIndexes"
51            | "dropIndexes"
52            | "modify"
53            | "shardCollection"
54            | "refineCollectionShardKey"
55            | "reshardCollection" => Self::Metadata,
56            _ => Self::Unknown,
57        }
58    }
59}
60
61/// Canonical Extended JSON v2 text for one BSON value. Canonical form keeps every BSON type
62/// distinct (`Int32` versus `Int64`, `ObjectId` versus string, `Decimal128`, dates).
63///
64/// # Errors
65/// Returns an error when the raw value is malformed.
66pub fn canonical_extjson(value: RawBsonRef<'_>) -> Result<String, ConnectorError> {
67    let value = Bson::try_from(value.to_raw_bson()).map_err(|error| {
68        ConnectorError::ReadError(format!("malformed MongoDB BSON value: {error}"))
69    })?;
70    Ok(value.into_canonical_extjson().to_string())
71}
72
73/// Canonical Extended JSON v2 text for one raw document.
74///
75/// # Errors
76/// Returns an error when the document is malformed.
77pub fn canonical_document_extjson(document: &RawDocument) -> Result<String, ConnectorError> {
78    canonical_extjson(RawBsonRef::Document(document))
79}
80
81#[cfg(test)]
82mod tests;
laminar_connectors/mongodb/change_event/mod.rs

laminar_connectors/mongodb/change_event/
mod.rs
