Go EventSourcing and CQRS microservice using EventStoreDB 👋⚡️💫
Tags: go, cqrs, eventsourcing, microservices
Published: 2022-02-28 07:10:07

In this article let's try to create closer to real world Event Sourcing CQRS microservice using: 🚀👨‍💻🙌
[EventStoreDB](https://www.eventstore.com/) The database built for Event Sourcing
[gRPC](https://github.com/grpc/grpc-go) Go implementation of gRPC
[MongoDB](https://github.com/mongodb/mongo-go-driver) Web and API based SMTP testing
[Elasticsearch](https://github.com/olivere/elastic) Elasticsearch client for Go.
[Jaeger](https://www.jaegertracing.io/) open source, end-to-end distributed [tracing](https://opentracing.io/)
[Prometheus](https://prometheus.io/) monitoring and alerting
[Grafana](https://grafana.com/) for to compose observability dashboards with everything from Prometheus
[swag](https://github.com/swaggo/swag) Swagger for Go
[Echo](https://github.com/labstack/echo) web framework
[Kibana](https://github.com/labstack/echo) Kibana is user interface that lets you visualize your Elasticsearch


Source code you can find in [GitHub repository](https://github.com/AleksK1NG/Go-EventSourcing-CQRS)<br/>
The main idea of this project is the implementation of Event Sourcing and CQRS using Go, [EventStoreDB](https://www.eventstore.com/), gRPC, MongoDB and ElasticSearch.
Didn't write in this article about Event Sourcing and CQRS patterns, because it makes the article huge and the best place to read is [microservices.io (https://microservices.io/patterns/data/cqrs.html), found [this article](https://www.eventstore.com/event-sourcing) is very good too, and highly recommend  **Alexey Zimarev** **"Hands-on Domain-Driven Design with .NET Core"** book and also his [blog](https://zimarev.com/blog/).

In this project we have microservice working with [EventStoreDB](https://www.eventstore.com/) using oficial [go client](https://github.com/EventStore/EventStore-Client-Go/), for [projections (https://zimarev.com/blog/event-sourcing/projections/) used [MongoDB](https://github.com/mongodb/mongo-go-driver) and [Elasticsearch](https://github.com/olivere/elastic) for search,
and communicate by gRPC and REST.
Did not implement here any interesting business logic and didn't cover tests, because don't have enough time,
the events list is very simple: create a new order, update shopping cart, pay, submit, cancel, change the delivery address, complete order, and of course in real-world better use more concrete and meaningfully events, but the target here is to show the idea and how it works.
Event Sourcing can be implemented in different ways, used here EventStoreDB, but we can do it with PostgreSQL and Kafka for example.
After trying both approaches, found EventStoreDB is a better solution because all required features are implemented out of the box, it is optimized and really very good engineers developing it.

For local development:
```
make local or docker_dev // for run docker compose
make run_es // run es microservice
```
or
```
make dev // run all in docker compose with hot reload
```

All UI interfaces will be available on ports:

#### EventStoreDB UI: http://localhost:2113
<img src="https://i.postimg.cc/7LQcWKZz/Event-Store-Dashboard-2022-01-03-17-06-38.png"  alt="EventStoreDB"/>

#### Jaeger UI: http://localhost:16686
<img src="https://i.postimg.cc/6pCpwGvT/Jaeger-UI-2022-01-03-19-37-09.png"  alt="Jaeger"/>

#### Prometheus UI: http://localhost:9090
<img src="https://i.postimg.cc/9z5HNJ1b/Prometheus-Time-Series-Collection-and-Processing-Server-2022-01-03-18-39-45.png"  alt="Prometheus"/>

#### Grafana UI: http://localhost:3005
<img src="https://i.postimg.cc/TPpG1Lyr/New-dashboard-Grafana-2022-01-03-19-31-56.png"  alt="Grafana"/>

#### Swagger UI: http://localhost:5007/swagger/index.html
<img src="https://i.postimg.cc/856ZYgBT/Swagger-UI-2022-02-21-10-57-28.png"  alt="Swagger"/>

#### Kibana UI: http://localhost:5601/app/home#/
<img src="https://i.postimg.cc/MKT9hygb/Dev-Tools-Elastic-2022-01-03-17-19-36.png"  alt="Kibana"/>


Docker compose file for this project,
depending on your environment, use image for eventstoredb: eventstore/eventstore:21.6.0-buster-slim,
and for M1: ghcr.io/eventstore/eventstore@sha256:ab30bf2a2629e2c8710f8f7fdcb74df5466c6b3b2200c8e9ad8a797ed138012a
```yaml
version: "3.8"

services:
  eventstore.db:
    image: eventstore/eventstore:21.6.0-buster-slim
    environment:
      - EVENTSTORE_CLUSTER_SIZE=1
      - EVENTSTORE_RUN_PROJECTIONS=All
      - EVENTSTORE_START_STANDARD_PROJECTIONS=true
      - EVENTSTORE_EXT_TCP_PORT=1113
      - EVENTSTORE_HTTP_PORT=2113
      - EVENTSTORE_INSECURE=true
      - EVENTSTORE_ENABLE_EXTERNAL_TCP=true
      - EVENTSTORE_ENABLE_ATOM_PUB_OVER_HTTP=true
    ports:
      - "1113:1113"
      - "2113:2113"
    volumes:
      - type: volume
        source: eventstore-volume-data
        target: /var/lib/eventstore
      - type: volume
        source: eventstore-volume-logs
        target: /var/log/eventstore
    networks: [ "microservices" ]

  prometheus:
    image: prom/prometheus:latest
    container_name: prometheus
    ports:
      - "9090:9090"
    command:
      - --config.file=/etc/prometheus/prometheus.yml
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro
    networks: [ "microservices" ]

  node_exporter:
    container_name: node_exporter_container
    restart: always
    image: prom/node-exporter
    ports:
      - '9101:9100'
    networks: [ "microservices" ]

  grafana:
    container_name: grafana_container
    restart: always
    image: grafana/grafana
    ports:
      - '3005:3000'
    networks: [ "microservices" ]

  mongodb:
    image: mongo:latest
    restart: always
    environment:
      MONGO_INITDB_ROOT_USERNAME: admin
      MONGO_INITDB_ROOT_PASSWORD: admin
      MONGODB_DATABASE: products
    ports:
      - "27017:27017"
    volumes:
      - mongodb_data_container:/data/db
    networks: [ "microservices" ]

  jaeger:
    container_name: jaeger_container
    restart: always
    image: jaegertracing/all-in-one:1.21
    environment:
      - COLLECTOR_ZIPKIN_HTTP_PORT=9411
    ports:
      - "5775:5775/udp"
      - "6831:6831/udp"
      - "6832:6832/udp"
      - "5778:5778"
      - "16686:16686"
      - "14268:14268"
      - "14250:14250"
      - "9411:9411"
    networks: [ "microservices" ]

  node01:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.11.1
    container_name: node01
    restart: always
    environment:
      - node.name=node01
      - cluster.name=es-cluster-7
      - discovery.type=single-node
      - "ES_JAVA_OPTS=-Xms128m -Xmx128m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - es-data01:/usr/share/elasticsearch/data
    ports:
      - "9200:9200"
    networks: [ "microservices" ]

  kibana:
    image: docker.elastic.co/kibana/kibana:7.11.1
    restart: always
    environment:
      ELASTICSEARCH_HOSTS: http://node01:9200
    ports:
      - "5601:5601"
    depends_on:
      - node01
    networks: [ "microservices" ]

volumes:
  mongodb_data_container:
  eventstore-volume-data:
  eventstore-volume-logs:
  es-data01:
    driver: local

networks:
  microservices:
    name: microservices
```

First, we look at EventStoreDB and then add some code for working with it, we can look at streams and events
**EventStoreDB ui**:
<img src="https://i.postimg.cc/gj3Bxbng/Event-Store-Stream-2022-02-21-11-13-29.png" />

AggregateRoot can be implemented in different ways, for this project and Go specific, the main aggregate root methods - load events and apply changes.
When we fetch the aggregate from the database, instead of reading its state as one record in a table or document, we read all events that were saved before and call the **When** method for each.
After all these steps, we will recover all the history of a given aggregate. By doing this, we will be bringing our aggregate to its latest state.
```go
type AggregateBase struct {
	ID                string
	Version           int64
	AppliedEvents     []Event
	UncommittedEvents []Event
	Type              AggregateType
	when              when
}

func (a *AggregateBase) Apply(event Event) error {
	if event.GetAggregateID() != a.GetID() {
		return ErrInvalidAggregateID
	}

	event.SetAggregateType(a.GetType())

	if err := a.when(event); err != nil {
		return err
	}

	a.Version++
	event.SetVersion(a.GetVersion())
	a.UncommittedEvents = append(a.UncommittedEvents, event)
	return nil
}

// RaiseEvent push event to aggregate applied events using When method, used for load directly from eventstore
func (a *AggregateBase) RaiseEvent(event Event) error {
	if event.GetAggregateID() != a.GetID() {
		return ErrInvalidAggregateID
	}
	if a.GetVersion() >= event.GetVersion() {
		return ErrInvalidEventVersion
	}

	event.SetAggregateType(a.GetType())

	if err := a.when(event); err != nil {
		return err
	}

	a.Version = event.GetVersion()
	return nil
}
```

The **Event** struct is abstraction under **esdb.RecordedEvent** and **esdb.EventData** for comfortable work with it.
An event represents a fact that took place in the domain. They are the source of truth; your current state is derived from the events. 
They are immutable and represent the business facts.
It means that we never change or remove anything in the database, we only append new events.
```go
// EventType is the type of any event, used as its unique identifier.
type EventType string

// Event is an internal representation of an event, returned when the Aggregate
// uses NewEvent to create a new event. The events loaded from the db is
// represented by each DBs internal event type, implementing Event.
type Event struct {
	EventID       string
	EventType     string
	Data          []byte
	Timestamp     time.Time
	AggregateType AggregateType
	AggregateID   string
	Version       int64
	Metadata      []byte
}
```

In this example we don't use snapshots, so the **AggregateStore** interface:
```go
// AggregateStore is responsible for loading and saving aggregates.
type AggregateStore interface {
	Load(ctx context.Context, aggregate Aggregate) error
	Save(ctx context.Context, aggregate Aggregate) error
	Exists(ctx context.Context, streamID string) error
}
```

Implementation of AggregateStore is Load, Save and Exists methods,
Load and Save accept aggregate then load or apply events using EventStoreDB client.
The **Load** method: find out the stream name for an aggregate, read all of the events from the aggregate stream, 
loop through all of the events, and call the RaiseEvent handler for each of them.
And the **Save** method persists aggregates by saving the history of changes, handling concurrency, when you retrieve a stream from EventStoreDB, you take note of the current version number, then when you save it back you can determine if somebody else has modified the record in the meantime.
```go
func (a *aggregateStore) Load(ctx context.Context, aggregate es.Aggregate) error {
	span, ctx := opentracing.StartSpanFromContext(ctx, "aggregateStore.Load")
	defer span.Finish()
	span.LogFields(log.String("AggregateID", aggregate.GetID()))

	stream, err := a.db.ReadStream(ctx, aggregate.GetID(), esdb.ReadStreamOptions{}, count)
	if err != nil {
		tracing.TraceErr(span, err)
		return errors.Wrap(err, "db.ReadStream")
	}
	defer stream.Close()

	for {
		event, err := stream.Recv()
		if errors.Is(err, esdb.ErrStreamNotFound) {
			tracing.TraceErr(span, err)
			return errors.Wrap(err, "stream.Recv")
		}
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			tracing.TraceErr(span, err)
			return errors.Wrap(err, "stream.Recv")
		}

		esEvent := es.NewEventFromRecorded(event.Event)
		if err := aggregate.RaiseEvent(esEvent); err != nil {
			tracing.TraceErr(span, err)
			return errors.Wrap(err, "RaiseEvent")
		}
		a.log.Debugf("(Load) esEvent: {%s}", esEvent.String())
	}

	a.log.Debugf("(Load) aggregate: {%s}", aggregate.String())
	return nil
}

func (a *aggregateStore) Save(ctx context.Context, aggregate es.Aggregate) error {
	span, ctx := opentracing.StartSpanFromContext(ctx, "aggregateStore.Save")
	defer span.Finish()
	span.LogFields(log.String("aggregate", aggregate.String()))

	if len(aggregate.GetUncommittedEvents()) == 0 {
		a.log.Debugf("(Save) [no uncommittedEvents] len: {%d}", len(aggregate.GetUncommittedEvents()))
		return nil
	}

	eventsData := make([]esdb.EventData, 0, len(aggregate.GetUncommittedEvents()))
	for _, event := range aggregate.GetUncommittedEvents() {
		eventsData = append(eventsData, event.ToEventData())
	}

	var expectedRevision esdb.ExpectedRevision
	if len(aggregate.GetAppliedEvents()) == 0 {
		expectedRevision = esdb.NoStream{}
		a.log.Debugf("(Save) expectedRevision: {%T}", expectedRevision)

		appendStream, err := a.db.AppendToStream(
			ctx,
			aggregate.GetID(),
			esdb.AppendToStreamOptions{ExpectedRevision: expectedRevision},
			eventsData...,
		)
		if err != nil {
			tracing.TraceErr(span, err)
			return errors.Wrap(err, "db.AppendToStream")
		}

		a.log.Debugf("(Save) stream: {%+v}", appendStream)
		return nil
	}

	readOps := esdb.ReadStreamOptions{Direction: esdb.Backwards, From: esdb.End{}}
	stream, err := a.db.ReadStream(context.Background(), aggregate.GetID(), readOps, 1)
	if err != nil {
		tracing.TraceErr(span, err)
		return errors.Wrap(err, "db.ReadStream")
	}
	defer stream.Close()

	lastEvent, err := stream.Recv()
	if err != nil {
		tracing.TraceErr(span, err)
		return errors.Wrap(err, "stream.Recv")
	}

	expectedRevision = esdb.Revision(lastEvent.OriginalEvent().EventNumber)
	a.log.Debugf("(Save) expectedRevision: {%T}", expectedRevision)

	appendStream, err := a.db.AppendToStream(
		ctx,
		aggregate.GetID(),
		esdb.AppendToStreamOptions{ExpectedRevision: expectedRevision},
		eventsData...,
	)
	if err != nil {
		tracing.TraceErr(span, err)
		return errors.Wrap(err, "db.AppendToStream")
	}

	a.log.Debugf("(Save) stream: {%+v}", appendStream)
	aggregate.ClearUncommittedEvents()
	return nil
}
```

For the next step let's create an order aggregate, we have to add es.AggregateBase struct and implement When interface:
:
```go
type OrderAggregate struct {
	*es.AggregateBase
	Order *models.Order
}

func (a *OrderAggregate) When(evt es.Event) error {

	switch evt.GetEventType() {

	case v1.OrderCreated:
		return a.onOrderCreated(evt)
	case v1.OrderPaid:
		return a.onOrderPaid(evt)
	case v1.OrderSubmitted:
		return a.onOrderSubmitted(evt)
	case v1.OrderCompleted:
		return a.onOrderCompleted(evt)
	case v1.OrderCanceled:
		return a.onOrderCanceled(evt)
	case v1.ShoppingCartUpdated:
		return a.onShoppingCartUpdated(evt)
	case v1.DeliveryAddressChanged:
		return a.onChangeDeliveryAddress(evt)
	default:
		return es.ErrInvalidEventType
	}
}


func (a *OrderAggregate) onOrderCreated(evt es.Event) error {
	var eventData v1.OrderCreatedEvent
	if err := evt.GetJsonData(&eventData); err != nil {
		return errors.Wrap(err, "GetJsonData")
	}

	a.Order.AccountEmail = eventData.AccountEmail
	a.Order.ShopItems = eventData.ShopItems
	a.Order.Created = true
	a.Order.TotalPrice = GetShopItemsTotalPrice(eventData.ShopItems)
	a.Order.DeliveryAddress = eventData.DeliveryAddress
	return nil
}

func (a *OrderAggregate) onOrderPaid(evt es.Event) error {
	var payment models.Payment
	if err := evt.GetJsonData(&payment); err != nil {
		return errors.Wrap(err, "GetJsonData")
	}

	a.Order.Paid = true
	a.Order.Payment = payment
	return nil
}

func (a *OrderAggregate) onOrderSubmitted(evt es.Event) error {
	a.Order.Submitted = true
	return nil
}

func (a *OrderAggregate) onOrderCompleted(evt es.Event) error {
	var eventData v1.OrderCompletedEvent
	if err := evt.GetJsonData(&eventData); err != nil {
		return errors.Wrap(err, "GetJsonData")
	}

	a.Order.Completed = true
	a.Order.DeliveredTime = eventData.DeliveryTimestamp
	a.Order.Canceled = false
	return nil
}

func (a *OrderAggregate) onOrderCanceled(evt es.Event) error {
	var eventData v1.OrderCanceledEvent
	if err := evt.GetJsonData(&eventData); err != nil {
		return errors.Wrap(err, "GetJsonData")
	}

	a.Order.Canceled = true
	a.Order.Completed = false
	a.Order.CancelReason = eventData.CancelReason
	return nil
}

func (a *OrderAggregate) onShoppingCartUpdated(evt es.Event) error {
	var eventData v1.ShoppingCartUpdatedEvent
	if err := evt.GetJsonData(&eventData); err != nil {
		return errors.Wrap(err, "GetJsonData")
	}

	a.Order.ShopItems = eventData.ShopItems
	a.Order.TotalPrice = GetShopItemsTotalPrice(eventData.ShopItems)
	return nil
}

func (a *OrderAggregate) onChangeDeliveryAddress(evt es.Event) error {
	var eventData v1.OrderDeliveryAddressChangedEvent
	if err := evt.GetJsonData(&eventData); err != nil {
		return errors.Wrap(err, "GetJsonData")
	}

	a.Order.DeliveryAddress = eventData.DeliveryAddress
	return nil
}
```

For example, let's look at, create order case, **onCreateOrderCommand** handle command, validate order state, serialize data and create **CreateOrderEvent**:
```go
func (a *OrderAggregate) CreateOrder(ctx context.Context, shopItems []*models.ShopItem, accountEmail, deliveryAddress string) error {
	span, _ := opentracing.StartSpanFromContext(ctx, "OrderAggregate.CreateOrder")
	defer span.Finish()
	span.LogFields(log.String("AggregateID", a.GetID()))

	if a.Order.Created {
		return ErrAlreadyCreated
	}
	if shopItems == nil {
		return ErrOrderShopItemsIsRequired
	}
	if deliveryAddress == "" {
		return ErrInvalidDeliveryAddress
	}

	event, err := eventsV1.NewOrderCreatedEvent(a, shopItems, accountEmail, deliveryAddress)
	if err != nil {
		tracing.TraceErr(span, err)
		return errors.Wrap(err, "NewOrderCreatedEvent")
	}

	if err := event.SetMetadata(tracing.ExtractTextMapCarrier(span.Context())); err != nil {
		tracing.TraceErr(span, err)
		return errors.Wrap(err, "SetMetadata")
	}

	return a.Apply(event)
}
```

Then aggregate handle event using **onOrderCreated** method which only applies changes to the state:
```go
func (a *OrderAggregate) onOrderCreated(evt es.Event) error {
	var eventData v1.OrderCreatedEvent
	if err := evt.GetJsonData(&eventData); err != nil {
		return errors.Wrap(err, "GetJsonData")
	}

	a.Order.AccountEmail = eventData.AccountEmail
	a.Order.ShopItems = eventData.ShopItems
	a.Order.Created = true
	a.Order.TotalPrice = GetShopItemsTotalPrice(eventData.ShopItems)
	a.Order.DeliveryAddress = eventData.DeliveryAddress
	return nil
}
```

Our microservice accept http or gRPC requests:
<img src="https://i.postimg.cc/zv1RdbHd/Swagger-UI-2022-02-18-13-10-04.png" alt="swagger"/>
For swagger used [swag](https://github.com/swaggo/swag) and let's look at create order handler code:
```go
// CreateOrder
// @Tags Orders
// @Summary Create order
// @Description Create new order
// @Param order body dto.CreateOrderReqDto true "create order"
// @Accept json
// @Produce json
// @Success 201 {string} id ""
// @Router /orders [post]
func (h *orderHandlers) CreateOrder() echo.HandlerFunc {
	return func(c echo.Context) error {
		ctx, span := tracing.StartHttpServerTracerSpan(c, "orderHandlers.CreateOrder")
		defer span.Finish()
		h.metrics.CreateOrderHttpRequests.Inc()

		var reqDto dto.CreateOrderReqDto
		if err := c.Bind(&reqDto); err != nil {
			h.log.Errorf("(Bind) err: {%v}", err)
			tracing.TraceErr(span, err)
			return httpErrors.ErrorCtxResponse(c, err, h.cfg.Http.DebugErrorsResponse)
		}

		if err := h.v.StructCtx(ctx, reqDto); err != nil {
			h.log.Errorf("(validate) err: {%v}", err)
			tracing.TraceErr(span, err)
			return httpErrors.ErrorCtxResponse(c, err, h.cfg.Http.DebugErrorsResponse)
		}

		id := uuid.NewV4().String()
		command := v1.NewCreateOrderCommand(id, reqDto.ShopItems, reqDto.AccountEmail, reqDto.DeliveryAddress)
		err := h.os.Commands.CreateOrder.Handle(ctx, command)
		if err != nil {
			h.log.Errorf("(CreateOrder.Handle) id: {%s}, err: {%v}", id, err)
			tracing.TraceErr(span, err)
			return httpErrors.ErrorCtxResponse(c, err, h.cfg.Http.DebugErrorsResponse)
		}

		h.log.Infof("(order created) id: {%s}", id)
		return c.JSON(http.StatusCreated, id)
	}
}
```

and gRPC **CreateOrder** handler does the same as http handler, validate request and call command.
For validation used [validator](https://github.com/go-playground/validator) because of implements value validations for structs and individual fields based on tags.
<img src="https://i.postimg.cc/X7p57bc7/Bloom-RPC-2022-02-18-13-03-36.png" alt="bloomrpc"/>
```go
func (s *orderGrpcService) CreateOrder(ctx context.Context, req *orderService.CreateOrderReq) (*orderService.CreateOrderRes, error) {
	ctx, span := tracing.StartGrpcServerTracerSpan(ctx, "orderGrpcService.CreateOrder")
	defer span.Finish()
	span.LogFields(log.String("req", req.String()))
	s.metrics.CreateOrderGrpcRequests.Inc()

	aggregateID := uuid.NewV4().String()
	command := v1.NewCreateOrderCommand(aggregateID, models.ShopItemsFromProto(req.GetShopItems()), req.GetAccountEmail(), req.GetDeliveryAddress())
	if err := s.v.StructCtx(ctx, command); err != nil {
		s.log.Errorf("(validate) aggregateID: {%s}, err: {%v}", aggregateID, err)
		tracing.TraceErr(span, err)
		return nil, s.errResponse(err)
	}

	if err := s.os.Commands.CreateOrder.Handle(ctx, command); err != nil {
		s.log.Errorf("(CreateOrder.Handle) orderID: {%s}, err: {%v}", aggregateID, err)
		return nil, s.errResponse(err)
	}

	s.log.Infof("(created order): orderID: {%s}", aggregateID)
	return &orderService.CreateOrderRes{AggregateID: aggregateID}, nil
}
```

http and gRPC handlers do the same, validate the incoming request and call command service with **CreateOrder command**,
which load OrderAggregate, call **HandleCommand** method and save it to event store.
The main reason for CQRS gaining popularity is the ability to handle reads and writes separately due to severe differences in optimization techniques for those much more distinct operations.
```go
func (c *createOrderHandler) Handle(ctx context.Context, command *CreateOrderCommand) error {
	span, ctx := opentracing.StartSpanFromContext(ctx, "createOrderHandler.Handle")
	defer span.Finish()
	span.LogFields(log.String("AggregateID", command.GetAggregateID()))

	order := aggregate.NewOrderAggregateWithID(command.AggregateID)
	err := c.es.Exists(ctx, order.GetID())
	if err != nil && !errors.Is(err, esdb.ErrStreamNotFound) {
		return err
	}

	if err := order.CreateOrder(ctx, command.ShopItems, command.AccountEmail, command.DeliveryAddress); err != nil {
		return err
	}

	span.LogFields(log.String("order", order.String()))
	return c.es.Save(ctx, order)
}
```
The process of building a piece of state from events is called a projection.
EventStoreDB has [subscriptions](https://developers.eventstore.com/clients/grpc/persistent-subscriptions.html), so we can subscribe our projections for order type stream events.
When we execute a command, the aggregate generates a new event that represents the state transitions of the aggregate. Those events are committed to the store, so the store appends them to the end of the aggregate stream. A subscription receives these events and updates its read models.
MongoDB's projection subscribes to events stream using the persistent subscription.
and process events using When method, like aggregate it applies changes depending on the event type:
```go
func (o *mongoProjection) ProcessEvents(ctx context.Context, stream *esdb.PersistentSubscription, workerID int) error {

	for {
		event := stream.Recv()
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		if event.SubscriptionDropped != nil {
			o.log.Errorf("(SubscriptionDropped) err: {%v}", event.SubscriptionDropped.Error)
			return errors.Wrap(event.SubscriptionDropped.Error, "Subscription Dropped")
		}

		if event.EventAppeared != nil {
			o.log.ProjectionEvent(constants.MongoProjection, o.cfg.Subscriptions.MongoProjectionGroupName, event.EventAppeared, workerID)

			err := o.When(ctx, es.NewEventFromRecorded(event.EventAppeared.Event))
			if err != nil {
				o.log.Errorf("(mongoProjection.when) err: {%v}", err)
				if err := stream.Nack(err.Error(), esdb.Nack_Retry, event.EventAppeared); err != nil {
					o.log.Errorf("(stream.Nack) err: {%v}", err)
					return errors.Wrap(err, "stream.Nack")
				}
			}
			err = stream.Ack(event.EventAppeared)
			if err != nil {
				o.log.Errorf("(stream.Ack) err: {%v}", err)
				return errors.Wrap(err, "stream.Ack")
			}
			o.log.Infof("(ACK) event commit: {%v}", *event.EventAppeared.Commit)
		}
	}
}

func (o *mongoProjection) When(ctx context.Context, evt es.Event) error {
	ctx, span := tracing.StartProjectionTracerSpan(ctx, "mongoProjection.When", evt)
	defer span.Finish()
	span.LogFields(log.String("AggregateID", evt.GetAggregateID()), log.String("EventType", evt.GetEventType()))

	switch evt.GetEventType() {

	case v1.OrderCreated:
		return o.onOrderCreate(ctx, evt)
	case v1.OrderPaid:
		return o.onOrderPaid(ctx, evt)
	case v1.OrderSubmitted:
		return o.onSubmit(ctx, evt)
	case v1.ShoppingCartUpdated:
		return o.onShoppingCartUpdate(ctx, evt)
	case v1.OrderCanceled:
		return o.onCancel(ctx, evt)
	case v1.OrderCompleted:
		return o.onCompleted(ctx, evt)
	case v1.DeliveryAddressUpdated:
		return o.onDeliveryAddressUpdated(ctx, evt)

	default:
		o.log.Warnf("(mongoProjection) [When unknown EventType] eventType: {%s}", evt.EventType)
		return es.ErrInvalidEventType
	}
}
```
and **onOrderCreate** method deserialize data and handle event calling MongoDB repository insert method:
```go
func (o *mongoProjection) onOrderCreate(ctx context.Context, evt es.Event) error {
	span, ctx := opentracing.StartSpanFromContext(ctx, "mongoProjection.onOrderCreate")
	defer span.Finish()
	span.LogFields(log.String("AggregateID", evt.GetAggregateID()))

	var eventData v1.OrderCreatedEvent
	if err := evt.GetJsonData(&eventData); err != nil {
		tracing.TraceErr(span, err)
		return errors.Wrap(err, "evt.GetJsonData")
	}
	span.LogFields(log.String("AccountEmail", eventData.AccountEmail))

	op := &models.OrderProjection{
		OrderID:         aggregate.GetOrderAggregateID(evt.AggregateID),
		ShopItems:       eventData.ShopItems,
		Created:         true,
		AccountEmail:    eventData.AccountEmail,
		TotalPrice:      aggregate.GetShopItemsTotalPrice(eventData.ShopItems),
		DeliveryAddress: eventData.DeliveryAddress,
	}

	_, err := o.mongoRepo.Insert(ctx, op)
	if err != nil {
		return err
	}

	return nil
}
```

The OrderMongoRepository insert method is simple:
```go
func (m *mongoRepository) Insert(ctx context.Context, order *models.OrderProjection) (string, error) {
	span, ctx := opentracing.StartSpanFromContext(ctx, "mongoRepository.Insert")
	defer span.Finish()
	span.LogFields(log.String("OrderID", order.OrderID))

	_, err := m.getOrdersCollection().InsertOne(ctx, order, &options.InsertOneOptions{})
	if err != nil {
		tracing.TraceErr(span, err)
		return "", err
	}

	return order.OrderID, nil
}
```


<img src="https://i.postimg.cc/RCTBYN0X/Dev-Tools-Elastic-2022-01-17-17-37-32.png" />
The same idea for ElasticSearch projection implementation, we can use it for searching orders, so let's look at the repository Search method.
At this project as elastic search client used [elastic](https://github.com/olivere/elastic).

```go
func (e *elasticRepository) Search(ctx context.Context, text string, pq *utils.Pagination) (*dto.OrderSearchResponseDto, error) {
	span, ctx := opentracing.StartSpanFromContext(ctx, "elasticRepository.Search")
	defer span.Finish()
	span.LogFields(log.String("Search", text))

	shouldMatch := v7.NewBoolQuery().
		Should(v7.NewMatchPhrasePrefixQuery(shopItemTitle, text), v7.NewMatchPhrasePrefixQuery(shopItemDescription, text)).
		MinimumNumberShouldMatch(minimumNumberShouldMatch)

	searchResult, err := e.elasticClient.Search(e.cfg.ElasticIndexes.Orders).
		Query(shouldMatch).
		From(pq.GetOffset()).
		Explain(e.cfg.Elastic.Explain).
		FetchSource(e.cfg.Elastic.FetchSource).
		Version(e.cfg.Elastic.Version).
		Size(pq.GetSize()).
		Pretty(e.cfg.Elastic.Pretty).
		Do(ctx)
	if err != nil {
		tracing.TraceErr(span, err)
		return nil, errors.Wrap(err, "elasticClient.Search")
	}

	orders := make([]*models.OrderProjection, 0, len(searchResult.Hits.Hits))
	for _, hit := range searchResult.Hits.Hits {
		jsonBytes, err := hit.Source.MarshalJSON()
		if err != nil {
			tracing.TraceErr(span, err)
			return nil, errors.Wrap(err, "Source.MarshalJSON")
		}
		var order models.OrderProjection
		if err := json.Unmarshal(jsonBytes, &order); err != nil {
			tracing.TraceErr(span, err)
			return nil, errors.Wrap(err, "json.Unmarshal")
		}
		orders = append(orders, &order)
	}

	return &dto.OrderSearchResponseDto{
		Pagination: dto.Pagination{
			TotalCount: searchResult.TotalHits(),
			TotalPages: int64(pq.GetTotalPages(int(searchResult.TotalHits()))),
			Page:       int64(pq.GetPage()),
			Size:       int64(pq.GetSize()),
			HasMore:    pq.GetHasMore(int(searchResult.TotalHits())),
		},
		Orders: mappers.OrdersFromProjections(orders),
	}, nil
}
```

More details and source code of the full project you can find [here](https://github.com/AleksK1NG/Go-EventSourcing-CQRS),
of course in real-world applications, we have to implement many more necessary features, like circuit breaker, retries, rate limiters, etc., 
depending on the project it can be implemented in different ways, for example, you can use Kubernetes and Istio for some of them. 
I hope this article is usefully and helpfully, and be happy to receive any feedback or questions, feel free to contact me by email or any messengers :)