Change Streams Overview

Notifikasi “pesanan baru masuk” di aplikasi toko biasanya lahir dari salah satu dari dua pola mahal: frontend mem-polling REST endpoint setiap beberapa detik, atau job terjadwal membaca ulang seluruh tabel. Keduanya membakar resource untuk menjawab pertanyaan yang hampir selalu berjawaban “tidak ada perubahan”. Change streams membalik modelnya: aplikasi membuka satu cursor ke MongoDB, lalu database mengirim change event begitu data berubah — tanpa polling, tanpa jeda jadwal. Section ini membahas fitur itu sampai ke arsitektur event-driven; chapter pembuka menjelaskan apa yang sebenarnya terjadi di baliknya.

Stream Perubahan di Atas Oplog #

Change stream adalah cursor agregasi khusus yang mengikuti aliran operasi tulis pada database. Sumber kebenarannya adalah oplog — log operasi replica set yang sudah kita bedah di Section Replication. Setiap insert, update, delete, dan replace yang majority-committed menghasilkan satu entri oplog, dan change stream menerjemahkan entri itu menjadi event terstruktur yang ramah aplikasi.

flowchart LR A[Aplikasi] -->|write| M[(Replica set rs0)] M --> OP[Oplog<br/>majority-committed] OP --> CS[Change stream cursor] CS -->|change event| B[Worker konsumen<br/>notifikasi, cache, sync]

Karena bergantung pada oplog, ada dua konsekuensi desain yang penting. Pertama, change stream hanya tersedia di replica set atau sharded cluster — server standalone tidak punya oplog, jadi watch() langsung ditolak. Kedua, umur event terikat pada jendela oplog: bila stream ditinggalkan terlalu lama dan titik terakhirnya sudah tertimpa, resume akan gagal. Lab kita memakai replica set rs0, jadi semua demo section ini berjalan nyata. Kita verifikasi status itu lalu menangkap event pertama:

js
use latihan_events
db.dropDatabase()
db.hello().setName
const cs = db.pesanan.watch()
db.pesanan.insertOne({ kode: "PSN-001", pelanggan: "Budi", total: 250000 })
const ev = cs.next()
({ replika: db.hello().setName, jenis: ev.operationType, koleksi: ev.ns.coll, pesanan: ev.fullDocument.kode })
cs.close()

Output yang Dihasilkan:

text
switched to db latihan_events
{ ok: 1, dropped: 'latihan_events' }
rs0

{
  acknowledged: true,
  insertedId: ObjectId('6ad3e2551d141c07369cd289')
}
Warning: If there are no documents in the batch, next will block. Use tryNext if you want to check if there are any documents without waiting, or cursor.disableBlockWarnings() if you want to disable this warning.

{
  replika: 'rs0',
  jenis: 'insert',
  koleksi: 'pesanan',
  pesanan: 'PSN-001'
}

Empat baris keluaran ringkas itu membuktikan seluruh rantainya: koneksi kita berada di replica set rs0, insert PSN-001 menghasilkan event insert pada collection pesanan, dan fullDocument membawa isi pesanan lengkap. Event baru mengalir hanya setelah tulisan diakui mayoritas anggota — aplikasi tidak akan pernah menerima kabar tentang data yang berpotensi dibatalkan saat failover.

Filter di Sisi Server #

Cursor stream menerima pipeline agregasi, sehingga penyaringan bisa digeser dari aplikasi ke server. Bandingkan: memfilter pesanan bernilai besar di sisi aplikasi berarti menerima SEMUA event lalu membuang sebagian; memfilter dengan $match di watch() berarti server hanya mengirim yang relevan:

js
const cs2 = db.pesanan.watch([ { $match: { "fullDocument.total": { $gte: 200000 } } } ])
cs2.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-002", pelanggan: "Siti", total: 90000 })
db.pesanan.insertOne({ kode: "PSN-003", pelanggan: "Dewi", total: 310000 })
cs2.next().fullDocument.kode
cs2.close()

Output yang Dihasilkan:

text
ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e2561d141c07369cd28a')
}
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e2561d141c07369cd28b')
}
PSN-003

Dua pesanan disisipkan, tetapi stream hanya mengeluarkan satu: PSN-003 dengan total Rp310.000. Event PSN-002 (Rp90.000) tersaring di server dan tidak pernah menempati jaringan maupun loop aplikasi. Pola ini menjadi fondasi desain worker yang efisien di bab-bab berikutnya.

Kenapa Bukan Polling #

Perbandingan langsungnya:

Aspek	Polling (`find` berulang)	Change stream
Latensi deteksi	Secepat interval polling	Mendekati seketika setelah majority-commit
Beban server	Setiap polling = query penuh, mayoritas hasil kosong	Satu cursor `getMore` idle yang murah
Perubahan terlewat	Ya — dua perubahan antara dua polling terlihat satu	Tidak — setiap operasi menghasilkan event sendiri
Titik waktu	Tidak ada jejak posisi	Resume token; bisa dilanjutkan dari posisi persis

Baris “perubahan terlewat” sering paling menohok: polling hanya melihat kondisi terkini, sehingga update yang saling menimpa dalam satu interval polling tidak pernah terdeteksi. Change stream mengalirkan setiap event secara berurut sesuai urutan penerapan di oplog, termasuk diff update (updateDescription) yang tidak bisa direkonstruksi dari snapshot.

Use Case dan Peta Section #

Empat pola yang paling sering dipakai di produksi:

Invalidasi cache — product service memicu penghapusan cache setiap kali document produk berubah; pembaca berikutnya selalu mendapat data segar.
Sinkronisasi sistem eksternal — perubahan di MongoDB mengalir ke Elasticsearch, data warehouse, atau sistem faktur.
Audit dan aktivitas real-time — log aktivitas pengguna atau jejak perubahan dokumen direkam begitu terjadi.
Event-driven architecture — perubahan menjadi pemicu proses bisnis lanjutan: notifikasi, provisioning, analitik.

Peta section ini mengikuti kurva itu: struktur anatomi event penuh (Chapter Struktur Change Event), menonton satu collection lalu seluruh database, resume token untuk keandalan, penanganan koneksi terputus, dan puncaknya pola arsitektur event-driven. Semua contoh memakai collection pesanan dari toko buku kita, jadi hasilnya mudah dibandingkan antar chapter.