Event-Driven Architecture

Peralatan sudah lengkap: kita tahu struktur event, cara menyimpan resume token, dan cara menyambung ulang stream. Sekarang saatnya mengubahnya menjadi arsitektur — dari aplikasi yang menanyakan data menjadi sistem yang bereaksi ketika data berubah. Bab penutup section ini merancang pola source-process-action di atas change streams, menguji dua implementasi mini yang berjalan (notifikasi pesanan dan invalidasi cache), lalu menutup dengan batas pola ini supaya kita tidak memaksakan change streams menjadi sesuatu yang bukan dirinya.

Pola Source - Process - Action #

Semua sistem event-driven di atas change streams berbagi satu kerangka: MongoDB sebagai source kebenaran, worker sebagai processor yang menyaring dan menerjemahkan event, dan efek samping sebagai action:

Empat pasangan proses-aksi yang paling sering dipakai di produksi:

Dibanding polling terjadwal, seluruh tabel di atas mendapat latensi mendekati seketika, tanpa query kosong berulang, dan tanpa perubahan yang tertimpa tanpa terlihat — perbandingan rinci ada di Chapter Change Streams Overview. Yang ditambah pola ini adalah disiplin: satu worker, satu tugas, satu filter.

Notifikasi Pesanan Baru #

Contoh pertama adalah kasus paling klasik: toko buku ingin setiap pesanan baru memicu notifikasi. Worker membuka stream dengan filter operationType: "insert" di sisi server, lalu menyerap semua event yang siap dalam satu batch:

use latihan_events
db.dropDatabase()
const cs = db.pesanan.watch([{ $match: { operationType: "insert" } }], { maxAwaitTimeMS: 2000 })
cs.disableBlockWarnings()
db.pesanan.insertMany([
  { kode: "PSN-301", pelanggan: "Budi", total: 250000 },
  { kode: "PSN-302", pelanggan: "Siti", total: 175000 },
  { kode: "PSN-303", pelanggan: "Dewi", total: 310000 }
])
let terkirim = 0
let ev
while ((ev = cs.tryNext()) !== null) {
  terkirim = terkirim + 1
  print("[notifikasi] " + terkirim + ". Pesanan baru " + ev.fullDocument.kode + " dari " + ev.fullDocument.pelanggan + " senilai Rp" + ev.fullDocument.total)
}
cs.close()

Output yang Dihasilkan:

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

ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedIds: {
    '0': ObjectId('6ad3eb7d1d9200c51f2142f6'),
    '1': ObjectId('6ad3eb7d1d9200c51f2142f7'),
    '2': ObjectId('6ad3eb7d1d9200c51f2142f8')
  }
}

[notifikasi] 1. Pesanan baru PSN-301 dari Budi senilai Rp250000
[notifikasi] 2. Pesanan baru PSN-302 dari Siti senilai Rp175000
[notifikasi] 3. Pesanan baru PSN-303 dari Dewi senilai Rp310000

Tiga pesanan masuk, tiga notifikasi keluar — tanpa satu pun polling. Empat keputusan desain ada di balik blok pendek ini:

1. Filter di server — $match di watch() membuat server hanya mengirim event insert; update status pesanan yang pasti menyusul tidak pernah membebani worker.
2. Penyerapan batch — tryNext() mengambil event yang sudah tersedia tanpa menunggu, berhenti saat null. Notifikasi yang pengirimannya lebih lambat dari lahirnya event tidak menumpuk di server, melainkan di loop worker — inilah penanganan backpressure paling sederhana: serap secepat side effect mengizinkan, biarkan cursor menahan sisanya.
3. maxAwaitTimeMS: 2000 — batas atas server menahan satu getMore; worker mendapat kesempatan bangun tiap dua detik untuk cek token/checkpoint meski event tak kunjung datang.
4. Resume siap pakai — blok ini bisa dibungkus loop reconnect dari Chapter Menangani Interruption tanpa mengubah isinya.

Invalidasi Cache #

Contoh kedua menyelesaikan masalah setua cache itu sendiri: pembaca mendapat data basi karena produk berharga baru namun cache masih memuat harga lama. Dengan stream pada produk, setiap update langsung menghapus kunci yang bersangkutan:

db.produk.insertMany([
  { _id: "produk-001", nama: "Kopi Gayo", harga: 85000 },
  { _id: "produk-002", nama: "Teh Melati", harga: 32000 }
])
const cache = new Map()
db.produk.find().forEach(p => cache.set(p._id, p.nama))
cache

Output yang Dihasilkan:

{
  acknowledged: true,
  insertedIds: { '0': 'produk-001', '1': 'produk-002' }
}

Map(2) { 'produk-001' => 'Kopi Gayo', 'produk-002' => 'Teh Melati' }

const csC = db.produk.watch([{ $match: { operationType: "update" } }])
csC.disableBlockWarnings()
db.produk.updateOne({ _id: "produk-001" }, { $set: { harga: 92000 } })
db.produk.insertOne({ _id: "produk-003", nama: "Gula Aren", harga: 15000 })
let evHarga
while ((evHarga = csC.tryNext()) !== null) {
  cache.delete(evHarga.documentKey._id)
  print("[cache] invalidate " + evHarga.documentKey._id)
}
({ sisaCache: [...cache.keys()] })
csC.close()

Output yang Dihasilkan:

ChangeStreamCursor on produk
{
  acknowledged: true,
  insertedId: null,
  matchedCount: 1,
  modifiedCount: 1,
  upsertedCount: 0
}
{ acknowledged: true, insertedId: 'produk-003' }

[cache] invalidate produk-001

{ sisaCache: [ 'produk-002' ] }

Harga Kopi Gayo berubah → kunci produk-001 dihapus dari cache → pembaca berikutnya dipaksa mengambil data segar dari MongoDB. Perhatikan dua detail: event update cukup documentKey — worker tidak butuh isi document untuk menghapus kunci cache, jadi opsi fullDocument tidak perlu; dan insert Gula Aren tidak memicu apa pun karena tersaring $match — cache baru akan diisi malas saat pembaca pertama datang (pola cache-aside).

Urutan, Backpressure, dan Batas Pola #

Urutan. MongoDB menjamin event untuk satu document tiba sesuai urutan penerapan operasinya — insert PSN-301 tidak pernah tiba setelah update-nya, dan updateMany yang menyentuh banyak document menghasilkan satu event per document. Pada replica set aliran mengikuti urutan oplog; pada sharded cluster jaminan penyusunan menyeluruh hanya berlaku per document. Konsekuensinya praktis: worker yang mempertahankan state per document (misalnya status pesanan terkini) bisa memproses event secara berurutan tanpa pengurutan ulang.

Backpressure. Dua tuas yang sudah kita lihat — penyerapan batch dengan tryNext() dan maxAwaitTimeMS — cukup untuk worker ringan. Untuk side effect yang berat (kirim email, panggil API pihak ketiga), pisahkan loop stream dari eksekusi: worker hanya memindahkan event ke antrean internal (worker queue), dan pool eksekutor memprosesnya dengan laju sendiri.

Batas pola. Change stream bukan message broker serba bisa. Tidak ada consumer groups ala Kafka — dua worker yang membaca stream sama dengan token masing-masing akan menerima event yang sama (fan-out, bukan load balancing). Tidak ada penyimpanan event jangka panjang — data hidup selama jendela oplog. Bila sistem butuh keduanya, letakkan broker antrean (RabbitMQ, Kafka) sebagai action dari pola source-process-action, dan biarkan change stream tetap sebagai detektor perubahan yang andal.

Rangkuman #

• Kerangka dasar: MongoDB (source) → worker dengan filter (process) → efek samping (action); satu worker satu tugas.
• Notifikasi pesanan: $match insert di server + penyerapan batch tryNext(); resume token membuatnya tahan restart.
• Invalidasi cache: documentKey saja sudah cukup; insert baru tidak perlu ikut karena cache bersifat malas.
• Urutan dijamin per document (dan menyeluruh pada replica set); backpressure ditangani batch + maxAwaitTimeMS, atau antrean internal untuk side effect berat.
• Tanpa consumer groups dan tanpa penyimpanan permanen — untuk kebutuhan itu, kombinasikan dengan message broker.