Repository: looplab/eventhorizon
Pull request #314 (2021-04-30 12:15:49)

Refactor / Clearer names for event-related methods
### Description

Minor refactoring of EventStore (and related) interface, mainly unrelated work from #311. Also splits out the event store maintenance functionality to separate files.

### Affected Components

- Event Store

### Related Issues

### Solution and Design

### Steps to test and verify



Patch:
diff --git a/codec/bson/codec.go b/codec/bson/codec.go
index 1a8e8a8..15f01f7 100644
--- a/codec/bson/codec.go
+++ b/codec/bson/codec.go
@@ -32,11 +32,11 @@ type EventCodec struct{}
 // MarshalEvent marshals an event into bytes in BSON format.
 func (c *EventCodec) MarshalEvent(ctx context.Context, event eh.Event) ([]byte, error) {
 	e := evt{
-		AggregateID:   event.AggregateID().String(),
-		AggregateType: event.AggregateType(),
 		EventType:     event.EventType(),
-		Version:       event.Version(),
 		Timestamp:     event.Timestamp(),
+		AggregateType: event.AggregateType(),
+		AggregateID:   event.AggregateID().String(),
+		Version:       event.Version(),
 		Metadata:      event.Metadata(),
 		Context:       eh.MarshalContext(ctx),
 	}
diff --git a/codec/json/codec.go b/codec/json/codec.go
index b56f1b2..5d8f77a 100644
--- a/codec/json/codec.go
+++ b/codec/json/codec.go
@@ -32,11 +32,11 @@ type EventCodec struct{}
 // MarshalEvent marshals an event into bytes in JSON format.
 func (c *EventCodec) MarshalEvent(ctx context.Context, event eh.Event) ([]byte, error) {
 	e := evt{
-		AggregateID:   event.AggregateID().String(),
-		AggregateType: event.AggregateType(),
 		EventType:     event.EventType(),
-		Version:       event.Version(),
 		Timestamp:     event.Timestamp(),
+		AggregateType: event.AggregateType(),
+		AggregateID:   event.AggregateID().String(),
+		Version:       event.Version(),
 		Metadata:      event.Metadata(),
 		Context:       eh.MarshalContext(ctx),
 	}
diff --git a/compare.go b/compare.go
index a2661a7..581cc77 100644
--- a/compare.go
+++ b/compare.go
@@ -64,15 +64,15 @@ func CompareEvents(e1, e2 Event, options ...CompareOption) error {
 			return fmt.Errorf("incorrect timestamp: %s (should be %s)", e1.Timestamp(), e2.Timestamp())
 		}
 	}
-	if e1.AggregateID() != e2.AggregateID() {
-		return fmt.Errorf("incorrect aggregate ID: %s (should be %s)", e1.AggregateID(), e2.AggregateID())
-	}
 	if e1.AggregateType() != e2.AggregateType() {
 		return fmt.Errorf("incorrect aggregate type: %s (should be %s)", e1.AggregateType(), e2.AggregateType())
 	}
+	if e1.AggregateID() != e2.AggregateID() {
+		return fmt.Errorf("incorrect aggregate ID: %s (should be %s)", e1.AggregateID(), e2.AggregateID())
+	}
 	if !opts.ignoreVersion {
 		if e1.Version() != e2.Version() {
-			return fmt.Errorf("incorrect aggregate version: %d (should be %d)", e1.Version(), e2.Version())
+			return fmt.Errorf("incorrect version: %d (should be %d)", e1.Version(), e2.Version())
 		}
 	}
 	if !reflect.DeepEqual(e1.Metadata(), e2.Metadata()) {
diff --git a/event.go b/event.go
index 5b684a9..dfc41a6 100644
--- a/event.go
+++ b/event.go
@@ -38,13 +38,12 @@ type Event interface {
 	// Timestamp of when the event was created.
 	Timestamp() time.Time
 
-	// AggregateType returns the type of the aggregate that the event can be
+	// AggregateType is the type of the aggregate that the event can be
 	// applied to.
 	AggregateType() AggregateType
-	// AggregateID returns the ID of the aggregate that the event should be
-	// applied to.
+	// AggregateID is the ID of the aggregate that the event belongs to.
 	AggregateID() uuid.UUID
-	// Version of the aggregate for this event (after it has been applied).
+	// Version is the version of the aggregate after the event has been applied.
 	Version() int
 
 	// Metadata is app-specific metadata such as request ID, originating user etc.
@@ -166,7 +165,7 @@ func (e event) AggregateType() AggregateType {
 	return e.aggregateType
 }
 
-// AggrgateID implements the AggrgateID method of the Event interface.
+// AggregateID implements the AggregateID method of the Event interface.
 func (e event) AggregateID() uuid.UUID {
 	return e.aggregateID
 }
diff --git a/eventmaintenance.go b/eventmaintenance.go
new file mode 100644
index 0000000..68b7550
--- /dev/null
+++ b/eventmaintenance.go
@@ -0,0 +1,30 @@
+// Copyright (c) 2014 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package eventhorizon
+
+import (
+	"context"
+)
+
+// EventStoreMaintenance is an interface with maintenance tools for an EventStore.
+// NOTE: Should not be used in apps, useful for migration tools etc.
+type EventStoreMaintenance interface {
+	// Replace an event, the version must match. Useful for maintenance actions.
+	// Returns ErrAggregateNotFound if there is no aggregate.
+	Replace(context.Context, Event) error
+
+	// RenameEvent renames all instances of the event type.
+	RenameEvent(ctx context.Context, from, to EventType) error
+}
diff --git a/eventsource.go b/eventsource.go
index aef8d87..f4b8ba1 100644
--- a/eventsource.go
+++ b/eventsource.go
@@ -17,7 +17,10 @@ package eventhorizon
 // EventSource is a source of events, used for getting events for handling,
 // storing, publishing etc. Mostly used in the aggregate stores.
 type EventSource interface {
-	// Events returns events. It should only return any new events since its
-	// last invocation.
-	Events() []Event
+	// UncommittedEvents returns events that are not committed to the event store,
+	// or handeled in other ways (depending on the caller).
+	UncommittedEvents() []Event
+	// ClearUncommittedEvents clears uncommitted events, used after they have been
+	// committed to the event store or handled in other ways (depending on the caller).
+	ClearUncommittedEvents()
 }
diff --git a/eventstore.go b/eventstore.go
index 66fd98b..e2fb84f 100644
--- a/eventstore.go
+++ b/eventstore.go
@@ -30,19 +30,6 @@ type EventStore interface {
 	Load(context.Context, uuid.UUID) ([]Event, error)
 }
 
-// EventStoreMaintainer is an interface for a maintainer of an EventStore.
-// NOTE: Should not be used in apps, useful for migration tools etc.
-type EventStoreMaintainer interface {
-	EventStore
-
-	// Replace an event, the version must match. Useful for maintenance actions.
-	// Returns ErrAggregateNotFound if there is no aggregate.
-	Replace(context.Context, Event) error
-
-	// RenameEvent renames all instances of the event type.
-	RenameEvent(ctx context.Context, from, to EventType) error
-}
-
 // EventStoreError is an error in the event store, with the namespace.
 type EventStoreError struct {
 	// Err is the error.
diff --git a/eventstore/memory/eventmaintenance.go b/eventstore/memory/eventmaintenance.go
new file mode 100644
index 0000000..6e0c756
--- /dev/null
+++ b/eventstore/memory/eventmaintenance.go
@@ -0,0 +1,100 @@
+// Copyright (c) 2014 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package memory
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+
+	eh "github.com/looplab/eventhorizon"
+)
+
+// Replace implements the Replace method of the eventhorizon.EventStore interface.
+func (s *EventStore) Replace(ctx context.Context, event eh.Event) error {
+	// Ensure that the namespace exists.
+	ns := s.namespace(ctx)
+
+	s.dbMu.RLock()
+	aggregate, ok := s.db[ns][event.AggregateID()]
+	if !ok {
+		s.dbMu.RUnlock()
+		return eh.ErrAggregateNotFound
+	}
+	s.dbMu.RUnlock()
+
+	// Create the event record for the Database.
+	e, err := copyEvent(ctx, event)
+	if err != nil {
+		return err
+	}
+
+	// Find the event to replace.
+	idx := -1
+	for i, e := range aggregate.Events {
+		if e.Version() == event.Version() {
+			idx = i
+			break
+		}
+	}
+	if idx == -1 {
+		return eh.ErrInvalidEvent
+	}
+
+	// Replace event.
+	s.dbMu.Lock()
+	defer s.dbMu.Unlock()
+	aggregate.Events[idx] = e
+
+	return nil
+}
+
+// RenameEvent implements the RenameEvent method of the eventhorizon.EventStore interface.
+func (s *EventStore) RenameEvent(ctx context.Context, from, to eh.EventType) error {
+	// Ensure that the namespace exists.
+	ns := s.namespace(ctx)
+
+	s.dbMu.Lock()
+	defer s.dbMu.Unlock()
+
+	updated := map[uuid.UUID]aggregateRecord{}
+	for id, aggregate := range s.db[ns] {
+		events := make([]eh.Event, len(aggregate.Events))
+		for i, e := range aggregate.Events {
+			if e.EventType() == from {
+				// Rename any matching event by duplicating.
+				events[i] = eh.NewEvent(
+					to,
+					e.Data(),
+					e.Timestamp(),
+					eh.ForAggregate(
+						e.AggregateType(),
+						e.AggregateID(),
+						e.Version(),
+					),
+					eh.WithMetadata(e.Metadata()),
+				)
+			}
+		}
+		aggregate.Events = events
+		updated[id] = aggregate
+	}
+
+	for id, aggregate := range updated {
+		s.db[ns][id] = aggregate
+	}
+
+	return nil
+}
diff --git a/eventstore/memory/eventstore.go b/eventstore/memory/eventstore.go
index c7de467..361fc35 100644
--- a/eventstore/memory/eventstore.go
+++ b/eventstore/memory/eventstore.go
@@ -180,83 +180,6 @@ func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 	return events, nil
 }
 
-// Replace implements the Replace method of the eventhorizon.EventStore interface.
-func (s *EventStore) Replace(ctx context.Context, event eh.Event) error {
-	// Ensure that the namespace exists.
-	ns := s.namespace(ctx)
-
-	s.dbMu.RLock()
-	aggregate, ok := s.db[ns][event.AggregateID()]
-	if !ok {
-		s.dbMu.RUnlock()
-		return eh.ErrAggregateNotFound
-	}
-	s.dbMu.RUnlock()
-
-	// Create the event record for the Database.
-	e, err := copyEvent(ctx, event)
-	if err != nil {
-		return err
-	}
-
-	// Find the event to replace.
-	idx := -1
-	for i, e := range aggregate.Events {
-		if e.Version() == event.Version() {
-			idx = i
-			break
-		}
-	}
-	if idx == -1 {
-		return eh.ErrInvalidEvent
-	}
-
-	// Replace event.
-	s.dbMu.Lock()
-	defer s.dbMu.Unlock()
-	aggregate.Events[idx] = e
-
-	return nil
-}
-
-// RenameEvent implements the RenameEvent method of the eventhorizon.EventStore interface.
-func (s *EventStore) RenameEvent(ctx context.Context, from, to eh.EventType) error {
-	// Ensure that the namespace exists.
-	ns := s.namespace(ctx)
-
-	s.dbMu.Lock()
-	defer s.dbMu.Unlock()
-
-	updated := map[uuid.UUID]aggregateRecord{}
-	for id, aggregate := range s.db[ns] {
-		events := make([]eh.Event, len(aggregate.Events))
-		for i, e := range aggregate.Events {
-			if e.EventType() == from {
-				// Rename any matching event by duplicating.
-				events[i] = eh.NewEvent(
-					to,
-					e.Data(),
-					e.Timestamp(),
-					eh.ForAggregate(
-						e.AggregateType(),
-						e.AggregateID(),
-						e.Version(),
-					),
-					eh.WithMetadata(e.Metadata()),
-				)
-			}
-		}
-		aggregate.Events = events
-		updated[id] = aggregate
-	}
-
-	for id, aggregate := range updated {
-		s.db[ns][id] = aggregate
-	}
-
-	return nil
-}
-
 // Helper to get the namespace and ensure that its data exists.
 func (s *EventStore) namespace(ctx context.Context) string {
 	s.dbMu.Lock()
diff --git a/eventstore/mongodb/eventmaintenance.go b/eventstore/mongodb/eventmaintenance.go
new file mode 100644
index 0000000..97e7861
--- /dev/null
+++ b/eventstore/mongodb/eventmaintenance.go
@@ -0,0 +1,107 @@
+// Copyright (c) 2015 - The Event Horizon authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mongodb
+
+import (
+	"context"
+
+	"go.mongodb.org/mongo-driver/bson"
+
+	// Register uuid.UUID as BSON type.
+	_ "github.com/looplab/eventhorizon/codec/bson"
+
+	eh "github.com/looplab/eventhorizon"
+)
+
+// Replace implements the Replace method of the eventhorizon.EventStore interface.
+func (s *EventStore) Replace(ctx context.Context, event eh.Event) error {
+	c := s.client.Database(s.dbName(ctx)).Collection("events")
+
+	// First check if the aggregate exists, the not found error in the update
+	// query can mean both that the aggregate or the event is not found.
+	if n, err := c.CountDocuments(ctx, bson.M{"_id": event.AggregateID()}); n == 0 {
+		return eh.ErrAggregateNotFound
+	} else if err != nil {
+		return eh.EventStoreError{
+			Err:       err,
+			Namespace: eh.NamespaceFromContext(ctx),
+		}
+	}
+
+	// Create the event record for the Database.
+	e, err := newEvt(ctx, event)
+	if err != nil {
+		return err
+	}
+
+	// Find and replace the event.
+	if r, err := c.UpdateOne(ctx,
+		bson.M{
+			"_id":            event.AggregateID(),
+			"events.version": event.Version(),
+		},
+		bson.M{
+			"$set": bson.M{"events.$": *e},
+		},
+	); err != nil {
+		return eh.EventStoreError{
+			Err:       eh.ErrCouldNotSaveEvents,
+			BaseErr:   err,
+			Namespace: eh.NamespaceFromContext(ctx),
+		}
+	} else if r.MatchedCount == 0 {
+		return eh.ErrInvalidEvent
+	}
+
+	return nil
+}
+
+// RenameEvent implements the RenameEvent method of the eventhorizon.EventStore interface.
+func (s *EventStore) RenameEvent(ctx context.Context, from, to eh.EventType) error {
+	c := s.client.Database(s.dbName(ctx)).Collection("events")
+
+	// Find and rename all events.
+	// TODO: Maybe use change info.
+	if _, err := c.UpdateMany(ctx,
+		bson.M{
+			"events.event_type": from.String(),
+		},
+		bson.M{
+			"$set": bson.M{"events.$.event_type": to.String()},
+		},
+	); err != nil {
+		return eh.EventStoreError{
+			Err:       eh.ErrCouldNotSaveEvents,
+			BaseErr:   err,
+			Namespace: eh.NamespaceFromContext(ctx),
+		}
+	}
+
+	return nil
+}
+
+// Clear clears the event storage.
+func (s *EventStore) Clear(ctx context.Context) error {
+	c := s.client.Database(s.dbName(ctx)).Collection("events")
+
+	if err := c.Drop(ctx); err != nil {
+		return eh.EventStoreError{
+			Err:       ErrCouldNotClearDB,
+			BaseErr:   err,
+			Namespace: eh.NamespaceFromContext(ctx),
+		}
+	}
+	return nil
+}
diff --git a/eventstore/mongodb/eventstore.go b/eventstore/mongodb/eventstore.go
index e16bd8c..1d72ad1 100644
--- a/eventstore/mongodb/eventstore.go
+++ b/eventstore/mongodb/eventstore.go
@@ -278,87 +278,6 @@ func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 	return events, nil
 }
 
-// Replace implements the Replace method of the eventhorizon.EventStore interface.
-func (s *EventStore) Replace(ctx context.Context, event eh.Event) error {
-	c := s.client.Database(s.dbName(ctx)).Collection("events")
-
-	// First check if the aggregate exists, the not found error in the update
-	// query can mean both that the aggregate or the event is not found.
-	if n, err := c.CountDocuments(ctx, bson.M{"_id": event.AggregateID()}); n == 0 {
-		return eh.ErrAggregateNotFound
-	} else if err != nil {
-		return eh.EventStoreError{
-			Err:       err,
-			Namespace: eh.NamespaceFromContext(ctx),
-		}
-	}
-
-	// Create the event record for the Database.
-	e, err := newEvt(ctx, event)
-	if err != nil {
-		return err
-	}
-
-	// Find and replace the event.
-	if r, err := c.UpdateOne(ctx,
-		bson.M{
-			"_id":            event.AggregateID(),
-			"events.version": event.Version(),
-		},
-		bson.M{
-			"$set": bson.M{"events.$": *e},
-		},
-	); err != nil {
-		return eh.EventStoreError{
-			Err:       eh.ErrCouldNotSaveEvents,
-			BaseErr:   err,
-			Namespace: eh.NamespaceFromContext(ctx),
-		}
-	} else if r.MatchedCount == 0 {
-		return eh.ErrInvalidEvent
-	}
-
-	return nil
-}
-
-// RenameEvent implements the RenameEvent method of the eventhorizon.EventStore interface.
-func (s *EventStore) RenameEvent(ctx context.Context, from, to eh.EventType) error {
-	c := s.client.Database(s.dbName(ctx)).Collection("events")
-
-	// Find and rename all events.
-	// TODO: Maybe use change info.
-	if _, err := c.UpdateMany(ctx,
-		bson.M{
-			"events.event_type": from.String(),
-		},
-		bson.M{
-			"$set": bson.M{"events.$.event_type": to.String()},
-		},
-	); err != nil {
-		return eh.EventStoreError{
-			Err:       eh.ErrCouldNotSaveEvents,
-			BaseErr:   err,
-			Namespace: eh.NamespaceFromContext(ctx),
-		}
-	}
-
-	return nil
-}
-
-// Clear clears the event storage.
-func (s *EventStore) Clear(ctx context.Context) error {
-	c := s.client.Database(s.dbName(ctx)).Collection("events")
-
-	if err := c.Drop(ctx); err != nil {
-		return eh.EventStoreError{
-			Err:       ErrCouldNotClearDB,
-			BaseErr:   err,
-			Namespace: eh.NamespaceFromContext(ctx),
-		}
-	}
-	return nil
-}
-
 // Close closes the database client.
 func (s *EventStore) Close(ctx context.Context) {
 	s.client.Disconnect(ctx)


Test patch:
diff --git a/aggregatestore/events/aggregate.go b/aggregatestore/events/aggregate.go
index 51cfa77..ce0ea69 100644
--- a/aggregatestore/events/aggregate.go
+++ b/aggregatestore/events/aggregate.go
@@ -20,24 +20,24 @@ import (
 	eh "github.com/looplab/eventhorizon"
 )
 
-// Aggregate is an interface representing a versioned data entity created from
-// events. It receives commands and generates events that are stored.
+// VersionedAggregate is an interface representing a versioned aggregate created
+// from events. It receives commands and generates events that are stored.
 //
 // The aggregate is created/loaded and saved by the Repository inside the
 // Dispatcher. A domain specific aggregate can either implement the full interface,
 // or more commonly embed *AggregateBase to take care of the common methods.
-type Aggregate interface {
+type VersionedAggregate interface {
 	// Provides all the basic aggregate data.
 	eh.Aggregate
 
 	// Provides events to persist and publish from the aggregate.
 	eh.EventSource
 
-	// Version returns the version of the aggregate.
-	Version() int
-	// Increment version increments the version of the aggregate. It should be
-	// called after an event has been successfully applied.
-	IncrementVersion()
+	// AggregateVersion returns the version of the aggregate.
+	AggregateVersion() int
+	// SetAggregateVersion sets the version of the aggregate. It should only be
+	// called after an event has been successfully applied, often by EH.
+	SetAggregateVersion(int)
 
 	// ApplyEvent applies an event on the aggregate by setting its values.
 	// If there are no errors the version should be incremented by calling
diff --git a/aggregatestore/events/aggregatebase.go b/aggregatestore/events/aggregatebase.go
index 87199f0..7eacba5 100644
--- a/aggregatestore/events/aggregatebase.go
+++ b/aggregatestore/events/aggregatebase.go
@@ -81,21 +81,26 @@ func (a *AggregateBase) AggregateType() eh.AggregateType {
 	return a.t
 }
 
-// Version implements the Version method of the Aggregate interface.
-func (a *AggregateBase) Version() int {
+// AggregateVersion implements the AggregateVersion method of the Aggregate interface.
+func (a *AggregateBase) AggregateVersion() int {
 	return a.v
 }
 
-// IncrementVersion implements the IncrementVersion method of the Aggregate interface.
-func (a *AggregateBase) IncrementVersion() {
-	a.v++
+// SetAggregateVersion implements the SetAggregateVersion method of the Aggregate interface.
+func (a *AggregateBase) SetAggregateVersion(v int) {
+	a.v = v
 }
 
-// Events implements the Events method of the eh.EventSource interface.
-func (a *AggregateBase) Events() []eh.Event {
-	events := a.events
+// UncommittedEvents implements the UncommittedEvents method of the eh.EventSource
+// interface.
+func (a *AggregateBase) UncommittedEvents() []eh.Event {
+	return a.events
+}
+
+// ClearUncommittedEvents implements the ClearUncommittedEvents method of the eh.EventSource
+// interface.
+func (a *AggregateBase) ClearUncommittedEvents() {
 	a.events = nil
-	return events
 }
 
 // AppendEvent appends an event for later retrieval by Events().
@@ -103,7 +108,7 @@ func (a *AggregateBase) AppendEvent(t eh.EventType, data eh.EventData, timestamp
 	options = append(options, eh.ForAggregate(
 		a.AggregateType(),
 		a.EntityID(),
-		a.Version()+len(a.events)+1),
+		a.AggregateVersion()+len(a.events)+1), // TODO: This will probably not work with a global version.
 	)
 	e := eh.NewEvent(t, data, timestamp, options...)
 	a.events = append(a.events, e)
diff --git a/aggregatestore/events/aggregatebase_test.go b/aggregatestore/events/aggregatebase_test.go
index 405829d..14931b5 100644
--- a/aggregatestore/events/aggregatebase_test.go
+++ b/aggregatestore/events/aggregatebase_test.go
@@ -36,20 +36,20 @@ func TestNewAggregateBase(t *testing.T) {
 	if agg.EntityID() != id {
 		t.Error("the entity ID should be correct: ", agg.EntityID(), id)
 	}
-	if agg.Version() != 0 {
-		t.Error("the version should be 0:", agg.Version())
+	if agg.AggregateVersion() != 0 {
+		t.Error("the version should be 0:", agg.AggregateVersion())
 	}
 }
 
 func TestAggregateVersion(t *testing.T) {
 	agg := NewAggregateBase(TestAggregateType, uuid.New())
-	if agg.Version() != 0 {
-		t.Error("the version should be 0:", agg.Version())
+	if agg.AggregateVersion() != 0 {
+		t.Error("the version should be 0:", agg.AggregateVersion())
 	}
 
-	agg.IncrementVersion()
-	if agg.Version() != 1 {
-		t.Error("the version should be 1:", agg.Version())
+	agg.SetAggregateVersion(51)
+	if agg.AggregateVersion() != 51 {
+		t.Error("the version should be 1:", agg.AggregateVersion())
 	}
 }
 
@@ -67,14 +67,14 @@ func TestAggregateEvents(t *testing.T) {
 	if !event1.Timestamp().Equal(timestamp) {
 		t.Error("the timestamp should not be zero:", event1.Timestamp())
 	}
-	if event1.Version() != 1 {
-		t.Error("the version should be 1:", event1.Version())
-	}
 	if event1.AggregateType() != TestAggregateType {
 		t.Error("the aggregate type should be correct:", event1.AggregateType())
 	}
 	if event1.AggregateID() != id {
-		t.Error("the aggregate id should be correct:", event1.AggregateID())
+		t.Error("the aggregate ID should be correct:", event1.AggregateID())
+	}
+	if event1.Version() != 1 {
+		t.Error("the version should be 1:", event1.Version())
 	}
 	if event1.String() != "TestAggregateEvent@1" {
 		t.Error("the string representation should be correct:", event1.String())
@@ -91,19 +91,21 @@ func TestAggregateEvents(t *testing.T) {
 	if event2.Version() != 2 {
 		t.Error("the version should be 2:", event2.Version())
 	}
-	events = agg.Events()
+	events = agg.UncommittedEvents()
 	if len(events) != 2 {
 		t.Error("there should be two events provided:", len(events))
 	}
+	agg.ClearUncommittedEvents()
 
 	event3 := agg.AppendEvent(TestAggregateEventType, &TestEventData{"event1"}, timestamp)
 	if event3.Version() != 1 {
 		t.Error("the version should be 1 after clearing uncommitted events (without applying any):", event3.Version())
 	}
-	events = agg.Events()
+	events = agg.UncommittedEvents()
 	if len(events) != 1 {
 		t.Error("there should be one new event provided:", len(events))
 	}
+	agg.ClearUncommittedEvents()
 
 	agg = NewTestAggregate(uuid.New())
 	event1 = agg.AppendEvent(TestAggregateEventType, &TestEventData{"event1"}, timestamp)
@@ -154,7 +156,7 @@ type TestAggregate struct {
 	event eh.Event
 }
 
-var _ = Aggregate(&TestAggregate{})
+var _ = VersionedAggregate(&TestAggregate{})
 
 func NewTestAggregate(id uuid.UUID) *TestAggregate {
 	return &TestAggregate{
diff --git a/aggregatestore/events/aggregatestore.go b/aggregatestore/events/aggregatestore.go
index b323512..af4f63f 100644
--- a/aggregatestore/events/aggregatestore.go
+++ b/aggregatestore/events/aggregatestore.go
@@ -32,8 +32,8 @@ type AggregateStore struct {
 var (
 	// ErrInvalidEventStore is when a dispatcher is created with a nil event store.
 	ErrInvalidEventStore = errors.New("invalid event store")
-	// ErrInvalidAggregateType is when  the aggregate does not implement event.Aggregte.
-	ErrInvalidAggregateType = errors.New("invalid aggregate type")
+	// ErrAggregateNotVersioned is when the aggregate does not implement the VersionedAggregte interface.
+	ErrAggregateNotVersioned = errors.New("aggregate is not versioned")
 	// ErrMismatchedEventType occurs when loaded events from ID does not match aggregate type.
 	ErrMismatchedEventType = errors.New("mismatched event type and aggregate type")
 )
@@ -84,9 +84,9 @@ func (r *AggregateStore) Load(ctx context.Context, aggregateType eh.AggregateTyp
 	if err != nil {
 		return nil, err
 	}
-	a, ok := agg.(Aggregate)
+	a, ok := agg.(VersionedAggregate)
 	if !ok {
-		return nil, ErrInvalidAggregateType
+		return nil, ErrAggregateNotVersioned
 	}
 
 	events, err := r.store.Load(ctx, a.EntityID())
@@ -104,19 +104,20 @@ func (r *AggregateStore) Load(ctx context.Context, aggregateType eh.AggregateTyp
 // Save implements the Save method of the eventhorizon.AggregateStore interface.
 // It saves all uncommitted events from an aggregate to the event store.
 func (r *AggregateStore) Save(ctx context.Context, agg eh.Aggregate) error {
-	a, ok := agg.(Aggregate)
+	a, ok := agg.(VersionedAggregate)
 	if !ok {
-		return ErrInvalidAggregateType
+		return ErrAggregateNotVersioned
 	}
 
 	// Retrieve any new events to store.
-	events := a.Events()
+	events := a.UncommittedEvents()
 	if len(events) == 0 {
 		return nil
 	}
-	if err := r.store.Save(ctx, events, a.Version()); err != nil {
+	if err := r.store.Save(ctx, events, a.AggregateVersion()); err != nil {
 		return err
 	}
+	a.ClearUncommittedEvents()
 
 	// Apply the events in case the aggregate needs to be further used
 	// after this save. Currently it is not reused.
@@ -127,7 +128,7 @@ func (r *AggregateStore) Save(ctx context.Context, agg eh.Aggregate) error {
 	return nil
 }
 
-func (r *AggregateStore) applyEvents(ctx context.Context, a Aggregate, events []eh.Event) error {
+func (r *AggregateStore) applyEvents(ctx context.Context, a VersionedAggregate, events []eh.Event) error {
 	for _, event := range events {
 		if event.AggregateType() != a.AggregateType() {
 			return ErrMismatchedEventType
@@ -139,7 +140,7 @@ func (r *AggregateStore) applyEvents(ctx context.Context, a Aggregate, events []
 				Err:   err,
 			}
 		}
-		a.IncrementVersion()
+		a.SetAggregateVersion(event.Version())
 	}
 
 	return nil
diff --git a/aggregatestore/events/aggregatestore_test.go b/aggregatestore/events/aggregatestore_test.go
index 2eb3c56..b599251 100644
--- a/aggregatestore/events/aggregatestore_test.go
+++ b/aggregatestore/events/aggregatestore_test.go
@@ -58,15 +58,15 @@ func TestAggregateStore_LoadNoEvents(t *testing.T) {
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
-	a, ok := agg.(Aggregate)
+	a, ok := agg.(VersionedAggregate)
 	if !ok {
 		t.Fatal("the aggregate shoud be of correct type")
 	}
 	if a.EntityID() != id {
 		t.Error("the aggregate ID should be correct: ", a.EntityID(), id)
 	}
-	if a.Version() != 0 {
-		t.Error("the version should be 0:", a.Version())
+	if a.AggregateVersion() != 0 {
+		t.Error("the aggregate version should be 0:", a.AggregateVersion())
 	}
 }
 
@@ -88,15 +88,15 @@ func TestAggregateStore_LoadEvents(t *testing.T) {
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
-	a, ok := loaded.(Aggregate)
+	a, ok := loaded.(VersionedAggregate)
 	if !ok {
-		t.Fatal("the aggregate shoud be of correct type")
+		t.Fatal("the aggregate should be of correct type")
 	}
 	if a.EntityID() != id {
 		t.Error("the aggregate ID should be correct: ", a.EntityID(), id)
 	}
-	if a.Version() != 1 {
-		t.Error("the version should be 1:", a.Version())
+	if a.AggregateVersion() != 1 {
+		t.Error("the aggregate version should be 1:", a.AggregateVersion())
 	}
 	if !reflect.DeepEqual(a.(*TestAggregate).event, event1) {
 		t.Error("the event should be correct:", a.(*TestAggregate).event)
@@ -170,12 +170,13 @@ func TestAggregateStore_SaveEvents(t *testing.T) {
 	if events[0] != event1 {
 		t.Error("the stored event should be correct:", events[0])
 	}
-	if len(agg.Events()) != 0 {
-		t.Error("there should be no uncommitted events:", agg.Events())
+	if len(agg.UncommittedEvents()) != 0 {
+		t.Error("there should be no uncommitted events:", agg.UncommittedEvents())
 	}
-	if agg.Version() != 1 {
-		t.Error("the version should be 1:", agg.Version())
+	if agg.AggregateVersion() != 1 {
+		t.Error("the aggregate version should be 1:", agg.AggregateVersion())
 	}
+	agg.ClearUncommittedEvents()
 
 	// Store error.
 	agg.AppendEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp)
@@ -239,7 +240,7 @@ type TestAggregateOther struct {
 	err error
 }
 
-var _ = Aggregate(&TestAggregateOther{})
+var _ = VersionedAggregate(&TestAggregateOther{})
 
 func NewTestAggregateOther(id uuid.UUID) *TestAggregateOther {
 	return &TestAggregateOther{
diff --git a/aggregatestore/model/aggregatestore.go b/aggregatestore/model/aggregatestore.go
index 7310341..45ba0ef 100644
--- a/aggregatestore/model/aggregatestore.go
+++ b/aggregatestore/model/aggregatestore.go
@@ -79,7 +79,9 @@ func (r *AggregateStore) Save(ctx context.Context, aggregate eh.Aggregate) error
 
 	// Handle any events optionally provided by the aggregate.
 	if a, ok := aggregate.(eh.EventSource); ok && r.eventHandler != nil {
-		for _, e := range a.Events() {
+		events := a.UncommittedEvents()
+		a.ClearUncommittedEvents()
+		for _, e := range events {
 			if err := r.eventHandler.HandleEvent(ctx, e); err != nil {
 				return err
 			}
diff --git a/aggregatestore/model/eventsource.go b/aggregatestore/model/eventsource.go
index 690a79b..43ab759 100644
--- a/aggregatestore/model/eventsource.go
+++ b/aggregatestore/model/eventsource.go
@@ -26,9 +26,13 @@ func (a *SliceEventSource) AppendEvent(e eh.Event) {
 	*a = append(*a, e)
 }
 
-// Events implements the Events method of the eh.EventSource interface.
-func (a *SliceEventSource) Events() []eh.Event {
-	events := *a
+// UncommittedEvents implements the UncommittedEvents method of the eh.EventSource interface.
+func (a *SliceEventSource) UncommittedEvents() []eh.Event {
+	return *a
+}
+
+// ClearUncommittedEvents implements the ClearUncommittedEvents method of the eh.EventSource
+// interface.
+func (a *SliceEventSource) ClearUncommittedEvents() {
 	*a = nil
-	return events
 }
diff --git a/event_test.go b/event_test.go
index 1bad4ce..494d55a 100644
--- a/event_test.go
+++ b/event_test.go
@@ -69,7 +69,7 @@ func TestNewEvent(t *testing.T) {
 		t.Error("the aggregate ID should be correct:", event.AggregateID())
 	}
 	if event.Version() != 3 {
-		t.Error("the version should be zero:", event.Version())
+		t.Error("the version should be correct:", event.Version())
 	}
 	if !reflect.DeepEqual(event.Metadata(), map[string]interface{}{
 		"meta":         "data",
@@ -89,6 +89,9 @@ func TestCreateEventData(t *testing.T) {
 	if !errors.Is(err, ErrEventDataNotRegistered) {
 		t.Error("there should be a event not registered error:", err)
 	}
+	if data != nil {
+		t.Error("the data should be nil")
+	}
 
 	RegisterEventData(TestEventRegisterType, func() EventData {
 		return &TestEventRegisterData{}
diff --git a/eventstore/acceptanece_testing.go b/eventstore/acceptanece_testing.go
index 969ee67..6a33d50 100644
--- a/eventstore/acceptanece_testing.go
+++ b/eventstore/acceptanece_testing.go
@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/mocks"
 )
@@ -130,6 +131,9 @@ func AcceptanceTest(t *testing.T, ctx context.Context, store eh.EventStore) []eh
 		event3,                 // Version 3
 		event4, event5, event6, // Version 4, 5 and 6
 	}
+	if len(events) != len(expectedEvents) {
+		t.Errorf("incorrect number of loaded events: %d", len(events))
+	}
 	for i, event := range events {
 		if err := eh.CompareEvents(event, expectedEvents[i], eh.IgnoreVersion()); err != nil {
 			t.Error("the event was incorrect:", err)
@@ -145,72 +149,8 @@ func AcceptanceTest(t *testing.T, ctx context.Context, store eh.EventStore) []eh
 		t.Error("there should be no error:", err)
 	}
 	expectedEvents = []eh.Event{event7}
-	for i, event := range events {
-		if err := eh.CompareEvents(event, expectedEvents[i], eh.IgnoreVersion()); err != nil {
-			t.Error("the event was incorrect:", err)
-		}
-		if event.Version() != i+1 {
-			t.Error("the event version should be correct:", event, event.Version())
-		}
-	}
-
-	return savedEvents
-}
-
-// MaintainerAcceptanceTest is the acceptance test that all implementations of
-// EventStoreMaintainer should pass. It should manually be called from a test
-// case in each implementation:
-//
-//   func TestEventStore(t *testing.T) {
-//       ctx := context.Background() // Or other when testing namespaces.
-//       store := NewEventStore()
-//       eventstore.AcceptanceTest(t, ctx, store)
-//   }
-//
-func MaintainerAcceptanceTest(t *testing.T, ctx context.Context, store eh.EventStoreMaintainer) {
-	ctx = context.WithValue(ctx, "testkey", "testval")
-
-	// Save some events.
-	id := uuid.New()
-	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
-	event1 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id, 1))
-	event2 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id, 2))
-	event3 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id, 3))
-	if err := store.Save(ctx, []eh.Event{event1, event2, event3}, 0); err != nil {
-		t.Error("there should be no error:", err)
-	}
-
-	// Replace event, no aggregate.
-	eventWithoutAggregate := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp,
-		eh.ForAggregate(mocks.AggregateType, uuid.New(), 1))
-	if err := store.Replace(ctx, eventWithoutAggregate); err != eh.ErrAggregateNotFound {
-		t.Error("there should be an aggregate not found error:", err)
-	}
-
-	// Replace event, no event version.
-	eventWithoutVersion := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event20"}, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id, 20))
-	if err := store.Replace(ctx, eventWithoutVersion); err != eh.ErrInvalidEvent {
-		t.Error("there should be an invalid event error:", err)
-	}
-
-	// Replace event.
-	event2Mod := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event2_mod"}, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id, 2))
-	if err := store.Replace(ctx, event2Mod); err != nil {
-		t.Error("there should be no error:", err)
-	}
-	events, err := store.Load(ctx, id)
-	if err != nil {
-		t.Error("there should be no error:", err)
-	}
-	expectedEvents := []eh.Event{
-		event1,    // Version 1
-		event2Mod, // Version 2, modified
-		event3,    // Version 3
+	if len(events) != len(expectedEvents) {
+		t.Errorf("incorrect number of loaded events: %d", len(events))
 	}
 	for i, event := range events {
 		if err := eh.CompareEvents(event, expectedEvents[i], eh.IgnoreVersion()); err != nil {
@@ -221,56 +161,15 @@ func MaintainerAcceptanceTest(t *testing.T, ctx context.Context, store eh.EventS
 		}
 	}
 
-	// Save events of the old type.
-	oldEventType := eh.EventType("old_event_type")
-	id1 := uuid.New()
-	oldEvent1 := eh.NewEvent(oldEventType, nil, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id1, 1))
-	if err := store.Save(ctx, []eh.Event{oldEvent1}, 0); err != nil {
-		t.Error("there should be no error:", err)
-	}
-	id2 := uuid.New()
-	oldEvent2 := eh.NewEvent(oldEventType, nil, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id2, 1))
-	if err := store.Save(ctx, []eh.Event{oldEvent2}, 0); err != nil {
-		t.Error("there should be no error:", err)
-	}
-
-	// Rename events to the new type.
-	newEventType := eh.EventType("new_event_type")
-	if err := store.RenameEvent(ctx, oldEventType, newEventType); err != nil {
-		t.Error("there should be no error:", err)
-	}
-	events, err = store.Load(ctx, id1)
-	if err != nil {
-		t.Error("there should be no error:", err)
-	}
-	newEvent1 := eh.NewEvent(newEventType, nil, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id1, 1))
-	if len(events) != 1 {
-		t.Fatal("there should be one event")
-	}
-	if err := eh.CompareEvents(events[0], newEvent1); err != nil {
-		t.Error("the event was incorrect:", err)
-	}
-	events, err = store.Load(ctx, id2)
-	if err != nil {
-		t.Error("there should be no error:", err)
-	}
-	newEvent2 := eh.NewEvent(newEventType, nil, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id2, 1))
-	if len(events) != 1 {
-		t.Fatal("there should be one event")
-	}
-	if err := eh.CompareEvents(events[0], newEvent2); err != nil {
-		t.Error("the event was incorrect:", err)
-	}
+	return savedEvents
 }
 
 func eventsToString(events []eh.Event) string {
 	parts := make([]string, len(events))
 	for i, e := range events {
-		parts[i] = fmt.Sprintf("%s:%s (%s)", e.AggregateType(), e.EventType(), e.AggregateID())
+		parts[i] = fmt.Sprintf("%s:%s (%s@%d)",
+			e.AggregateType(), e.EventType(),
+			e.AggregateID(), e.Version())
 	}
 	return strings.Join(parts, ", ")
 }
diff --git a/eventstore/maintenance_testing.go b/eventstore/maintenance_testing.go
new file mode 100644
index 0000000..008584c
--- /dev/null
+++ b/eventstore/maintenance_testing.go
@@ -0,0 +1,136 @@
+// Copyright (c) 2016 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package eventstore
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	eh "github.com/looplab/eventhorizon"
+	"github.com/looplab/eventhorizon/mocks"
+)
+
+// MaintenanceAcceptanceTest is the acceptance test that all implementations of
+// EventStoreMaintenance should pass. It should manually be called from a test
+// case in each implementation:
+//
+//   func TestEventStore(t *testing.T) {
+//       ctx := context.Background() // Or other when testing namespaces.
+//       store := NewEventStore()
+//       eventstore.AcceptanceTest(t, ctx, store)
+//   }
+//
+func MaintenanceAcceptanceTest(t *testing.T, ctx context.Context, store eh.EventStore, storeMaintenance eh.EventStoreMaintenance) {
+	ctx = context.WithValue(ctx, "testkey", "testval")
+
+	// Save some events.
+	id := uuid.New()
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+	event1 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 1))
+	event2 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 2))
+	event3 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 3))
+	if err := store.Save(ctx, []eh.Event{event1, event2, event3}, 0); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	// Replace event, no aggregate.
+	eventWithoutAggregate := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, uuid.New(), 1))
+	if err := storeMaintenance.Replace(ctx, eventWithoutAggregate); err != eh.ErrAggregateNotFound {
+		t.Error("there should be an aggregate not found error:", err)
+	}
+
+	// Replace event, no event version.
+	eventWithoutVersion := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event20"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 20))
+	if err := storeMaintenance.Replace(ctx, eventWithoutVersion); err != eh.ErrInvalidEvent {
+		t.Error("there should be an invalid event error:", err)
+	}
+
+	// Replace event.
+	event2Mod := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event2_mod"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 2))
+	if err := storeMaintenance.Replace(ctx, event2Mod); err != nil {
+		t.Error("there should be no error:", err)
+	}
+	events, err := store.Load(ctx, id)
+	if err != nil {
+		t.Error("there should be no error:", err)
+	}
+	expectedEvents := []eh.Event{
+		event1,    // Version 1
+		event2Mod, // Version 2, modified
+		event3,    // Version 3
+	}
+	for i, event := range events {
+		if err := eh.CompareEvents(event, expectedEvents[i], eh.IgnoreVersion()); err != nil {
+			t.Error("the event was incorrect:", err)
+		}
+		if event.Version() != i+1 {
+			t.Error("the event version should be correct:", event, event.Version())
+		}
+	}
+
+	// Save events of the old type.
+	oldEventType := eh.EventType("old_event_type")
+	id1 := uuid.New()
+	oldEvent1 := eh.NewEvent(oldEventType, nil, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id1, 1))
+	if err := store.Save(ctx, []eh.Event{oldEvent1}, 0); err != nil {
+		t.Error("there should be no error:", err)
+	}
+	id2 := uuid.New()
+	oldEvent2 := eh.NewEvent(oldEventType, nil, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id2, 1))
+	if err := store.Save(ctx, []eh.Event{oldEvent2}, 0); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	// Rename events to the new type.
+	newEventType := eh.EventType("new_event_type")
+	if err := storeMaintenance.RenameEvent(ctx, oldEventType, newEventType); err != nil {
+		t.Error("there should be no error:", err)
+	}
+	events, err = store.Load(ctx, id1)
+	if err != nil {
+		t.Error("there should be no error:", err)
+	}
+	newEvent1 := eh.NewEvent(newEventType, nil, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id1, 1))
+	if len(events) != 1 {
+		t.Fatal("there should be one event")
+	}
+	if err := eh.CompareEvents(events[0], newEvent1); err != nil {
+		t.Error("the event was incorrect:", err)
+	}
+	events, err = store.Load(ctx, id2)
+	if err != nil {
+		t.Error("there should be no error:", err)
+	}
+	newEvent2 := eh.NewEvent(newEventType, nil, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id2, 1))
+	if len(events) != 1 {
+		t.Fatal("there should be one event")
+	}
+	if err := eh.CompareEvents(events[0], newEvent2); err != nil {
+		t.Error("the event was incorrect:", err)
+	}
+}
diff --git a/eventstore/memory/eventmaintenance_test.go b/eventstore/memory/eventmaintenance_test.go
new file mode 100644
index 0000000..53ccbc9
--- /dev/null
+++ b/eventstore/memory/eventmaintenance_test.go
@@ -0,0 +1,34 @@
+// Copyright (c) 2014 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package memory
+
+import (
+	"context"
+	"testing"
+
+	"github.com/looplab/eventhorizon/eventstore"
+)
+
+func TestEventStoreMaintenance(t *testing.T) {
+	store, err := NewEventStore()
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+	if store == nil {
+		t.Fatal("there should be a store")
+	}
+
+	eventstore.MaintenanceAcceptanceTest(t, context.Background(), store, store)
+}
diff --git a/eventstore/memory/eventstore_test.go b/eventstore/memory/eventstore_test.go
index 2732a40..dace9c4 100644
--- a/eventstore/memory/eventstore_test.go
+++ b/eventstore/memory/eventstore_test.go
@@ -38,7 +38,6 @@ func TestEventStore(t *testing.T) {
 	eventstore.AcceptanceTest(t, context.Background(), store)
 	ctx := eh.NewContextWithNamespace(context.Background(), "ns")
 	eventstore.AcceptanceTest(t, ctx, store)
-	eventstore.MaintainerAcceptanceTest(t, context.Background(), store)
 }
 
 func TestWithEventHandler(t *testing.T) {
diff --git a/eventstore/mongodb/eventmaintenance_test.go b/eventstore/mongodb/eventmaintenance_test.go
new file mode 100644
index 0000000..d1f78be
--- /dev/null
+++ b/eventstore/mongodb/eventmaintenance_test.go
@@ -0,0 +1,54 @@
+// Copyright (c) 2015 - The Event Horizon authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mongodb
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/looplab/eventhorizon/eventstore"
+)
+
+func TestEventStoreMaintenanceIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	// Use MongoDB in Docker with fallback to localhost.
+	addr := os.Getenv("MONGODB_ADDR")
+	if addr == "" {
+		addr = "localhost:27017"
+	}
+	url := "mongodb://" + addr
+
+	store, err := NewEventStore(url, "test")
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+	if store == nil {
+		t.Fatal("there should be a store")
+	}
+
+	defer store.Close(context.Background())
+	defer func() {
+		if err = store.Clear(context.Background()); err != nil {
+			t.Fatal("there should be no error:", err)
+		}
+	}()
+
+	// Additional tests for the maintenance, implemented by the store.
+	eventstore.MaintenanceAcceptanceTest(t, context.Background(), store, store)
+}
diff --git a/eventstore/mongodb/eventstore_test.go b/eventstore/mongodb/eventstore_test.go
index 60b6f40..fb590aa 100644
--- a/eventstore/mongodb/eventstore_test.go
+++ b/eventstore/mongodb/eventstore_test.go
@@ -16,11 +16,14 @@ package mongodb
 
 import (
 	"context"
+	"crypto/rand"
+	"encoding/hex"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
+
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventstore"
 	"github.com/looplab/eventhorizon/mocks"
@@ -38,30 +41,28 @@ func TestEventStoreIntegration(t *testing.T) {
 	}
 	url := "mongodb://" + addr
 
-	store, err := NewEventStore(url, "test")
+	// Get a random DB name.
+	b := make([]byte, 4)
+	if _, err := rand.Read(b); err != nil {
+		t.Fatal(err)
+	}
+	db := "test-" + hex.EncodeToString(b)
+	t.Log("using DB:", db)
+
+	store, err := NewEventStore(url, db)
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 	if store == nil {
 		t.Fatal("there should be a store")
 	}
+	defer store.Close(context.Background())
 
 	customNamespaceCtx := eh.NewContextWithNamespace(context.Background(), "ns")
 
-	defer store.Close(context.Background())
-	defer func() {
-		if err = store.Clear(context.Background()); err != nil {
-			t.Fatal("there should be no error:", err)
-		}
-		if err = store.Clear(customNamespaceCtx); err != nil {
-			t.Fatal("there should be no error:", err)
-		}
-	}()
-
 	// Run the actual test suite, both for default and custom namespace.
 	eventstore.AcceptanceTest(t, context.Background(), store)
 	eventstore.AcceptanceTest(t, customNamespaceCtx, store)
-	eventstore.MaintainerAcceptanceTest(t, context.Background(), store)
 }
 
 func TestWithEventHandlerIntegration(t *testing.T) {
@@ -76,9 +77,17 @@ func TestWithEventHandlerIntegration(t *testing.T) {
 	}
 	url = "mongodb://" + url
 
+	// Get a random DB name.
+	b := make([]byte, 4)
+	if _, err := rand.Read(b); err != nil {
+		t.Fatal(err)
+	}
+	db := "test-" + hex.EncodeToString(b)
+	t.Log("using DB:", db)
+
 	h := &mocks.EventBus{}
 
-	store, err := NewEventStore(url, "test",
+	store, err := NewEventStore(url, db,
 		WithEventHandler(h),
 	)
 	if err != nil {
@@ -87,6 +96,7 @@ func TestWithEventHandlerIntegration(t *testing.T) {
 	if store == nil {
 		t.Fatal("there should be a store")
 	}
+	defer store.Close(context.Background())
 
 	ctx := context.Background()
 
@@ -106,6 +116,9 @@ func TestWithEventHandlerIntegration(t *testing.T) {
 		t.Error("there should be no error:", err)
 	}
 	// The stored events should be ok.
+	if len(events) != len(expected) {
+		t.Errorf("incorrect number of loaded events: %d", len(events))
+	}
 	for i, event := range events {
 		if err := eh.CompareEvents(event, expected[i], eh.IgnoreVersion()); err != nil {
 			t.Error("the stored event was incorrect:", err)
@@ -115,6 +128,9 @@ func TestWithEventHandlerIntegration(t *testing.T) {
 		}
 	}
 	// The handled events should be ok.
+	if len(h.Events) != len(expected) {
+		t.Errorf("incorrect number of loaded events: %d", len(events))
+	}
 	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)
diff --git a/examples/todomvc/backend/domains/todo/aggregate_test.go b/examples/todomvc/backend/domains/todo/aggregate_test.go
index 4b69e31..d402ed3 100644
--- a/examples/todomvc/backend/domains/todo/aggregate_test.go
+++ b/examples/todomvc/backend/domains/todo/aggregate_test.go
@@ -438,7 +438,7 @@ func TestAggregateHandleCommand(t *testing.T) {
 				t.Log("exp:", tc.expectedErr)
 				t.Log("got:", err)
 			}
-			events := tc.agg.Events()
+			events := tc.agg.UncommittedEvents()
 			if !reflect.DeepEqual(events, tc.expectedEvents) {
 				t.Errorf("test case '%s': incorrect events", name)
 				t.Log("exp:\n", pretty.Sprint(tc.expectedEvents))
diff --git a/repo/mongodb/repo_test.go b/repo/mongodb/repo_test.go
index 63be18e..901bc5b 100644
--- a/repo/mongodb/repo_test.go
+++ b/repo/mongodb/repo_test.go
@@ -16,6 +16,8 @@ package mongodb
 
 import (
 	"context"
+	"crypto/rand"
+	"encoding/hex"
 	"errors"
 	"os"
 	"reflect"
@@ -43,13 +45,22 @@ func TestReadRepoIntegration(t *testing.T) {
 	}
 	url := "mongodb://" + addr
 
-	r, err := NewRepo(url, "test", "mocks.Model")
+	// Get a random DB name.
+	b := make([]byte, 4)
+	if _, err := rand.Read(b); err != nil {
+		t.Fatal(err)
+	}
+	db := "test-" + hex.EncodeToString(b)
+	t.Log("using DB:", db)
+
+	r, err := NewRepo(url, db, "mocks.Model")
 	if err != nil {
 		t.Error("there should be no error:", err)
 	}
 	if r == nil {
 		t.Error("there should be a repository")
 	}
+	defer r.Close(context.Background())
 
 	r.SetEntityFactory(func() eh.Entity {
 		return &mocks.Model{}
@@ -60,21 +71,10 @@ func TestReadRepoIntegration(t *testing.T) {
 
 	customNamespaceCtx := eh.NewContextWithNamespace(context.Background(), "ns")
 
-	defer r.Close(context.Background())
-	defer func() {
-		if err = r.Clear(context.Background()); err != nil {
-			t.Fatal("there should be no error:", err)
-		}
-		if err = r.Clear(customNamespaceCtx); err != nil {
-			t.Fatal("there should be no error:", err)
-		}
-	}()
-
 	repo.AcceptanceTest(t, context.Background(), r)
 	extraRepoTests(t, context.Background(), r)
 	repo.AcceptanceTest(t, customNamespaceCtx, r)
 	extraRepoTests(t, customNamespaceCtx, r)
-
 }
 
 func extraRepoTests(t *testing.T, ctx context.Context, r *Repo) {
@@ -164,7 +164,6 @@ func extraRepoTests(t *testing.T, ctx context.Context, r *Repo) {
 	if err != nil {
 		t.Error("there should be no error:", err)
 	}
-
 }
 
 func TestRepository(t *testing.T) {