Menonton satu collection adalah bentuk change stream yang paling umum: worker notifikasi hanya peduli pesanan, worker pengiriman hanya peduli pengiriman. Cakupan sempit berarti event yang mengalir seratus persen relevan, dan filter yang kita pasang bekerja pada volume terkecil. Bab ini membuka stream pada collection pesanan, membaca event insert secara utuh, menyaring jenis event dengan pipeline, lalu menghadapi kasus yang hampir selalu mengejutkan pendatang baru: event update yang tidak membawa isi document.

Membuka Stream dan Membaca Event #

Satu collection ditonton dengan db.<collection>.watch(). Seperti biasa di lab, kita buka stream dulu, melakukan tulisan, lalu membaca event yang dihasilkan:

js 
use latihan_events
db.dropDatabase()
db.pesanan.insertOne({ kode: "PSN-000", pelanggan: "Eka", total: 120000 })
const cs = db.pesanan.watch()
db.pesanan.insertOne({ kode: "PSN-001", pelanggan: "Budi", total: 250000 })
cs.next()

Output yang Dihasilkan:

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

{
  acknowledged: true,
  insertedId: ObjectId('6ad3e297349f527dc9ccabdf')
}
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.
{
  _id: {
    _data: '826AD3E297000000012B042C0100296E5A100494709451B73C4727A20A3E7AE2819C51463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F696400646AD3E297349F527DC9CCABDF000004'
  },
  operationType: 'insert',
  clusterTime: Timestamp({ t: 1792270999, i: 1 }),
  wallTime: ISODate('2026-10-17T21:03:19.383Z'),
  fullDocument: {
    _id: ObjectId('6ad3e297349f527dc9ccabdf'),
    kode: 'PSN-001',
    pelanggan: 'Budi',
    total: 250000
  },
  ns: { db: 'latihan_events', coll: 'pesanan' },
  documentKey: { _id: ObjectId('6ad3e297349f527dc9ccabdf') }
}

Perhatikan bahwa PSN-000 sengaja disisipkan SEBELUM watch() dibuka — eventnya tidak pernah muncul. Change stream hanya mengalirkan perubahan setelah cursor dibuka; data yang sudah ada bukan urusannya. Enam field utama event insert:

  1. _id — resume token (string _data heksadesimal); posisi event di aliran oplog, kuncinya di Chapter Resume Tokens.
  2. operationType — jenis perubahan: insert, update, replace, delete, drop, invalidate.
  3. clusterTime / wallTime — waktu logis dari logical clock cluster dan waktu dinding saat operasi diterapkan.
  4. fullDocument — isi document utuh; untuk insert ini memang versi yang barusan disisipkan.
  5. ns — namespace: database latihan_events, collection pesanan.
  6. documentKey — _id document yang berubah; kunci untuk mengambil ulang datanya.

Peringatan mongosh soal next yang menunggu adalah perilaku yang diinginkan: next() memblokir sampai event tiba. Di worker produksi, panggilan ini hidup di dalam loop; di REPL, kita cukup memastikan tulisannya sudah terjadi sebelum memanggil next(). Untuk keluaran yang lebih bersih, cs.disableBlockWarnings() mematikan peringatan itu — kita pakai mulai blok berikutnya.

Menyaring Jenis Event dengan Pipeline #

Argumen pertama watch() adalah pipeline agregasi. Stage $match di posisi pertama menyaring event sebelum meninggalkan server:

js 
const cs2 = db.pesanan.watch([ { $match: { operationType: "insert" } } ])
cs2.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-002", pelanggan: "Siti", total: 175000 })
db.pesanan.updateOne({ kode: "PSN-002" }, { $set: { status: "dibayar" } })
cs2.next().fullDocument
cs2.close()

Output yang Dihasilkan:

text 
ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e298349f527dc9ccabe0')
}
{
  acknowledged: true,
  insertedId: null,
  matchedCount: 1,
  modifiedCount: 1,
  upsertedCount: 0
}
{
  _id: ObjectId('6ad3e298349f527dc9ccabe0'),
  kode: 'PSN-002',
  pelanggan: 'Siti',
  total: 175000
}

Dua operasi tulis terjadi — insert PSN-002 dan update statusnya — tetapi next() hanya mengembalikan event insert. Event update tersaring $match dan menghilang dari stream; close() menutup cursor. Worker notifikasi “pesanan baru” biasanya persis seperti ini: operationType: "insert" saja, sisanya diabaikan.

Kejutan Event Update: Tanpa fullDocument #

Sekarang kita tonton tanpa filter dan memicu update. Inilah perilaku yang mengejutkan:

js 
const cs3 = db.pesanan.watch()
cs3.disableBlockWarnings()
db.pesanan.updateOne({ kode: "PSN-001" }, { $set: { status: "diproses" } })
cs3.next()
cs3.close()

Output yang Dihasilkan:

text 
ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: null,
  matchedCount: 1,
  modifiedCount: 1,
  upsertedCount: 0
}
{
  _id: {
    _data: '826AD3E299000000012B042C0100296E5A100494709451B73C4727A20A3E7AE2819C51463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046645F696400646AD3E297349F527DC9CCABDF000004'
  },
  operationType: 'update',
  clusterTime: Timestamp({ t: 1792271001, i: 1 }),
  wallTime: ISODate('2026-10-17T21:03:21.562Z'),
  ns: { db: 'latihan_events', coll: 'pesanan' },
  documentKey: { _id: ObjectId('6ad3e297349f527dc9ccabdf') },
  updateDescription: {
    updatedFields: { status: 'diproses' },
    removedFields: [],
    truncatedArrays: []
  }
}

Field fullDocument tidak ada. Sebagai gantinya, updateDescription memuat diff: updatedFields berisi field yang berubah beserta nilai barunya, removedFields berisi yang dihapus. Perilaku ini masuk akal dari kacamata oplog — oplog menyimpan diff operasi, bukan salinan document, dan menyalin seluruh isi document ke setiap event akan menggandakan beban untuk update kecil pada document besar.

Bila worker kita butuh document lengkap (misalnya untuk membangun ulang cache), minta secara eksplisit lewat opsi fullDocument: "updateLookup":

js 
const cs4 = db.pesanan.watch([], { fullDocument: "updateLookup" })
cs4.disableBlockWarnings()
db.pesanan.updateOne({ kode: "PSN-001" }, { $set: { kurir: "JNE" } })
cs4.next().fullDocument
cs4.close()

Output yang Dihasilkan:

text 
ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: null,
  matchedCount: 1,
  modifiedCount: 1,
  upsertedCount: 0
}
{
  _id: ObjectId('6ad3e297349f527dc9ccabdf'),
  kode: 'PSN-001',
  pelanggan: 'Budi',
  total: 250000,
  status: 'diproses',
  kurir: 'JNE'
}

Dengan updateLookup, event update tetap membawa updateDescription (diff tetap ada di event aslinya), tetapi fullDocument kini terisi versi terbaru document hasil pembacaan ulang oleh server pada saat event diproses. Satu konsekuensi wajar dari “hasil pembacaan ulang”: fullDocument adalah kondisi saat event dibaca, bukan kondisi persis saat operasi — bila dua update beruntun sangat cepat, event pertama bisa menampilkan hasil yang sudah memuat perubahan kedua.

Rangkuman #

  • db.<collection>.watch() membuka stream untuk satu collection; event hanya mengalir untuk perubahan setelah cursor dibuka.
  • Pipeline [{ $match: { operationType: ... } }] menyaring jenis event di sisi server sebelum mencapai aplikasi.
  • Event update default tidak membawa fullDocument — hanya updateDescription berisi updatedFields/removedFields (diff dari oplog).
  • Opsi { fullDocument: "updateLookup" } mengisi fullDocument dengan versi terbaru document saat event diproses; delete tidak pernah membawanya.