ego

README

eGo

eGo is a minimal library that help build event-sourcing and CQRS application through a simple interface, and it allows developers to describe their commands, events and states are defined using google protocol buffers. Under the hood, ego leverages Go-Akt to scale out and guarantee performant, reliable persistence.

Features

Domain Entity/Aggregate Root

The aggregate root is crucial for maintaining data consistency, especially in distributed systems. It defines how to handle the various commands (requests to perform actions) that are always directed at the aggregate root. In eGo commands sent the aggregate root are processed in order. When a command is processed, it may result in the generation of events, which are then stored in an event store. Every event persisted has a revision number and timestamp that can help track it. The aggregate root in eGo is responsible for defining how to handle events that are the result of command handlers. The end result of events handling is to build the new state of the aggregate root. When running in cluster mode, aggregate root are sharded.

• Commands handler: The command handlers define how to handle each incoming command, which validations must be applied, and finally, which events will be persisted if any. When there is no event to be persisted a nil can be returned as a no-op. Command handlers are the meat of the event sourced actor. They encode the business rules of your event sourced actor and act as a guardian of the Aggregate consistency. The command handler must first validate that the incoming command can be applied to the current model state. Any decision should be solely based on the data passed in the commands and the state of the Behavior. In case of successful validation, one or more events expressing the mutations are persisted. Once the events are persisted, they are applied to the state producing a new valid state.
• Events handler: The event handlers are used to mutate the state of the Aggregate by applying the events to it. Event handlers must be pure functions as they will be used when instantiating the Aggregate and replaying the event store.

Howto

To define an Aggregate Root, one needs to:

1. the state of the aggregate root using google protocol buffers message
2. the various commands that will be handled by the aggregate root
3. the various events that are result of the command handlers and that will be handled by the aggregate root to return the new state of the aggregate root
4. implements the EntityBehavior interface.

Events Stream

Every event handled by Aggregate Root are pushed to an events stream. That enables real-time processing of events without having to interact with the events store

Projection

One can add a projection to the eGo engine to help build a read model. Projections in eGo rely on an offset store to track how far they have consumed events persisted by the write model. The offset used in eGo is a timestamp-based offset. One can also:

• remove a given projection: this will stop the projection and remove it from the system
• check the status of a given projection

Events Store

One can implement a custom events store. See EventsStore. eGo comes packaged with two events store:

• Postgres: Schema can be found here
• Memory (for testing purpose only)

Offsets Store

One can implement a custom offsets store. See OffsetStore. eGo comes packaged with two offset store:

• Postgres: Schema can be found here
• Memory (for testing purpose only)

Cluster

The cluster mode heavily relies on Go-Akt clustering.

Mocks

eGo ships in some mocks

Examples

Check the examples

Installation

go get github.com/tochemey/ego

Sample

package main

import (
  "context"
  "errors"
  "log"
  "os"
  "os/signal"
  "syscall"
  "time"

  "github.com/google/uuid"
  "google.golang.org/protobuf/proto"

  "github.com/tochemey/ego/v3"
  "github.com/tochemey/ego/v3/eventstore/memory"
  samplepb "github.com/tochemey/ego/v3/example/pbs/sample/pb/v1"
)

func main() {
  // create the go context
  ctx := context.Background()
  // create the event store
  eventStore := memory.NewEventsStore()
  // connect the event store
  _ = eventStore.Connect(ctx)
  // create the ego engine
  engine := ego.NewEngine("Sample", eventStore)
  // start ego engine
  _ = engine.Start(ctx)
  // create a persistence id
  entityID := uuid.NewString()
  // create an entity behavior with a given id
  behavior := NewAccountBehavior(entityID)
  // create an entity
  _ = engine.Entity(ctx, behavior)

  // send some commands to the pid
  var command proto.Message
  // create an account
  command = &samplepb.CreateAccount{
    AccountId:      entityID,
    AccountBalance: 500.00,
  }
  // send the command to the actor. Please don't ignore the error in production grid code
  reply, _, _ := engine.SendCommand(ctx, entityID, command, time.Minute)
  account := reply.(*samplepb.Account)
  log.Printf("current balance: %v", account.GetAccountBalance())

  // send another command to credit the balance
  command = &samplepb.CreditAccount{
    AccountId: entityID,
    Balance:   250,
  }

  reply, _, _ = engine.SendCommand(ctx, entityID, command, time.Minute)
  account = reply.(*samplepb.Account)
  log.Printf("current balance: %v", account.GetAccountBalance())

  // capture ctrl+c
  interruptSignal := make(chan os.Signal, 1)
  signal.Notify(interruptSignal, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
  <-interruptSignal

  // disconnect the event store
  _ = eventStore.Disconnect(ctx)
  // stop the actor system
  _ = engine.Stop(ctx)
  os.Exit(0)
}

// AccountBehavior implements EntityBehavior
type AccountBehavior struct {
  id string
}

// make sure that AccountBehavior is a true persistence behavior
var _ ego.EntityBehavior = (*AccountBehavior)(nil)

// NewAccountBehavior creates an instance of AccountBehavior
func NewAccountBehavior(id string) *AccountBehavior {
  return &AccountBehavior{id: id}
}

// ID returns the id
func (a *AccountBehavior) ID() string {
  return a.id
}

// InitialState returns the initial state
func (a *AccountBehavior) InitialState() ego.State {
  return ego.State(new(samplepb.Account))
}

// HandleCommand handles every command that is sent to the persistent behavior
func (a *AccountBehavior) HandleCommand(_ context.Context, command ego.Command, _ ego.State) (events []ego.Event, err error) {
  switch cmd := command.(type) {
  case *samplepb.CreateAccount:
    // TODO in production grid app validate the command using the prior state
    return []ego.Event{
      &samplepb.AccountCreated{
        AccountId:      cmd.GetAccountId(),
        AccountBalance: cmd.GetAccountBalance(),
      },
    }, nil

  case *samplepb.CreditAccount:
    // TODO in production grid app validate the command using the prior state
    return []ego.Event{
      &samplepb.AccountCredited{
        AccountId:      cmd.GetAccountId(),
        AccountBalance: cmd.GetBalance(),
      },
    }, nil

  default:
    return nil, errors.New("unhandled command")
  }
}

// HandleEvent handles every event emitted
func (a *AccountBehavior) HandleEvent(_ context.Context, event ego.Event, priorState ego.State) (state ego.State, err error) {
  switch evt := event.(type) {
  case *samplepb.AccountCreated:
    return &samplepb.Account{
      AccountId:      evt.GetAccountId(),
      AccountBalance: evt.GetAccountBalance(),
    }, nil

  case *samplepb.AccountCredited:
    account := priorState.(*samplepb.Account)
    bal := account.GetAccountBalance() + evt.GetAccountBalance()
    return &samplepb.Account{
      AccountId:      evt.GetAccountId(),
      AccountBalance: bal,
    }, nil

  default:
    return nil, errors.New("unhandled event")
  }
}

Versioning

The version system adopted in eGo deviates a bit from the standard semantic versioning system. The version format is as follows:

• The MAJOR part of the version will stay at v3 for the meantime.
• The MINOR part of the version will cater for any new features, breaking changes with a note on the breaking changes.
• The PATCH part of the version will cater for dependencies upgrades, bug fixes, security patches and co.

The versioning will remain like v3.x.x until further notice.

Contribution

Contributions are welcome! The project adheres to Semantic Versioning and Conventional Commits. This repo uses Earthly.

There are two ways you can become a contributor:

1. Request to become a collaborator, and then you can just open pull requests against the repository without forking it.
2. Follow these steps

• Fork the repository
• Create a feature branch
• Set your docker credentials on your fork using the following secret names: DOCKER_USER and DOCKER_PASS
• Submit a pull request

Test & Linter

Prior to submitting a pull request, please run:

earthly +test

Documentation

Index

• Variables
• type Command
• type Engine
  • func NewEngine(name string, eventsStore eventstore.EventsStore, opts ...Option) *Engine
  • func (engine *Engine) AddProjection(ctx context.Context, name string, handler projection.Handler, ...) error
  • func (engine *Engine) Entity(ctx context.Context, behavior EntityBehavior) error
  • func (engine *Engine) IsProjectionRunning(ctx context.Context, name string) (bool, error)
  • func (engine *Engine) RemoveProjection(ctx context.Context, name string) error
  • func (engine *Engine) SendCommand(ctx context.Context, entityID string, cmd Command, timeout time.Duration) (resultingState State, revision uint64, err error)
  • func (engine *Engine) Start(ctx context.Context) error
  • func (engine *Engine) Started() bool
  • func (engine *Engine) Stop(ctx context.Context) error
  • func (engine *Engine) Subscribe() (eventstream.Subscriber, error)
• type EntityBehavior
• type Event
• type Option
  • func WithCluster(provider discovery.Provider, partitionCount uint64, minimumPeersQuorum uint16, ...) Option
  • func WithLogger(logger log.Logger) Option
• type OptionFunc
  • func (f OptionFunc) Apply(e *Engine)
• type State

Variables

var (
	// ErrEngineNotStarted is returned when the eGo engine has not started
	ErrEngineNotStarted = errors.New("eGo engine has not started")
	// ErrUndefinedEntityID is returned when sending a command to an undefined entity
	ErrUndefinedEntityID = errors.New("eGo entity id is not defined")
	// ErrCommandReplyUnmarshalling is returned when unmarshalling command reply failed
	ErrCommandReplyUnmarshalling = errors.New("failed to parse command reply")
)

Types

type Command

type Command proto.Message

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

Engine represents the engine that empowers the various entities

func NewEngine

func NewEngine(name string, eventsStore eventstore.EventsStore, opts ...Option) *Engine

NewEngine creates an instance of Engine

func (*Engine) AddProjection

func (engine *Engine) AddProjection(ctx context.Context, name string, handler projection.Handler, offsetStore offsetstore.OffsetStore, opts ...projection.Option) error

AddProjection add a projection to the running eGo engine and starts it

func (*Engine) Entity

func (engine *Engine) Entity(ctx context.Context, behavior EntityBehavior) error

Entity creates an entity. This will return the entity path that can be used to send command to the entity

func (*Engine) IsProjectionRunning

func (engine *Engine) IsProjectionRunning(ctx context.Context, name string) (bool, error)

IsProjectionRunning returns true when the projection is active and running One needs to check the error to see whether this function does not return a false negative

func (*Engine) RemoveProjection

func (engine *Engine) RemoveProjection(ctx context.Context, name string) error

RemoveProjection stops and removes a given projection from the engine

func (*Engine) SendCommand

func (engine *Engine) SendCommand(ctx context.Context, entityID string, cmd Command, timeout time.Duration) (resultingState State, revision uint64, err error)

SendCommand sends command to a given entity ref. This will return: 1. the resulting state after the command has been handled and the emitted event persisted 2. nil when there is no resulting state or no event persisted 3. an error in case of error

func (*Engine) Start

func (engine *Engine) Start(ctx context.Context) error

Start starts the ego engine

func (*Engine) Started

func (engine *Engine) Started() bool

Started returns true when the eGo engine has started

func (*Engine) Stop

func (engine *Engine) Stop(ctx context.Context) error

Stop stops the ego engine

func (*Engine) Subscribe

func (engine *Engine) Subscribe() (eventstream.Subscriber, error)

Subscribe creates an events subscriber

type EntityBehavior

type EntityBehavior interface {
	// ID defines the id that will be used in the event journal.
	// This helps track the entity in the events store.
	ID() string
	// InitialState returns the event sourced actor initial state.
	// This is set as the initial state when there are no snapshots found the entity
	InitialState() State
	// HandleCommand helps handle commands received by the event sourced actor. The command handlers define how to handle each incoming command,
	// which validations must be applied, and finally, which events will be persisted if any. When there is no event to be persisted a nil can
	// be returned as a no-op. Command handlers are the meat of the event sourced actor.
	// They encode the business rules of your event sourced actor and act as a guardian of the event sourced actor consistency.
	// The command eventSourcedHandler must first validate that the incoming command can be applied to the current model state.
	//  Any decision should be solely based on the data passed in the commands and the state of the Behavior.
	// In case of successful validation, one or more events expressing the mutations are persisted.
	// Once the events are persisted, they are applied to the state producing a new valid state.
	// Every event emitted are processed one after the other in the same order they were emitted to guarantee consistency.
	// It is at the discretion of the application developer to know in which order a given command should return the list of events
	// This is really powerful when a command needs to return two events. For instance, an OpenAccount command can result in two events: one is AccountOpened and the second is AccountCredited
	HandleCommand(ctx context.Context, command Command, priorState State) (events []Event, err error)
	// HandleEvent handle events emitted by the command handlers. The event handlers are used to mutate the state of the event sourced actor by applying the events to it.
	// Event handlers must be pure functions as they will be used when instantiating the event sourced actor and replaying the event journal.
	HandleEvent(ctx context.Context, event Event, priorState State) (state State, err error)
}

EntityBehavior defines an event sourced behavior when modeling a CQRS EntityBehavior.

type Event

type Event proto.Message

type Option

type Option interface {
	// Apply sets the Option value of a config.
	Apply(e *Engine)
}

Option is the interface that applies a configuration option.

func WithCluster

func WithCluster(provider discovery.Provider, partitionCount uint64, minimumPeersQuorum uint16, host string, remotingPort, gossipPort, peersPort int) Option

WithCluster enables cluster mode

func WithLogger

func WithLogger(logger log.Logger) Option

WithLogger sets the logger

type OptionFunc

type OptionFunc func(e *Engine)

OptionFunc implements the Option interface.

func (OptionFunc) Apply

func (f OptionFunc) Apply(e *Engine)

Apply applies the options to Engine

type State

type State proto.Message