Message Contract #

In event-driven architectures, one of the most common perception errors among developer teams is treating RabbitMQ messages purely as free, dynamic JSON payloads. Because JSON formats naturally support flexible schema-less structures, developers often casually modify message data structures—changing field names, deleting elements, or swapping data types—without mature coordination with other service developer teams. This freedom often ends in runtime chaos in production, triggering mass panic when dozens of consumer instances crash in chains from being unable to parse unfamiliar new data formats.

Architecturally, every message sent through RabbitMQ isn’t just an ordinary data carrier, but a formal public API contract that’s asynchronous. This message contract connects various independent services loosely (loose coupling). Therefore, we must manage message contract schemas with the same discipline, governance, and standardization levels as HTTP REST APIs (using OpenAPI/Swagger) or gRPC APIs (using Protobuf). Maintaining this contract integrity is the key to preventing integration failures, stopping corrupted message pile-ups in Dead Letter Queues (DLQs), and enabling developer team scaling independently without locking each other.

Standardized Routing Key Naming Conventions #

The first step in composing disciplined message contract governance is determining routing key naming conventions. Routing keys are messages’ logical addresses determining where messages flow. Without standard rules, we’ll find confusing, inconsistent routing keys like kirim-data, user_update, or event123.

The best routing key naming rule uses a hierarchical format based on domain taxonomy:

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

Let’s break down every component of the structure above:

• [Domain]: The main business context name (e.g., sales, identity, inventory).
• [Subdomain]: The specific domain part below it (e.g., order, user, stock).
• [Entity]: The concrete business object name experiencing status changes (e.g., cart, profile, item).
• [Action]: A past-tense verb representing business facts that have occurred (e.g., created, updated, deleted, added).

Correct Implementation Examples: #

• sales.order.cart.added (The event when users add items to shopping carts).
• identity.user.profile.updated (The event when users update their profile information).
• inventory.stock.item.depleted (The event when warehouse item stock runs out).

Additional Rules: #

1. Use Lowercase: Always use lowercase letters to avoid case sensitivity across various client library makers.
2. Use Dots (.) as Separators: The dot character is the AMQP standard for dividing routing token levels, easing wildcard pattern matching on Topic Exchanges.

Standardized Payload Envelopes #

Messages sent between services must not only contain raw business data. We must wrap those payloads into a standard message envelope structure (Payload Envelope). This envelope separates system administrative information (metadata) from functional business data.

A healthy message envelope structure is divided into two main parts:

1. Metadata (Envelope): Contains message identity information for tracing, audit, and de-duplication needs.
2. Data (Payload): Specific objects containing original transaction data.

Here is an example of a standard JSON envelope schema that must be implemented across our entire microservices cluster:

{
  "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
      }
    ]
  }
}

Metadata Attribute Explanations: #

• event_id: A unique UUID string for every message. This must be used by consumers for data de-duplication checks (idempotency guarantees).
• event_version: An integer pointer to schema versions. Helps consumers identify which parser versions to use.
• occurred_at: A timestamp of when this event occurred in the real world, formatted using the ISO 8601 UTC standard.
• correlation_id: A global transaction identifier. This value must be forwarded in every asynchronous call chain to ease cross-microservice data flow tracking using log aggregators (like ELK Stacks or Jaeger Tracing).

Design Pattern: Separating Database Entities from Event DTOs #

One of the worst design errors most often made by developer teams is publishing ORM database models directly into RabbitMQ payloads. For example, the User Service team directly serializes the User database struct (mapped using GORM/Hibernate) into JSON formats and sends it to the broker.

Why Is This Pattern Very Dangerous? #

1. Internal Detail Leaks (Internal Leaking): Our internal database table structures are freely exposed to external services.
2. Monolithic Coupling: If database teams want to change table structures (e.g., changing database field types, deleting unused columns, or normalizing tables), those changes automatically break all external consumer services reading the messages. This destroys microservices independence essence.

The Correct Solution Pattern: Event Data Transfer Objects (DTOs) #

We must apply strict separation. Internal database models are private to the service itself. When state changes occur, applications must map that database data into special Event DTO structs minimally designed per public consumer needs, then publish them to the broker.

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

Event Schema Versioning Strategies #

As businesses grow, message data schemas will certainly undergo modifications. We need mature versioning strategy management so system damage isn’t triggered:

1. Backward Compatible Evolution #

As much as possible, design schema changes to be backward compatible. Old consumers must still be able to read new message formats without crashing.

• Addition Rules: We’re only allowed to add new optional fields. Old consumers automatically ignore these new fields.
• Prohibition Rules: We’re strictly forbidden from deleting old required fields, and forbidden from changing existing field data types (e.g., changing customer_id from integer to UUID string types).

2. Routing Key Versioning for Breaking Changes #

If we’re forced to make breaking changes, we must explicitly declare new schema versions through routing keys and maintain both versions side by side during deprecation periods.

• Old Version Messages: Published with the v1.sales.order.created routing key.
• New Version Messages: Published with the v2.sales.order.created routing key.

Consumer services not ready for updates can keep listening to queues bound to v1, while new consumer services can connect to v2 queues. After all consumers migrate to v2, producers can cleanly stop sending v1 versions.

JSON Schema Validation Implementations in Go Applications #

To ensure no producers send corrupted data to brokers, we must perform schema validation at runtime before messages are published. The best choice for JSON schemas is using the JSON Schema standard.

Here is a complete Go implementation using the xeipuuv/gojsonschema library to validate message payloads before sending them to RabbitMQ:

package main

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

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

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

// StandardEnvelope represents the universal message envelope structure
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"` // Dynamic business data
}

// OrderCreatedPayload represents the specific business data for order created events
type OrderCreatedPayload struct {
	OrderID     string  `json:"order_id"`
	CustomerID  string  `json:"customer_id"`
	TotalAmount float64 `json:"total_amount"`
}

// JSON Schema message contract definition in JSON string format
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 and compile the contract JSON Schema at 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 validates payloads against the contract before sending to the broker
func (mp *MessagePublisher) ValidateAndPublish(ctx context.Context, envelope StandardEnvelope) error {
	// 1. Serialize the envelope to JSON bytes format
	jsonBytes, err := json.Marshal(envelope)
	if err != nil {
		return err
	}

	// 2. Perform validation against the JSON Schema
	documentLoader := gojsonschema.NewBytesLoader(jsonBytes)
	result, err := mp.schema.Validate(documentLoader)
	if err != nil {
		return err
	}

	// If validation fails, reject deliveries from the producer application side
	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. Publish to RabbitMQ if validation passes
	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. Test the CORRECT Payload Scenario
	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. Test the WRONG Payload Scenario (Violating Contracts)
	invalidOrder := StandardEnvelope{
		EventID:       "bukan-uuid", // WRONG: invalid UUID format
		EventType:     "sales.order.created",
		EventVersion:  1,
		ProducerName:  "sales-service",
		OccurredAt:    "bukan-tanggal-format-iso", // WRONG: corrupted timestamp format
		CorrelationID: "corr-12345",
		Data: OrderCreatedPayload{
			OrderID:     "ORD-99801",
			CustomerID:  "", // WRONG: empty customer ID
			TotalAmount: -15000.00, // WRONG: negative total amount (violates 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).")
	}
}

Comparison: Without Contracts vs With Disciplined Contracts #

Here is an architectural evaluation matrix table between systems not applying message contract discipline and systems adhering to schema contract governance:

Summary #

• Asynchronous Public APIs — Consider every asynchronous event in RabbitMQ a public API contract. Apply the same design discipline levels as REST or gRPC.
• Use Standard Routing Keys — Apply hierarchical naming conventions using the Domain.Subdomain.Entity.Action taxonomy format with lowercase letters separated by dots.
• Wrap with Envelopes — Always use standard message envelopes carrying administrative metadata (event_id, occurred_at, correlation_id) alongside business data.
• Separate DB Entities from DTOs — Never broadcast internal ORM database models directly to brokers. Use special Event DTOs as domain boundary barriers.
• Validate Before Sending — Compile contract schemas using JSON Schema, and perform producer-side runtime payload validation before messages flow to broker networks.