05 CQRS & More

In this step, we'll refactor our User service to implement the Command Query Responsibility Segregation (CQRS) pattern. CQRS separates read (query) and write (command) operations for improved performance, scalability, and flexibility. We'll split our User service into two distinct services: User Command Service and User Query Service.

Goals

By the end of this branch, you'll have:

A User Command Service responsible for handling write operations (creating, updating, deleting users).
A User Query Service responsible for handling read operations (retrieving users).
Both services will interact with the same database but will have their own separate data models and handlers.
Extract common functionality to global package

diagram

Prepare Repo

Checkout Branch:
```
git checkout 04-cqrs 
```

Update Project Structure:

Adjust the user-service directory structure:

user-service/
├── cmd
│   ├── command  // User Command Service
│   │   └── main.go
│   └── query   // User Query Service
│       └── main.go
├── internal
│   ├── command    // Command handlers and logic
│   │   └── ... 
│   ├── handlers   // HTTP handlers for both services
│   │   └── ...
│   ├── models     // Data models for both services
│   │   └── ...
│   ├── query      // Query handlers and logic
│   │   └── ...
└── ...            // Other files (Dockerfile.command, Dockerfile.query, go.mod)

mkdir -p pkg/{db,utils,middleware} user-service/cmd/{command,query} user-service/internal/{command,query}
touch pkg/db/mysql.go pkg/utils/tracing.go pkg/middleware/tracing.go
touch user-service/cmd/command/main.go user-service/cmd/query/main.go 
touch user-service/internal/command/user.go user-service/internal/query/user.go
touch user-service/internal/handlers/query_user.go 
touch user-service/internal/handlers/command_user.go
touch user-service/Dockerfile.command
touch user-service/Dockerfile.query

Common Pkg

Let's extract common functionality that will be used through our future services

db/mysql.go will contain functionality to connect to mysql server and will be used amonth other services

package db

import (
  "database/sql"
  "fmt"
  "log"

  _ "github.com/go-sql-driver/mysql"
  "github.com/spf13/viper"
)

func MySqlConnect() *sql.DB {

  // Database Configuration
  viper.SetEnvPrefix("USER_SERVICE") // Set environment variable prefix
  viper.AutomaticEnv()               // Enable Viper to read from environment variables.

  // Database Configuration from Environment Variables
  dbHost := viper.GetString("DB_HOST")
  dbPort := viper.GetString("DB_PORT")
  dbUser := viper.GetString("DB_USER")
  dbPass := viper.GetString("DB_PASS")
  dbName := viper.GetString("DB_NAME")

  connectionString := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", dbUser, dbPass, dbHost, dbPort, dbName)
  db, err := sql.Open("mysql", connectionString)
  if err != nil {
    log.Fatalf("Unable to connect to database: %v\n", err)
  }
  return db
}

func SeedData(conn *sql.DB) {
  // Check if the users table is empty
  var count int
  err := conn.QueryRow("SELECT COUNT(*) FROM users").Scan(&count)
  if err != nil {
    log.Fatalf("Failed to check users table: %v\n", err)
  }

  if count == 0 {
    // Insert sample data
    _, err = conn.Exec(`
            INSERT INTO users (name, email, password) VALUES
            ('Alice Johnson', 'alice.johnson@example.com', 'hashed_password_1'),
            ('Bob Smith', 'bob.smith@example.com', 'hashed_password_2'),
            ('Charlie Brown', 'charlie.brown@example.com', 'hashed_password_3'),
            ('Diana Prince', 'diana.prince@example.com', 'hashed_password_4'),
            ('Ethan Hunt', 'ethan.hunt@example.com', 'hashed_password_5');
        `)
    if err != nil {
      log.Fatalf("Failed to seed users table: %v\n", err)
    }

    log.Println("Seed data inserted successfully.")
  } else {
    log.Println("Users table already seeded.")
  }
}

middleware/tracing.go is a middleware to monitor and observe requests as they flow through various services in a microservices architecture

package middleware

import (
  "github.com/gofiber/fiber/v2"
  "go.opentelemetry.io/otel"
)

func TracingMiddleware(name string) fiber.Handler {
  tracer := otel.Tracer(name)

  return func(c *fiber.Ctx) error {
    // Start a new span
    ctx, span := tracer.Start(c.UserContext(), c.Path())
    defer span.End()
    // Pass the context with the span to the next handler
    c.SetUserContext(ctx)
    return c.Next()
  }
}

utils/tracing.go includes functionality to initialize a tracing system using OpenTelemetry, which is a set of APIs, libraries, agents, and instrumentation to provide observability for applications.

package utils

import (
  "context"
  "log"

  "go.opentelemetry.io/otel"
  "go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
  "go.opentelemetry.io/otel/sdk/trace"
)

func InitTracer() func() {
  exporter, err := stdouttrace.New(stdouttrace.WithPrettyPrint())
  if err != nil {
    log.Fatalf("failed to initialize stdouttrace exporter: %v", err)
  }

  tp := trace.NewTracerProvider(
    trace.WithBatcher(exporter),
  )
  otel.SetTracerProvider(tp)

  return func() {
    if err := tp.Shutdown(context.Background()); err != nil {
      log.Fatalf("failed to shutdown TracerProvider: %v", err)
    }
  }
}

Init project and import packages

go mod init github.com/rfashwall/go-task
go mod tidy

For our services to be able to use the base package, we need to import We commit the pkg and fetch the commit hash, then import to user-service
```
cd user-service 
go get github.com/rfashwall/go-task@62c46d1ecf43e0e991568464ec072d0643862d92
```

Separate Models (`user-service/internal/models`)

Create distinct models for commands and queries. For simplicity, we'll keep the models the same for now, but in a real-world application, they might differ significantly.

Update Repositories

Update both repos according to CQRS Patterm where we now separate the repository into to separate pkgs

user-service/internal/command/user.go 
  ```go
  package command

  import (
    "context"
    "database/sql"

    _ "github.com/go-sql-driver/mysql"
    "github.com/rfashwall/user-service/internal/models"
  )

  type UserCommand interface {
    CreateUser(ctx context.Context, user *models.User) error
    UpdateUser(ctx context.Context, user *models.User) error
    DeleteUser(ctx context.Context, id int) error
  }

  type MySQLUserCommand struct {
    Conn *sql.DB
  }

  func NewMySQLUserCommand(conn *sql.DB) *MySQLUserCommand {
    return &MySQLUserCommand{Conn: conn}
  }

  func (c *MySQLUserCommand) CreateUser(ctx context.Context, user *models.User) error {
    _, err := c.Conn.ExecContext(ctx, "INSERT INTO users (name, email, password) VALUES (?, ?, ?)", user.Name, user.Email, user.Password)
    return err
  }

  func (c *MySQLUserCommand) UpdateUser(ctx context.Context, user *models.User) error {
    _, err := c.Conn.ExecContext(ctx, "UPDATE users SET name=?, email=?, password=? WHERE id=?", user.Name, user.Email, user.Password, user.ID)
    return err
  }

  func (c *MySQLUserCommand) DeleteUser(ctx context.Context, id int) error {
    _, err := c.Conn.ExecContext(ctx, "DELETE FROM users WHERE id=?", id)
    return err
  }

  ```

 user-service/internal/query/user.go
  ```go
  package query

  import (
    "context"
    "database/sql"

    _ "github.com/go-sql-driver/mysql"
    "github.com/rfashwall/user-service/internal/models"
  )

  type UserQuery interface {
    GetUserByID(ctx context.Context, id int) (*models.User, error)
    ListUsers(ctx context.Context) ([]*models.User, error)
  }

  type MySQLUserQuery struct {
    Conn *sql.DB
  }

  func NewMySQLUserQuery(conn *sql.DB) *MySQLUserQuery {
    return &MySQLUserQuery{Conn: conn}
  }

  func (q *MySQLUserQuery) GetUserByID(ctx context.Context, id int) (*models.User, error) {
    var user models.User
    err := q.Conn.QueryRowContext(ctx, "SELECT id, name, email, password FROM users WHERE id=?", id).Scan(&user.ID, &user.Name, &user.Email, &user.Password)
    if err != nil {
      return nil, err
    }
    return &user, nil
  }

  func (q *MySQLUserQuery) ListUsers(ctx context.Context) ([]*models.User, error) {
    rows, err := q.Conn.QueryContext(ctx, "SELECT id, name, email, password FROM users")
    if err != nil {
      return nil, err
    }
    defer rows.Close()

    var users []*models.User
    for rows.Next() {
      var user models.User
      if err := rows.Scan(&user.ID, &user.Name, &user.Email, &user.Password); err != nil {
        return nil, err
      }
      users = append(users, &user)
    }
    return users, nil
  }

  ```

Update Handlers

Update user-service/internal/handlers/command_user.go

package handlers

import (
  "github.com/gofiber/fiber/v2"
  "github.com/rfashwall/user-service/internal/command"
  "github.com/rfashwall/user-service/internal/models"
  "go.uber.org/zap"
)

type UserCommandHandler struct {
  UserCommand command.UserCommand
  logger      *zap.Logger
}

func NewUserCommandHandler(userCommand command.UserCommand, logger *zap.Logger) *UserCommandHandler {
  return &UserCommandHandler{
    UserCommand: userCommand,
    logger:      logger,
  }
}

func (h *UserCommandHandler) SetupRoutes(app *fiber.App) {
  app.Post("/users", h.createUser)
  app.Put("/users/:id", h.updateUser)
  app.Delete("/users/:id", h.deleteUser)
}

func (h *UserCommandHandler) createUser(c *fiber.Ctx) error {
  user := new(models.User)
  if err := c.BodyParser(user); err != nil {
    h.logger.Error("Invalid input", zap.Error(err))
    return c.Status(fiber.StatusBadRequest).SendString("Invalid input")
  }

  err := h.UserCommand.CreateUser(c.Context(), user)
  if err != nil {
    h.logger.Error("Failed to create user", zap.Error(err))
    return c.Status(fiber.StatusInternalServerError).SendString(err.Error())
  }

  return c.Status(fiber.StatusCreated).JSON(user)
}

func (h *UserCommandHandler) updateUser(c *fiber.Ctx) error {
  id, err := c.ParamsInt("id")
  if err != nil {
    h.logger.Error("Invalid user ID", zap.Error(err))
    return c.Status(fiber.StatusBadRequest).SendString("Invalid user ID")
  }

  user := new(models.User)
  if err := c.BodyParser(user); err != nil {
    h.logger.Error("Invalid user input", zap.Error(err))
    return c.Status(fiber.StatusBadRequest).SendString("Invalid input")
  }
  user.ID = id

  err = h.UserCommand.UpdateUser(c.Context(), user)
  if err != nil {
    h.logger.Error("Failed to update user", zap.Error(err), zap.Int("user_id", id))
    return c.Status(fiber.StatusInternalServerError).SendString(err.Error())
  }

  h.logger.Info("User updated", zap.Int("id", id))
  return c.JSON(user)
}

func (h *UserCommandHandler) deleteUser(c *fiber.Ctx) error {
  id, err := c.ParamsInt("id")
  if err != nil {
    h.logger.Error("Invalid user ID", zap.Error(err))
    return c.Status(fiber.StatusBadRequest).SendString("Invalid user ID")
  }

  err = h.UserCommand.DeleteUser(c.Context(), id)
  if err != nil {
    h.logger.Error("Failed to delete user", zap.Error(err), zap.Int("user_id", id))
    return c.Status(fiber.StatusInternalServerError).SendString(err.Error())
  }

  h.logger.Info("User deleted", zap.Int("id", id))
  return c.SendStatus(fiber.StatusNoContent)
}

Update user-service/internal/handlers/query_user.go handler

package handlers

import (
  "github.com/gofiber/fiber/v2"
  "github.com/rfashwall/user-service/internal/query"
  "go.uber.org/zap"
)

type UserQueryHandler struct {
  UserQuery query.UserQuery
  logger    *zap.Logger
}

func NewUserQueryHandler(userQuery query.UserQuery, logger *zap.Logger) *UserQueryHandler {
  return &UserQueryHandler{
    UserQuery: userQuery,
    logger:    logger,
  }
}

func (h *UserQueryHandler) SetupRoutes(app *fiber.App) {
  app.Get("/users/:id", h.getUser)
  app.Get("/users", h.listUsers)
}

func (h *UserQueryHandler) getUser(c *fiber.Ctx) error {
  id, err := c.ParamsInt("id")
  if err != nil {
    h.logger.Error("Invalid user ID", zap.Error(err))
    return c.Status(fiber.StatusBadRequest).SendString("Invalid user ID")
  }

  user, err := h.UserQuery.GetUserByID(c.UserContext(), id)
  if err != nil {
    h.logger.Error("Failed to get user", zap.Error(err), zap.Int("user_id", id))
    return c.Status(fiber.StatusInternalServerError).SendString(err.Error())
  }

  return c.JSON(user)
}

func (h *UserQueryHandler) listUsers(c *fiber.Ctx) error {
  users, err := h.UserQuery.ListUsers(c.UserContext())
  if err != nil {
    h.logger.Error("Failed to list users", zap.Error(err))
    return c.Status(fiber.StatusInternalServerError).SendString(err.Error())
  }

  h.logger.Info("Listed users", zap.Int("count", len(users)))
  return c.JSON(users)
}

Create Independent Services

(user-service/cmd/command/main.go and user-service/cmd/query/main.go)

We'll need two main application files: one for the User Command Service and one for the User Query Service.

user-service/cmd/command/main.go:

package main

import (
  "log"

  "github.com/gofiber/fiber/v2"
  "github.com/gofiber/fiber/v2/middleware/healthcheck"
  "github.com/gofiber/fiber/v2/middleware/logger"
  "github.com/rfashwall/go-task/pkg/db"
  "github.com/rfashwall/go-task/pkg/middleware"
  "github.com/rfashwall/go-task/pkg/utils"
  "github.com/rfashwall/user-service/internal/command"
  "github.com/rfashwall/user-service/internal/handlers"
  "go.uber.org/zap"
)

func main() {
  shutdown := utils.InitTracer()
  defer shutdown()

  app := fiber.New()
  app.Use(logger.New())
  app.Use(healthcheck.New())
  app.Use(middleware.TracingMiddleware("user-command-service"))

  logger, err := zap.NewDevelopment()
  if err != nil {
    log.Fatal(err)
  }
  defer logger.Sync()

  // Connect to the database
  logger.Info("Connecting to MySQL")
  conn := db.MySqlConnect()
  defer conn.Close()

  logger.Info("Seeding data")
  db.SeedData(conn)

  // Initialize the repository
  userCommand := command.NewMySQLUserCommand(conn)

  // Initialize the handler
  logger.Debug("Initializing user command handler")
  userHandler := handlers.NewUserCommandHandler(userCommand, logger)

  // Set up routes
  logger.Info("Setting up routes")
  userHandler.SetupRoutes(app)

  logger.Info("Listening on port 3001")
  log.Fatal(app.Listen(":3001"))
}

user-service/cmd/query/main.go:

package main

import (
  "log"

  "github.com/gofiber/fiber/v2"
  "github.com/gofiber/fiber/v2/middleware/healthcheck"
  "github.com/gofiber/fiber/v2/middleware/logger"
  "github.com/rfashwall/go-task/pkg/db"
  "github.com/rfashwall/go-task/pkg/middleware"
  "github.com/rfashwall/go-task/pkg/utils"
  "github.com/rfashwall/user-service/internal/handlers"
  "github.com/rfashwall/user-service/internal/query"
  "go.uber.org/zap"
)

func main() {
  shutdown := utils.InitTracer()
  defer shutdown()

  app := fiber.New()
  app.Use(logger.New())
  app.Use(healthcheck.New())
  app.Use(middleware.TracingMiddleware("user-query-service"))

  logger, err := zap.NewDevelopment()
  if err != nil {
    log.Fatal(err)
  }
  defer logger.Sync()

  // Connect to the database
  logger.Info("Connecting to MySQL")
  conn := db.MySqlConnect()
  defer conn.Close()

  // Initialize the repository
  userQuery := query.NewMySQLUserQuery(conn)

  // Initialize the handler
  logger.Debug("Initializing user query handler")
  userHandler := handlers.NewUserQueryHandler(userQuery, logger)

  // Set up routes
  logger.Info("Setting up routes")
  userHandler.SetupRoutes(app)

  logger.Info("Listening on port 3000")
  log.Fatal(app.Listen(":3000"))
}

Dockerfiles

Create two separate Dockerfiles for the command and query services:

user-service/Dockerfile.command:

FROM golang:1.23.4-alpine

WORKDIR /app

COPY . .

RUN go mod tidy
RUN go build -o user-command-service ./cmd/command

CMD ["./user-command-service"]

user-service/Dockerfile.query:

FROM golang:1.23.4-alpine

WORKDIR /app

COPY . .

RUN go mod tidy
RUN go build -o user-query-service ./cmd/query

CMD ["./user-query-service"]

Docker Compose

Update your docker-compose.yaml to include separate services for command and query:

version: '3.8'

services:
  mysql-server:
      image: mysql:latest
      container_name: mysql-server
      environment:
        MYSQL_ROOT_PASSWORD: root
        MYSQL_DATABASE: user_tracker
      ports:
        - "3306:3306"
      volumes:
        - mysql-data:/var/lib/mysql
        - ./init.sh:/docker-entrypoint-initdb.d/init.sh
      healthcheck:
        test: ["CMD-SHELL", "mysqladmin ping -h localhost"]
        interval: 10s
        retries: 5
        start_period: 30s
        timeout: 10s

  user-command-service:
    build:
      context: ./user-service
      dockerfile: Dockerfile.command
    container_name: user-command-service
    environment:
      - USER_SERVICE_DB_HOST=mysql-server 
      - USER_SERVICE_DB_PORT=3306
      - USER_SERVICE_DB_USER=root
      - USER_SERVICE_DB_PASS=root
      - USER_SERVICE_DB_NAME=user_tracker
    ports:
      - "3001:3001"
    depends_on:
      mysql-server:
        condition: service_healthy

  user-query-service:
    build:
      context: ./user-service
      dockerfile: Dockerfile.query
    container_name: user-query-service
    ports:
      - "3000:3000"
    environment:
      - USER_SERVICE_DB_HOST=mysql-server 
      - USER_SERVICE_DB_PORT=3306
      - USER_SERVICE_DB_USER=root
      - USER_SERVICE_DB_PASS=root
      - USER_SERVICE_DB_NAME=user_tracker
    depends_on:
      mysql-server:
        condition: service_healthy

volumes:
  mysql-data:

Run and Test

Build and run the services using Docker Compose:
```
docker compose up -d --build --force-recreate
```

Test your API endpoints using curl or a REST client, remembering to send commands to port 3001 (command service) and queries to port 3000 (query service).

curl -X GET -H "Content-Type: application/json"  http://localhost:3000/users

curl -X POST http://localhost:3001/users \
  -H "Content-Type: application/json" \
  -d '{
      "name": "John Doe",
      "email": "john.doe@example.com",
      "password": "hashed_password"
  }'

Navigation Menu

Repository navigation

05 CQRS & More

Goals

Prepare Repo

Common Pkg

Separate Models (`user-service/internal/models`)

Update Repositories

Update Handlers

Create Independent Services

Dockerfiles

Docker Compose

Run and Test

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally

05 CQRS & More

Goals

Prepare Repo

Common Pkg

Separate Models (user-service/internal/models)

Update Repositories

Update Handlers

Create Independent Services

Dockerfiles

Docker Compose

Run and Test

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally

Separate Models (`user-service/internal/models`)