Menonton Satu Database

Worker audit atau sinkronisasi search index jarang hanya peduli satu collection — perubahan di pesanan, pelanggan, dan produk sama-sama harus mereka tangkap. Membuka tiga stream terpisah bisa saja, tetapi MongoDB menyediakan cakupan yang lebih rapi: satu stream untuk seluruh collection dalam satu database, dengan field ns (namespace) sebagai pembeda asal event. Bab ini menaikkan level menonton dari collection ke database latihan_events, menyaring per collection dari dalam stream, lalu menutup dengan cakupan terluas: seluruh cluster.

Satu Stream untuk Semua Collection #

Panggilan db.watch() — pada objek database, bukan collection — membuka stream untuk semua collection non-system di database aktif:

js
use latihan_events
db.dropDatabase()
db.pesanan.insertOne({ kode: "PSN-001", pelanggan: "Budi", total: 250000 })
const cs = db.watch()
cs.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-002", pelanggan: "Siti", total: 175000 })
const ev1 = cs.next()
db.pengiriman.insertOne({ pesanan: "PSN-002", kurir: "JNE", estimasiHari: 3 })
const ev2 = cs.next()
({ sumberPertama: ev1.ns.coll, sumberKedua: ev2.ns.coll, jenis: [ ev1.operationType, ev2.operationType ] })
cs.close()

Output yang Dihasilkan:

text
switched to db latihan_events
{ ok: 1, dropped: 'latihan_events' }
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e32c349f527dc9ccabe1')
}

{
  acknowledged: true,
  insertedId: ObjectId('6ad3e32c349f527dc9ccabe2')
}
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e32c349f527dc9ccabe3')
}
{
  sumberPertama: 'pesanan',
  sumberKedua: 'pengiriman',
  jenis: [ 'insert', 'insert' ]
}

Satu cursor, dua sumber: event pertama datang dari pesanan, event kedua dari pengiriman — keduanya dibedakan lewat field ns (namespace) yang memuat db dan coll. Inilah pola yang tepat untuk worker lintas collection: satu koneksi, satu loop, dan perutean berdasarkan ns.coll di dalam kode konsumen. Seperti sebelumnya, insert PSN-001 yang terjadi sebelum watch() dibuka tidak menghasilkan event.

Urutan event juga terjaga: keduanya tiba sesuai urutan penulisannya di oplog, bukan acak. Untuk collection yang berbeda, urutan ini mengikuti urutan global operasi pada cluster — jadi konsistensi antara pesanan dan pengiriman yang ditulis berurutan tetap terlihat berurutan di stream.

Menyaring per Collection di Sisi Server #

Cakupan luas tidak berarti konsumen wajib memproses semuanya. Pipeline $match pada ns.coll menyaring asal event sebelum meninggalkan server — worker pengiriman hanya menerima event pengiriman meski stream-nya mencakup seluruh database:

js
const cs2 = db.watch([ { $match: { "ns.coll": "pengiriman" } } ])
cs2.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-003", pelanggan: "Dewi", total: 320000 })
db.pengiriman.insertOne({ pesanan: "PSN-003", kurir: "SiCepat", estimasiHari: 2 })
cs2.next().fullDocument
cs2.close()

Output yang Dihasilkan:

text
ChangeStreamCursor on latihan_events
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e32d349f527dc9ccabe4')
}
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e32d349f527dc9ccabe5')
}
{
  _id: ObjectId('6ad3e32d349f527dc9ccabe5'),
  pesanan: 'PSN-003',
  kurir: 'SiCepat',
  estimasiHari: 2
}

Insert PSN-003 ke pesanan dan insert pengirimannya sama-sama terjadi, tetapi next() hanya mengeluarkan document dari pengiriman — event pesanan tersaring di server. Dibandingkan membuka dua stream collection terpisah, pola ini menghemat satu koneksi dan menjadikan filter sebagai data konfigurasi: daftar collection yang diminati tinggal ditulis dalam satu $match ({ "ns.coll": { $in: ["pesanan", "pengiriman"] } }).

Menonton Seluruh Cluster #

Level terluas adalah cluster: db.getMongo().watch() membuka stream yang mencakup semua database dan collection non-system di seluruh deployment. Kita uji dengan menulis ke dua database berbeda sekaligus:

js
const csx = db.getMongo().watch()
csx.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-004", pelanggan: "Eka", total: 98000 })
db.getSiblingDB("latihan_events_lain").aktivitas.insertOne({ aksi: "uji-cluster" })
({ dbPertama: csx.next().ns.db, dbKedua: csx.next().ns.db })
csx.close()
db.getSiblingDB("latihan_events_lain").dropDatabase()

Output yang Dihasilkan:

text
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e32e349f527dc9ccabe6')
}
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e32e349f527dc9ccabe7')
}
{
  dbPertama: 'latihan_events',
  dbKedua: 'latihan_events_lain'
}
{ ok: 1, dropped: 'latihan_events_lain' }

Dua event dari dua database berbeda mengalir lewat satu cursor cluster. Use case level ini biasanya bersifat platform: audit lintas tenant, pipeline ETL global, atau alat observability yang mencatat semua perubahan. Kolom ns.db menjadi pembeda utamanya.

Rangkuman #

db.watch() menonton semua collection non-system dalam satu database; db.getMongo().watch() menaikkan cakupan ke seluruh cluster.
Field ns (db + coll) membedakan asal setiap event; perutean konsumen dilakukan berdasarkan ns.coll.
Filter { $match: { "ns.coll": ... } } menyaring collection tertentu di sisi server — alternatif efisien untuk membuka banyak stream terpisah.
Cakupan yang lebih luas menuntut privilege changeStream + find yang lebih luas pula; berikan worker hanya cakupan minimum.