Event-Driven Architecture #

// ✓ CORRECT: An event with a clear and stable structure
// Past-tense name, immutable, carries enough data

package events

import "time"

// OrderPlaced is published when an order is successfully created
// Version 1 — all fields backward compatible
type OrderPlaced struct {
	// Event metadata — required in all events
	EventID    string    `json:"event_id"`    // unique event ID (for idempotency)
	EventName  string    `json:"event_name"`  // "order.placed"
	Version    int       `json:"version"`     // 1
	OccurredAt time.Time `json:"occurred_at"`

	// Event payload
	OrderID    string      `json:"order_id"`
	CustomerID string      `json:"customer_id"`
	Items      []OrderItem `json:"items"`
	TotalCents int64       `json:"total_cents"`
	Currency   string      `json:"currency"`
}

type OrderItem struct {
	ProductID  string `json:"product_id"`
	Name       string `json:"name"`
	Quantity   int    `json:"quantity"`
	PriceCents int64  `json:"price_cents"`
}

// PaymentProcessed is published when a payment succeeds
type PaymentProcessed struct {
	EventID       string    `json:"event_id"`
	EventName     string    `json:"event_name"` // "payment.processed"
	Version       int       `json:"version"`
	OccurredAt    time.Time `json:"occurred_at"`

	PaymentID  string `json:"payment_id"`
	OrderID    string `json:"order_id"`
	AmountCents int64 `json:"amount_cents"`
	Method     string `json:"method"` // "credit_card", "bank_transfer"
}

// ✗ ANTI-PATTERN: an event with an imperative name and minimal data
type CreateOrder struct { // ✗ a command name, not an event
	ID string // ✗ only an ID — consumers must make API calls for details
}

// pkg/kafka/producer.go — generic Kafka producer
package kafka

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

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

// Producer is a type-safe Kafka writer wrapper
type Producer struct {
	writer *kafka.Writer
}

func NewProducer(brokers []string, topic string) *Producer {
	return &Producer{
		writer: &kafka.Writer{
			Addr:         kafka.TCP(brokers...),
			Topic:        topic,
			Balancer:     &kafka.LeastBytes{},
			WriteTimeout: 5 * time.Second,
			RequiredAcks: kafka.RequireAll, // ✓ wait for all replicas
		},
	}
}

// Publish publishes an event to Kafka
func (p *Producer) Publish(ctx context.Context, key string, event interface{}) error {
	payload, err := json.Marshal(event)
	if err != nil {
		return fmt.Errorf("failed to marshal event: %w", err)
	}

	err = p.writer.WriteMessages(ctx, kafka.Message{
		Key:   []byte(key),
		Value: payload,
		Headers: []kafka.Header{
			{Key: "content-type", Value: []byte("application/json")},
		},
	})
	if err != nil {
		return fmt.Errorf("failed to publish to kafka: %w", err)
	}
	return nil
}

func (p *Producer) Close() error {
	return p.writer.Close()
}

// internal/order/service.go — the Order Service publishes events
package order

import (
	"context"
	"time"

	"myapp/events"
	"myapp/pkg/kafka"
)

type EventPublisher interface {
	Publish(ctx context.Context, key string, event interface{}) error
}

type OrderService struct {
	repo      OrderRepository
	publisher EventPublisher
}

func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) (*Order, error) {
	order := buildOrder(input)

	if err := s.repo.Save(ctx, order); err != nil {
		return nil, err
	}

	// Publish the event AFTER it is successfully saved
	// ✓ Event Carried State Transfer — carries all the data consumers need
	event := events.OrderPlaced{
		EventID:    generateEventID(),
		EventName:  "order.placed",
		Version:    1,
		OccurredAt: time.Now(),
		OrderID:    order.ID,
		CustomerID: order.CustomerID,
		Items:      toEventItems(order.Items),
		TotalCents: order.TotalCents,
		Currency:   "IDR",
	}

	// Key = OrderID to ensure events for the same order go to the same partition
	// ✓ Preserves per-order event ordering
	if err := s.publisher.Publish(ctx, order.ID, event); err != nil {
		// Log the error but do not fail the order — use the outbox pattern for reliability
		slog.ErrorContext(ctx, "failed to publish OrderPlaced event",
			"order_id", order.ID,
			"error", err.Error(),
		)
	}

	return order, nil
}

// internal/notification/consumer.go — Notification Service consumer
package notification

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

	"github.com/segmentio/kafka-go"

	"myapp/events"
)

// ProcessedEventRepository stores events that have already been processed
type ProcessedEventRepository interface {
	IsProcessed(ctx context.Context, eventID string) (bool, error)
	MarkProcessed(ctx context.Context, eventID string) error
}

type Consumer struct {
	reader       *kafka.Reader
	notifier     Notifier
	processedRepo ProcessedEventRepository
}

func NewConsumer(brokers []string, topic, groupID string, notifier Notifier, processedRepo ProcessedEventRepository) *Consumer {
	return &Consumer{
		reader: kafka.NewReader(kafka.ReaderConfig{
			Brokers:        brokers,
			Topic:          topic,
			GroupID:        groupID,        // ✓ consumer group for load balancing
			MinBytes:       10e3,           // 10KB
			MaxBytes:       10e6,           // 10MB
			CommitInterval: time.Second,    // auto-commit every 1 second
		}),
		notifier:     notifier,
		processedRepo: processedRepo,
	}
}

// Start begins the consumer loop
func (c *Consumer) Start(ctx context.Context) error {
	for {
		msg, err := c.reader.FetchMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil // context cancelled — normal shutdown
			}
			slog.Error("failed to fetch message", "error", err)
			continue
		}

		if err := c.processMessage(ctx, msg); err != nil {
			slog.Error("failed to process message",
				"offset", msg.Offset,
				"error", err,
			)
			// Do not commit the offset — the message will be reprocessed
			continue
		}

		// Commit only after successful processing
		if err := c.reader.CommitMessages(ctx, msg); err != nil {
			slog.Error("failed to commit offset", "error", err)
		}
	}
}

func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) error {
	var event events.OrderPlaced
	if err := json.Unmarshal(msg.Value, &event); err != nil {
		// Invalid event — skip (no point retrying)
		slog.Warn("invalid event, skipping", "error", err)
		return nil
	}

	// ✓ IDEMPOTENCY CHECK — check whether the event was already processed
	processed, err := c.processedRepo.IsProcessed(ctx, event.EventID)
	if err != nil {
		return err
	}
	if processed {
		slog.Info("event already processed, skipping", "event_id", event.EventID)
		return nil // idempotent — no side effects on the second time
	}

	// Process the event
	if err := c.notifier.SendOrderConfirmation(ctx, event); err != nil {
		return err
	}

	// Mark as processed — AFTER success
	return c.processedRepo.MarkProcessed(ctx, event.EventID)
}

func (c *Consumer) Close() error {
	return c.reader.Close()
}

// pkg/kafka/dlq_consumer.go — Consumer with DLQ support
package kafka

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	kgo "github.com/segmentio/kafka-go"
)

type ProcessFunc func(ctx context.Context, msg kgo.Message) error

type DLQConsumer struct {
	reader      *kgo.Reader
	dlqWriter   *kgo.Writer // writer to the DLQ topic
	process     ProcessFunc
	maxRetries  int
}

func NewDLQConsumer(
	brokers []string,
	topic, groupID, dlqTopic string,
	process ProcessFunc,
	maxRetries int,
) *DLQConsumer {
	return &DLQConsumer{
		reader: kgo.NewReader(kgo.ReaderConfig{
			Brokers: brokers,
			Topic:   topic,
			GroupID: groupID,
		}),
		dlqWriter: &kgo.Writer{
			Addr:  kgo.TCP(brokers...),
			Topic: dlqTopic,
		},
		process:    process,
		maxRetries: maxRetries,
	}
}

func (c *DLQConsumer) Start(ctx context.Context) error {
	for {
		msg, err := c.reader.FetchMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			continue
		}

		var lastErr error
		for attempt := 1; attempt <= c.maxRetries; attempt++ {
			lastErr = c.process(ctx, msg)
			if lastErr == nil {
				break
			}

			slog.Warn("failed to process event, retrying",
				"attempt", attempt,
				"max_retries", c.maxRetries,
				"error", lastErr,
			)

			// Exponential backoff before retrying
			if attempt < c.maxRetries {
				select {
				case <-ctx.Done():
					return nil
				case <-time.After(time.Duration(attempt*attempt) * 100 * time.Millisecond):
				}
			}
		}

		if lastErr != nil {
			// ✓ Send to the DLQ after all retries fail
			dlqMsg := c.buildDLQMessage(msg, lastErr)
			if err := c.dlqWriter.WriteMessages(ctx, dlqMsg); err != nil {
				slog.Error("failed to send to DLQ", "error", err)
				continue // do not commit — try again later
			}
			slog.Error("event sent to DLQ",
				"topic", msg.Topic,
				"offset", msg.Offset,
				"error", lastErr,
			)
		}

		c.reader.CommitMessages(ctx, msg)
	}
}

func (c *DLQConsumer) buildDLQMessage(original kgo.Message, err error) kgo.Message {
	type DLQEnvelope struct {
		OriginalTopic   string    `json:"original_topic"`
		OriginalOffset  int64     `json:"original_offset"`
		OriginalPayload string    `json:"original_payload"`
		ErrorMessage    string    `json:"error_message"`
		FailedAt        time.Time `json:"failed_at"`
	}

	envelope, _ := json.Marshal(DLQEnvelope{
		OriginalTopic:   original.Topic,
		OriginalOffset:  original.Offset,
		OriginalPayload: string(original.Value),
		ErrorMessage:    err.Error(),
		FailedAt:        time.Now(),
	})

	return kgo.Message{
		Key:   original.Key,
		Value: envelope,
		Headers: append(original.Headers, kgo.Header{
			Key:   "x-dlq-reason",
			Value: []byte(err.Error()),
		}),
	}
}

// Outbox pattern: store the event in the database in the same transaction as the main operation
// A separate goroutine (relay) reads the outbox and publishes to the broker

// internal/order/service.go — with the outbox pattern
func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) (*Order, error) {
	order := buildOrder(input)
	event := buildOrderPlacedEvent(order)

	// ✓ One database transaction: save the order AND the event to the outbox
	err := s.db.WithTransaction(ctx, func(tx *sql.Tx) error {
		// Save the order
		if err := s.repo.SaveTx(ctx, tx, order); err != nil {
			return err
		}

		// Save the event to the outbox table — in the same transaction
		eventPayload, _ := json.Marshal(event)
		_, err := tx.ExecContext(ctx,
			`INSERT INTO outbox_events (id, topic, key, payload, created_at, published)
			 VALUES ($1, $2, $3, $4, NOW(), FALSE)`,
			event.EventID, "order.events", order.ID, eventPayload,
		)
		return err
	})
	if err != nil {
		return nil, err
	}

	return order, nil
}

// internal/outbox/relay.go — goroutine that publishes events from the outbox to Kafka
type OutboxRelay struct {
	db        *sql.DB
	publisher EventPublisher
	interval  time.Duration
}

func (r *OutboxRelay) Start(ctx context.Context) {
	ticker := time.NewTicker(r.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			r.publishPendingEvents(ctx)
		}
	}
}

func (r *OutboxRelay) publishPendingEvents(ctx context.Context) {
	// Fetch events that have not been published yet
	rows, err := r.db.QueryContext(ctx,
		`SELECT id, topic, key, payload FROM outbox_events
		 WHERE published = FALSE
		 ORDER BY created_at
		 LIMIT 100`,
	)
	if err != nil {
		return
	}
	defer rows.Close()

	for rows.Next() {
		var id, topic, key string
		var payload []byte
		rows.Scan(&id, &topic, &key, &payload)

		// Publish to Kafka
		if err := r.publisher.PublishRaw(ctx, topic, key, payload); err != nil {
			slog.Error("failed to publish outbox event", "id", id, "error", err)
			continue
		}

		// Mark as published
		r.db.ExecContext(ctx,
			`UPDATE outbox_events SET published = TRUE, published_at = NOW() WHERE id = $1`,
			id,
		)
	}
}

// events/order_placed.go — versioning event schemas
package events

// V1 — initial schema
type OrderPlacedV1 struct {
	EventID    string    `json:"event_id"`
	Version    int       `json:"version"`    // = 1
	OrderID    string    `json:"order_id"`
	CustomerID string    `json:"customer_id"`
	TotalCents int64     `json:"total_cents"`
	OccurredAt time.Time `json:"occurred_at"`
}

// V2 — adds new fields (backward compatible: new fields optional)
type OrderPlacedV2 struct {
	EventID      string    `json:"event_id"`
	Version      int       `json:"version"`    // = 2
	OrderID      string    `json:"order_id"`
	CustomerID   string    `json:"customer_id"`
	TotalCents   int64     `json:"total_cents"`
	OccurredAt   time.Time `json:"occurred_at"`
	// New fields — old consumers will ignore these
	DiscountCode string    `json:"discount_code,omitempty"`
	DeliveryAddr string    `json:"delivery_address,omitempty"`
}

// A consumer supporting multiple versions
func processOrderPlaced(ctx context.Context, payload []byte) error {
	// Decode the version first
	var meta struct {
		Version int `json:"version"`
	}
	json.Unmarshal(payload, &meta)

	switch meta.Version {
	case 1:
		var event OrderPlacedV1
		json.Unmarshal(payload, &event)
		return handleV1(ctx, event)
	case 2:
		var event OrderPlacedV2
		json.Unmarshal(payload, &event)
		return handleV2(ctx, event)
	default:
		slog.Warn("unknown event version", "version", meta.Version)
		return nil // skip — do not panic for unknown versions
	}
}

// ✗ Events sent BEFORE the database commit — can be inconsistent
func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) error {
	order := buildOrder(input)

	// ✗ Publish first, then save — if Save fails, the event was already sent
	s.publisher.Publish(ctx, order.ID, OrderPlaced{OrderID: order.ID})
	return s.repo.Save(ctx, order) // ✗ may fail after the event was sent
}

// ✓ Publish AFTER the database commit succeeds (or use the outbox pattern)
func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) error {
	order := buildOrder(input)
	if err := s.repo.Save(ctx, order); err != nil { // ✓ save first
		return err
	}
	s.publisher.Publish(ctx, order.ID, OrderPlaced{OrderID: order.ID}) // ✓ publish after success
	return nil
}

// ✗ A non-idempotent consumer — duplicate events cause double processing
func (c *Consumer) handleOrderPlaced(ctx context.Context, event OrderPlaced) error {
	// ✗ send the notification directly without checking for duplicates
	return c.mailer.Send(ctx, event.CustomerID, "Order Confirmed", buildBody(event))
}

// ✓ An idempotent consumer — check the event ID before processing
func (c *Consumer) handleOrderPlaced(ctx context.Context, event OrderPlaced) error {
	if ok, _ := c.repo.IsProcessed(ctx, event.EventID); ok {
		return nil // ✓ already processed, skip
	}
	if err := c.mailer.Send(ctx, event.CustomerID, "Order Confirmed", buildBody(event)); err != nil {
		return err
	}
	return c.repo.MarkProcessed(ctx, event.EventID) // ✓ mark after success
}

// ✗ Business logic in the broker or event schema
// An event used as a command (imperative) dictating what consumers must do
type ProcessPaymentCommand struct { // ✗ this is a command, not an event
	OrderID string
	Action  string // "process_payment" — logic in the event
}

// ✓ An event only states what happened — consumers decide how to react
type OrderPlaced struct { // ✓ past tense, facts, no instructions
	OrderID    string
	CustomerID string
	TotalCents int64
}