CQRS Architecture #

// command/handler.go — all Commands and CommandHandlers
package command

import (
	"context"
	"errors"
	"time"
)

// PlaceOrderCommand is the Command to create a new order
// ✓ Command = instruction to change state
type PlaceOrderCommand struct {
	CustomerID string
	Items      []PlaceOrderItem
}

type PlaceOrderItem struct {
	ProductID  string
	Quantity   int
	PriceCents int64
}

// PlaceOrderResult is the minimal command result — only an ID, not domain data
type PlaceOrderResult struct {
	OrderID string
}

// OrderCommandHandler handles all order-related commands
type OrderCommandHandler struct {
	repo    OrderWriteRepository
	userSvc UserValidator
}

func NewOrderCommandHandler(repo OrderWriteRepository, userSvc UserValidator) *OrderCommandHandler {
	return &OrderCommandHandler{repo: repo, userSvc: userSvc}
}

// Handle processes PlaceOrderCommand — business validation lives here and in the domain
func (h *OrderCommandHandler) Handle(ctx context.Context, cmd PlaceOrderCommand) (*PlaceOrderResult, error) {
	// Validate command input
	if cmd.CustomerID == "" {
		return nil, errors.New("customer ID is required")
	}
	if len(cmd.Items) == 0 {
		return nil, errors.New("an order must have at least one item")
	}

	// Cross-service validation
	if err := h.userSvc.ValidateActive(ctx, cmd.CustomerID); err != nil {
		return nil, errors.New("invalid customer: " + err.Error())
	}

	// Create the domain entity — business rules inside the entity
	order, err := domain.NewOrder(cmd.CustomerID, toOrderItems(cmd.Items))
	if err != nil {
		return nil, err
	}

	// Save to the write model
	if err := h.repo.Save(ctx, order); err != nil {
		return nil, err
	}

	// ✓ A Command only returns an ID — it does not return full domain data
	return &PlaceOrderResult{OrderID: order.ID()}, nil
}

// CancelOrderCommand
type CancelOrderCommand struct {
	OrderID string
	Reason  string
}

func (h *OrderCommandHandler) HandleCancel(ctx context.Context, cmd CancelOrderCommand) error {
	order, err := h.repo.FindByID(ctx, cmd.OrderID)
	if err != nil {
		return err
	}

	if err := order.Cancel(cmd.Reason); err != nil {
		return err
	}

	return h.repo.Save(ctx, order)
}

// query/handler.go — all Queries and QueryHandlers
package query

import "context"

// GetOrderQuery is the Query to get one order
// ✓ Query = data read request
type GetOrderQuery struct {
	OrderID string
}

// OrderDetailView is the Read Model — optimized for display, not the domain
// May contain denormalized data (product names, customer names, etc.)
type OrderDetailView struct {
	OrderID      string          `json:"order_id"`
	CustomerName string          `json:"customer_name"`
	CustomerEmail string         `json:"customer_email"`
	Items        []OrderItemView `json:"items"`
	TotalCents   int64           `json:"total_cents"`
	Status       string          `json:"status"`
	CreatedAt    string          `json:"created_at"`
}

type OrderItemView struct {
	ProductName string `json:"product_name"`
	Quantity    int    `json:"quantity"`
	PriceCents  int64  `json:"price_cents"`
	SubtotalCents int64 `json:"subtotal_cents"`
}

// ListOrdersByCustomerQuery
type ListOrdersByCustomerQuery struct {
	CustomerID string
	Page       int
	PageSize   int
}

// OrderSummaryView is a compact Read Model for lists
type OrderSummaryView struct {
	OrderID    string `json:"order_id"`
	Status     string `json:"status"`
	TotalCents int64  `json:"total_cents"`
	ItemCount  int    `json:"item_count"`
	CreatedAt  string `json:"created_at"`
}

// OrderQueryHandler handles all order-related queries
type OrderQueryHandler struct {
	readRepo OrderReadRepository
}

func NewOrderQueryHandler(readRepo OrderReadRepository) *OrderQueryHandler {
	return &OrderQueryHandler{readRepo: readRepo}
}

// Handle processes GetOrderQuery — purely reading, no side effects
func (h *OrderQueryHandler) Handle(ctx context.Context, q GetOrderQuery) (*OrderDetailView, error) {
	return h.readRepo.FindDetailByID(ctx, q.OrderID)
}

// HandleList processes ListOrdersByCustomerQuery
func (h *OrderQueryHandler) HandleList(
	ctx context.Context,
	q ListOrdersByCustomerQuery,
) ([]*OrderSummaryView, int, error) {
	if q.Page < 1 {
		q.Page = 1
	}
	if q.PageSize < 1 || q.PageSize > 100 {
		q.PageSize = 20
	}
	return h.readRepo.FindSummariesByCustomer(ctx, q.CustomerID, q.Page, q.PageSize)
}

// infrastructure/write/postgres_order_repo.go — Write Repository
package write

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

	"myapp/domain"
)

type PostgresOrderWriteRepo struct {
	db *sql.DB
}

var _ query.OrderWriteRepository = (*PostgresOrderWriteRepo)(nil)

func (r *PostgresOrderWriteRepo) Save(ctx context.Context, order *domain.Order) error {
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	// Save the order to the write database (normalized)
	_, err = tx.ExecContext(ctx,
		`INSERT INTO orders (id, customer_id, status, total_cents, created_at)
		 VALUES ($1, $2, $3, $4, NOW())
		 ON CONFLICT (id) DO UPDATE
		 SET status = $3, total_cents = $4`,
		order.ID(), order.CustomerID(), order.Status(), order.TotalCents(),
	)
	if err != nil {
		return fmt.Errorf("failed to save order: %w", err)
	}

	// Save the items
	for _, item := range order.Items() {
		_, err = tx.ExecContext(ctx,
			`INSERT INTO order_items (id, order_id, product_id, quantity, price_cents)
			 VALUES ($1, $2, $3, $4, $5)
			 ON CONFLICT (id) DO NOTHING`,
			item.ID(), order.ID(), item.ProductID(), item.Quantity(), item.PriceCents(),
		)
		if err != nil {
			return fmt.Errorf("failed to save order item: %w", err)
		}
	}

	// Save domain events to the outbox to be sent to the projector
	for _, event := range order.PullEvents() {
		payload, _ := json.Marshal(event)
		_, err = tx.ExecContext(ctx,
			`INSERT INTO outbox_events (id, event_name, payload, created_at, published)
			 VALUES ($1, $2, $3, NOW(), FALSE)`,
			event.EventID(), event.EventName(), payload,
		)
		if err != nil {
			return fmt.Errorf("failed to save outbox event: %w", err)
		}
	}

	return tx.Commit()
}

// infrastructure/read/redis_order_repo.go — Read Repository using Redis
package read

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

	"github.com/redis/go-redis/v9"

	"myapp/query"
)

type RedisOrderReadRepo struct {
	client *redis.Client
	ttl    time.Duration
}

func NewRedisOrderReadRepo(client *redis.Client) *RedisOrderReadRepo {
	return &RedisOrderReadRepo{
		client: client,
		ttl:    24 * time.Hour,
	}
}

func (r *RedisOrderReadRepo) FindDetailByID(
	ctx context.Context,
	orderID string,
) (*query.OrderDetailView, error) {
	key := fmt.Sprintf("order:detail:%s", orderID)

	data, err := r.client.Get(ctx, key).Bytes()
	if err == redis.Nil {
		return nil, fmt.Errorf("order not found: %s", orderID)
	}
	if err != nil {
		return nil, fmt.Errorf("failed to read from redis: %w", err)
	}

	var view query.OrderDetailView
	if err := json.Unmarshal(data, &view); err != nil {
		return nil, err
	}
	return &view, nil
}

// Store saves the read model to Redis — called by the projector
func (r *RedisOrderReadRepo) Store(ctx context.Context, view *query.OrderDetailView) error {
	key := fmt.Sprintf("order:detail:%s", view.OrderID)
	data, err := json.Marshal(view)
	if err != nil {
		return err
	}
	return r.client.Set(ctx, key, data, r.ttl).Err()
}

// Invalidate removes the read model from the cache — called when an order is updated
func (r *RedisOrderReadRepo) Invalidate(ctx context.Context, orderID string) error {
	key := fmt.Sprintf("order:detail:%s", orderID)
	return r.client.Del(ctx, key).Err()
}

// infrastructure/projector/order_projector.go
package projector

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

	"myapp/events"
	"myapp/query"
	"myapp/infrastructure/read"
)

// OrderProjector builds and updates the read model based on domain events
type OrderProjector struct {
	writeDB  *sql.DB         // to fetch data the read model needs
	readRepo *read.RedisOrderReadRepo
	interval time.Duration
}

func NewOrderProjector(writeDB *sql.DB, readRepo *read.RedisOrderReadRepo) *OrderProjector {
	return &OrderProjector{
		writeDB:  writeDB,
		readRepo: readRepo,
		interval: 500 * time.Millisecond,
	}
}

// Start runs the projector loop — read the outbox, update the read model
func (p *OrderProjector) Start(ctx context.Context) {
	ticker := time.NewTicker(p.interval)
	defer ticker.Stop()

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

func (p *OrderProjector) processOutbox(ctx context.Context) {
	rows, err := p.writeDB.QueryContext(ctx,
		`SELECT id, event_name, payload
		 FROM outbox_events
		 WHERE published = FALSE
		 ORDER BY created_at
		 LIMIT 50`,
	)
	if err != nil {
		return
	}
	defer rows.Close()

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

		if err := p.handleEvent(ctx, eventName, payload); err != nil {
			slog.Error("projector failed to handle event",
				"event_id", id,
				"event_name", eventName,
				"error", err,
			)
			continue
		}

		p.writeDB.ExecContext(ctx,
			`UPDATE outbox_events SET published = TRUE, published_at = NOW() WHERE id = $1`,
			id,
		)
	}
}

func (p *OrderProjector) handleEvent(ctx context.Context, eventName string, payload []byte) error {
	switch eventName {
	case "order.placed":
		var event events.OrderPlaced
		json.Unmarshal(payload, &event)
		return p.projectOrderPlaced(ctx, event)

	case "order.cancelled":
		var event events.OrderCancelled
		json.Unmarshal(payload, &event)
		return p.projectOrderCancelled(ctx, event)

	default:
		return nil // ignore irrelevant events
	}
}

// projectOrderPlaced builds an OrderDetailView from the OrderPlaced event
func (p *OrderProjector) projectOrderPlaced(ctx context.Context, event events.OrderPlaced) error {
	// Fetch additional data the read model needs (customer name, product names)
	customerName, _ := p.fetchCustomerName(ctx, event.CustomerID)

	items := make([]query.OrderItemView, len(event.Items))
	for i, item := range event.Items {
		productName, _ := p.fetchProductName(ctx, item.ProductID)
		items[i] = query.OrderItemView{
			ProductName:   productName,
			Quantity:      item.Quantity,
			PriceCents:    item.PriceCents,
			SubtotalCents: item.PriceCents * int64(item.Quantity),
		}
	}

	view := &query.OrderDetailView{
		OrderID:      event.OrderID,
		CustomerName: customerName,
		Items:        items,
		TotalCents:   event.TotalCents,
		Status:       "pending",
		CreatedAt:    event.OccurredAt.Format("2006-01-02 15:04:05"),
	}

	return p.readRepo.Store(ctx, view)
}

func (p *OrderProjector) projectOrderCancelled(ctx context.Context, event events.OrderCancelled) error {
	// Invalidate the read model — it will be rebuilt on the next query
	return p.readRepo.Invalidate(ctx, event.OrderID)
}

func (p *OrderProjector) fetchCustomerName(ctx context.Context, customerID string) (string, error) {
	var name string
	err := p.writeDB.QueryRowContext(ctx,
		`SELECT full_name FROM users WHERE id = $1`, customerID,
	).Scan(&name)
	return name, err
}

func (p *OrderProjector) fetchProductName(ctx context.Context, productID string) (string, error) {
	var name string
	err := p.writeDB.QueryRowContext(ctx,
		`SELECT name FROM products WHERE id = $1`, productID,
	).Scan(&name)
	return name, err
}

// domain/order_sourced.go — an Order that uses Event Sourcing
type Order struct {
	id         string
	customerID string
	status     OrderStatus
	items      []OrderItem
	events     []DomainEvent // accumulated events not yet persisted
}

// Apply applies one event to the order state
func (o *Order) Apply(event DomainEvent) {
	switch e := event.(type) {
	case OrderPlacedEvent:
		o.id = e.OrderID
		o.customerID = e.CustomerID
		o.items = e.Items
		o.status = StatusPending
	case OrderConfirmedEvent:
		o.status = StatusConfirmed
	case OrderCancelledEvent:
		o.status = StatusCancelled
	}
}

// Reconstruct builds an Order from event history
func Reconstruct(events []DomainEvent) *Order {
	order := &Order{}
	for _, event := range events {
		order.Apply(event)
	}
	return order
}

// Cancel produces a new event — it does not directly modify state
func (o *Order) Cancel(reason string) error {
	if o.status != StatusPending {
		return errors.New("only pending orders can be cancelled")
	}
	event := OrderCancelledEvent{
		OrderID:    o.id,
		Reason:     reason,
		OccurredAt: time.Now(),
	}
	o.events = append(o.events, event) // record the event
	o.Apply(event)                     // apply to local state
	return nil
}

// ✗ A Command returning full domain data — violates CQRS
func (h *OrderCommandHandler) PlaceOrder(ctx context.Context, cmd PlaceOrderCommand) (*Order, error) {
	order := buildOrder(cmd)
	h.repo.Save(ctx, order)
	return order, nil // ✗ Commands must not return domain objects
}

// ✓ A Command only returns an ID or void
func (h *OrderCommandHandler) PlaceOrder(ctx context.Context, cmd PlaceOrderCommand) (string, error) {
	order := buildOrder(cmd)
	h.repo.Save(ctx, order)
	return order.ID(), nil // ✓ only an ID
}

// ✗ A Query handler changing state — violates CQRS
func (h *OrderQueryHandler) GetOrder(ctx context.Context, orderID string) (*Order, error) {
	order, _ := h.repo.FindByID(ctx, orderID)
	order.ViewCount++ // ✗ queries must not change state
	h.repo.Save(ctx, order)
	return order, nil
}

// ✓ A Query handler purely reading
func (h *OrderQueryHandler) GetOrder(ctx context.Context, orderID string) (*OrderDetailView, error) {
	return h.readRepo.FindDetailByID(ctx, orderID) // ✓ read-only
}

// ✗ Using a domain entity as a read model — the model is not optimal for display
func (h *OrderQueryHandler) ListOrders(ctx context.Context) ([]*domain.Order, error) {
	return h.writeRepo.FindAll(ctx) // ✗ domain entities have many fields the display does not need
}

// ✓ Use a read model optimized for display
func (h *OrderQueryHandler) ListOrders(ctx context.Context) ([]*OrderSummaryView, error) {
	return h.readRepo.FindSummaries(ctx) // ✓ optimized flat view
}

// ✗ One handler mixing commands and queries
type OrderService struct{}

func (s *OrderService) PlaceOrder(cmd PlaceOrderCommand) (*Order, error) { ... }
func (s *OrderService) GetOrder(id string) (*Order, error) { ... }
func (s *OrderService) ListOrders() ([]*Order, error) { ... }
// ✗ one service containing everything — hard to optimize and test separately

// ✓ Separate handlers per responsibility
type OrderCommandHandler struct { writeRepo OrderWriteRepository }
type OrderQueryHandler struct { readRepo OrderReadRepository }