Stage 12 - Events with Kafka

func(msg []byte) error

  redpanda:
    image: redpandadata/redpanda:v24.2.18
    container_name: minimart-redpanda
    command:
      - redpanda
      - start
      - --mode=dev-container
      - --kafka-addr=PLAINTEXT://0.0.0.0:9092
      - --advertise-kafka-addr=PLAINTEXT://localhost:9092
    ports:
      - "9092:9092"

// KafkaBrokers is the comma-separated list of Kafka bootstrap brokers.
func KafkaBrokers() []string {
	return strings.Split(env("KAFKA_BROKERS", "localhost:9092"), ",")
}

# Kafka bootstrap brokers (comma-separated). Redpanda speaks the Kafka protocol.
KAFKA_BROKERS="localhost:9092"

docker compose up -d

docker compose ps

docker compose exec redpanda rpk topic list
# (no topics yet)

// Package kafka is MiniMart's thin wrapper over sarama + watermill. Domains depend
// only on the small Publisher interface and the func(msg []byte) error consumer
// signature - never on sarama or watermill directly.
package kafka

// Topic is a Kafka topic name.
type Topic string

func (t Topic) String() string { return string(t) }

// Topics used by MiniMart.
const (
	OrderPlacedTopic Topic = "order-placed"
)

// Consumer groups. Two different groups subscribed to the same topic EACH receive
// every message - that is how one published event fans out to several independent
// handlers. Here one OrderPlaced event drives both an order-side confirmation and
// a notification email.
const (
	OrderConfirmationConsumer = "order-confirmation"
	NotificationConsumer      = "order-notification"
)

// Package message holds the event payloads published across domains. An event is
// a published contract, like an HTTP response body: keep it small and stable, and
// change it only in backward-compatible ways.
package message

// OrderPlaced is emitted after an order is successfully saved. Consumers in other
// domains react to it without the order domain knowing they exist.
type OrderPlaced struct {
	OrderID    string `json:"order_id"`
	UserID     string `json:"user_id"`
	TotalCents int64  `json:"total_cents"`
	ItemCount  int    `json:"item_count"`
}

package kafka

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

	"github.com/IBM/sarama"
	"github.com/ThreeDotsLabs/watermill"
	wkafka "github.com/ThreeDotsLabs/watermill-kafka/v3/pkg/kafka"
	"github.com/ThreeDotsLabs/watermill/message"
)

// Publisher is the output port the application depends on to emit events. The
// application imports this interface, not sarama - so a test can pass a fake.
type Publisher interface {
	PublishMessage(ctx context.Context, topic Topic, key string, payload interface{}) error
}

// Producer is the Kafka implementation of Publisher.
type Producer struct {
	pub message.Publisher
}

func NewProducer(brokers []string) (*Producer, error) {
	cfg := wkafka.DefaultSaramaSyncPublisherConfig()
	// V2_1_0_0 keeps the client on non-flexible protocol encodings, which every
	// Kafka-compatible broker (including Redpanda in dev mode) accepts.
	cfg.Version = sarama.V2_1_0_0

	pub, err := wkafka.NewPublisher(
		wkafka.PublisherConfig{
			Brokers:               brokers,
			Marshaler:             wkafka.NewWithPartitioningMarshaler(partitionByKey),
			OverwriteSaramaConfig: cfg,
		},
		watermill.NewStdLogger(false, false),
	)
	if err != nil {
		return nil, fmt.Errorf("new kafka publisher: %w", err)
	}
	return &Producer{pub: pub}, nil
}

// partitionByKey sends all messages sharing a key to the same partition, so events
// for one order stay ordered.
func partitionByKey(topic string, msg *message.Message) (string, error) {
	return msg.Metadata.Get("key"), nil
}

func (p *Producer) PublishMessage(ctx context.Context, topic Topic, key string, payload interface{}) error {
	var body []byte
	switch v := payload.(type) {
	case []byte:
		body = v
	default:
		var err error
		body, err = json.Marshal(payload)
		if err != nil {
			return fmt.Errorf("marshal event: %w", err)
		}
	}
	msg := message.NewMessage(watermill.NewUUID(), body)
	msg.SetContext(ctx)
	msg.Metadata.Set("key", key)
	if err := p.pub.Publish(topic.String(), msg); err != nil {
		return fmt.Errorf("publish to %s: %w", topic, err)
	}
	return nil
}

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

go mod tidy      # resolves sarama + watermill; updates go.sum
go build ./pkg/...

package application

import (
	"minimart/order/domain"
	"minimart/pkg/kafka"
	"minimart/pkg/module/contracts"
)

// Service is the order application layer. It depends on the order repository port,
// on other domains through the contracts.Modules INTERFACE, and on the Kafka
// Publisher INTERFACE - never on concrete infrastructure. Depending on interfaces
// is what lets tests pass fakes for all three.
type Service struct {
	repo      domain.Repository
	modules   contracts.Modules
	publisher kafka.Publisher
}

func NewService(repo domain.Repository, modules contracts.Modules, publisher kafka.Publisher) *Service {
	return &Service{repo: repo, modules: modules, publisher: publisher}
}

package application

import (
	"context"
	"fmt"
	"log"

	"minimart/order/domain"
	"minimart/pkg/kafka"
	"minimart/pkg/message"
)

// PlaceOrderRequest is the input to PlaceOrder. It carries NO prices or names:
// those are authoritative from the catalog, fetched below. The client only says
// which product and how many.
type PlaceOrderRequest struct {
	UserID string
	Items  []PlaceOrderItem
}

type PlaceOrderItem struct {
	ProductID string
	Quantity  int
}

// PlaceOrder resolves each product's real price from the catalog (through the
// published contract), builds and saves the order, then publishes an OrderPlaced
// event for other domains to react to.
func (s *Service) PlaceOrder(ctx context.Context, req PlaceOrderRequest) (domain.Order, error) {
	if len(req.Items) == 0 {
		return domain.Order{}, domain.ErrNoItems
	}
	items := make([]domain.Item, 0, len(req.Items))
	for _, it := range req.Items {
		if it.Quantity <= 0 {
			return domain.Order{}, domain.ErrInvalidQuantity
		}
		p, err := s.modules.GetProduct(ctx, it.ProductID)
		if err != nil {
			return domain.Order{}, fmt.Errorf("resolve product %s: %w", it.ProductID, err)
		}
		items = append(items, domain.Item{
			ProductID:      p.ID,
			Name:           p.Name,
			UnitPriceCents: p.PriceCents,
			Quantity:       it.Quantity,
		})
	}
	order, err := domain.NewOrder(newID(), req.UserID, items)
	if err != nil {
		return domain.Order{}, err
	}
	if err := s.repo.Save(ctx, order); err != nil {
		return domain.Order{}, err
	}

	// The order is committed. Publishing is best-effort: a broker hiccup must not
	// fail a request whose order already exists, so we log and continue. (A system
	// needing an at-least-once guarantee writes the event to a transactional outbox
	// in the same DB transaction and relays it separately.)
	if err := s.publisher.PublishMessage(ctx, kafka.OrderPlacedTopic, order.ID, message.OrderPlaced{
		OrderID:    order.ID,
		UserID:     order.UserID,
		TotalCents: order.TotalCents,
		ItemCount:  len(order.Items),
	}); err != nil {
		log.Printf("order %s: publish %s failed: %v", order.ID, kafka.OrderPlacedTopic, err)
	}
	return order, nil
}

go build ./order/...

// Package event holds the order domain's own message input adapters. Consuming an
// event your own domain published is a normal pattern - here it stands in for a
// follow-up step like starting fulfilment.
package event

import (
	"encoding/json"
	"log"

	"minimart/pkg/message"
)

type Handler struct{}

func NewHandler() *Handler { return &Handler{} }

// OrderConfirmation reacts to the order-placed topic. It logs a confirmation; a
// real system might reserve stock or start fulfilment.
func (h *Handler) OrderConfirmation(msg []byte) error {
	var event message.OrderPlaced
	if err := json.Unmarshal(msg, &event); err != nil {
		return err
	}
	log.Printf("[order] confirming order %s for user %s (%d cents, %d items)",
		event.OrderID, event.UserID, event.TotalCents, event.ItemCount)
	return nil
}

package message

import (
	"context"

	"minimart/order/presentation/message/event"
	"minimart/pkg/kafka"
	"minimart/pkg/worker"
)

// Workers registers the order domain's Kafka consumers onto the worker.
func Workers(ctx context.Context, w *worker.Worker) {
	h := event.NewHandler()
	w.AttachKafka(ctx, kafka.OrderPlacedTopic, kafka.OrderConfirmationConsumer, h.OrderConfirmation)
}

// Package application is the notification domain's use-case layer. In a real
// system it would send email or push notifications; here it logs, standing in for
// any side effect a downstream domain performs in reaction to an event.
package application

import (
	"context"
	"log"
)

type Service struct{}

func NewService() *Service { return &Service{} }

func (s *Service) SendOrderConfirmation(ctx context.Context, orderID, userID string, totalCents int64) error {
	log.Printf("[notification] emailing user %s: order %s confirmed, total %d cents",
		userID, orderID, totalCents)
	return nil
}

// Package event holds the notification domain's message input adapters. A handler
// here is structurally identical to an HTTP controller: it deserializes a transport
// payload (JSON bytes off a Kafka topic) into a request and calls the application
// service. It knows nothing about Kafka.
package event

import (
	"context"
	"encoding/json"

	"minimart/notification/application"
	"minimart/pkg/message"
)

type Handler struct {
	svc *application.Service
}

func NewHandler(svc *application.Service) *Handler { return &Handler{svc: svc} }

// OrderPlaced is the input adapter for the order-placed topic.
func (h *Handler) OrderPlaced(msg []byte) error {
	var event message.OrderPlaced
	if err := json.Unmarshal(msg, &event); err != nil {
		return err
	}
	return h.svc.SendOrderConfirmation(context.Background(), event.OrderID, event.UserID, event.TotalCents)
}

package message

import (
	"context"

	"minimart/notification/application"
	"minimart/notification/presentation/message/event"
	"minimart/pkg/kafka"
	"minimart/pkg/worker"
)

// Workers registers the notification domain's Kafka consumers onto the worker.
func Workers(ctx context.Context, w *worker.Worker, svc *application.Service) {
	h := event.NewHandler(svc)
	w.AttachKafka(ctx, kafka.OrderPlacedTopic, kafka.NotificationConsumer, h.OrderPlaced)
}

package notification

import (
	"context"
	"net/http"

	"minimart/notification/application"
	"minimart/notification/presentation/message"
	"minimart/pkg/module"
	"minimart/pkg/module/contracts"
	"minimart/pkg/worker"
)

// Module is a worker-only domain: it registers no HTTP routes and publishes no
// contract. It exists to react to events, demonstrating that a consumer is a
// first-class input adapter.
type Module struct {
	svc *application.Service
}

var _ module.Module = (*Module)(nil)

func NewModule() *Module { return &Module{} }

func (m *Module) Name() string { return "notification" }

func (m *Module) Init(ctx context.Context) error {
	m.svc = application.NewService()
	return nil
}

func (m *Module) RegisterContracts(ctx context.Context, c *contracts.Contracts) error { return nil }

func (m *Module) RegisterHTTP(ctx context.Context, mux *http.ServeMux) error { return nil }

func (m *Module) RegisterWorkers(ctx context.Context, w *worker.Worker) error {
	message.Workers(ctx, w, m.svc)
	return nil
}

go build ./order/presentation/message/event ./notification/application ./notification/presentation/message/event

package kafka

import (
	"context"
	"fmt"

	"github.com/IBM/sarama"
	"github.com/ThreeDotsLabs/watermill"
	wkafka "github.com/ThreeDotsLabs/watermill-kafka/v3/pkg/kafka"
)

// Consumer subscribes to topics and dispatches raw message bodies to handlers.
type Consumer struct {
	brokers []string
	dlq     Publisher // optional; failed messages are routed here
	logger  watermill.LoggerAdapter
}

func NewConsumer(brokers []string, dlq Publisher) *Consumer {
	return &Consumer{
		brokers: brokers,
		dlq:     dlq,
		logger:  watermill.NewStdLogger(false, false),
	}
}

// Subscribe joins consumerGroup on topic and calls handler for each message body.
// It blocks until ctx is cancelled. The handler signature - func(msg []byte) error
// - is deliberately the same shape as reading an HTTP request body: a queue message
// is just another input.
//
// A handler error routes the message to a dead-letter topic (<topic>.dlq) if a DLQ
// publisher is configured, then acknowledges it so one poison message cannot block
// the partition forever.
func (c *Consumer) Subscribe(ctx context.Context, topic Topic, consumerGroup string, handler func(msg []byte) error) error {
	cfg := wkafka.DefaultSaramaSubscriberConfig()
	// Same version pin as the producer - see producer.go.
	cfg.Version = sarama.V2_1_0_0
	cfg.Consumer.Offsets.Initial = sarama.OffsetOldest

	sub, err := wkafka.NewSubscriber(
		wkafka.SubscriberConfig{
			Brokers:               c.brokers,
			Unmarshaler:           wkafka.DefaultMarshaler{},
			OverwriteSaramaConfig: cfg,
			ConsumerGroup:         consumerGroup,
		},
		c.logger,
	)
	if err != nil {
		return fmt.Errorf("new kafka subscriber: %w", err)
	}
	defer sub.Close()

	messages, err := sub.Subscribe(ctx, topic.String())
	if err != nil {
		return fmt.Errorf("subscribe %s: %w", topic, err)
	}

	for msg := range messages {
		if err := handler(msg.Payload); err != nil {
			c.logger.Error("handler failed; routing to DLQ", err, watermill.LogFields{
				"topic": topic.String(), "group": consumerGroup,
			})
			if c.dlq != nil {
				_ = c.dlq.PublishMessage(ctx, Topic(topic.String()+".dlq"), msg.Metadata.Get("key"), msg.Payload)
			}
		}
		msg.Ack()
	}
	return nil
}

// Package worker runs a domain's Kafka consumers. It is the worker-process
// equivalent of an http.ServeMux: modules register handlers onto it during the
// RegisterWorkers boot phase, and each handler runs as one consumer-group loop.
package worker

import (
	"context"
	"log"
	"sync"

	"minimart/pkg/kafka"
)

type Worker struct {
	consumer *kafka.Consumer
	producer kafka.Publisher
	wg       sync.WaitGroup
}

func New(consumer *kafka.Consumer, producer kafka.Publisher) *Worker {
	return &Worker{consumer: consumer, producer: producer}
}

// Producer exposes the shared publisher to modules that also emit events.
func (w *Worker) Producer() kafka.Publisher { return w.producer }

// AttachKafka registers one consumer: a handler for a topic under a consumer
// group. Each runs in its own goroutine until ctx is cancelled.
func (w *Worker) AttachKafka(ctx context.Context, topic kafka.Topic, group string, handler func(msg []byte) error) {
	w.wg.Add(1)
	go func() {
		defer w.wg.Done()
		if err := w.consumer.Subscribe(ctx, topic, group, handler); err != nil {
			log.Printf("worker: %s/%s stopped: %v", topic, group, err)
		}
	}()
}

// Wait blocks until every consumer goroutine has exited.
func (w *Worker) Wait() { w.wg.Wait() }

package module

import (
	"context"
	"net/http"

	"minimart/pkg/module/contracts"
	"minimart/pkg/worker"
)

// Module is one self-contained domain in the modular monolith.
//
// The boot sequence in cmd/ calls these methods in phases, for every module:
//
//  1. Init             build the module's own internals (service, adapters).
//  2. RegisterContracts publish this module's contract into the shared registry.
//  3. (Verify)         the boot fails if any contract slot is still nil.
//     4a. RegisterHTTP    (server) wire this module's routes onto the mux.
//     4b. RegisterWorkers (worker) wire this module's Kafka consumers onto the worker.
//
// The server process runs phase 4a; the worker process runs phase 4b. No module
// may call another module before phase 3 completes.
type Module interface {
	Name() string
	Init(ctx context.Context) error
	RegisterContracts(ctx context.Context, c *contracts.Contracts) error
	RegisterHTTP(ctx context.Context, mux *http.ServeMux) error
	RegisterWorkers(ctx context.Context, w *worker.Worker) error
}

// RegisterWorkers is a no-op: the catalog domain has no Kafka consumers.
func (m *Module) RegisterWorkers(ctx context.Context, w *worker.Worker) error { return nil }

package order

import (
	"context"
	"net/http"

	"github.com/jackc/pgx/v5/pgxpool"

	"minimart/order/application"
	"minimart/order/infrastructure/postgres"
	orderhttp "minimart/order/presentation/http"
	ordermessage "minimart/order/presentation/message"
	"minimart/pkg/kafka"
	"minimart/pkg/module"
	"minimart/pkg/module/contracts"
	"minimart/pkg/worker"
)

// Module wires the order domain. It holds the shared contracts registry (to call
// other domains), the DB pools, and the Kafka publisher.
type Module struct {
	registry  *contracts.Contracts
	readPool  *pgxpool.Pool
	writePool *pgxpool.Pool
	publisher kafka.Publisher
	svc       *application.Service
	handlers  *orderhttp.Handlers
}

var _ module.Module = (*Module)(nil)

func NewModule(registry *contracts.Contracts, readPool, writePool *pgxpool.Pool, publisher kafka.Publisher) *Module {
	return &Module{registry: registry, readPool: readPool, writePool: writePool, publisher: publisher}
}

func (m *Module) Name() string { return "order" }

func (m *Module) Init(ctx context.Context) error {
	repo := postgres.NewOrderRepository(m.readPool, m.writePool)
	m.svc = application.NewService(repo, m.registry, m.publisher)
	m.handlers = orderhttp.NewHandlers(m.svc)
	return nil
}

// RegisterContracts is a no-op: the order domain publishes no contract of its own.
func (m *Module) RegisterContracts(ctx context.Context, c *contracts.Contracts) error { return nil }

func (m *Module) RegisterHTTP(ctx context.Context, mux *http.ServeMux) error {
	m.handlers.Register(mux)
	return nil
}

// RegisterWorkers wires the order domain's Kafka consumers.
func (m *Module) RegisterWorkers(ctx context.Context, w *worker.Worker) error {
	ordermessage.Workers(ctx, w)
	return nil
}

package cmd

import (
	"context"
	"log"

	"minimart/catalog"
	"minimart/conf"
	"minimart/notification"
	"minimart/order"
	kafkapkg "minimart/pkg/kafka"
	"minimart/pkg/module"
	"minimart/pkg/module/contracts"
	"minimart/pkg/worker"
)

// StartWorker runs the background consumers. It boots the same modules as the
// server, but wires their Kafka handlers (RegisterWorkers) instead of HTTP routes,
// then blocks while the consumer goroutines run.
func StartWorker() error {
	ctx := context.Background()

	readPool, writePool, err := newPools(ctx)
	if err != nil {
		return err
	}
	defer writePool.Close()
	defer readPool.Close()

	producer, err := kafkapkg.NewProducer(conf.KafkaBrokers())
	if err != nil {
		return err
	}
	defer producer.Close()

	// The producer doubles as the dead-letter publisher for failed messages.
	consumer := kafkapkg.NewConsumer(conf.KafkaBrokers(), producer)
	w := worker.New(consumer, producer)

	registry := &contracts.Contracts{}
	modules := []module.Module{
		catalog.NewModule(readPool, writePool),
		order.NewModule(registry, readPool, writePool, producer),
		notification.NewModule(),
	}

	for _, m := range modules {
		if err := m.Init(ctx); err != nil {
			return err
		}
	}
	for _, m := range modules {
		if err := m.RegisterContracts(ctx, registry); err != nil {
			return err
		}
	}
	if err := registry.Verify(); err != nil {
		return err
	}
	for _, m := range modules {
		if err := m.RegisterWorkers(ctx, w); err != nil {
			return err
		}
	}

	log.Println("minimart worker started; consuming events")
	w.Wait()
	return nil
}

	// after building the pools:
	producer, err := kafkapkg.NewProducer(conf.KafkaBrokers())
	if err != nil {
		return err
	}
	defer producer.Close()

	// ...and pass it into the order module:
	order.NewModule(registry, readPool, writePool, producer),