Message Contract #

Dalam arsitektur terdistribusi berbasis peristiwa (event-driven architecture), salah satu kesalahan persepsi yang paling sering terjadi di kalangan tim pengembang adalah memperlakukan pesan (message) RabbitMQ murni sebagai payload JSON dinamis yang bebas. Karena format JSON secara natural mendukung skema fleksibel tanpa skema ketat (schema-less), developer sering kali dengan santai memodifikasi struktur data pesan—mengubah nama field, menghapus elemen, atau mengganti tipe data—tanpa melakukan koordinasi yang matang dengan tim pengembang layanan lainnya. Kebebasan ini sering kali berujung pada kekacauan runtime di lingkungan produksi, yang memicu kepanikan massal saat puluhan instans konsumen crash secara berantai akibat tidak mampu mengurai (parsing) data format baru yang tidak dikenal.

Secara arsitektural, setiap pesan yang dikirimkan melewati RabbitMQ bukan sekadar pembawa data biasa, melainkan kontrak API publik formal yang bersifat asinkron. Kontrak pesan ini menghubungkan berbagai layanan mandiri secara longgar (loose coupling). Oleh karena itu, kita harus mengelola skema kontrak pesan dengan tingkat disiplin, tata kelola, dan standardisasi yang sama ketatnya dengan API HTTP REST (menggunakan OpenAPI/Swagger) atau API gRPC (menggunakan Protobuf). Menjaga integritas kontrak ini adalah kunci utama untuk mencegah kegagalan integrasi, menghentikan penumpukan pesan rusak di Dead Letter Queue (DLQ), dan memungkinkan penskalaan tim pengembang secara mandiri tanpa saling mengunci satu sama lain.

Konvensei Penamaan Routing Key Terstandarisasi #

Langkah pertama dalam menyusun tata kelola kontrak pesan yang disiplin adalah menentukan konvensi penamaan kunci perutean (routing key naming convention). Routing key adalah alamat logis pesan yang menentukan ke mana pesan akan dialirkan. Tanpa aturan yang baku, kita akan mendapati routing key yang membingungkan dan tidak konsisten, seperti kirim-data, user_update, atau event123.

Aturan penamaan routing key terbaik adalah menggunakan format bertingkat yang berbasis pada taksonomi domain:

[Domain].[Subdomain].[Entity].[Action]

Mari kita urai setiap komponen dari struktur di atas:

• [Domain]: Nama konteks bisnis utama (misalnya sales, identity, inventory).
• [Subdomain]: Bagian domain spesifik di bawahnya (misalnya order, user, stock).
• [Entity]: Nama objek bisnis konkret yang mengalami perubahan status (misalnya cart, profile, item).
• [Action]: Kata kerja masa lampau (past-tense verb) yang merepresentasikan fakta bisnis yang telah terjadi (misalnya created, updated, deleted, added).

Contoh Penerapan yang Benar: #

• sales.order.cart.added (Peristiwa ketika pengguna menambahkan item ke keranjang belanja).
• identity.user.profile.updated (Peristiwa ketika pengguna memperbarui informasi profil mereka).
• inventory.stock.item.depleted (Peristiwa ketika stok barang di gudang telah habis).

Aturan Tambahan: #

1. Gunakan Huruf Kecil (Lowercase): Selalu gunakan huruf kecil untuk menghindari sensitivitas huruf (case sensitivity) di berbagai pustaka pembuat client.
2. Gunakan Titik (.) sebagai Pemisah: Karakter titik adalah standar AMQP untuk membagi tingkatan token perutean, yang mempermudah pencocokan pola wildcard pada Topic Exchange.

Payload Envelope yang Terstandarisasi #

Pesan yang dikirimkan antar-layanan tidak boleh hanya berisi data bisnis mentah. Kita harus membungkus (wrap) payload tersebut ke dalam struktur amplop pesan standar (Payload Envelope). Amplop ini memisahkan informasi administratif sistem (metadata) dengan data fungsional bisnis.

Struktur amplop pesan yang sehat dibagi menjadi dua bagian utama:

1. Metadata (Envelope): Berisi informasi identitas pesan untuk kebutuhan pelacakan (tracing), audit, dan de-duplikasi.
2. Data (Payload): Objek spesifik yang berisi data transaksi asli.

Berikut adalah contoh skema envelope JSON standar yang wajib diimplementasikan di seluruh klaster microservices kita:

{
  "event_id": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
  "event_type": "sales.order.created",
  "event_version": 1,
  "producer_name": "sales-service-api",
  "occurred_at": "2026-06-09T05:15:30Z",
  "correlation_id": "tx-8890-adc-09",
  "data": {
    "order_id": "ORD-77621",
    "customer_id": "CUST-992",
    "total_amount": 150000.00,
    "items": [
      {
        "sku": "SKU-9982",
        "quantity": 2,
        "price": 75000.00
      }
    ]
  }
}

Penjelasan Atribut Metadata: #

• event_id: String UUID unik untuk setiap pesan. Ini wajib digunakan oleh konsumen untuk melakukan pengecekan de-duplikasi data (jaminan idempotensi).
• event_version: Integer penunjuk versi skema data. Membantu konsumen mengidentifikasi versi parser yang harus digunakan.
• occurred_at: Timestamp kapan peristiwa ini terjadi di dunia nyata, diformat menggunakan standar ISO 8601 UTC.
• correlation_id: Identifier transaksi global. Nilai ini harus diteruskan di setiap rantai panggilan asinkron untuk mempermudah pelacakan alur data lintas microservices menggunakan log aggregator (seperti ELK Stack atau Jaeger Tracing).

Pola Desain: Pemisahan Database Entity dengan Event DTO #

Salah satu kesalahan desain terburuk yang paling sering dilakukan oleh tim developer adalah mempublikasikan model database ORM secara langsung ke dalam payload RabbitMQ. Sebagai contoh, tim Layanan Pengguna menyerialisasikan langsung struct database User (yang dipetakan menggunakan GORM/Hibernate) ke format JSON dan mengirimkannya ke broker.

Mengapa Pola Ini Sangat Berbahaya? #

1. Kebocoran Detail Internal (Internal Leaking): Struktur tabel database internal kita terekspos secara bebas ke layanan luar.
2. Monolithic Coupling (Keterikatan Erat): Jika tim database ingin mengubah struktur tabel (misalnya mengubah tipe field database, menghapus kolom yang tidak terpakai, atau melakukan normalisasi tabel), perubahan tersebut otomatis merusak seluruh layanan konsumen eksternal yang membaca pesan. Hal ini menghancurkan esensi independensi microservices.

Pola Solusi yang Benar: Event Data Transfer Object (DTO) #

Kita harus menerapkan pemisahan yang tegas. Model database internal bersifat privat untuk layanan itu sendiri. Ketika terjadi perubahan state, aplikasi harus memetakan (mapping) data database tersebut ke dalam struct Event DTO khusus yang didesain secara minimalis sesuai kebutuhan konsumen publik, baru kemudian mempublikasikannya ke broker.

flowchart LR
    Database[(Database Transaksi)] -->|"1. Query Internal Model"| ServiceApp["Aplikasi Service"]
    ServiceApp -->|"2. Map Database Model ke Event DTO"| EventDTO["Event DTO Struct"]
    EventDTO -->|"3. Serialisasi & Kirim"| RabbitMQ(("RabbitMQ Broker"))
    
    style Database stroke:#0288d1,stroke-width:2px
    style ServiceApp stroke:#7b1fa2,stroke-width:2px
    style RabbitMQ stroke:#388e3c,stroke-width:2px

Strategi Versioning Skema Event #

Seiring berkembangnya bisnis, skema data pesan pasti akan mengalami modifikasi. Kita membutuhkan strategi pengelolaan versi (versioning strategy) yang matang agar tidak memicu kerusakan sistem:

1. Evolusi yang Kompatibel ke Belakang (Backward Compatible) #

Sebisa mungkin, rancang perubahan skema agar bersifat kompatibel ke belakang. Konsumen lama harus tetap dapat membaca pesan format baru tanpa crash.

• Aturan Tambah: Kita hanya diperbolehkan menambahkan field baru yang bersifat opsional. Konsumen lama akan mengabaikan field baru ini secara otomatis.
• Aturan Larang: Kita dilarang keras menghapus field lama yang bersifat wajib, dan dilarang mengubah tipe data field yang sudah ada (misalnya mengubah customer_id dari tipe integer menjadi string UUID).

2. Versioning Lewat Routing Key untuk Breaking Changes #

Jika kita terpaksa harus melakukan perubahan yang merusak kompatibilitas (breaking changes), kita harus mendeklarasikan versi skema baru secara eksplisit melalui routing key dan memelihara kedua versi tersebut secara berdampingan selama masa transisi (deprecation period).

• Pesan Versi Lama: Dipublikasikan dengan routing key v1.sales.order.created.
• Pesan Versi Baru: Dipublikasikan dengan routing key v2.sales.order.created.

Layanan konsumen yang belum siap diperbarui dapat tetap mendengarkan antrean yang terikat ke v1, sementara layanan konsumen baru dapat terhubung ke antrean v2. Setelah seluruh konsumen bermigrasi ke v2, produsen dapat menghentikan pengiriman versi v1 secara bersih.

Implementasi Validasi JSON Schema pada Aplikasi Go #

Untuk memastikan tidak ada produsen yang mengirimkan data cacat ke broker, kita harus melakukan validasi skema (schema validation) di runtime sebelum pesan dipublikasikan. Pilihan terbaik untuk skema JSON adalah menggunakan standar JSON Schema.

Berikut adalah implementasi lengkap di Go menggunakan library xeipuuv/gojsonschema untuk memvalidasi payload pesan sebelum dikirimkan ke RabbitMQ:

package main

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
	"github.com/xeipuuv/gojsonschema"
)

const (
	amqpURI      = "amqp://admin:securepassword@rabbitmq-cluster:5672/"
	exchangeName = "exchange.sales.direct"
	routingKey   = "sales.order.created"
)

// StandardEnvelope mewakili struktur amplop pesan universal
type StandardEnvelope struct {
	EventID       string      `json:"event_id"`
	EventType     string      `json:"event_type"`
	EventVersion  int         `json:"event_version"`
	ProducerName  string      `json:"producer_name"`
	OccurredAt    string      `json:"occurred_at"`
	CorrelationID string      `json:"correlation_id"`
	Data          interface{} `json:"data"` // Data bisnis dinamis
}

// OrderCreatedPayload mewakili data bisnis khusus untuk event order created
type OrderCreatedPayload struct {
	OrderID     string  `json:"order_id"`
	CustomerID  string  `json:"customer_id"`
	TotalAmount float64 `json:"total_amount"`
}

// JSON Schema definisi kontrak pesan dalam format string JSON
const orderCreatedSchema = `{
	"$schema": "http://json-schema.org/draft-07/schema#",
	"title": "OrderCreatedEvent",
	"type": "object",
	"properties": {
		"event_id": {"type": "string", "format": "uuid"},
		"event_type": {"type": "string"},
		"event_version": {"type": "integer", "minimum": 1},
		"producer_name": {"type": "string"},
		"occurred_at": {"type": "string", "format": "date-time"},
		"correlation_id": {"type": "string"},
		"data": {
			"type": "object",
			"properties": {
				"order_id": {"type": "string"},
				"customer_id": {"type": "string"},
				"total_amount": {"type": "number", "minimum": 0}
			},
			"required": ["order_id", "customer_id", "total_amount"]
		}
	},
	"required": ["event_id", "event_type", "event_version", "producer_name", "occurred_at", "correlation_id", "data"]
}`

type MessagePublisher struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	schema  *gojsonschema.Schema
}

func (mp *MessagePublisher) Init() error {
	var err error
	mp.conn, err = amqp.Dial(amqpURI)
	if err != nil {
		return err
	}

	mp.channel, err = mp.conn.Channel()
	if err != nil {
		mp.conn.Close()
		return err
	}

	// Load dan kompilasi JSON Schema kontrak di awal startup
	schemaLoader := gojsonschema.NewStringLoader(orderCreatedSchema)
	mp.schema, err = gojsonschema.NewSchema(schemaLoader)
	if err != nil {
		mp.channel.Close()
		mp.conn.Close()
		return err
	}

	return nil
}

// ValidateAndPublish memvalidasi payload berdasarkan kontrak sebelum mengirimkan ke broker
func (mp *MessagePublisher) ValidateAndPublish(ctx context.Context, envelope StandardEnvelope) error {
	// 1. Serialisasi envelope ke format JSON bytes
	jsonBytes, err := json.Marshal(envelope)
	if err != nil {
		return err
	}

	// 2. Lakukan validasi terhadap JSON Schema
	documentLoader := gojsonschema.NewBytesLoader(jsonBytes)
	result, err := mp.schema.Validate(documentLoader)
	if err != nil {
		return err
	}

	// Jika validasi gagal, tolak pengiriman sejak di sisi aplikasi produsen
	if !result.Valid() {
		log.Println("[✗ VALIDASI GAGAL] Payload melanggar kontrak pesan:")
		for _, desc := range result.Errors() {
			log.Printf("  - %s\n", desc.String())
		}
		return errors.New("payload melanggar spesifikasi kontrak skema JSON")
	}

	log.Println("[✓ VALIDASI SUKSES] Payload memenuhi standar kontrak pesan.")

	// 3. Publikasikan ke RabbitMQ jika validasi lulus
	return mp.channel.PublishWithContext(ctx,
		exchangeName,
		routingKey,
		false,
		false,
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent,
			Body:         jsonBytes,
			CorrelationId: envelope.CorrelationID,
		},
	)
}

func (mp *MessagePublisher) Close() {
	if mp.channel != nil {
		mp.channel.Close()
	}
	if mp.conn != nil {
		mp.conn.Close()
	}
}

func main() {
	pub := &MessagePublisher{}
	err := pub.Init()
	if err != nil {
		log.Fatalf("Inisialisasi publisher gagal: %v\n", err)
	}
	defer pub.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// 1. Uji Coba Skenario Payload yang BENAR
	validOrder := StandardEnvelope{
		EventID:       "6ba7b810-9dad-11d1-80b4-00c04fd430c8", // Valid UUID
		EventType:     "sales.order.created",
		EventVersion:  1,
		ProducerName:  "sales-service",
		OccurredAt:    time.Now().UTC().Format(time.RFC3339), // Valid date-time format
		CorrelationID: "corr-12345",
		Data: OrderCreatedPayload{
			OrderID:     "ORD-99801",
			CustomerID:  "CUST-880",
			TotalAmount: 250000.00,
		},
	}

	log.Println("Menguji pengiriman payload valid...")
	err = pub.ValidateAndPublish(ctx, validOrder)
	if err != nil {
		log.Printf("Gagal mempublikasikan: %v\n", err)
	} else {
		log.Println("Pesan valid berhasil terkirim ke broker.")
	}

	log.Println("--------------------------------------------------")

	// 2. Uji Coba Skenario Payload yang SALAH (Melanggar Kontrak)
	invalidOrder := StandardEnvelope{
		EventID:       "bukan-uuid", // SALAH: format UUID tidak valid
		EventType:     "sales.order.created",
		EventVersion:  1,
		ProducerName:  "sales-service",
		OccurredAt:    "bukan-tanggal-format-iso", // SALAH: format timestamp rusak
		CorrelationID: "corr-12345",
		Data: OrderCreatedPayload{
			OrderID:     "ORD-99801",
			CustomerID:  "", // SALAH: customer ID kosong
			TotalAmount: -15000.00, // SALAH: total amount bernilai negatif (melanggar minimum 0)
		},
	}

	log.Println("Menguji pengiriman payload tidak valid...")
	err = pub.ValidateAndPublish(ctx, invalidOrder)
	if err != nil {
		log.Printf("[PENCEGAHAN] Pengiriman diblokir oleh aplikasi: %v\n", err)
	} else {
		log.Println("Pesan terkirim (ini seharusnya tidak terjadi).")
	}
}

Perbandingan: Tanpa Kontrak vs Dengan Kontrak Disiplin #

Berikut adalah tabel matriks evaluasi arsitektural antara sistem yang tidak menerapkan disiplin kontrak pesan dengan sistem yang patuh pada tata kelola kontrak skema:

Ringkasan #

• API Publik Asinkron — Anggap setiap event asinkron di RabbitMQ sebagai kontrak API publik. Terapkan tingkat kedisiplinan desain yang sama seperti pada REST atau gRPC.
• Gunakan Routing Key Baku — Terapkan konvensi penamaan bertingkat menggunakan format taksonomi Domain.Subdomain.Entity.Action dengan huruf kecil terpisah titik.
• Bungkus dengan Envelope — Selalu gunakan envelope pesan standar yang membawa metadata administratif (event_id, occurred_at, correlation_id) berdampingan dengan data bisnis.
• Pisahkan Entity DB dengan DTO — Jangan pernah menyiarkan model database ORM internal langsung ke broker. Gunakan Event DTO khusus sebagai sekat pembatas domain.
• Validasi Sebelum Mengirim — Kompilasi skema kontrak menggunakan JSON Schema, dan lakukan validasi payload di runtime sisi produsen sebelum pesan dialirkan ke jaringan broker.

← Sebelumnya: Monitoring