Belajar Software Architecture and Design Patterns - Event Sourcing & Outbox Pattern

#Pendahuluan

Dua pola terakhir trio event menjawab dua masalah berbeda. Event sourcing menyimpan state bukan sebagai baris terakhir, melainkan sebagai rangkaian peristiwa — membuka audit trail sempurna dan time-travel. Transactional Outbox memperbaiki lubang kecil namun fatal di episode 21: bagaimana menjamin event benar-benar terkirim bersamaan dengan commit data.

#Konsep

#Event Sourcing: State = Replay Events

TRADISIONAL : tabel orders -> UPDATE row (nilai lama HILANG)
EVENT SOURCING:
  stream order#o1:
    [1] OrderPlaced{customerId, items:[A x1]}
    [2] ItemAdded{sku:B, qty:2}
    [3] ItemRemoved{sku:A}
  state sekarang = fold/replay semua event dari awal

Manfaatnya:

• Audit trail alami — setiap perubahan tercatat siapa/apa/kapan; kewajiban regulator terpenuhi gratis.
• Time travel & debug — rekonstruksi state pada titik waktu mana pun.
• Proyeksi bebas — read model baru dibangun ulang tanpa migrasi data lama (proses kons = consumer yang membangun proyeksi).

Biayanya jujur untuk disebut: query butuh proyeksi, evolusi schema event butuh upcasting, dan volume event besar butuh snapshotting — simpan state hasil fold tiap N event, replay hanya dari snapshot terakhir.

#Dual-Write Problem → Transactional Outbox

Bug klasik EDA: commit DB sukses, publish broker gagal (atau sebaliknya) → sistem tidak sinkron. Solusinya elegan:

Event tersimpan di tabel outbox dalam transaksi yang sama dengan data — atomik pasti. Relay (scheduler/ticker/worker) mengangkutnya ke broker. Konsumen tetap harus idempotent (ep.21) karena relay bisa dobel-kirim.

#Real-World Implementasi

// Simpan bersama transaksi:
await this.uow.transaction(async () => {
  await this.orders.save(order);                       // state tradisional ATAU stream
  await this.outbox.append(order.pullEvents());        // events masuk satu transaksi
});
 
// Relay via @nestjs/schedule:
@Interval(1000)
async relay() {
  const batch = await this.outbox.unsent(50);
  for (const e of batch) {
    await this.broker.publish(e.type, e.payload);      // at-least-once
    await this.outbox.markSent(e.id);
  }
}

func StartRelay(ctx context.Context, db *sql.DB, js nats.JetStream) {
    t := time.NewTicker(time.Second)
    go func() {
        for {
            select {
            case <-ctx.Done():
                return
            case <-t.C:
                rows := fetchUnsent(ctx, db, 50)
                for _, ev := range rows {
                    if err := publish(js, ev.Type, ev.Payload); err != nil {
                        continue // biarkan diambil lagi tick berikutnya
                    }
                    markSent(ctx, db, ev.ID)
                }
            }
        }
    }()
}

// app/Console/Kernel.php:
$schedule->command('outbox:relay')->everySecond();
 
// app/Console/Commands/OutboxRelay.php:
public function handle(OutboxRepository $outbox, Broker $broker): void {
    foreach ($outbox->unsent(50) as $event) {          // FOR UPDATE SKIP LOCKED
        $broker->publish($event->type, $event->payload);
        $outbox->markSent($event);
    }
}

#Event Stream & Rebuild State

function rebuild(events: StoredEvent[]): OrderState {
  return events.reduce<OrderState>(
    (s, e) => match (e.type) {
      'OrderPlaced'  => place(s, e),
      'ItemAdded'    => addItem(s, e),
      'OrderShipped' => ship(s),
      default        => s,
    },
    emptyState(),
  );
}

#Praktik

Target outline: simpan event stream OrderPlaced/ItemAdded + rebuild state, dan outbox untuk publish aman.

1. Tabel event_store(stream_id, version, type, payload JSONB, occurred_at)
   Tabel outbox(id PK, type, payload, created_at, sent_at NULL)
2. PlaceOrder use case:
   a. append OrderPlaced ke event_store + outbox DALAM SATU transaksi
   b. tambah ItemAdded x2 lewat method aggregate
3. Rebuild test: baca stream order o1 -> fold -> assert total/status
4. Snapshotting mini: simpan snapshot tiap 5 event;
   rebuild mulai dari snapshot + sisa event (ukur selisih waktu!)
5. Relay: jalankan scheduler/ticker/worker ->
   amati event sampai broker (consumer log ep.21)
6. Uji kegagalan: matikan broker saat relay jalan ->
   event TETAP di outbox (sent_at masih null) ->
   nyalakan broker -> terkirim. Itu bukti atomicity.

[x] Append stream + outbox satu transaksi (bukti rollback: gagal sengaja,
     KEDUANYA tidak masuk)
[x] Rebuild state dari N event identik dengan state live
[x] Snapshot mempercepat rebuild (ukur ms sebelum/sesudah)
[x] Relay at-least-once + konsumen idempotent tetap aman
[x] Audit trail demo: "tunjukkan isi keranjang user kemarin jam 14:00"
     dijawab dari replay - bukan dari tabel
[x] Commit "ep22: event store + outbox"

#Penutup

Rangkuman episode ini:

• Event sourcing: state = fold event; manfaat audit/time-travel/proyeksi; biaya proyeksi+upcasting+snapshot.
• Outbox menyelesaikan dual-write: event & data atomik di satu transaksi, relay mengangkut at-least-once.
• Praktik membuktikan keduanya: rollback bersama, rebuild identik, relay selamat broker mati.

Episode 23 menjaga semuanya tetap benar: Testing Strategy per Arsitektur — piramida test, unit domain murni, mock port, integration adapter, contract test, dan architecture tests. Sampai jumpa!