DocumentDB support

lambda-events/Cargo.toml

@@ -76,6 +76,7 @@ default = [
   "sns",
   "sqs",
   "streams",
+  "documentdb",
 ]
 
 activemq = []
@@ -117,3 +118,4 @@ ses = ["chrono"]
 sns = ["chrono", "serde_with"]
 sqs = ["serde_with"]
 streams = []
+documentdb = []

lambda-events/src/event/documentdb/events/commom_types.rs

@@ -0,0 +1,44 @@
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+pub type AnyDocument = HashMap<String, Value>;
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DatabaseCollection {
+    db: String,
+    #[serde(default)]
+    coll: Option<String>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+pub struct DocumentId {
+    #[serde(rename = "_data")]
+    pub data: String,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+pub struct DocumentKeyIdOid {
+    #[serde(rename = "$oid")]
+    pub oid: String,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+pub struct DocumentKeyId {
+    #[serde(rename = "_id")]
+    pub id: DocumentKeyIdOid,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+pub struct InnerTimestamp {
+    t: usize,
+    i: usize,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+pub struct Timestamp {
+    #[serde(rename = "$timestamp")]
+    pub timestamp: InnerTimestamp,
+}

lambda-events/src/event/documentdb/events/delete_event.rs

@@ -0,0 +1,20 @@
+use serde::{Deserialize, Serialize};
+
+use super::commom_types::{AnyDocument, DatabaseCollection, DocumentId, DocumentKeyId, Timestamp};
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ChangeDeleteEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    #[serde(default)]
+    cluster_time: Option<Timestamp>,
+    document_key: DocumentKeyId,
+    #[serde(default)]
+    #[serde(rename = "lsid")]
+    ls_id: Option<AnyDocument>,
+    ns: DatabaseCollection,
+    // operation_type: String,
+    #[serde(default)]
+    txn_number: Option<String>,
+}

lambda-events/src/event/documentdb/events/drop_database_event.rs

@@ -0,0 +1,18 @@
+use serde::{Deserialize, Serialize};
+
+use super::commom_types::{DocumentId, Timestamp, DatabaseCollection, AnyDocument};
+
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ChangeDropDatabaseEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    cluster_time: Timestamp,
+    #[serde(rename = "lsid")]
+    ls_id: Option<AnyDocument>,
+    ns: DatabaseCollection,
+    // operation_type: String,
+    #[serde(default)]
+    txn_number: Option<String>,
+}

lambda-events/src/event/documentdb/events/drop_event.rs

@@ -0,0 +1,17 @@
+use super::commom_types::{AnyDocument, DatabaseCollection, DocumentId, Timestamp};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ChangeDropEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    cluster_time: Timestamp,
+    #[serde(default)]
+    #[serde(rename = "lsid")]
+    ls_id: Option<AnyDocument>,
+    ns: DatabaseCollection,
+    // operation_type: String,
+    #[serde(default)]
+    txn_number: Option<String>,
+}

lambda-events/src/event/documentdb/events/insert_event.rs

@@ -0,0 +1,21 @@
+use serde::{Deserialize, Serialize};
+
+use super::commom_types::{AnyDocument, DatabaseCollection, DocumentId, DocumentKeyId, Timestamp};
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+
+pub struct ChangeInsertEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    #[serde(default)]
+    cluster_time: Option<Timestamp>,
+    document_key: DocumentKeyId,
+    #[serde(default)]
+    #[serde(rename = "lsid")]
+    ls_id: Option<String>,
+    ns: DatabaseCollection,
+    //operation_type: String,
+    #[serde(default)]
+    txn_number: Option<AnyDocument>,
+}

lambda-events/src/event/documentdb/events/invalidate_event.rs

@@ -0,0 +1,13 @@
+use serde::{Deserialize, Serialize};
+
+use super::commom_types::{DocumentId, Timestamp};
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ChangeInvalidateEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    #[serde(default)]
+    cluster_time: Option<Timestamp>,
+    // operation_type: String,
+}

lambda-events/src/event/documentdb/events/mod.rs

@@ -0,0 +1,9 @@
+pub mod commom_types;
+pub mod delete_event;
+pub mod drop_event;
+pub mod drop_database_event;
+pub mod insert_event;
+pub mod invalidate_event;
+pub mod rename_event;
+pub mod replace_event;
+pub mod update_event;

lambda-events/src/event/documentdb/events/rename_event.rs

@@ -0,0 +1,21 @@
+use serde::{Deserialize, Serialize};
+
+use super::commom_types::{AnyDocument, DatabaseCollection, DocumentId, Timestamp};
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ChangeRenameEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    #[serde(default)]
+    cluster_time: Option<Timestamp>,
+
+    #[serde(default)]
+    #[serde(rename = "lsid")]
+    ls_id: Option<AnyDocument>,
+    ns: DatabaseCollection,
+    //operation_type: String,
+    #[serde(default)]
+    txn_number: Option<String>,
+    to: DatabaseCollection,
+}

lambda-events/src/event/documentdb/events/replace_event.rs

@@ -0,0 +1,20 @@
+use serde::{Deserialize, Serialize};
+
+use super::commom_types::{AnyDocument, DatabaseCollection, DocumentId, DocumentKeyId, Timestamp};
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ChangeReplaceEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    #[serde(default)]
+    cluster_time: Option<Timestamp>,
+    document_key: DocumentKeyId,
+    #[serde(default)]
+    #[serde(rename = "lsid")]
+    ls_id: Option<String>,
+    ns: DatabaseCollection,
+    // operation_type: String,
+    #[serde(default)]
+    txn_number: Option<AnyDocument>,
+}

lambda-events/src/event/documentdb/events/update_event.rs

@@ -0,0 +1,19 @@
+use serde::{Deserialize, Serialize};
+
+use super::commom_types::{AnyDocument, DocumentId, DocumentKeyId, DatabaseCollection, Timestamp};
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ChangeUpdateEvent {
+    #[serde(rename = "_id")]
+    id: DocumentId,
+    #[serde(default)]
+    cluster_time: Option<Timestamp>,
+    document_key: DocumentKeyId,
+    #[serde(rename = "lsid")]
+    ls_id: Option<String>,
+    ns: DatabaseCollection,
+    // operation_type: String,
+    #[serde(default)]
+    txn_number: Option<AnyDocument>,
+}

lambda-events/src/event/documentdb/mod.rs

@@ -0,0 +1,97 @@
+pub mod events;
+
+use self::events::{
+    delete_event::ChangeDeleteEvent, 
+    drop_event::ChangeDropEvent, 
+    insert_event::ChangeInsertEvent,
+    invalidate_event::ChangeInvalidateEvent,
+    replace_event::ChangeReplaceEvent, 
+    update_event::ChangeUpdateEvent,
+    rename_event::ChangeRenameEvent,
+    drop_database_event::ChangeDropDatabaseEvent,
+};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "operationType", rename_all = "camelCase")]
+pub enum ChangeEvent {
+    Insert(ChangeInsertEvent),
+    Delete(ChangeDeleteEvent),
+    Drop(ChangeDropEvent),
+    DropDatabase(ChangeDropDatabaseEvent),
+    Invalidate(ChangeInvalidateEvent),
+    Replace(ChangeReplaceEvent),
+    Update(ChangeUpdateEvent),
+    Rename(ChangeRenameEvent),
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct DocumentDbInnerEvent {
+    pub event: ChangeEvent,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DocumentDbEvent {
+    #[serde(default)]
+    pub event_source_arn: Option<String>,
+    pub events: Vec<DocumentDbInnerEvent>,
+    #[serde(default)]
+    pub event_source: Option<String>,
+}
+
+#[cfg(test)]
+#[cfg(feature = "documentdb")]
+mod test {
+    use super::*;
+
+    pub type Event = DocumentDbEvent;
+
+    fn test_example(data: &[u8]) {
+        let parsed: Event = serde_json::from_slice(data).unwrap();
+        let output: String = serde_json::to_string(&parsed).unwrap();
+        let reparsed: Event = serde_json::from_slice(output.as_bytes()).unwrap();
+
+        assert_eq!(parsed, reparsed);
+    }
+
+    #[test]
+    fn example_documentdb_insert_event() {
+        test_example(include_bytes!( "../../fixtures/example-documentdb-insert-event.json"));
+    }
+
+    #[test]
+    fn example_documentdb_delete_event() {
+        test_example(include_bytes!("../../fixtures/example-documentdb-delete-event.json"));
+    }
+
+    #[test]
+    fn example_documentdb_drop_event() {
+        test_example(include_bytes!("../../fixtures/example-documentdb-drop-event.json"));
+    }
+
+    #[test]
+    fn example_documentdb_replace_event() {
+        test_example(include_bytes!("../../fixtures/example-documentdb-replace-event.json"));
+    }
+
+    #[test]
+    fn example_documentdb_update_event() {
+        test_example(include_bytes!("../../fixtures/example-documentdb-update-event.json"));
+    }
+
+    #[test]
+    fn example_documentdb_rename_event() {
+        test_example(include_bytes!("../../fixtures/example-documentdb-rename-event.json"));
+    }
+
+    #[test]
+    fn example_documentdb_invalidate_event() {
+        test_example(include_bytes!("../../fixtures/example-documentdb-invalidate-event.json"));
+    }
+
+    #[test]
+    fn example_documentdb_drop_database_event(){
+        test_example(include_bytes!("../../fixtures/example-documentdb-drop-database-event.json"));
+    }
+}