[Feature] Add MongoDB CDC source

Cargo.toml

@@ -8,7 +8,7 @@ description = "A Change Data Capture (CDC) pipeline in Rust"
 repository = "https://github.com/manfredcml/cdcflow"
 homepage = "https://github.com/manfredcml/cdcflow"
 readme = "README.md"
-keywords = ["cdc", "change-data-capture", "replication", "postgres", "mysql"]
+keywords = ["cdc", "change-data-capture", "replication", "postgres", "mongodb"]
 categories = ["database", "command-line-utilities"]
 exclude = ["benchmark/"]
 
@@ -33,6 +33,7 @@ arrow-schema = "57"
 parquet = { version = "57", default-features = false, features = ["arrow"] }
 async-trait = "0.1"
 rdkafka = { version = "0.39.0", features = ["cmake-build", "tokio"] }
+mongodb = "3"
 ordered-float = { version = "5.1.0", features = ["serde"] }
 chrono = "0.4"
 axum = { version = "0.8", features = ["json"] }
@@ -50,4 +51,4 @@ approx_constant = "allow"
 tower = "0.5"
 tempfile = "3"
 testcontainers = "0.27.1"
-testcontainers-modules = { version = "0.15.0", features = ["postgres", "mysql", "kafka"] }
+testcontainers-modules = { version = "0.15.0", features = ["postgres", "mysql", "kafka", "mongo"] }

README.md

@@ -9,7 +9,7 @@ destinations.
 
 ## Features
 
-- **Database sources**: PostgreSQL and MySQL
+- **Database sources**: PostgreSQL, MySQL, and MongoDB
 - **Sink destinations**: Stdout, Kafka, PostgreSQL, Apache Iceberg
 - **Offset stores**: SQLite, Memoryfor resumable streaming after restarts
 - **Sync modes**: CDC (append-only changelog) and Replication (live replica of source tables)
@@ -20,7 +20,7 @@ destinations.
 ```
 ┌────────────┐     ┌────────────┐     ┌──────────────┐
 │   Source   │────▶│  Pipeline  │────▶│     Sink     │
-│ PG / MySQL │     │ (batch +   │     │ Kafka / PG / │
+│PG/MySQL/Mon│     │ (batch +   │     │ Kafka / PG / │
 │            │     │  flush)    │     │ Iceberg / Out│
 └────────────┘     └─────┬──────┘     └──────────────┘
                          │
@@ -37,12 +37,32 @@ checkpoints progress in the offset store. On restart, the pipeline resumes from
 - **Platform**: Linux and macOS only
 - **CMake**: Required for building the bundled librdkafka (Kafka dependency)
 
+### Source requirements
+
+| Source | Requirements |
+| --- | --- |
+| PostgreSQL | Logical replication (`wal_level=logical`), a replication slot and publication |
+| MySQL | Row-based binlog (`binlog_format=ROW`, `binlog_row_image=FULL`) |
+| MongoDB | Replica set (or sharded cluster) — change streams do not work on standalone `mongod` |
+
+**MongoDB pre-images (optional).** Enabling `changeStreamPreAndPostImages` on a
+collection makes the full "before" row available on updates and deletes:
+
+```js
+db.runCommand({ collMod: "users", changeStreamPreAndPostImages: { enabled: true } });
+```
+
+Without it, update and delete events fall back to the change event's
+`documentKey`, so the `_id` primary key is always captured — replication mode
+still applies updates and deletes correctly, but CDC-mode `old` values contain
+only `_id`.
+
 ## Quick Start - Standalone Mode
 
 The standalone mode runs a single pipeline without the admin server. This is ideal for local testing and development.
 
 ```bash
-# Start local infrastructure (PostgreSQL, MySQL, Kafka, MinIO, Iceberg, Trino)
+# Start local infrastructure (PostgreSQL, MySQL, MongoDB, Kafka, MinIO, Iceberg, Trino)
 # You can optionally start just the components you need for testing a specific source/sink combo
 cd example && docker compose up -d && cd ..
 

example/configs/mongo-to-iceberg-replication.json

@@ -0,0 +1,31 @@
+{
+  "mode": "replication",
+  "source": {
+    "type": "mongodb",
+    "connection_string": "mongodb://localhost:27017/?replicaSet=rs0&directConnection=true",
+    "database": "demo",
+    "collections": ["users", "orders"]
+  },
+  "sink": {
+    "type": "iceberg",
+    "catalog": {
+      "type": "rest",
+      "uri": "http://localhost:8181",
+      "warehouse": "s3://warehouse/iceberg",
+      "properties": {
+        "s3.endpoint": "http://localhost:9000",
+        "s3.access-key-id": "minioadmin",
+        "s3.secret-access-key": "minioadmin",
+        "s3.path-style-access": "true",
+        "s3.region": "us-east-1"
+      }
+    },
+    "namespace": ["demo"],
+    "table_prefix": ""
+  },
+  "offset": {
+    "type": "sqlite",
+    "path": "/tmp/cdc-offsets.db",
+    "key": "mongo-iceberg-replication"
+  }
+}

example/configs/mongo-to-iceberg.json

@@ -0,0 +1,30 @@
+{
+  "source": {
+    "type": "mongodb",
+    "connection_string": "mongodb://localhost:27017/?replicaSet=rs0&directConnection=true",
+    "database": "demo",
+    "collections": ["users", "orders"]
+  },
+  "sink": {
+    "type": "iceberg",
+    "catalog": {
+      "type": "rest",
+      "uri": "http://localhost:8181",
+      "warehouse": "s3://warehouse/iceberg",
+      "properties": {
+        "s3.endpoint": "http://localhost:9000",
+        "s3.access-key-id": "minioadmin",
+        "s3.secret-access-key": "minioadmin",
+        "s3.path-style-access": "true",
+        "s3.region": "us-east-1"
+      }
+    },
+    "namespace": ["demo"],
+    "table_prefix": ""
+  },
+  "offset": {
+    "type": "sqlite",
+    "path": "/tmp/cdc-offsets.db",
+    "key": "mongo-iceberg"
+  }
+}

example/configs/mongo-to-kafka.json

@@ -0,0 +1,18 @@
+{
+  "source": {
+    "type": "mongodb",
+    "connection_string": "mongodb://localhost:27017/?replicaSet=rs0&directConnection=true",
+    "database": "demo",
+    "collections": ["users", "orders"]
+  },
+  "sink": {
+    "type": "kafka",
+    "brokers": "localhost:9092",
+    "topic_prefix": "cdc"
+  },
+  "offset": {
+    "type": "sqlite",
+    "path": "/tmp/cdc-offsets.db",
+    "key": "mongo-kafka"
+  }
+}

example/configs/mongo-to-pg-replication.json

@@ -0,0 +1,23 @@
+{
+  "mode": "replication",
+  "source": {
+    "type": "mongodb",
+    "connection_string": "mongodb://localhost:27017/?replicaSet=rs0&directConnection=true",
+    "database": "demo",
+    "collections": ["users", "orders"]
+  },
+  "sink": {
+    "type": "postgres",
+    "host": "localhost",
+    "port": 5433,
+    "user": "cdc_user",
+    "password": "cdc_password",
+    "database": "demo_dest",
+    "schema": "public"
+  },
+  "offset": {
+    "type": "sqlite",
+    "path": "/tmp/cdc-offsets.db",
+    "key": "mongo-pg-replication"
+  }
+}

example/configs/mongo-to-pg.json

@@ -0,0 +1,22 @@
+{
+  "source": {
+    "type": "mongodb",
+    "connection_string": "mongodb://localhost:27017/?replicaSet=rs0&directConnection=true",
+    "database": "demo",
+    "collections": ["users", "orders"]
+  },
+  "sink": {
+    "type": "postgres",
+    "host": "localhost",
+    "port": 5433,
+    "user": "cdc_user",
+    "password": "cdc_password",
+    "database": "demo_dest",
+    "schema": "public"
+  },
+  "offset": {
+    "type": "sqlite",
+    "path": "/tmp/cdc-offsets.db",
+    "key": "mongo-pg"
+  }
+}

example/configs/mongo-to-stdout.json

@@ -0,0 +1,16 @@
+{
+  "source": {
+    "type": "mongodb",
+    "connection_string": "mongodb://localhost:27017/?replicaSet=rs0&directConnection=true",
+    "database": "demo",
+    "collections": ["users", "orders"]
+  },
+  "sink": {
+    "type": "stdout"
+  },
+  "offset": {
+    "type": "sqlite",
+    "path": "/tmp/cdc-offsets.db",
+    "key": "mongo-stdout"
+  }
+}

example/docker-compose.yml

@@ -50,6 +50,28 @@ services:
       timeout: 5s
       retries: 10
 
+  mongodb:
+    image: mongo:8
+    container_name: cdc-mongodb
+    ports:
+      - "27017:27017"
+    command: ["mongod", "--replSet", "rs0", "--bind_ip_all"]
+    healthcheck:
+      test: ["CMD", "mongosh", "--eval", "db.adminCommand('ping')"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+  mongodb-init:
+    image: mongo:8
+    container_name: cdc-mongodb-init
+    depends_on:
+      mongodb:
+        condition: service_healthy
+    volumes:
+      - ./init/mongodb/init.js:/init.js
+    entrypoint: ["mongosh", "mongodb://mongodb:27017/?directConnection=true", "--file", "/init.js"]
+
   # ──────────────────────────────────────────────
   # Infrastructure
   # ──────────────────────────────────────────────

example/init/mongodb/init.js

@@ -0,0 +1,56 @@
+// Initialize replica set (required for change streams).
+// Re-running the script against an already-initialized set is not an error.
+try {
+  rs.initiate({ _id: "rs0", members: [{ _id: 0, host: "localhost:27017" }] });
+} catch (e) {
+  print("replSetInitiate: " + e);
+}
+
+// Wait for primary election
+let attempts = 0;
+while (!db.hello().isWritablePrimary && attempts < 30) {
+  sleep(1000);
+  attempts++;
+}
+
+if (!db.hello().isWritablePrimary) {
+  print("ERROR: Failed to elect primary after 30 seconds");
+  quit(1);
+}
+
+print("Replica set initialized, primary elected");
+
+// Create demo database and collections.
+// NOTE: `const db = db.getSiblingDB(...)` is a self-referencing declaration and
+// throws a ReferenceError — the handle must use a different name.
+const demoDb = db.getSiblingDB("demo");
+
+// Enable pre/post images for change streams (MongoDB 6.0+).
+// Optional: without them, update/delete events fall back to `documentKey`, so
+// the `_id` primary key is still captured but the full "before" row is not.
+demoDb.createCollection("users", {
+  changeStreamPreAndPostImages: { enabled: true },
+});
+demoDb.createCollection("orders", {
+  changeStreamPreAndPostImages: { enabled: true },
+});
+
+// Insert sample data
+demoDb.users.insertMany([
+  { _id: ObjectId(), name: "Alice", email: "alice@example.com", age: 30, active: true },
+  { _id: ObjectId(), name: "Bob", email: "bob@example.com", age: 25, active: true },
+  { _id: ObjectId(), name: "Charlie", email: "charlie@example.com", age: 35, active: false },
+]);
+
+demoDb.orders.insertMany([
+  { _id: ObjectId(), user: "Alice", product: "Widget", quantity: 2, price: 19.99 },
+  { _id: ObjectId(), user: "Bob", product: "Gadget", quantity: 1, price: 49.99 },
+]);
+
+print(
+  "Demo data inserted: " +
+    demoDb.users.countDocuments() +
+    " users, " +
+    demoDb.orders.countDocuments() +
+    " orders",
+);

src/config/mod.rs

@@ -1,5 +1,6 @@
 mod iceberg_sink;
 mod kafka_sink;
+mod mongodb_source;
 mod mysql_source;
 mod offset;
 mod postgres_sink;
@@ -9,6 +10,7 @@ mod source;
 
 pub use self::iceberg_sink::{IcebergCatalogConfig, IcebergSinkConfig, RestCatalogConfig};
 pub use self::kafka_sink::KafkaSinkConfig;
+pub use self::mongodb_source::MongodbSourceConfig;
 pub use self::mysql_source::MySqlSourceConfig;
 pub use self::offset::OffsetConfig;
 pub use self::postgres_sink::PostgresSinkConfig;
@@ -37,6 +39,8 @@ pub enum SourceConnectionConfig {
     Postgres { url: String },
     #[serde(rename = "mysql")]
     Mysql { url: String },
+    #[serde(rename = "mongodb")]
+    Mongodb { url: String },
 }
 
 /// Top-level configuration for the CDC agent.
@@ -53,6 +57,28 @@ pub struct Config {
 mod tests {
     use super::*;
 
+    /// Every shipped example config must deserialize, so a config that drifts
+    /// from the schema (wrong field type, renamed key) fails the build rather
+    /// than only failing when a user runs it.
+    #[test]
+    fn test_example_configs_deserialize() {
+        let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("example/configs");
+        let mut checked = 0;
+
+        for entry in std::fs::read_dir(&dir).expect("example/configs must exist") {
+            let path = entry.expect("readable dir entry").path();
+            if path.extension().and_then(|e| e.to_str()) != Some("json") {
+                continue;
+            }
+            let contents = std::fs::read_to_string(&path).expect("readable config");
+            serde_json::from_str::<Config>(&contents)
+                .unwrap_or_else(|e| panic!("{} failed to parse: {e}", path.display()));
+            checked += 1;
+        }
+
+        assert!(checked > 0, "no example configs found in {}", dir.display());
+    }
+
     #[test]
     fn test_sink_mode_default_is_cdc() {
         let json = r#"{