Membangun Real-time Notification Pipeline dengan Event-Driven Architecture di Go

Dalam 15 menit ke depan, kita akan merakit stack notifikasi real-time berskala masif dari nol sampai berjalan mulus. Pengalaman di lapangan membuktikan bahwa ketika jutaan pengguna aktif bersamaan, arsitektur request-response konvensional langsung kolaps. Seringkali kita mendapati klien mengeluh notifikasi terlambat datang, server WebSocket kehabisan memori, atau database utama jebol karena connection storm. Di sinilah seni merancang Event-Driven Architecture (EDA) dengan memadukan ketangguhan Apache Kafka sebagai pusat distribusi pesan, performa Go untuk menangani jutaan koneksi WebSocket, serta strategi retry dan Dead Letter Queue (DLQ) yang kokoh berperan menyelamatkan sistem kita. Mari kita langsung masuk ke dapur produksi.

1. Menu Kita Hari Ini & Alat yang Perlu Disiapkan

Bayangkan kita sedang mengelola sistem e-commerce raksasa yang harus mengirimkan status pembayaran dan kurir secara instan ke jutaan browser pengguna tanpa jeda. Untuk meracik pipeline ini, kita butuh beberapa bahan utama:

• Go (Golang) v1.22+: Mesin utama kita karena konkurensinya yang sangat ringan lewat goroutines.
• Apache Kafka: Sebagai event backbone atau pusat saraf yang menampung aliran data mentah dari berbagai microservices.
• Redis: Berfungsi sebagai state store cepat dan pengelola sesi atau broker transisional ringan.
• Docker & Docker Compose: Agar kita bisa langsung menaikkan seluruh ekosistem dalam satu perintah tanpa ribet instalasi manual.

Pastikan Docker dan Go sudah terpasang di mesin lokal Anda. Begitu semua siap, mari kita petakan bagaimana bahan-bahan ini saling berinteraksi secara mulus.

2. Peta Jalan Singkat: Apa yang Sedang Kita Rakit

Sebelum tangan kita kotor oleh barisan kode, mari kita lihat cetak biru perjalanannya. Kita akan memecah waktu 15 menit ini menjadi tiga fase utama yang mengalir tanpa celah:

1. Arsitektur Pesan Berbasis Kafka: Kita bangun fondasi event backbone untuk menerima dan mendistribusikan notifikasi secara asinkron.
2. Manajemen Koneksi WebSocket di Go: Kita racik server Go yang sanggup menjaga puluhan ribu koneksi TCP tetap hidup dan mendistribusikan pesan dari Kafka secara instan.
3. Strategi Retry dan Dead Letter Queue: Kita amankan pipeline dari kegagalan sistem pengiriman dengan mekanisme pentalan otomatis dan kotak sampah investigasi.

[!TIP] Jangan khawatir jika Anda belum terbiasa dengan Kafka. Kita akan menggunakan konfigurasi Docker Compose yang langsung jadi agar kita bisa fokus pada logika kodenya.

Mari kita mulai langkah pertama dengan membangun fondasi lingkungannya.

3. Langkah 1: Mempersiapkan Fondasi & Environment

Kita butuh infrastruktur dasar yang mencakup Kafka, Zookeeper, dan Redis. Buat sebuah direktori kerja, lalu tuliskan file docker-compose.yml berikut di dalamnya. Konfigurasi ini dirancang agar siap menerima beban tinggi di lingkungan lokal:

version: '3.8'

services:
  zookeeper:
  	image: confluentinc/cp-zookeeper:7.3.0
  	environment:
  	  ZOOKEEPER_CLIENT_PORT: 2181
  	  ZOOKEEPER_TICK_TIME: 2000

  kafka:
  	image: confluentinc/cp-kafka:7.3.0
  	depend_on:
  	  - zookeeper
  	ports:
  	  - "9092:9092"
  	environment:
  	  KAFKA_BROKER_ID: 1
  	  KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
  	  KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
  	  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  redis:
  	image: redis:7-alpine
  	ports:
  	  - "6379:6379"

  minio:
  	image: minio/minio
  	ports:
  	  - "9000:9000"
  	  - "9001:9001"
  	environment:
  	  MINIO_ROOT_USER: 'admin'
  	  MINIO_ROOT_PASSWORD: 'password123'
  	command: server /data --console-address ":9001"

Jalankan perintah docker compose up -d di terminal Anda. Lihat responsnya—ketika kontainer-kontainer tersebut menyala satu per satu, itu tanda bahwa jantung infrastruktur kita sudah berdenyut. Dari fondasi ini, kita langsung melangkah ke peracikan kode inti di sisi backend.

4. Langkah 2: Meracik Konfigurasi Inti Tanpa Basa-basi

Sekarang kita masuk ke bagian paling krusial: bagaimana Go membaca pesan dari Kafka dan menyebarkannya lewat WebSocket ke klien. Kita gunakan pustaka segmentio/kafka-go dan gorilla/websocket. Inisialisasi proyek Go Anda dengan go mod init notification-pipeline, lalu buat file main.go yang berisi orkestrasi lengkap:

package main

import (
	"context"
	"encoding/json"
	"log"
	"net/http"

	"github.com/gorilla/websocket"
	"github.com/segmentio/kafka-go"
)

// NotificationPayload mendefinisikan struktur data pesan real-time kita
type NotificationPayload struct {
	UserID  string `json:"user_id"`
	Message string `json:"message"`
	Type    string `json:"type"`
}

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool { return true },
}

// Hub menyimpan pemetaan koneksi klien aktif
var clients = make(map[string]*websocket.Conn)

func handleConnections(w http.ResponseWriter, r *http.Request) {
	ws, err := upgrader.Upgrade(w, r, nil)
	ifi err != nil {
		log.Printf("Gagal melakukan upgrade WebSocket: %v", err)
		return
	}
	defer ws.Close()

	// Simulasi identifikasi user sederhana dari query param
	userId := r.URL.Query().Get("user_id")
	if userId == "" {
		return
	}

	clients[userId] = ws
	log.Printf("User %s terhubung via WebSocket", userId)

	for {
		// Menjaga koneksi tetap hidup dan mendeteksi diseksi
		_, _, err := ws.ReadMessage()
		ifi err != nil {
			delete(clients, userId)
			log.Printf("User %s terputus", userId)
			break
		}
	}
}

func startKafkaConsumer() {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  []string{"localhost:9092"},
		Topic:    "user-notifications",
		GroupID:  "notification-group",
		MinBytes: 10e3, // 10KB
		MaxBytes: 10e6, // 10MB
	})

	ctx := context.Background()
	for {
		m, err := r.ReadMessage(ctx)
		ifi err != nil {
			log.Printf("Gagal membaca pesan Kafka: %v", err)
			// Masuk ke logika Retry / DLQ jika diperlukan
			continue
		}

		var notif NotificationPayload
		if err := json.Unmarshal(m.Value, &notif); err != nil {
			log.Printf("Format JSON tidak valid: %v", err)
			continue
		}

		// Salurkan pesan ke klien WebSocket yang sesuai
		if conn, ok := clients[notif.UserID]; ok {
			err := conn.WriteJSON(notif)
			ifi err != nil {
				log.Printf("Gagal mengirim pesan ke WebSocket: %v", err)
				conn.Close()
				delete(clients, notif.UserID)
			}
		}
	}
}

func main() {
	go startKafkaConsumer()

	http.HandleFunc("/ws", handleConnections)
	log.Println("Server WebSocket berjalan di port 8080...")
	ifi err := http.ListenAndServe(":8080", nil); err != nil {
		log.Fatal("Server error: ", err)
	}
}

Kode di atas langsung mengawinkan konsumsi pesan Kafka dengan pengiriman instan ke WebSocket. Namun, sistem produksi nyata tidak boleh berhenti hanya sampai di sini. Kita perlu mengantisipasi skenario saat pengiriman gagal berkali-kali.

5. Langkah 3: Sentuhan Akhir Keamanan & Optimasi

Di lapangan, jaringan bisa terputus, klien bisa mengalami timeout, atau layanan tujuan bisa macet total. Kita butuh Strategi Retry dan Dead Letter Queue (DLQ). Jika pesan gagal dikirim ke WebSocket sebanyak tiga kali, kita tidak boleh membuangnya begitu saja atau membuat konsumer Kafka macet. Kita harus melemparnya ke topik Kafka khusus bernama user-notifications-dlq untuk dianalisis ulang oleh tim teknik nanti.

[!IMPORTANT] Selalu terapkan Exponential Backoff pada strategi retry agar kita tidak membebani sistem hilir (downstream) saat terjadi lonjakan lonjakan error.

Tambahkan fungsi pengiriman DLQ sederhana pada konsumer Kafka kita:

func sendToDLQ(message []byte) {
	// Inisialisasi Kafka writer khusus untuk DLQ
	writer := &kafka.Writer{
		Addr:     kafka.TCP("localhost:9092"),
		Topic:    "user-notifications-dlq",
		Balancer: &kafka.Hash{},
	}
	defer writer.Close()

	err := writer.WriteMessages(context.Background(), kafka.Message{
		Value: message,
	})
	ifi err != nil {
		log.Printf("Kritis: Gagal menulis ke DLQ: %v", err)
	}
}

Dengan tambahan ini, pipeline kita kini memiliki ketahanan enterprise-grade. Mari kita buktikan ketangguhannya dalam sesi uji coba kilat.

6. Tes Rasa: Memastikan Semuanya Berjalan Sempurna

Waktunya pembuktian. Jalankan aplikasi Go Anda dengan perintah go run main.go. Untuk menguji apakah pipa pesan ini berfungsi dengan benar, kita bisa mempublikasikan pesan langsung ke Kafka melalui CLI dalam container Docker.

1. Masuk ke dalam kontainer Kafka:

   docker exec -it <container_id_kafka> kafka-console-producer --bootstrap-server localhost:9092 --topic user-notifications
   

2. Masukkan payload JSON mentah berikut:

   {"user_id": "123", "message": "Pembayaran Anda berhasil diverifikasi!", "type": "success"}
   

3. Buka browser atau hubungkan klien WebSocket uji coba ke ws://localhost:8080/ws?user_id=123. Anda akan melihat pesan tersebut muncul secara instan tanpa jeda yang terasa.

Jika pesan langsung mendarat di layar klien, itu tanda racikan arsitektur kita sukses besar!

7. Pertolongan Pertama: Jika Terjadi Kendala Kecil

Sebagai penutup, berikut beberapa gangguan klasik yang sering ditemui di lapangan beserta penawarnya:

• Koneksi WebSocket langsung terputus (Error 1006): Biasanya disebabkan oleh proxy atau Nginx di depan server Go yang belum dikonfigurasi untuk mendukung Upgrade Connection. Pastikan Anda menambahkan direktif proxy_set_header Upgrade $http_upgrade; dan proxy_set_header Connection "upgrade"; pada konfigurasi Nginx Anda.
• Kafka Consumergroup Rebalancing Loop: Terjadi jika proses pengolahan pesan terlalu lama (MaxPollIntervalMs terlampaui). Solusinya, pisahkan proses berat di dalam konsumer menjadi asynchronous worker pool di Go.

Nikmati pipeline baru Anda, dan sampai jumpa di resep arsitektur sistem berikutnya!