Repository: looplab/eventhorizon
Pull request #309 (2021-03-29 10:13:01)

Feature / Publish on bus from the Event Store
### Description

Publish on bus from the Event Store. Should make for a more flexible setup when implementing an atomic save/publish later.

### Affected Components

- Event Aggregate Store
- Event Store
- Examples/setup

### Related Issues

### Solution and Design

The Event Store now has an option called `WithEventHandle` which adds an event handler to call after saving events. Typically this event handler is the event bus.

### Steps to test and verify



Patch:
diff --git a/eventbus/local/eventbus.go b/eventbus/local/eventbus.go
index b84ef0a..ee667fd 100644
--- a/eventbus/local/eventbus.go
+++ b/eventbus/local/eventbus.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"log"
 	"sync"
+	"time"
 
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/codec/json"
@@ -143,6 +144,9 @@ func (b *EventBus) handle(ctx context.Context, m eh.EventMatcher, h eh.EventHand
 	for {
 		select {
 		case data := <-ch:
+			// Artificial delay to simulate network.
+			time.Sleep(10 * time.Millisecond)
+
 			event, ctx, err := b.codec.UnmarshalEvent(ctx, data)
 			if err != nil {
 				err = fmt.Errorf("could not unmarshal event: %w", err)
diff --git a/eventstore.go b/eventstore.go
index 9a09afe..66fd98b 100644
--- a/eventstore.go
+++ b/eventstore.go
@@ -72,11 +72,15 @@ func (e EventStoreError) Cause() error {
 	return e.Unwrap()
 }
 
-// ErrNoEventsToAppend is when no events are available to append.
-var ErrNoEventsToAppend = errors.New("no events to append")
-
-// ErrInvalidEvent is when an event does not implement the Event interface.
-var ErrInvalidEvent = errors.New("invalid event")
-
-// ErrIncorrectEventVersion is when an event is for an other version of the aggregate.
-var ErrIncorrectEventVersion = errors.New("mismatching event version")
+var (
+	// ErrNoEventsToAppend is when no events are available to append.
+	ErrNoEventsToAppend = errors.New("no events to append")
+	// ErrInvalidEvent is when an event does not implement the Event interface.
+	ErrInvalidEvent = errors.New("invalid event")
+	// ErrIncorrectEventVersion is when an event is for an other version of the aggregate.
+	ErrIncorrectEventVersion = errors.New("mismatching event version")
+	// ErrCouldNotSaveEvents is when events could not be saved.
+	ErrCouldNotSaveEvents = errors.New("could not save events")
+	// ErrCouldNotHandleEvents is when the events could not be handeled after saving.
+	ErrCouldNotHandleEvents = errors.New("could not handle events")
+)
diff --git a/eventstore/memory/eventstore.go b/eventstore/memory/eventstore.go
index 7f2a7a6..c7de467 100644
--- a/eventstore/memory/eventstore.go
+++ b/eventstore/memory/eventstore.go
@@ -17,6 +17,7 @@ package memory
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 
 	"github.com/google/uuid"
@@ -26,8 +27,6 @@ import (
 )
 
 var (
-	// ErrCouldNotSaveAggregate is when an aggregate could not be saved.
-	ErrCouldNotSaveAggregate = errors.New("could not save aggregate")
 	// ErrCouldNotCreateEvent is when event data could not be created.
 	ErrCouldNotCreateEvent = errors.New("could not create event")
 )
@@ -35,16 +34,34 @@ var (
 // EventStore implements EventStore as an in memory structure.
 type EventStore struct {
 	// The outer map is with namespace as key, the inner with aggregate ID.
-	db   map[string]map[uuid.UUID]aggregateRecord
-	dbMu sync.RWMutex
+	db           map[string]map[uuid.UUID]aggregateRecord
+	dbMu         sync.RWMutex
+	eventHandler eh.EventHandler
 }
 
 // NewEventStore creates a new EventStore using memory as storage.
-func NewEventStore() *EventStore {
+func NewEventStore(options ...Option) (*EventStore, error) {
 	s := &EventStore{
 		db: map[string]map[uuid.UUID]aggregateRecord{},
 	}
-	return s
+	for _, option := range options {
+		if err := option(s); err != nil {
+			return nil, fmt.Errorf("error while applying option: %v", err)
+		}
+	}
+	return s, nil
+}
+
+// Option is an option setter used to configure creation.
+type Option func(*EventStore) error
+
+// WithEventHandler adds an event handler that will be called when saving events.
+// An example would be to add an event bus to publish events.
+func WithEventHandler(h eh.EventHandler) Option {
+	return func(s *EventStore) error {
+		s.eventHandler = h
+		return nil
+	}
 }
 
 // Save implements the Save method of the eventhorizon.EventStore interface.
@@ -106,7 +123,8 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		if aggregate, ok := s.db[ns][aggregateID]; ok {
 			if aggregate.Version != originalVersion {
 				return eh.EventStoreError{
-					Err:       ErrCouldNotSaveAggregate,
+					Err:       eh.ErrCouldNotSaveEvents,
+					BaseErr:   fmt.Errorf("invalid original version %d", originalVersion),
 					Namespace: eh.NamespaceFromContext(ctx),
 				}
 			}
@@ -118,6 +136,20 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		}
 	}
 
+	// Let the optional event handler handle the events. Aborts the transaction
+	// in case of error.
+	if s.eventHandler != nil {
+		for _, e := range events {
+			if err := s.eventHandler.HandleEvent(ctx, e); err != nil {
+				return eh.EventStoreError{
+					Err:       eh.ErrCouldNotHandleEvents,
+					BaseErr:   err,
+					Namespace: eh.NamespaceFromContext(ctx),
+				}
+			}
+		}
+	}
+
 	return nil
 }
 
diff --git a/eventstore/mongodb/eventstore.go b/eventstore/mongodb/eventstore.go
index 22c5000..e16bd8c 100644
--- a/eventstore/mongodb/eventstore.go
+++ b/eventstore/mongodb/eventstore.go
@@ -45,17 +45,14 @@ var (
 	ErrCouldNotMarshalEvent = errors.New("could not marshal event")
 	// ErrCouldNotUnmarshalEvent is when an event could not be unmarshaled into a concrete type.
 	ErrCouldNotUnmarshalEvent = errors.New("could not unmarshal event")
-	// ErrCouldNotLoadAggregate is when an aggregate could not be loaded.
-	ErrCouldNotLoadAggregate = errors.New("could not load aggregate")
-	// ErrCouldNotSaveAggregate is when an aggregate could not be saved.
-	ErrCouldNotSaveAggregate = errors.New("could not save aggregate")
 )
 
 // EventStore implements an EventStore for MongoDB.
 type EventStore struct {
-	client   *mongo.Client
-	dbPrefix string
-	dbName   func(ctx context.Context) string
+	client       *mongo.Client
+	dbPrefix     string
+	dbName       func(ctx context.Context) string
+	eventHandler eh.EventHandler
 }
 
 // NewEventStore creates a new EventStore with a MongoDB URI: `mongodb://hostname`.
@@ -90,8 +87,7 @@ func NewEventStoreWithClient(client *mongo.Client, dbPrefix string, options ...O
 	}
 
 	for _, option := range options {
-		err := option(s)
-		if err != nil {
+		if err := option(s); err != nil {
 			return nil, fmt.Errorf("error while applying option: %w", err)
 		}
 	}
@@ -120,6 +116,15 @@ func WithDBName(dbName func(context.Context) string) Option {
 	}
 }
 
+// WithEventHandler adds an event handler that will be called when saving events.
+// An example would be to add an event bus to publish events.
+func WithEventHandler(h eh.EventHandler) Option {
+	return func(s *EventStore) error {
+		s.eventHandler = h
+		return nil
+	}
+}
+
 // Save implements the Save method of the eventhorizon.EventStore interface.
 func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersion int) error {
 	if len(events) == 0 {
@@ -170,7 +175,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 
 		if _, err := c.InsertOne(ctx, aggregate); err != nil {
 			return eh.EventStoreError{
-				Err:       ErrCouldNotSaveAggregate,
+				Err:       eh.ErrCouldNotSaveEvents,
 				BaseErr:   err,
 				Namespace: eh.NamespaceFromContext(ctx),
 			}
@@ -190,19 +195,32 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 			},
 		); err != nil {
 			return eh.EventStoreError{
-				Err:       ErrCouldNotSaveAggregate,
+				Err:       eh.ErrCouldNotSaveEvents,
 				BaseErr:   err,
 				Namespace: eh.NamespaceFromContext(ctx),
 			}
 		} else if r.MatchedCount == 0 {
 			return eh.EventStoreError{
-				Err:       ErrCouldNotSaveAggregate,
+				Err:       eh.ErrCouldNotSaveEvents,
 				BaseErr:   fmt.Errorf("invalid original version %d", originalVersion),
 				Namespace: eh.NamespaceFromContext(ctx),
 			}
 		}
 	}
 
+	// Let the optional event handler handle the events.
+	if s.eventHandler != nil {
+		for _, e := range events {
+			if err := s.eventHandler.HandleEvent(ctx, e); err != nil {
+				return eh.EventStoreError{
+					Err:       eh.ErrCouldNotHandleEvents,
+					BaseErr:   err,
+					Namespace: eh.NamespaceFromContext(ctx),
+				}
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -292,7 +310,7 @@ func (s *EventStore) Replace(ctx context.Context, event eh.Event) error {
 		},
 	); err != nil {
 		return eh.EventStoreError{
-			Err:       ErrCouldNotSaveAggregate,
+			Err:       eh.ErrCouldNotSaveEvents,
 			BaseErr:   err,
 			Namespace: eh.NamespaceFromContext(ctx),
 		}
@@ -318,7 +336,7 @@ func (s *EventStore) RenameEvent(ctx context.Context, from, to eh.EventType) err
 		},
 	); err != nil {
 		return eh.EventStoreError{
-			Err:       ErrCouldNotSaveAggregate,
+			Err:       eh.ErrCouldNotSaveEvents,
 			BaseErr:   err,
 			Namespace: eh.NamespaceFromContext(ctx),
 		}
diff --git a/examples/guestlist/domains/guestlist/setup.go b/examples/guestlist/domains/guestlist/setup.go
index 7abf13f..8aefa26 100644
--- a/examples/guestlist/domains/guestlist/setup.go
+++ b/examples/guestlist/domains/guestlist/setup.go
@@ -42,7 +42,7 @@ func Setup(
 		eh.UseEventHandlerMiddleware(&Logger{}, observer.Middleware))
 
 	// Create the aggregate repository.
-	aggregateStore, err := events.NewAggregateStore(eventStore, eventBus)
+	aggregateStore, err := events.NewAggregateStore(eventStore)
 	if err != nil {
 		log.Fatalf("could not create aggregate store: %s", err)
 	}
diff --git a/examples/todomvc/backend/domains/todo/setup.go b/examples/todomvc/backend/domains/todo/setup.go
index 2c41a1f..cb24c68 100644
--- a/examples/todomvc/backend/domains/todo/setup.go
+++ b/examples/todomvc/backend/domains/todo/setup.go
@@ -43,7 +43,7 @@ func SetupDomain(
 	}, projector)
 
 	// Create the event sourced aggregate repository.
-	aggregateStore, err := events.NewAggregateStore(eventStore, eventBus)
+	aggregateStore, err := events.NewAggregateStore(eventStore)
 	if err != nil {
 		return fmt.Errorf("could not create aggregate store: %w", err)
 	}
diff --git a/examples/todomvc/backend/main.go b/examples/todomvc/backend/main.go
index ba728d6..22fdc02 100644
--- a/examples/todomvc/backend/main.go
+++ b/examples/todomvc/backend/main.go
@@ -34,7 +34,6 @@ import (
 	mongoRepo "github.com/looplab/eventhorizon/repo/mongodb"
 	tracingRepo "github.com/looplab/eventhorizon/repo/tracing"
 	"github.com/looplab/eventhorizon/repo/version"
-	versionRepo "github.com/looplab/eventhorizon/repo/version"
 
 	"github.com/looplab/eventhorizon/examples/todomvc/backend/domains/todo"
 	"github.com/looplab/eventhorizon/examples/todomvc/backend/handler"
@@ -44,11 +43,11 @@ func main() {
 	log.Println("starting TodoMVC backend")
 
 	// Use MongoDB in Docker with fallback to localhost.
-	dbAddr := os.Getenv("MONGODB_ADDR")
-	if dbAddr == "" {
-		dbAddr = "localhost:27017"
+	addr := os.Getenv("MONGODB_ADDR")
+	if addr == "" {
+		addr = "localhost:27017"
 	}
-	dbURL := "mongodb://" + dbAddr
+	url := "mongodb://" + addr
 	dbPrefix := "todomvc-example"
 
 	// Connect to localhost if not running inside docker
@@ -70,13 +69,6 @@ func main() {
 	// Create an event bus.
 	commandBus := bus.NewCommandHandler()
 
-	// Create the event store.
-	var eventStore eh.EventStore
-	if eventStore, err = mongoEventStore.NewEventStore(dbURL, dbPrefix); err != nil {
-		log.Fatal("could not create event store: ", err)
-	}
-	eventStore = tracingEventStore.NewEventStore(eventStore)
-
 	// Create the event bus that distributes events.
 	var eventBus eh.EventBus
 	if eventBus, err = gcpEventBus.NewEventBus("project-id", dbPrefix); err != nil {
@@ -91,6 +83,15 @@ func main() {
 	// Wrap the event bus to add tracing.
 	eventBus = tracingEventBus.NewEventBus(eventBus)
 
+	// Create the event store.
+	var eventStore eh.EventStore
+	if eventStore, err = mongoEventStore.NewEventStore(url, dbPrefix,
+		mongoEventStore.WithEventHandler(eventBus), // Add the event bus as a handler after save.
+	); err != nil {
+		log.Fatal("could not create event store: ", err)
+	}
+	eventStore = tracingEventStore.NewEventStore(eventStore)
+
 	ctx, cancel := context.WithCancel(context.Background())
 
 	// Add an event logger as an observer.
@@ -105,10 +106,10 @@ func main() {
 
 	// Create the repository and wrap in a version repository.
 	var todoRepo eh.ReadWriteRepo
-	if todoRepo, err = mongoRepo.NewRepo(dbURL, dbPrefix, "todos"); err != nil {
+	if todoRepo, err = mongoRepo.NewRepo(url, dbPrefix, "todos"); err != nil {
 		log.Fatal("could not create invitation repository: ", err)
 	}
-	todoRepo = versionRepo.NewRepo(todoRepo)
+	todoRepo = version.NewRepo(todoRepo)
 	todoRepo = tracingRepo.NewRepo(todoRepo)
 
 	// Setup the Todo domain.


Test patch:
diff --git a/aggregatestore/events/aggregatestore.go b/aggregatestore/events/aggregatestore.go
index cb435df..b323512 100644
--- a/aggregatestore/events/aggregatestore.go
+++ b/aggregatestore/events/aggregatestore.go
@@ -26,15 +26,12 @@ import (
 // uses an event store for loading and saving events used to build the aggregate
 // and an event handler to handle resulting events.
 type AggregateStore struct {
-	store        eh.EventStore
-	eventHandler eh.EventHandler
+	store eh.EventStore
 }
 
 var (
 	// ErrInvalidEventStore is when a dispatcher is created with a nil event store.
 	ErrInvalidEventStore = errors.New("invalid event store")
-	// ErrInvalidEventBus is when a dispatcher is created with a nil event bus.
-	ErrInvalidEventBus = errors.New("invalid event bus")
 	// ErrInvalidAggregateType is when  the aggregate does not implement event.Aggregte.
 	ErrInvalidAggregateType = errors.New("invalid aggregate type")
 	// ErrMismatchedEventType occurs when loaded events from ID does not match aggregate type.
@@ -68,18 +65,12 @@ func (e ApplyEventError) Cause() error {
 // NewAggregateStore creates a aggregate store with an event store and an event
 // handler that will handle resulting events (for example by publishing them
 // on an event bus).
-func NewAggregateStore(store eh.EventStore, eventHandler eh.EventHandler) (*AggregateStore, error) {
+func NewAggregateStore(store eh.EventStore) (*AggregateStore, error) {
 	if store == nil {
 		return nil, ErrInvalidEventStore
 	}
-
-	if eventHandler == nil {
-		return nil, ErrInvalidEventBus
-	}
-
 	d := &AggregateStore{
-		store:        store,
-		eventHandler: eventHandler,
+		store: store,
 	}
 	return d, nil
 }
@@ -133,12 +124,6 @@ func (r *AggregateStore) Save(ctx context.Context, agg eh.Aggregate) error {
 		return err
 	}
 
-	for _, e := range events {
-		if err := r.eventHandler.HandleEvent(ctx, e); err != nil {
-			return err
-		}
-	}
-
 	return nil
 }
 
diff --git a/aggregatestore/events/aggregatestore_test.go b/aggregatestore/events/aggregatestore_test.go
index 9354e87..2eb3c56 100644
--- a/aggregatestore/events/aggregatestore_test.go
+++ b/aggregatestore/events/aggregatestore_test.go
@@ -30,11 +30,8 @@ func TestNewAggregateStore(t *testing.T) {
 	eventStore := &mocks.EventStore{
 		Events: make([]eh.Event, 0),
 	}
-	bus := &mocks.EventBus{
-		Events: make([]eh.Event, 0),
-	}
 
-	store, err := NewAggregateStore(nil, bus)
+	store, err := NewAggregateStore(nil)
 	if !errors.Is(err, ErrInvalidEventStore) {
 		t.Error("there should be a ErrInvalidEventStore error:", err)
 	}
@@ -42,15 +39,7 @@ func TestNewAggregateStore(t *testing.T) {
 		t.Error("there should be no aggregate store:", store)
 	}
 
-	store, err = NewAggregateStore(eventStore, nil)
-	if !errors.Is(err, ErrInvalidEventBus) {
-		t.Error("there should be a ErrInvalidEventBus error:", err)
-	}
-	if store != nil {
-		t.Error("there should be no aggregate store:", store)
-	}
-
-	store, err = NewAggregateStore(eventStore, bus)
+	store, err = NewAggregateStore(eventStore)
 	if err != nil {
 		t.Error("there should be no error:", err)
 	}
@@ -60,7 +49,7 @@ func TestNewAggregateStore(t *testing.T) {
 }
 
 func TestAggregateStore_LoadNoEvents(t *testing.T) {
-	store, _, _ := createStore(t)
+	store, _ := createStore(t)
 
 	ctx := context.Background()
 
@@ -82,7 +71,7 @@ func TestAggregateStore_LoadNoEvents(t *testing.T) {
 }
 
 func TestAggregateStore_LoadEvents(t *testing.T) {
-	store, eventStore, _ := createStore(t)
+	store, eventStore := createStore(t)
 
 	ctx := context.Background()
 
@@ -124,7 +113,7 @@ func TestAggregateStore_LoadEvents(t *testing.T) {
 }
 
 func TestAggregateStore_LoadEvents_MismatchedEventType(t *testing.T) {
-	store, eventStore, _ := createStore(t)
+	store, eventStore := createStore(t)
 
 	ctx := context.Background()
 
@@ -153,7 +142,7 @@ func TestAggregateStore_LoadEvents_MismatchedEventType(t *testing.T) {
 }
 
 func TestAggregateStore_SaveEvents(t *testing.T) {
-	store, eventStore, bus := createStore(t)
+	store, eventStore := createStore(t)
 
 	ctx := context.Background()
 
@@ -188,12 +177,8 @@ func TestAggregateStore_SaveEvents(t *testing.T) {
 		t.Error("the version should be 1:", agg.Version())
 	}
 
-	if !reflect.DeepEqual(bus.Events, []eh.Event{event1}) {
-		t.Error("there should be an event on the bus:", bus.Events)
-	}
-
 	// Store error.
-	event1 = agg.AppendEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp)
+	agg.AppendEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp)
 	aggregateErr := errors.New("aggregate error")
 	eventStore.Err = aggregateErr
 	err = store.Save(ctx, agg)
@@ -203,26 +188,17 @@ func TestAggregateStore_SaveEvents(t *testing.T) {
 	eventStore.Err = nil
 
 	// Aggregate error.
-	event1 = agg.AppendEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp)
+	agg.AppendEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp)
 	agg.err = errors.New("error")
 	err = store.Save(ctx, agg)
 	if _, ok := err.(ApplyEventError); !ok {
 		t.Error("there should be an error of type ApplyEventError:", err)
 	}
 	agg.err = nil
-
-	// Bus error.
-	event1 = agg.AppendEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp)
-	busErr := errors.New("bus error")
-	bus.Err = busErr
-	err = store.Save(ctx, agg)
-	if !errors.Is(err, busErr) {
-		t.Error("the error should be correct:", err)
-	}
 }
 
 func TestAggregateStore_AggregateNotRegistered(t *testing.T) {
-	store, _, _ := createStore(t)
+	store, _ := createStore(t)
 
 	ctx := context.Background()
 
@@ -236,21 +212,18 @@ func TestAggregateStore_AggregateNotRegistered(t *testing.T) {
 	}
 }
 
-func createStore(t *testing.T) (*AggregateStore, *mocks.EventStore, *mocks.EventBus) {
+func createStore(t *testing.T) (*AggregateStore, *mocks.EventStore) {
 	eventStore := &mocks.EventStore{
 		Events: make([]eh.Event, 0),
 	}
-	bus := &mocks.EventBus{
-		Events: make([]eh.Event, 0),
-	}
-	store, err := NewAggregateStore(eventStore, bus)
+	store, err := NewAggregateStore(eventStore)
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 	if store == nil {
 		t.Fatal("there should be a aggregate store")
 	}
-	return store, eventStore, bus
+	return store, eventStore
 }
 
 func init() {
diff --git a/eventstore/memory/eventstore_test.go b/eventstore/memory/eventstore_test.go
index 3a0a807..2732a40 100644
--- a/eventstore/memory/eventstore_test.go
+++ b/eventstore/memory/eventstore_test.go
@@ -17,13 +17,19 @@ package memory
 import (
 	"context"
 	"testing"
+	"time"
 
+	"github.com/google/uuid"
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventstore"
+	"github.com/looplab/eventhorizon/mocks"
 )
 
 func TestEventStore(t *testing.T) {
-	store := NewEventStore()
+	store, err := NewEventStore()
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
 	if store == nil {
 		t.Fatal("there should be a store")
 	}
@@ -34,3 +40,51 @@ func TestEventStore(t *testing.T) {
 	eventstore.AcceptanceTest(t, ctx, store)
 	eventstore.MaintainerAcceptanceTest(t, context.Background(), store)
 }
+
+func TestWithEventHandler(t *testing.T) {
+	h := &mocks.EventBus{}
+
+	store, err := NewEventStore(WithEventHandler(h))
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+	if store == nil {
+		t.Fatal("there should be a store")
+	}
+
+	ctx := context.Background()
+
+	// The event handler should be called.
+	id1 := uuid.New()
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+	event1 := eh.NewEventForAggregate(mocks.EventType, &mocks.EventData{Content: "event1"},
+		timestamp, mocks.AggregateType, id1, 1)
+	err = store.Save(ctx, []eh.Event{event1}, 0)
+	if err != nil {
+		t.Error("there should be no error:", err)
+	}
+	expected := []eh.Event{event1}
+	// The saved events should be ok.
+	events, err := store.Load(ctx, id1)
+	if err != nil {
+		t.Error("there should be no error:", err)
+	}
+	// The stored events should be ok.
+	for i, event := range events {
+		if err := eh.CompareEvents(event, expected[i], eh.IgnoreVersion()); err != nil {
+			t.Error("the stored event was incorrect:", err)
+		}
+		if event.Version() != i+1 {
+			t.Error("the event version should be correct:", event, event.Version())
+		}
+	}
+	// The handeled events should be ok.
+	for i, event := range h.Events {
+		if err := eh.CompareEvents(event, expected[i], eh.IgnoreVersion()); err != nil {
+			t.Error("the handeled event was incorrect:", err)
+		}
+		if event.Version() != i+1 {
+			t.Error("the event version should be correct:", event, event.Version())
+		}
+	}
+}
diff --git a/eventstore/mongodb/eventstore_test.go b/eventstore/mongodb/eventstore_test.go
index 9f42b6f..60b6f40 100644
--- a/eventstore/mongodb/eventstore_test.go
+++ b/eventstore/mongodb/eventstore_test.go
@@ -18,9 +18,12 @@ import (
 	"context"
 	"os"
 	"testing"
+	"time"
 
+	"github.com/google/uuid"
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventstore"
+	"github.com/looplab/eventhorizon/mocks"
 )
 
 func TestEventStoreIntegration(t *testing.T) {
@@ -60,3 +63,64 @@ func TestEventStoreIntegration(t *testing.T) {
 	eventstore.AcceptanceTest(t, customNamespaceCtx, store)
 	eventstore.MaintainerAcceptanceTest(t, context.Background(), store)
 }
+
+func TestWithEventHandlerIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	// Use MongoDB in Docker with fallback to localhost.
+	url := os.Getenv("MONGO_HOST")
+	if url == "" {
+		url = "localhost:27017"
+	}
+	url = "mongodb://" + url
+
+	h := &mocks.EventBus{}
+
+	store, err := NewEventStore(url, "test",
+		WithEventHandler(h),
+	)
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+	if store == nil {
+		t.Fatal("there should be a store")
+	}
+
+	ctx := context.Background()
+
+	// The event handler should be called.
+	id1 := uuid.New()
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+	event1 := eh.NewEventForAggregate(mocks.EventType, &mocks.EventData{Content: "event1"},
+		timestamp, mocks.AggregateType, id1, 1)
+	err = store.Save(ctx, []eh.Event{event1}, 0)
+	if err != nil {
+		t.Error("there should be no error:", err)
+	}
+	expected := []eh.Event{event1}
+	// The saved events should be ok.
+	events, err := store.Load(ctx, id1)
+	if err != nil {
+		t.Error("there should be no error:", err)
+	}
+	// The stored events should be ok.
+	for i, event := range events {
+		if err := eh.CompareEvents(event, expected[i], eh.IgnoreVersion()); err != nil {
+			t.Error("the stored event was incorrect:", err)
+		}
+		if event.Version() != i+1 {
+			t.Error("the event version should be correct:", event, event.Version())
+		}
+	}
+	// The handled events should be ok.
+	for i, event := range h.Events {
+		if err := eh.CompareEvents(event, expected[i], eh.IgnoreVersion()); err != nil {
+			t.Error("the handeled event was incorrect:", err)
+		}
+		if event.Version() != i+1 {
+			t.Error("the event version should be correct:", event, event.Version())
+		}
+	}
+}
diff --git a/eventstore/recorder/eventstore_test.go b/eventstore/recorder/eventstore_test.go
index 23be935..3a91254 100644
--- a/eventstore/recorder/eventstore_test.go
+++ b/eventstore/recorder/eventstore_test.go
@@ -28,7 +28,10 @@ import (
 
 // NOTE: Not named "Integration" to enable running with the unit tests.
 func TestEventStore(t *testing.T) {
-	baseStore := memory.NewEventStore()
+	baseStore, err := memory.NewEventStore()
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
 	store := NewEventStore(baseStore)
 	if store == nil {
 		t.Fatal("there should be a store")
@@ -66,7 +69,7 @@ func TestEventStore(t *testing.T) {
 	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
 	event7 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
 		eh.ForAggregate(mocks.AggregateType, event1.AggregateID(), 7))
-	err := store.Save(ctx, []eh.Event{event7}, 6)
+	err = store.Save(ctx, []eh.Event{event7}, 6)
 	if err != nil {
 		t.Error("there should be no error:", err)
 	}
diff --git a/eventstore/tracing/eventstore_test.go b/eventstore/tracing/eventstore_test.go
index 621993f..4afcc67 100644
--- a/eventstore/tracing/eventstore_test.go
+++ b/eventstore/tracing/eventstore_test.go
@@ -25,7 +25,10 @@ import (
 
 // NOTE: Not named "Integration" to enable running with the unit tests.
 func TestEventStore(t *testing.T) {
-	innerStore := memory.NewEventStore()
+	innerStore, err := memory.NewEventStore()
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
 	if innerStore == nil {
 		t.Fatal("there should be a store")
 	}
diff --git a/examples/guestlist/memory/memory_test.go b/examples/guestlist/memory/memory_test.go
index ee22ce1..9531eba 100644
--- a/examples/guestlist/memory/memory_test.go
+++ b/examples/guestlist/memory/memory_test.go
@@ -25,32 +25,37 @@ import (
 	"github.com/google/uuid"
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/commandhandler/bus"
-	eventbus "github.com/looplab/eventhorizon/eventbus/local"
-	eventstore "github.com/looplab/eventhorizon/eventstore/memory"
+	localEventBus "github.com/looplab/eventhorizon/eventbus/local"
+	memoryEventStore "github.com/looplab/eventhorizon/eventstore/memory"
 	"github.com/looplab/eventhorizon/examples/guestlist/domains/guestlist"
-	repo "github.com/looplab/eventhorizon/repo/memory"
+	"github.com/looplab/eventhorizon/repo/memory"
 )
 
 // NOTE: Not named "Integration" to enable running with the unit tests.
 func Example() {
-	// Create the event store.
-	eventStore := eventstore.NewEventStore()
-
 	// Create the event bus that distributes events.
-	eventBus := eventbus.NewEventBus()
+	eventBus := localEventBus.NewEventBus()
 	go func() {
 		for e := range eventBus.Errors() {
 			log.Printf("eventbus: %s", e.Error())
 		}
 	}()
 
+	// Create the event store.
+	eventStore, err := memoryEventStore.NewEventStore(
+		memoryEventStore.WithEventHandler(eventBus), // Add the event bus as a handler after save.
+	)
+	if err != nil {
+		log.Fatalf("could not create event store: %s", err)
+	}
+
 	// Create the command bus.
 	commandBus := bus.NewCommandHandler()
 
 	// Create the read repositories.
-	invitationRepo := repo.NewRepo()
+	invitationRepo := memory.NewRepo()
 	invitationRepo.SetEntityFactory(func() eh.Entity { return &guestlist.Invitation{} })
-	guestListRepo := repo.NewRepo()
+	guestListRepo := memory.NewRepo()
 	guestListRepo.SetEntityFactory(func() eh.Entity { return &guestlist.GuestList{} })
 
 	// Set the namespace to use.
diff --git a/examples/guestlist/mongodb/mongodb_test.go b/examples/guestlist/mongodb/mongodb_test.go
index 8560864..55d488d 100644
--- a/examples/guestlist/mongodb/mongodb_test.go
+++ b/examples/guestlist/mongodb/mongodb_test.go
@@ -25,13 +25,15 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/commandhandler/bus"
-	eventbus "github.com/looplab/eventhorizon/eventbus/local"
-	eventstore "github.com/looplab/eventhorizon/eventstore/mongodb"
-	"github.com/looplab/eventhorizon/examples/guestlist/domains/guestlist"
-	repo "github.com/looplab/eventhorizon/repo/mongodb"
+	localEventBus "github.com/looplab/eventhorizon/eventbus/local"
+	mongoEventStore "github.com/looplab/eventhorizon/eventstore/mongodb"
+	"github.com/looplab/eventhorizon/repo/mongodb"
 	"github.com/looplab/eventhorizon/repo/version"
+
+	"github.com/looplab/eventhorizon/examples/guestlist/domains/guestlist"
 )
 
 func ExampleIntegration() {
@@ -51,33 +53,37 @@ guest list: 4 invited - 3 accepted, 1 declined - 2 confirmed, 1 denied`)
 		addr = "localhost:27017"
 	}
 	url := "mongodb://" + addr
-
-	// Create the event store.
-	eventStore, err := eventstore.NewEventStore(url, "demo")
-	if err != nil {
-		log.Fatalf("could not create event store: %s", err)
-	}
+	dbPrefix := "guestlist-example"
 
 	// Create the event bus that distributes events.
-	eventBus := eventbus.NewEventBus()
+	eventBus := localEventBus.NewEventBus(nil)
 	go func() {
 		for e := range eventBus.Errors() {
 			log.Printf("eventbus: %s", e.Error())
 		}
 	}()
 
+	// Create the event store.
+	eventStore, err := mongoEventStore.NewEventStore(url, dbPrefix,
+		mongoEventStore.WithEventHandler(eventBus), // Add the event bus as a handler after save.
+	)
+	if err != nil {
+		log.Fatalf("could not create event store: %s", err)
+	}
+
 	// Create the command bus.
 	commandBus := bus.NewCommandHandler()
 
 	// Create the read repositories.
-	invitationRepo, err := repo.NewRepo(url, "demo", "invitations")
+	invitationRepo, err := mongodb.NewRepo(url, dbPrefix, "invitations")
 	if err != nil {
 		log.Fatalf("could not create invitation repository: %s", err)
 	}
 	invitationRepo.SetEntityFactory(func() eh.Entity { return &guestlist.Invitation{} })
 	// A version repo is needed for the projector to handle eventual consistency.
 	invitationVersionRepo := version.NewRepo(invitationRepo)
-	guestListRepo, err := repo.NewRepo(url, "demo", "guest_lists")
+
+	guestListRepo, err := mongodb.NewRepo(url, dbPrefix, "guest_lists")
 	if err != nil {
 		log.Fatalf("could not create guest list repository: %s", err)
 	}
diff --git a/examples/todomvc/backend/handler/handler_test.go b/examples/todomvc/backend/handler/handler_test.go
index 1bda7d0..24294c0 100644
--- a/examples/todomvc/backend/handler/handler_test.go
+++ b/examples/todomvc/backend/handler/handler_test.go
@@ -733,13 +733,23 @@ func NewTestSession(ctx context.Context) (
 	eh.EventBus,
 	eh.ReadWriteRepo,
 ) {
+	// Create the event bus that distributes events.
+	eventBus := localEventBus.NewEventBus(nil)
+
+	// Create the event store.
+	eventStore, _ := memoryEventStore.NewEventStore(
+		memoryEventStore.WithEventHandler(eventBus), // Add the event bus as a handler after save.
+	)
+
+	// Create the command bus.
 	commandBus := bus.NewCommandHandler()
-	eventStore := memoryEventStore.NewEventStore()
-	eventBus := localEventBus.NewEventBus()
+
+	// Create the read repositories.
 	todoRepo := memory.NewRepo()
 	if err := todo.SetupDomain(ctx, commandBus, eventStore, eventBus, todoRepo); err != nil {
 		log.Println("could not setup domain:", err)
 	}
+
 	return commandBus, eventBus, todoRepo
 }
 
@@ -749,20 +759,15 @@ func NewIntegrationTestSession(ctx context.Context) (
 	eh.ReadWriteRepo,
 ) {
 	// Use MongoDB in Docker with fallback to localhost.
-	dbAddr := os.Getenv("MONGODB_ADDR")
-	if dbAddr == "" {
-		dbAddr = "localhost:27017"
+	addr := os.Getenv("MONGODB_ADDR")
+	if addr == "" {
+		addr = "localhost:27017"
 	}
-	dbURL := "mongodb://" + dbAddr
+	url := "mongodb://" + addr
 	dbPrefix := "todomvc-example"
 
 	commandBus := bus.NewCommandHandler()
 
-	eventStore, err := mongoEventStore.NewEventStore(dbURL, dbPrefix)
-	if err != nil {
-		log.Fatalf("could not create event store: %s", err)
-	}
-
 	eventBus, err := gcpEventBus.NewEventBus("project-id", dbPrefix)
 	if err != nil {
 		log.Fatalf("could not create event bus: %s", err)
@@ -773,7 +778,14 @@ func NewIntegrationTestSession(ctx context.Context) (
 		}
 	}()
 
-	repo, err := mongodb.NewRepo(dbURL, dbPrefix, "todos")
+	eventStore, err := mongoEventStore.NewEventStore(url, dbPrefix,
+		mongoEventStore.WithEventHandler(eventBus), // Add the event bus as a handler after save.
+	)
+	if err != nil {
+		log.Fatalf("could not create event store: %s", err)
+	}
+
+	repo, err := mongodb.NewRepo(url, dbPrefix, "todos")
 	if err != nil {
 		log.Fatalf("could not create invitation repository: %s", err)
 	}
@@ -788,6 +800,7 @@ func NewIntegrationTestSession(ctx context.Context) (
 		log.Println("could not clear DB:", err)
 	}
 
+	// Setup the todo domain.
 	if err := todo.SetupDomain(ctx, commandBus, eventStore, eventBus, todoRepo); err != nil {
 		log.Println("could not setup domain:", err)
 	}