MongoDB #

MongoDB is a document database that stores data in BSON (Binary JSON) format — every record is a document that can have a different structure without a rigid schema. In Rust, the official mongodb driver from MongoDB Inc. provides complete async access. The advantage of using Rust with MongoDB is seamless serde integration: Rust structs with #[derive(Serialize, Deserialize)] can be stored and read from collections directly without manual mapping. This article covers everything from basic connections to aggregation pipelines, Change Streams for real-time, and patterns commonly used in production.

Installation #

[dependencies]
mongodb = { version = "2", features = ["tokio-runtime"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
bson = { version = "2", features = ["chrono-0_4"] }
chrono = { version = "0.4", features = ["serde"] }

MongoDB Basic Concepts #

flowchart LR
    subgraph MongoDB
        DB["Database\n(contoh)"]
        DB --> C1["Collection\npengguna"]
        DB --> C2["Collection\nartikel"]
        C1 --> D1["Document\n{_id: ObjectId, nama: ...}"]
        C1 --> D2["Document\n{_id: ObjectId, nama: ...}"]
        C2 --> D3["Document\n{_id: ObjectId, judul: ...}"]
    end

SQL Concept	MongoDB Concept
Database	Database
Table	Collection
Row	Document
Column	Field
Primary key	`_id` (ObjectId by default)
JOIN	`$lookup` in aggregation
Index	Index
Transaction	Multi-document transaction (replica set)

Connections and Clients #

use mongodb::{Client, options::ClientOptions};

async fn buat_client(uri: &str) -> Result<Client, mongodb::error::Error> {
    let mut options = ClientOptions::parse(uri).await?;

    // Pool configuration
    options.max_pool_size = Some(20);
    options.min_pool_size = Some(2);
    options.connect_timeout = Some(std::time::Duration::from_secs(5));
    options.server_selection_timeout = Some(std::time::Duration::from_secs(5));

    // Application name — appears in MongoDB logs
    options.app_name = Some("aplikasi-rust".to_string());

    Client::with_options(options)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Format: mongodb://user:***@host:port/database
    // or MongoDB Atlas: mongodb+srv://user:***@cluster.mongodb.net/database
    let uri = std::env::var("MONGODB_URI")
        .unwrap_or_else(|_| "mongodb://localhost:27017".to_string());

    let client = buat_client(&uri).await?;

    // Verify the connection
    client
        .database("admin")
        .run_command(bson::doc! {"ping": 1}, None)
        .await?;
    println!("Connected to MongoDB!");

    // Access a database and collection
    let db = client.database("contoh");
    let koleksi = db.collection::<bson::Document>("pengguna");

    println!("Collection: {:?}", koleksi.name());
    Ok(())
}

Structs and Serde #

Rust structs with #[derive(Serialize, Deserialize)] can be stored in and read from MongoDB directly:

use bson::{oid::ObjectId, DateTime as BsonDateTime};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct Pengguna {
    // _id uses ObjectId — MongoDB's special field
    #[serde(rename = "_id", skip_serializing_if = "Option::is_none")]
    pub id: Option<ObjectId>,
    pub nama: String,
    pub email: String,
    #[serde(skip_serializing)]  // don't include in JSON responses
    pub password: String,
    pub peran: String,
    pub aktif: bool,
    pub tag: Vec<String>,
    pub dibuat_pada: DateTime<Utc>,
}

impl Pengguna {
    pub fn baru(nama: &str, email: &str, password_hash: &str) -> Self {
        Pengguna {
            id: None,  // MongoDB fills it in automatically
            nama: nama.to_string(),
            email: email.to_string(),
            password: password_hash.to_string(),
            peran: "user".to_string(),
            aktif: true,
            tag: Vec::new(),
            dibuat_pada: Utc::now(),
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct Artikel {
    #[serde(rename = "_id", skip_serializing_if = "Option::is_none")]
    pub id: Option<ObjectId>,
    pub pengguna_id: ObjectId,  // reference to a user
    pub judul: String,
    pub slug: String,
    pub konten: String,
    pub tag: Vec<String>,
    pub diterbitkan: bool,
    pub views: u64,
    pub dibuat_pada: DateTime<Utc>,
}

Typed Collections #

A Collection<T> parameterized with a struct type provides type-safe operations:

use mongodb::{Client, Collection};

fn koleksi_pengguna(client: &Client) -> Collection<Pengguna> {
    client.database("contoh").collection("pengguna")
}

fn koleksi_artikel(client: &Client) -> Collection<Artikel> {
    client.database("contoh").collection("artikel")
}

Create — Storing Documents #

use bson::oid::ObjectId;
use mongodb::Collection;

async fn buat_pengguna(
    col: &Collection<Pengguna>,
    pengguna: Pengguna,
) -> Result<ObjectId, mongodb::error::Error> {
    let hasil = col.insert_one(pengguna, None).await?;

    // inserted_id is bson::Bson — convert to ObjectId
    let id = hasil.inserted_id
        .as_object_id()
        .ok_or_else(|| mongodb::error::Error::custom("ID is not an ObjectId"))?;

    Ok(id)
}

// Insert many at once
async fn buat_banyak_pengguna(
    col: &Collection<Pengguna>,
    pengguna_list: Vec<Pengguna>,
) -> Result<Vec<ObjectId>, mongodb::error::Error> {
    let hasil = col.insert_many(pengguna_list, None).await?;

    let ids: Vec<ObjectId> = hasil.inserted_ids
        .values()
        .filter_map(|id| id.as_object_id())
        .collect();

    Ok(ids)
}

Read — Finding Documents #

use bson::{doc, oid::ObjectId};
use mongodb::{Collection, options::FindOptions};
use futures::TryStreamExt;

async fn cari_by_id(
    col: &Collection<Pengguna>,
    id: ObjectId,
) -> Result<Option<Pengguna>, mongodb::error::Error> {
    col.find_one(doc! {"_id": id}, None).await
}

async fn cari_by_email(
    col: &Collection<Pengguna>,
    email: &str,
) -> Result<Option<Pengguna>, mongodb::error::Error> {
    col.find_one(doc! {"email": email}, None).await
}

async fn cari_pengguna_aktif(
    col: &Collection<Pengguna>,
) -> Result<Vec<Pengguna>, mongodb::error::Error> {
    let opsi = FindOptions::builder()
        .sort(doc! {"dibuat_pada": -1})  // -1 = descending
        .limit(50)
        .build();

    let mut cursor = col.find(doc! {"aktif": true}, opsi).await?;
    let mut hasil = Vec::new();

    // A cursor is an async stream
    while let Some(pengguna) = cursor.try_next().await? {
        hasil.push(pengguna);
    }

    Ok(hasil)
}

// With pagination
async fn daftar_pengguna(
    col: &Collection<Pengguna>,
    halaman: u64,
    per_halaman: u64,
) -> Result<Vec<Pengguna>, mongodb::error::Error> {
    let opsi = FindOptions::builder()
        .sort(doc! {"_id": -1})
        .skip(Some((halaman.saturating_sub(1)) * per_halaman))
        .limit(Some(per_halaman as i64))
        .build();

    col.find(doc! {}, opsi)
        .await?
        .try_collect()
        .await
}

// Filter with many conditions
async fn cari_dengan_filter(
    col: &Collection<Pengguna>,
    aktif: bool,
    peran: &str,
    tag: &str,
) -> Result<Vec<Pengguna>, mongodb::error::Error> {
    let filter = doc! {
        "aktif": aktif,
        "peran": peran,
        "tag": tag,          // array field: checks whether tag is in the array
    };

    col.find(filter, None)
        .await?
        .try_collect()
        .await
}

Update — Updating Documents #

use bson::{doc, oid::ObjectId};
use mongodb::{Collection, options::UpdateOptions};

async fn perbarui_nama(
    col: &Collection<Pengguna>,
    id: ObjectId,
    nama_baru: &str,
) -> Result<bool, mongodb::error::Error> {
    let hasil = col.update_one(
        doc! {"_id": id},
        doc! {
            "$set": {
                "nama": nama_baru,
                "diperbarui": chrono::Utc::now()
            }
        },
        None,
    )
    .await?;

    Ok(hasil.modified_count > 0)
}

// Push an element to an array field
async fn tambah_tag(
    col: &Collection<Pengguna>,
    id: ObjectId,
    tag: &str,
) -> Result<bool, mongodb::error::Error> {
    let hasil = col.update_one(
        doc! {"_id": id},
        doc! {
            "$addToSet": {"tag": tag}  // addToSet: adds only if not already present
        },
        None,
    )
    .await?;

    Ok(hasil.modified_count > 0)
}

// Increment a numeric field
async fn increment_views(
    col: &Collection<Artikel>,
    id: ObjectId,
) -> Result<(), mongodb::error::Error> {
    col.update_one(
        doc! {"_id": id},
        doc! {"$inc": {"views": 1}},
        None,
    )
    .await?;
    Ok(())
}

// Upsert — update if present, insert if not
async fn upsert_konfigurasi(
    col: &Collection<bson::Document>,
    kunci: &str,
    nilai: &str,
) -> Result<(), mongodb::error::Error> {
    let opsi = UpdateOptions::builder().upsert(true).build();
    col.update_one(
        doc! {"kunci": kunci},
        doc! {
            "$set": {"nilai": nilai},
            "$setOnInsert": {"dibuat_pada": chrono::Utc::now()}
        },
        opsi,
    )
    .await?;
    Ok(())
}

Delete — Removing Documents #

use bson::{doc, oid::ObjectId};
use mongodb::Collection;

async fn hapus_pengguna(
    col: &Collection<Pengguna>,
    id: ObjectId,
) -> Result<bool, mongodb::error::Error> {
    let hasil = col.delete_one(doc! {"_id": id}, None).await?;
    Ok(hasil.deleted_count > 0)
}

// Delete many documents
async fn hapus_pengguna_tidak_aktif(
    col: &Collection<Pengguna>,
) -> Result<u64, mongodb::error::Error> {
    let hasil = col.delete_many(doc! {"aktif": false}, None).await?;
    Ok(hasil.deleted_count)
}

Aggregation Pipelines #

The aggregation pipeline is MongoDB’s most powerful way to do complex queries — the equivalent of JOIN, GROUP BY, and subqueries in SQL:

use bson::doc;
use futures::TryStreamExt;
use mongodb::{Collection, options::AggregateOptions};

async fn statistik_pengguna(
    col: &Collection<Pengguna>,
) -> Result<Vec<bson::Document>, mongodb::error::Error> {
    let pipeline = vec![
        // $match — filter documents
        doc! {"$match": {"aktif": true}},

        // $group — group by role, count
        doc! {
            "$group": {
                "_id": "$peran",
                "jumlah": {"$sum": 1},
                "rata_tag": {"$avg": {"$size": "$tag"}}
            }
        },

        // $sort — sort by count
        doc! {"$sort": {"jumlah": -1}},

        // $project — select fields to display
        doc! {
            "$project": {
                "peran": "$_id",
                "jumlah": 1,
                "rata_tag": {"$round": ["$rata_tag", 1]},
                "_id": 0
            }
        }
    ];

    col.aggregate(pipeline, None)
        .await?
        .try_collect()
        .await
}

// Aggregation with $lookup (JOIN between collections)
async fn artikel_dengan_penulis(
    col_artikel: &Collection<Artikel>,
) -> Result<Vec<bson::Document>, mongodb::error::Error> {
    let pipeline = vec![
        doc! {"$match": {"diterbitkan": true}},
        doc! {
            "$lookup": {
                "from": "pengguna",          // target collection name
                "localField": "pengguna_id", // field in artikel
                "foreignField": "_id",       // field in pengguna
                "as": "penulis"              // JOIN result field name
            }
        },
        doc! {
            "$unwind": "$penulis"  // turn the penulis array into an object
        },
        doc! {
            "$project": {
                "judul": 1,
                "tag": 1,
                "views": 1,
                "dibuat_pada": 1,
                "nama_penulis": "$penulis.nama",
                "email_penulis": "$penulis.email",
            }
        },
        doc! {"$sort": {"dibuat_pada": -1}},
        doc! {"$limit": 20}
    ];

    col_artikel.aggregate(pipeline, None)
        .await?
        .try_collect()
        .await
}

// Aggregation with $facet — several pipelines at once
async fn dashboard_artikel(
    col: &Collection<Artikel>,
) -> Result<bson::Document, mongodb::error::Error> {
    let pipeline = vec![
        doc! {
            "$facet": {
                "total": [
                    {"$count": "jumlah"}
                ],
                "per_status": [
                    {"$group": {"_id": "$diterbitkan", "jumlah": {"$sum": 1}}}
                ],
                "top_viewed": [
                    {"$sort": {"views": -1}},
                    {"$limit": 5},
                    {"$project": {"judul": 1, "views": 1}}
                ]
            }
        }
    ];

    let mut cursor = col.aggregate(pipeline, None).await?;
    cursor.try_next().await?.ok_or_else(|| {
        mongodb::error::Error::custom("Aggregation produced no documents")
    })
}

Indexes #

use mongodb::{Collection, IndexModel, options::IndexOptions};
use bson::doc;

async fn buat_index(
    col: &Collection<Pengguna>,
) -> Result<(), mongodb::error::Error> {
    // Unique index on email
    let opsi_email = IndexOptions::builder().unique(true).build();
    col.create_index(
        IndexModel::builder()
            .keys(doc! {"email": 1})
            .options(opsi_email)
            .build(),
        None,
    )
    .await?;

    // Compound index for frequent queries
    col.create_index(
        IndexModel::builder()
            .keys(doc! {"aktif": 1, "peran": 1, "dibuat_pada": -1})
            .build(),
        None,
    )
    .await?;

    // Text index for full-text search
    let col_artikel: Collection<Artikel> = col.clone_with_type();
    col_artikel.create_index(
        IndexModel::builder()
            .keys(doc! {"judul": "text", "konten": "text"})
            .build(),
        None,
    )
    .await?;

    println!("Indexes created successfully");
    Ok(())
}

// Full-text search using $text
async fn cari_artikel_fts(
    col: &Collection<Artikel>,
    query: &str,
) -> Result<Vec<Artikel>, mongodb::error::Error> {
    col.find(
        doc! {
            "$text": {"$search": query},
            "diterbitkan": true
        },
        mongodb::options::FindOptions::builder()
            .sort(doc! {"score": {"$meta": "textScore"}})
            .projection(doc! {"score": {"$meta": "textScore"}})
            .build(),
    )
    .await?
    .try_collect()
    .await
}

Transactions #

MongoDB transactions require a replica set or sharded cluster (not available on standalone):

use mongodb::{Client, ClientSession};

async fn transfer_dengan_transaksi(
    client: &Client,
    dari_id: bson::oid::ObjectId,
    ke_id: bson::oid::ObjectId,
    jumlah: f64,
) -> Result<(), mongodb::error::Error> {
    let mut session = client.start_session(None).await?;

    // Start the transaction
    session.start_transaction(None).await?;

    let db = client.database("contoh");
    let akun = db.collection::<bson::Document>("akun");

    // Deduct the sender's balance
    let hasil = akun.update_one_with_session(
        doc! {"_id": dari_id, "saldo": {"$gte": jumlah}},
        doc! {"$inc": {"saldo": -jumlah}},
        None,
        &mut session,
    )
    .await?;

    if hasil.modified_count == 0 {
        session.abort_transaction().await?;
        return Err(mongodb::error::Error::custom("Insufficient balance or account not found"));
    }

    // Add to the recipient's balance
    akun.update_one_with_session(
        doc! {"_id": ke_id},
        doc! {"$inc": {"saldo": jumlah}},
        None,
        &mut session,
    )
    .await?;

    session.commit_transaction().await?;
    println!("Transfer succeeded");
    Ok(())
}

Change Streams — Real-Time Data #

Change Streams let you listen to changes on a collection in real time — like LISTEN/NOTIFY in PostgreSQL:

use bson::doc;
use futures::TryStreamExt;
use mongodb::{Collection, options::ChangeStreamOptions};

async fn pantau_pengguna_baru(
    col: &Collection<Pengguna>,
) -> Result<(), mongodb::error::Error> {
    // Filter only insert events
    let pipeline = vec![
        doc! {"$match": {"operationType": "insert"}}
    ];

    let opsi = ChangeStreamOptions::builder()
        .full_document(Some(mongodb::options::FullDocumentType::UpdateLookup))
        .build();

    let mut change_stream = col.watch(pipeline, opsi).await?;

    println!("Watching for new users...");

    while let Some(event) = change_stream.try_next().await? {
        println!("Event: {:?}", event.operation_type);
        if let Some(dokumen) = event.full_document {
            println!("New user: {} ({})", dokumen.nama, dokumen.email);
        }
    }

    Ok(())
}

Summary #

Collection<T> for type-safe operations — parameterize the collection with a Rust struct with #[derive(Serialize, Deserialize)] for CRUD without manual mapping.
_id uses Option<ObjectId> — use skip_serializing_if = "Option::is_none" so inserts don’t send _id: null to MongoDB.
The doc! {} macro — the idiomatic way to build filters, updates, and BSON pipelines. Supports all MongoDB operators ($match, $set, $inc, $addToSet, etc.).
Cursors are async streams — use try_next().await? in a loop or .try_collect().await? to gather all documents.
Aggregation pipelines replace JOINs — $lookup for relations between collections, $group for aggregation, $facet for multiple pipelines at once.
$addToSet vs $push — $addToSet only adds if not already in the array (set semantics), $push always adds (duplicates possible).
Indexes are mandatory for frequent queries — MongoDB doesn’t have a query planner as good as PostgreSQL’s; without an index, every query does a full collection scan.
Transactions require a replica set — standalone MongoDB doesn’t support multi-document transactions. Use Docker with a replica set or MongoDB Atlas for testing transactions.
Change Streams for real-time — watch collection changes without polling, similar to PostgreSQL’s LISTEN/NOTIFY. Requires a replica set or Atlas.

← Previous: PostgreSQL Next: Elasticsearch →