Repository: looplab/eventhorizon
Pull request #396 (2022-11-25 09:23:53)

Eventstore snapshot
### Description

This adds snapshot capability to eventstore and aggregatestore. With given strategy saves a snapshot of aggregate and loads next time and remaining events wich can increase load times of long-living aggregates

### Affected Components

- Event Store(Mongodb, memory)
- AggregateSttore

### Related Issues

None

### Solution and Design
A eventstore implemtation can now implement `eh.SnapshotStore` an now store snaphots of current state in aggregate.

every aggregate that want to be snapshotted needs to implement `eh.Snapshotable` 

```go
type Snapshotable interface {
	CreateSnapshot() *Snapshot
	ApplySnapshot(snapshot *Snapshot)
}
```
#### Loading
When aggretate i requested to be loaded the store checks if aggretate implements `eh.Snapshotable` and tries to find latest snapshot for aggregate. Loads and applies snapshot then loads all events that is not applies in current snapshot.

#### Saving
When aggretate i requested to be saved the store checks if aggretate implements `eh.Snapshotable` and if the strategy return true we create a snapshot ans saves it alongside the events.

### Steps to test and verify

1. Add Strategy to aggregatestore
    - `EveryNumberEventSnapshotStrateg`  <- set to every *1*  
    - `PeriodSnapshotStrategy` 
    - `NoSnapshotStrategy`
2. Implement `ehSnapshotable` in longliving aggregate
3. check mongo db snapshot collection



Patch:
diff --git a/aggregate.go b/aggregate.go
index e6dd885..2cc4be0 100644
--- a/aggregate.go
+++ b/aggregate.go
@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/looplab/eventhorizon/uuid"
 )
@@ -58,6 +59,13 @@ type AggregateStore interface {
 	Save(context.Context, Aggregate) error
 }
 
+// SnapshotStrategy determines if a snapshot should be taken or not.
+type SnapshotStrategy interface {
+	ShouldTakeSnapshot(lastSnapshotVersion int,
+		lastSnapshotTimestamp time.Time,
+		event Event) bool
+}
+
 var (
 	// ErrAggregateNotFound is when no aggregate can be found.
 	ErrAggregateNotFound = errors.New("aggregate not found")
@@ -148,7 +156,8 @@ func (e *AggregateError) Cause() error {
 // used to create concrete aggregate types when loading from the database.
 //
 // An example would be:
-//     RegisterAggregate(func(id UUID) Aggregate { return &MyAggregate{id} })
+//
+//	RegisterAggregate(func(id UUID) Aggregate { return &MyAggregate{id} })
 func RegisterAggregate(factory func(uuid.UUID) Aggregate) {
 	// Check that the created aggregate matches the registered type.
 	// TODO: Explore the use of reflect/gob for creating concrete types without
diff --git a/eventbus/gcp/eventbus.go b/eventbus/gcp/eventbus.go
index 7923b30..17a0b8b 100644
--- a/eventbus/gcp/eventbus.go
+++ b/eventbus/gcp/eventbus.go
@@ -36,7 +36,6 @@ type EventBus struct {
 	client       *pubsub.Client
 	clientOpts   []option.ClientOption
 	topic        *pubsub.Topic
-	topicConfig  *pubsub.TopicConfig
 	registered   map[eh.EventHandlerType]struct{}
 	registeredMu sync.RWMutex
 	errCh        chan error
@@ -85,14 +84,8 @@ func NewEventBus(projectID, appID string, options ...Option) (*EventBus, error)
 	if ok, err := b.topic.Exists(b.cctx); err != nil {
 		return nil, err
 	} else if !ok {
-		if b.topicConfig != nil {
-			if b.topic, err = b.client.CreateTopicWithConfig(b.cctx, name, b.topicConfig); err != nil {
-				return nil, err
-			}
-		} else {
-			if b.topic, err = b.client.CreateTopic(b.cctx, name); err != nil {
-				return nil, err
-			}
+		if b.topic, err = b.client.CreateTopic(b.cctx, name); err != nil {
+			return nil, err
 		}
 	}
 
@@ -122,16 +115,6 @@ func WithPubSubOptions(opts ...option.ClientOption) Option {
 	}
 }
 
-// WithTopicOptions adds the options to the pubsub.TopicConfig.
-// This allows control over the Topic creation, including message retention.
-func WithTopicOptions(topicConfig *pubsub.TopicConfig) Option {
-	return func(b *EventBus) error {
-		b.topicConfig = topicConfig
-
-		return nil
-	}
-}
-
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (b *EventBus) HandlerType() eh.EventHandlerType {
 	return "eventbus"
diff --git a/eventbus/kafka/eventbus.go b/eventbus/kafka/eventbus.go
index 3e65d9d..477197c 100644
--- a/eventbus/kafka/eventbus.go
+++ b/eventbus/kafka/eventbus.go
@@ -19,7 +19,6 @@ import (
 	"errors"
 	"fmt"
 	"log"
-	"strings"
 	"sync"
 	"time"
 
@@ -33,7 +32,7 @@ import (
 // to all matching registered handlers, in order of registration.
 type EventBus struct {
 	// TODO: Support multiple brokers.
-	addresses    []string
+	addr         string
 	appID        string
 	topic        string
 	startOffset  int64
@@ -49,13 +48,11 @@ type EventBus struct {
 }
 
 // NewEventBus creates an EventBus, with optional GCP connection settings.
-func NewEventBus(addressList, appID string, options ...Option) (*EventBus, error) {
+func NewEventBus(addr, appID string, options ...Option) (*EventBus, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	addrSplit := strings.Split(addressList, ",")
-
 	b := &EventBus{
-		addresses:   addrSplit,
+		addr:        addr,
 		appID:       appID,
 		topic:       appID + "_events",
 		startOffset: kafka.LastOffset, // Default: Don't read old messages.
@@ -79,7 +76,7 @@ func NewEventBus(addressList, appID string, options ...Option) (*EventBus, error
 
 	// Get or create the topic.
 	b.client = &kafka.Client{
-		Addr: kafka.TCP(addrSplit...),
+		Addr: kafka.TCP(addr),
 	}
 
 	var resp *kafka.CreateTopicsResponse
@@ -117,7 +114,7 @@ func NewEventBus(addressList, appID string, options ...Option) (*EventBus, error
 	}
 
 	b.writer = &kafka.Writer{
-		Addr:         kafka.TCP(addrSplit...),
+		Addr:         kafka.TCP(addr),
 		Topic:        b.topic,
 		BatchSize:    1,                // Write every event to the bus without delay.
 		RequiredAcks: kafka.RequireOne, // Stronger consistency.
@@ -152,16 +149,6 @@ func WithStartOffset(startOffset int64) Option {
 	}
 }
 
-// WithTopic uses the specified topic for the event bus topic name
-//
-// Defaults to: appID + "_events"
-func WithTopic(topic string) Option {
-	return func(b *EventBus) error {
-		b.topic = topic
-		return nil
-	}
-}
-
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (b *EventBus) HandlerType() eh.EventHandlerType {
 	return "eventbus"
@@ -219,9 +206,8 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 
 	// Get or create the subscription.
 	groupID := b.appID + "_" + h.HandlerType().String()
-
 	r := kafka.NewReader(kafka.ReaderConfig{
-		Brokers:               b.addresses,
+		Brokers:               []string{b.addr},
 		Topic:                 b.topic,
 		GroupID:               groupID,     // Send messages to only one subscriber per group.
 		MaxWait:               time.Second, // Allow to exit readloop in max 1s.
diff --git a/eventbus/nats/eventbus.go b/eventbus/nats/eventbus.go
index 6501dde..7dede32 100644
--- a/eventbus/nats/eventbus.go
+++ b/eventbus/nats/eventbus.go
@@ -25,7 +25,6 @@ import (
 
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/codec/json"
-	"github.com/looplab/eventhorizon/middleware/eventhandler/ephemeral"
 )
 
 // EventBus is a NATS Jetstream event bus that delegates handling of published
@@ -37,7 +36,6 @@ type EventBus struct {
 	js           nats.JetStreamContext
 	stream       *nats.StreamInfo
 	connOpts     []nats.Option
-	streamConfig *nats.StreamConfig
 	registered   map[eh.EventHandlerType]struct{}
 	registeredMu sync.RWMutex
 	errCh        chan error
@@ -45,7 +43,6 @@ type EventBus struct {
 	cancel       context.CancelFunc
 	wg           sync.WaitGroup
 	codec        eh.EventCodec
-	unsubscribe  []func()
 }
 
 // NewEventBus creates an EventBus, with optional settings.
@@ -91,15 +88,9 @@ func NewEventBus(url, appID string, options ...Option) (*EventBus, error) {
 	// Create the stream, which stores messages received on the subject.
 	subjects := b.streamName + ".*.*"
 	cfg := &nats.StreamConfig{
-		Name:      b.streamName,
-		Subjects:  []string{subjects},
-		Storage:   nats.FileStorage,
-		Retention: nats.InterestPolicy,
-	}
-
-	// Use the custom stream config if provided.
-	if b.streamConfig != nil {
-		cfg = b.streamConfig
+		Name:     b.streamName,
+		Subjects: []string{subjects},
+		Storage:  nats.FileStorage,
 	}
 
 	if b.stream, err = b.js.AddStream(cfg); err != nil {
@@ -130,14 +121,6 @@ func WithNATSOptions(opts ...nats.Option) Option {
 	}
 }
 
-// WithStreamConfig can customize the config for created NATS JetStream.
-func WithStreamConfig(opts *nats.StreamConfig) Option {
-	return func(b *EventBus) error {
-		b.streamConfig = opts
-		return nil
-	}
-}
-
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (b *EventBus) HandlerType() eh.EventHandlerType {
 	return "eventbus"
@@ -181,6 +164,7 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 	consumerName := fmt.Sprintf("%s_%s", b.appID, h.HandlerType())
 
 	sub, err := b.js.QueueSubscribe(subject, consumerName, b.handler(b.cctx, m, h),
+		nats.Durable(consumerName),
 		nats.DeliverNew(),
 		nats.ManualAck(),
 		nats.AckExplicit(),
@@ -191,11 +175,6 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 		return fmt.Errorf("could not subscribe to queue: %w", err)
 	}
 
-	// capture the subscription of ephemeral consumers so we can unsubscribe when we exit.
-	if b.handlerIsEphemeral(h) {
-		b.unsubscribe = append(b.unsubscribe, func() { sub.Unsubscribe() })
-	}
-
 	// Register handler.
 	b.registered[h.HandlerType()] = struct{}{}
 
@@ -212,31 +191,11 @@ func (b *EventBus) Errors() <-chan error {
 	return b.errCh
 }
 
-// handlerIsEphemeral traverses the middleware chain and checks for the
-// ephemeral middleware and quires it's status.
-func (b *EventBus) handlerIsEphemeral(h eh.EventHandler) bool {
-	for {
-		if obs, ok := h.(ephemeral.EphemeralHandler); ok {
-			return obs.IsEphemeralHandler()
-		} else if c, ok := h.(eh.EventHandlerChain); ok {
-			if h = c.InnerHandler(); h != nil {
-				continue
-			}
-		}
-		return false
-	}
-}
-
 // Close implements the Close method of the eventhorizon.EventBus interface.
 func (b *EventBus) Close() error {
 	b.cancel()
 	b.wg.Wait()
 
-	// unsubscribe any ephemeral subscribers we created.
-	for _, unSub := range b.unsubscribe {
-		unSub()
-	}
-
 	b.conn.Close()
 
 	return nil
diff --git a/eventstore.go b/eventstore.go
index 7e51848..34d6ed2 100644
--- a/eventstore.go
+++ b/eventstore.go
@@ -31,10 +31,19 @@ type EventStore interface {
 	// Load loads all events for the aggregate id from the store.
 	Load(context.Context, uuid.UUID) ([]Event, error)
 
+	// LoadFrom loads all events from version for the aggregate id from the store.
+	LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]Event, error)
+
 	// Close closes the EventStore.
 	Close() error
 }
 
+// SnapshotStore is an interface for snapshot store.
+type SnapshotStore interface {
+	LoadSnapshot(ctx context.Context, id uuid.UUID) (*Snapshot, error)
+	SaveSnapshot(ctx context.Context, id uuid.UUID, snapshot Snapshot) error
+}
+
 var (
 	// Missing events for save operation.
 	ErrMissingEvents = errors.New("missing events")
@@ -64,6 +73,11 @@ const (
 	EventStoreOpRename = "rename"
 	// Errors during clearing of the event store.
 	EventStoreOpClear = "clear"
+
+	// Errors during loading of snapshot.
+	EventStoreOpLoadSnapshot = "load_snapshot"
+	// Errors during saving of snapshot.
+	EventStoreOpSaveSnapshot = "save_snapshot"
 )
 
 // EventStoreError is an error in the event store.
@@ -107,12 +121,8 @@ func (e *EventStoreError) Error() string {
 
 	if len(e.Events) > 0 {
 		var es []string
-		for _, ev := range e.Events {
-			if ev != nil {
-				es = append(es, ev.String())
-			} else {
-				es = append(es, "nil event")
-			}
+		for _, e := range e.Events {
+			es = append(es, e.String())
 		}
 
 		str += " [" + strings.Join(es, ", ") + "]"
diff --git a/eventstore/memory/eventstore.go b/eventstore/memory/eventstore.go
index 7d00da4..0e3b0ef 100644
--- a/eventstore/memory/eventstore.go
+++ b/eventstore/memory/eventstore.go
@@ -190,6 +190,11 @@ func (s *EventStore) save(ctx context.Context, events []eh.Event, originalVersio
 
 // Load implements the Load method of the eventhorizon.EventStore interface.
 func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error) {
+	return s.LoadFrom(ctx, id, 1)
+}
+
+// LoadFrom loads all events from version for the aggregate id from the store.
+func (s *EventStore) LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]eh.Event, error) {
 	s.dbMu.RLock()
 	defer s.dbMu.RUnlock()
 
@@ -205,6 +210,10 @@ func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 	events := make([]eh.Event, len(aggregate.Events))
 
 	for i, event := range aggregate.Events {
+		if event.Version() < version {
+			continue
+		}
+
 		e, err := copyEvent(ctx, event)
 		if err != nil {
 			return nil, &eh.EventStoreError{
diff --git a/eventstore/mongodb/eventstore.go b/eventstore/mongodb/eventstore.go
index dcf81cb..a6efd07 100644
--- a/eventstore/mongodb/eventstore.go
+++ b/eventstore/mongodb/eventstore.go
@@ -324,6 +324,11 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 
 // Load implements the Load method of the eventhorizon.EventStore interface.
 func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error) {
+	return s.LoadFrom(ctx, id, 1)
+}
+
+// LoadFrom loads all events from version for the aggregate id from the store.
+func (s *EventStore) LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]eh.Event, error) {
 	var aggregate aggregateRecord
 	if err := s.aggregates.FindOne(ctx, bson.M{"_id": id}).Decode(&aggregate); err != nil {
 		// Translate to our own not found error.
@@ -341,6 +346,10 @@ func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 	events := make([]eh.Event, len(aggregate.Events))
 
 	for i, e := range aggregate.Events {
+		if e.Version < version {
+			continue
+		}
+
 		// Create an event of the correct type and decode from raw BSON.
 		if len(e.RawData) > 0 {
 			var err error
diff --git a/eventstore/mongodb_v2/eventstore.go b/eventstore/mongodb_v2/eventstore.go
index e37c9d2..0e347aa 100644
--- a/eventstore/mongodb_v2/eventstore.go
+++ b/eventstore/mongodb_v2/eventstore.go
@@ -15,12 +15,18 @@
 package mongodb_v2
 
 import (
+	"bytes"
+	"compress/gzip"
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
 	mongoOptions "go.mongodb.org/mongo-driver/mongo/options"
 	"go.mongodb.org/mongo-driver/mongo/readconcern"
 	"go.mongodb.org/mongo-driver/mongo/readpref"
@@ -37,13 +43,15 @@ import (
 // for all events and another to keep track of all aggregates/streams. It also
 // keep tracks of the global position of events, stored as metadata.
 type EventStore struct {
-	client                *mongo.Client
-	clientOwnership       clientOwnership
-	db                    *mongo.Database
-	events                *mongo.Collection
-	streams               *mongo.Collection
-	eventHandlerAfterSave eh.EventHandler
-	eventHandlerInTX      eh.EventHandler
+	client                  *mongo.Client
+	clientOwnership         clientOwnership
+	db                      *mongo.Database
+	events                  *mongo.Collection
+	streams                 *mongo.Collection
+	snapshots               *mongo.Collection
+	eventHandlerAfterSave   eh.EventHandler
+	eventHandlerInTX        eh.EventHandler
+	skipNonRegisteredEvents bool
 }
 
 type clientOwnership int
@@ -85,6 +93,7 @@ func newEventStoreWithClient(client *mongo.Client, clientOwnership clientOwnersh
 		db:              db,
 		events:          db.Collection("events"),
 		streams:         db.Collection("streams"),
+		snapshots:       db.Collection("snapshots"),
 	}
 
 	for _, option := range options {
@@ -105,6 +114,24 @@ func newEventStoreWithClient(client *mongo.Client, clientOwnership clientOwnersh
 		return nil, fmt.Errorf("could not ensure events index: %w", err)
 	}
 
+	if _, err := s.events.Indexes().CreateOne(ctx, mongo.IndexModel{
+		Keys: bson.M{"version": 1},
+	}); err != nil {
+		return nil, fmt.Errorf("could not ensure events index: %w", err)
+	}
+
+	if _, err := s.snapshots.Indexes().CreateOne(ctx, mongo.IndexModel{
+		Keys: bson.M{"aggregate_id": 1},
+	}); err != nil {
+		return nil, fmt.Errorf("could not ensure snapshot aggregate_id index: %w", err)
+	}
+
+	if _, err := s.snapshots.Indexes().CreateOne(ctx, mongo.IndexModel{
+		Keys: bson.M{"version": 1},
+	}); err != nil {
+		return nil, fmt.Errorf("could not ensure snapshot version index: %w", err)
+	}
+
 	// Make sure the $all stream exists.
 	if err := s.streams.FindOne(ctx, bson.M{
 		"_id": "$all",
@@ -354,7 +381,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 
 		if s.eventHandlerInTX != nil {
 			for _, e := range events {
-				if err := s.eventHandlerInTX.HandleEvent(txCtx, e); err != nil {
+				if err := s.eventHandlerInTX.HandleEvent(ctx, e); err != nil {
 					return nil, fmt.Errorf("could not handle event in transaction: %w", err)
 				}
 			}
@@ -398,6 +425,24 @@ func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 		}
 	}
 
+	return s.loadFromCursor(ctx, id, cursor)
+}
+
+// LoadFrom implements LoadFrom method of the eventhorizon.SnapshotStore interface.
+func (s *EventStore) LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]eh.Event, error) {
+	cursor, err := s.events.Find(ctx, bson.M{"aggregate_id": id, "version": bson.M{"$gte": version}})
+	if err != nil {
+		return nil, &eh.EventStoreError{
+			Err:         fmt.Errorf("could not find event: %w", err),
+			Op:          eh.EventStoreOpLoad,
+			AggregateID: id,
+		}
+	}
+
+	return s.loadFromCursor(ctx, id, cursor)
+}
+
+func (s *EventStore) loadFromCursor(ctx context.Context, id uuid.UUID, cursor *mongo.Cursor) ([]eh.Event, error) {
 	var events []eh.Event
 
 	for cursor.Next(ctx) {
@@ -464,6 +509,109 @@ func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 	return events, nil
 }
 
+func (s *EventStore) LoadSnapshot(ctx context.Context, id uuid.UUID) (*eh.Snapshot, error) {
+	result := s.snapshots.FindOne(ctx, bson.M{"aggregate_id": id}, options.FindOne().SetSort(bson.M{"version": -1}))
+	if err := result.Err(); err != nil {
+		if errors.Is(err, mongo.ErrNoDocuments) {
+			return nil, nil
+		}
+
+		return nil, err
+	}
+
+	var (
+		record   = new(SnapshotRecord)
+		snapshot = new(eh.Snapshot)
+		err      error
+	)
+
+	if err := result.Decode(record); err != nil {
+		return nil, &eh.EventStoreError{
+			Err:         fmt.Errorf("could not decode snapshot: %w", err),
+			Op:          eh.EventStoreOpLoadSnapshot,
+			AggregateID: id,
+		}
+	}
+
+	if snapshot.State, err = eh.CreateSnapshotData(record.AggregateID, record.AggregateType); err != nil {
+		return nil, &eh.EventStoreError{
+			Err:         fmt.Errorf("could not decode snapshot: %w", err),
+			Op:          eh.EventStoreOpLoadSnapshot,
+			AggregateID: id,
+		}
+	}
+
+	if err = record.decompress(); err != nil {
+		return nil, &eh.EventStoreError{
+			Err:         fmt.Errorf("could not decompress snapshot: %w", err),
+			Op:          eh.EventStoreOpLoadSnapshot,
+			AggregateID: id,
+		}
+	}
+
+	if err = json.Unmarshal(record.RawData, snapshot); err != nil {
+		return nil, &eh.EventStoreError{
+			Err:         fmt.Errorf("could not decode snapshot: %w", err),
+			Op:          eh.EventStoreOpLoadSnapshot,
+			AggregateID: id,
+		}
+	}
+
+	return snapshot, nil
+}
+
+func (s *EventStore) SaveSnapshot(ctx context.Context, id uuid.UUID, snapshot eh.Snapshot) (err error) {
+	if snapshot.AggregateType == "" {
+		return &eh.EventStoreError{
+			Err:           fmt.Errorf("aggregate type is empty"),
+			Op:            eh.EventStoreOpSaveSnapshot,
+			AggregateID:   id,
+			AggregateType: snapshot.AggregateType,
+		}
+	}
+
+	if snapshot.State == nil {
+		return &eh.EventStoreError{
+			Err:           fmt.Errorf("snapshots state is nil"),
+			Op:            eh.EventStoreOpSaveSnapshot,
+			AggregateID:   id,
+			AggregateType: snapshot.AggregateType,
+		}
+	}
+
+	record := SnapshotRecord{
+		AggregateID:   id,
+		AggregateType: snapshot.AggregateType,
+		Timestamp:     time.Now(),
+		Version:       snapshot.Version,
+	}
+
+	if record.RawData, err = json.Marshal(snapshot); err != nil {
+		return
+	}
+
+	if err = record.compress(); err != nil {
+		return &eh.EventStoreError{
+			Err:         fmt.Errorf("could not compress snapshot: %w", err),
+			Op:          eh.EventStoreOpSaveSnapshot,
+			AggregateID: id,
+		}
+	}
+
+	if _, err := s.snapshots.InsertOne(ctx,
+		record,
+		options.InsertOne(),
+	); err != nil {
+		return &eh.EventStoreError{
+			Err:         fmt.Errorf("could not save snapshot: %w", err),
+			Op:          eh.EventStoreOpSaveSnapshot,
+			AggregateID: id,
+		}
+	}
+
+	return nil
+}
+
 // Close implements the Close method of the eventhorizon.EventStore interface.
 func (s *EventStore) Close() error {
 	if s.clientOwnership == externalClient {
@@ -474,6 +622,48 @@ func (s *EventStore) Close() error {
 	return s.client.Disconnect(context.Background())
 }
 
+type SnapshotRecord struct {
+	AggregateID   uuid.UUID        `bson:"aggregate_id"`
+	RawData       []byte           `bson:"data"`
+	Timestamp     time.Time        `bson:"timestamp"`
+	Version       int              `bson:"version"`
+	AggregateType eh.AggregateType `bson:"aggregate_type"`
+}
+
+func (r *SnapshotRecord) decompress() error {
+	byteReader := bytes.NewReader(r.RawData)
+	reader, err := gzip.NewReader(byteReader)
+
+	if err != nil {
+		return err
+	}
+
+	r.RawData, err = io.ReadAll(reader)
+
+	return err
+}
+
+func (r *SnapshotRecord) compress() error {
+	var b bytes.Buffer
+	w := gzip.NewWriter(&b)
+
+	if _, err := w.Write(r.RawData); err != nil {
+		return err //nolint:wrapcheck
+	}
+
+	if err := w.Flush(); err != nil {
+		return err //nolint:wrapcheck
+	}
+
+	if err := w.Close(); err != nil {
+		return err //nolint:wrapcheck
+	}
+
+	r.RawData = b.Bytes()
+
+	return nil
+}
+
 // stream is a stream of events, often containing the events for an aggregate.
 type stream struct {
 	ID            uuid.UUID        `bson:"_id"`
diff --git a/go.mod b/go.mod
index 3704554..f3f83fe 100644
--- a/go.mod
+++ b/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/nats-io/nats.go v1.13.0
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/segmentio/kafka-go v0.4.25
+	github.com/stretchr/testify v1.7.0
 	github.com/uber/jaeger-client-go v2.29.1+incompatible
 	go.mongodb.org/mongo-driver v1.8.0
 	google.golang.org/api v0.61.0
@@ -27,6 +28,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403 // indirect
 	github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0 // indirect
 	github.com/envoyproxy/protoc-gen-validate v0.1.0 // indirect
@@ -42,10 +44,11 @@ require (
 	github.com/nats-io/nats-server/v2 v2.6.2 // indirect
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/pierrec/lz4 v2.6.0+incompatible // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rogpeppe/go-internal v1.6.1 // indirect
-	github.com/stretchr/testify v1.7.0 // indirect
 	github.com/uber/jaeger-lib v2.4.1+incompatible // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.0.2 // indirect
@@ -63,4 +66,6 @@ require (
 	google.golang.org/genproto v0.0.0-20211118181313-81c1377c94b1 // indirect
 	google.golang.org/grpc v1.40.0 // indirect
 	google.golang.org/protobuf v1.27.1 // indirect
+	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect
+	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )
diff --git a/go.sum b/go.sum
index a1ba950..0d81669 100644
--- a/go.sum
+++ b/go.sum
@@ -227,6 +227,7 @@ github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
@@ -699,6 +700,7 @@ google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+Rur
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
diff --git a/middleware.go b/middleware.go
index 5a76d8f..496d31a 100644
--- a/middleware.go
+++ b/middleware.go
@@ -33,17 +33,6 @@ func UseCommandHandlerMiddleware(h CommandHandler, middleware ...CommandHandlerM
 // able to chain.
 type EventHandlerMiddleware func(EventHandler) EventHandler
 
-// EventHandlerChain declares InnerHandler that returns the inner handler of a event handler middleware.
-// This enables an endpoint or other middlewares to traverse the chain of handlers
-// in order to find a specific middleware that can be interacted with.
-//
-// For handlers who's intrinsic properties requires them to be the last responder of a chain, or
-// can't produce an InnerHandler, a nil response can be implemented thereby hindering any
-// further attempt to traverse the chain.
-type EventHandlerChain interface {
-	InnerHandler() EventHandler
-}
-
 // UseEventHandlerMiddleware wraps a EventHandler in one or more middleware.
 func UseEventHandlerMiddleware(h EventHandler, middleware ...EventHandlerMiddleware) EventHandler {
 	// Apply in reverse order.
diff --git a/middleware/eventhandler/async/middleware.go b/middleware/eventhandler/async/middleware.go
index a81f0d5..1d04f7a 100644
--- a/middleware/eventhandler/async/middleware.go
+++ b/middleware/eventhandler/async/middleware.go
@@ -36,11 +36,6 @@ type eventHandler struct {
 	errCh chan *Error
 }
 
-// InnerHandler implements EventHandlerChain
-func (h *eventHandler) InnerHandler() eh.EventHandler {
-	return h.EventHandler
-}
-
 // HandleEvent implements the HandleEvent method of the EventHandler.
 func (h *eventHandler) HandleEvent(ctx context.Context, event eh.Event) error {
 	go func() {
diff --git a/middleware/eventhandler/ephemeral/middleware.go b/middleware/eventhandler/ephemeral/middleware.go
deleted file mode 100644
index b21c10b..0000000
--- a/middleware/eventhandler/ephemeral/middleware.go
+++ /dev/null
@@ -1,55 +0,0 @@
-// Copyright (c) 2017 - 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 ephemeral
-
-import (
-	eh "github.com/looplab/eventhorizon"
-)
-
-// EphemeralHandler is used to check for an ephemeral observer middleware in a chain.
-type EphemeralHandler interface {
-	IsEphemeralHandler() bool
-}
-
-type eventHandler struct {
-	eh.EventHandler
-}
-
-// NewMiddleware creates a new middleware that can be examined for ephemeral status.
-// A handler can be ephemeral if it never cares about events created before the handler,
-// or care about events that might occur when the handler is offline.
-//
-// Such handlers can be for instance handlers that create their initial state on startup
-// but needs to update their internal state based on events as they happen.
-//
-// Marking a handler as ephemeral enables event publishers to optimize operations
-// and clean up subscriptions when they are no longer needed.
-func NewMiddleware() func(eh.EventHandler) eh.EventHandler {
-	return func(h eh.EventHandler) eh.EventHandler {
-		return &eventHandler{
-			EventHandler: h,
-		}
-	}
-}
-
-// IsEphemeralHandler returns true if the handler should be ephemeral if possible.
-func (h *eventHandler) IsEphemeralHandler() bool {
-	return true
-}
-
-// InnerHandler implements MiddlewareChain
-func (h *eventHandler) InnerHandler() eh.EventHandler {
-	return h.EventHandler
-}
diff --git a/middleware/eventhandler/observer/middleware.go b/middleware/eventhandler/observer/middleware.go
index 2861e2b..a634b51 100644
--- a/middleware/eventhandler/observer/middleware.go
+++ b/middleware/eventhandler/observer/middleware.go
@@ -73,11 +73,6 @@ func (h *eventHandler) HandlerType() eh.EventHandlerType {
 	return h.handlerType
 }
 
-// InnerHandler implements MiddlewareChain
-func (h *eventHandler) InnerHandler() eh.EventHandler {
-	return h.EventHandler
-}
-
 // NewMiddleware creates a middleware that lets multiple handlers handle an event
 // depending on their group. It works by suffixing the group name to the handler type.
 // To create an observer that is unique for every added handler use the RandomGroup.
diff --git a/mocks/mocks.go b/mocks/mocks.go
index 470fd5c..d657202 100644
--- a/mocks/mocks.go
+++ b/mocks/mocks.go
@@ -95,6 +95,19 @@ func (a *Aggregate) HandleCommand(ctx context.Context, cmd eh.Command) error {
 	return nil
 }
 
+func (a *Aggregate) CreateSnapshot() *eh.Snapshot {
+	return &eh.Snapshot{
+		Timestamp: time.Now(),
+		State:     a,
+	}
+}
+
+func (a *Aggregate) ApplySnapshot(snapshot *eh.Snapshot) {
+	agg := snapshot.State.(*Aggregate)
+	a.ID = agg.ID
+	a.Commands = agg.Commands
+}
+
 // EventData is a mocked event data, useful in testing.
 type EventData struct {
 	Content string
@@ -268,6 +281,7 @@ func (m *EventHandler) Wait(d time.Duration) bool {
 // AggregateStore is a mocked AggregateStore, useful in testing.
 type AggregateStore struct {
 	Aggregates map[uuid.UUID]eh.Aggregate
+	Snapshots  map[uuid.UUID]eh.Snapshot
 	Context    context.Context
 	// Used to simulate errors in HandleCommand.
 	Err error
@@ -298,11 +312,25 @@ func (m *AggregateStore) Save(ctx context.Context, aggregate eh.Aggregate) error
 	return nil
 }
 
+func (m *AggregateStore) TakeSnapshot(ctx context.Context, agg eh.Aggregate) error {
+	if m.Err != nil {
+		return m.Err
+	}
+
+	if agg2, ok := agg.(eh.Snapshotable); ok {
+		m.Context = ctx
+		m.Snapshots[agg.EntityID()] = *agg2.CreateSnapshot()
+	}
+
+	return nil
+}
+
 // EventStore is a mocked eventhorizon.EventStore, useful in testing.
 type EventStore struct {
-	Events  []eh.Event
-	Loaded  uuid.UUID
-	Context context.Context
+	Events   []eh.Event
+	Snapshot eh.Snapshot
+	Loaded   uuid.UUID
+	Context  context.Context
 	// Used to simulate errors in the store.
 	Err error
 }
@@ -333,6 +361,26 @@ func (m *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 	return m.Events, nil
 }
 
+// LoadFrom loads all events from version for the aggregate id from the store.
+func (m *EventStore) LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]eh.Event, error) {
+	if m.Err != nil {
+		return nil, m.Err
+	}
+
+	m.Loaded = id
+	m.Context = ctx
+
+	var events []eh.Event
+
+	for _, e := range m.Events {
+		if e.Version() >= version {
+			events = append(events, e)
+		}
+	}
+
+	return events, nil
+}
+
 // Replace implements the Replace method of the eventhorizon.EventStore interface.
 func (m *EventStore) Replace(ctx context.Context, event eh.Event) error {
 	if m.Err != nil {
@@ -350,6 +398,24 @@ func (m *EventStore) Close() error {
 	return nil
 }
 
+func (m *EventStore) LoadSnapshot(ctx context.Context, id uuid.UUID) (*eh.Snapshot, error) {
+	m.Context = ctx
+	m.Loaded = id
+
+	return &m.Snapshot, nil
+}
+
+func (m *EventStore) SaveSnapshot(ctx context.Context, id uuid.UUID, snapshot eh.Snapshot) error {
+	if m.Err != nil {
+		return m.Err
+	}
+
+	m.Snapshot = snapshot
+	m.Context = ctx
+
+	return nil
+}
+
 // EventBus is a mocked eventhorizon.EventBus, useful in testing.
 type EventBus struct {
 	Events  []eh.Event
diff --git a/namespace/eventstore.go b/namespace/eventstore.go
index f7feb15..a8eb047 100644
--- a/namespace/eventstore.go
+++ b/namespace/eventstore.go
@@ -35,25 +35,27 @@ type EventStore struct {
 // function to create new event stores for the provided namespace.
 //
 // Usage:
-//    eventStore := NewEventStore(func(ns string) (eh.EventStore, error) {
-//        s, err := mongodb.NewEventStore("mongodb://", ns)
-//        if err != nil {
-//            return nil, err
-//        }
-//        return s, nil
-//    })
+//
+//	eventStore := NewEventStore(func(ns string) (eh.EventStore, error) {
+//	    s, err := mongodb.NewEventStore("mongodb://", ns)
+//	    if err != nil {
+//	        return nil, err
+//	    }
+//	    return s, nil
+//	})
 //
 // Usage shared DB client:
-//    client, err := mongo.Connect(ctx)
-//    ...
 //
-//    eventStore := NewEventStore(func(ns string) (eh.EventStore, error) {
-//        s, err := mongodb.NewEventStoreWithClient(client, ns)
-//        if err != nil {
-//            return nil, err
-//        }
-//        return s, nil
-//    })
+//	client, err := mongo.Connect(ctx)
+//	...
+//
+//	eventStore := NewEventStore(func(ns string) (eh.EventStore, error) {
+//	    s, err := mongodb.NewEventStoreWithClient(client, ns)
+//	    if err != nil {
+//	        return nil, err
+//	    }
+//	    return s, nil
+//	})
 func NewEventStore(factory func(ns string) (eh.EventStore, error)) *EventStore {
 	return &EventStore{
 		eventStores:   map[string]eh.EventStore{},
@@ -93,6 +95,16 @@ func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
 	return store.Load(ctx, id)
 }
 
+// LoadFrom loads all events from version for the aggregate id from the store.
+func (s *EventStore) LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]eh.Event, error) {
+	store, err := s.eventStore(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return store.LoadFrom(ctx, id, version)
+}
+
 // Close implements the Close method of the eventhorizon.EventStore interface.
 func (s *EventStore) Close() error {
 	s.eventStoresMu.RLock()
@@ -127,7 +139,7 @@ func (s *EventStore) eventStore(ctx context.Context) (eh.EventStore, error) {
 
 		// Perform an additional existence check within the write lock in the
 		// unlikely event that someone else added the event store right before us.
-		if eventStore, ok = s.eventStores[ns]; !ok {
+		if _, ok := s.eventStores[ns]; !ok {
 			var err error
 
 			eventStore, err = s.newEventStore(ns)
diff --git a/snapshot.go b/snapshot.go
new file mode 100644
index 0000000..1a54311
--- /dev/null
+++ b/snapshot.go
@@ -0,0 +1,79 @@
+// Copyright (c) 2021 - 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 (
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/looplab/eventhorizon/uuid"
+)
+
+// Snapshotable is an interface for creating and applying a Snapshot record.
+type Snapshotable interface {
+	CreateSnapshot() *Snapshot
+	ApplySnapshot(snapshot *Snapshot)
+}
+
+// Snapshot is a recording of the state of an aggregate at a point in time
+type Snapshot struct {
+	Version       int
+	AggregateType AggregateType
+	Timestamp     time.Time
+	State         interface{}
+}
+
+var snapshotDataFactories = make(map[AggregateType]func(uuid2 uuid.UUID) SnapshotData)
+
+type SnapshotData interface{}
+
+var snapshotDataFactoriesMu sync.RWMutex
+
+var ErrSnapshotDataNotRegistered = errors.New("snapshot data not registered")
+
+// RegisterSnapshotData registers an snapshot factory for a type. The factory is
+// used to create concrete snapshot state type when unmarshalling.
+//
+// An example would be:
+//
+//	RegisterSnapshotData("aggregateType1", func() SnapshotData { return &MySnapshotData{} })
+func RegisterSnapshotData(aggregateType AggregateType, factory func(id uuid.UUID) SnapshotData) {
+	if aggregateType == AggregateType("") {
+		panic("eventhorizon: attempt to register empty aggregate type")
+	}
+
+	snapshotDataFactoriesMu.Lock()
+	defer snapshotDataFactoriesMu.Unlock()
+
+	if _, ok := snapshotDataFactories[aggregateType]; ok {
+		panic(fmt.Sprintf("eventhorizon: registering duplicate types for %q", aggregateType))
+	}
+
+	snapshotDataFactories[aggregateType] = factory
+}
+
+// CreateSnapshotData create a concrete instance using the registered snapshot factories.
+func CreateSnapshotData(AggregateID uuid.UUID, aggregateType AggregateType) (SnapshotData, error) {
+	snapshotDataFactoriesMu.RLock()
+	defer snapshotDataFactoriesMu.RUnlock()
+
+	if factory, ok := snapshotDataFactories[aggregateType]; ok {
+		return factory(AggregateID), nil
+	}
+
+	return nil, ErrSnapshotDataNotRegistered
+}
diff --git a/tracing/eventhandler.go b/tracing/eventhandler.go
index d223b34..6191412 100644
--- a/tracing/eventhandler.go
+++ b/tracing/eventhandler.go
@@ -34,11 +34,6 @@ type eventHandler struct {
 	eh.EventHandler
 }
 
-// InnerHandler implements MiddlewareChain
-func (h *eventHandler) InnerHandler() eh.EventHandler {
-	return h.EventHandler
-}
-
 // HandleEvent implements the HandleEvent method of the EventHandler.
 func (h *eventHandler) HandleEvent(ctx context.Context, event eh.Event) error {
 	opName := fmt.Sprintf("%s.Event(%s)", h.HandlerType(), event.EventType())


Test patch:
diff --git a/aggregatestore/events/aggregatestore.go b/aggregatestore/events/aggregatestore.go
index ad10732..def3420 100644
--- a/aggregatestore/events/aggregatestore.go
+++ b/aggregatestore/events/aggregatestore.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/uuid"
@@ -27,7 +28,10 @@ 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
+	store            eh.EventStore
+	snapshotStore    eh.SnapshotStore
+	isSnapshotStore  bool
+	snapshotStrategy eh.SnapshotStrategy
 }
 
 var (
@@ -39,10 +43,10 @@ var (
 	ErrMismatchedEventType = errors.New("mismatched event type and aggregate type")
 )
 
-// NewAggregateStore creates a aggregate store with an event store and an event
+// NewAggregateStore creates an 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) (*AggregateStore, error) {
+func NewAggregateStore(store eh.EventStore, options ...Option) (*AggregateStore, error) {
 	if store == nil {
 		return nil, ErrInvalidEventStore
 	}
@@ -51,9 +55,31 @@ func NewAggregateStore(store eh.EventStore) (*AggregateStore, error) {
 		store: store,
 	}
 
+	d.snapshotStrategy = &NoSnapshotStrategy{}
+
+	for _, option := range options {
+		if err := option(d); err != nil {
+			return nil, fmt.Errorf("error while applying option: %w", err)
+		}
+	}
+
+	d.snapshotStore, d.isSnapshotStore = store.(eh.SnapshotStore)
+
 	return d, nil
 }
 
+// Option is an option setter used to configure creation.
+type Option func(*AggregateStore) error
+
+// WithSnapshotStrategy add the strategy to use when determining if a snapshot should be taken
+func WithSnapshotStrategy(s eh.SnapshotStrategy) Option {
+	return func(as *AggregateStore) error {
+		as.snapshotStrategy = s
+
+		return nil
+	}
+}
+
 // Load implements the Load method of the eventhorizon.AggregateStore interface.
 // It loads an aggregate from the event store by creating a new aggregate of the
 // type with the ID and then applies all events to it, thus making it the most
@@ -79,7 +105,26 @@ func (r *AggregateStore) Load(ctx context.Context, aggregateType eh.AggregateTyp
 		}
 	}
 
-	events, err := r.store.Load(ctx, a.EntityID())
+	fromVersion := 1
+
+	if sa, ok := a.(eh.Snapshotable); ok && r.isSnapshotStore {
+		snapshot, err := r.snapshotStore.LoadSnapshot(ctx, id)
+		if err != nil {
+			return nil, &eh.AggregateStoreError{
+				Err:           err,
+				Op:            eh.AggregateStoreOpLoad,
+				AggregateType: aggregateType,
+				AggregateID:   id,
+			}
+		}
+
+		if snapshot != nil {
+			sa.ApplySnapshot(snapshot)
+			fromVersion = snapshot.Version + 1
+		}
+	}
+
+	events, err := r.store.LoadFrom(ctx, a.EntityID(), fromVersion)
 	if err != nil && !errors.Is(err, eh.ErrAggregateNotFound) {
 		return nil, &eh.AggregateStoreError{
 			Err:           err,
@@ -142,6 +187,44 @@ func (r *AggregateStore) Save(ctx context.Context, agg eh.Aggregate) error {
 		}
 	}
 
+	return r.takeSnapshot(ctx, agg, events[len(events)-1])
+}
+
+func (r *AggregateStore) takeSnapshot(ctx context.Context, agg eh.Aggregate, lastEvent eh.Event) error {
+	a, ok := agg.(eh.Snapshotable)
+	if !ok || !r.isSnapshotStore {
+		return nil
+	}
+
+	s, err := r.snapshotStore.LoadSnapshot(ctx, agg.EntityID())
+	if err != nil {
+		return &eh.AggregateStoreError{
+			Err:           err,
+			Op:            eh.AggregateStoreOpSave,
+			AggregateType: agg.AggregateType(),
+			AggregateID:   agg.EntityID(),
+		}
+	}
+
+	version := 0
+	timestamp := time.Now()
+
+	if s != nil {
+		version = s.Version
+		timestamp = s.Timestamp
+	}
+
+	if res := r.snapshotStrategy.ShouldTakeSnapshot(version, timestamp, lastEvent); res {
+		if err = r.snapshotStore.SaveSnapshot(ctx, agg.EntityID(), *a.CreateSnapshot()); err != nil {
+			return &eh.AggregateStoreError{
+				Err:           err,
+				Op:            eh.AggregateStoreOpSave,
+				AggregateType: agg.AggregateType(),
+				AggregateID:   agg.EntityID(),
+			}
+		}
+	}
+
 	return nil
 }
 
diff --git a/aggregatestore/events/aggregatestore_test.go b/aggregatestore/events/aggregatestore_test.go
index 86f924d..c850016 100644
--- a/aggregatestore/events/aggregatestore_test.go
+++ b/aggregatestore/events/aggregatestore_test.go
@@ -17,6 +17,7 @@ package events
 import (
 	"context"
 	"errors"
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -24,6 +25,7 @@ import (
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/mocks"
 	"github.com/looplab/eventhorizon/uuid"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestNewAggregateStore(t *testing.T) {
@@ -253,6 +255,46 @@ func TestAggregateStore_SaveEvents(t *testing.T) {
 	agg.err = nil
 }
 
+func TestAggregateStore_TakeSnapshot(t *testing.T) {
+	eventStore := &mocks.EventStore{
+		Events: make([]eh.Event, 0),
+	}
+
+	store, err := NewAggregateStore(eventStore, WithSnapshotStrategy(NewEveryNumberEventSnapshotStrategy(2)))
+	if err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+
+	ctx := context.Background()
+
+	id := uuid.New()
+	agg := NewTestAggregateOther(id)
+
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+
+	for i := 0; i < 3; i++ {
+		agg.AppendEvent(mocks.EventType, &mocks.EventData{Content: fmt.Sprintf("event%d", i)}, timestamp)
+
+		if err := store.Save(ctx, agg); err != nil {
+			t.Error("should not be an error")
+		}
+	}
+
+	assert.NotNil(t, eventStore.Snapshot, "snapshot should be taken")
+
+	agg2, err := store.Load(ctx, agg.AggregateType(), agg.EntityID())
+	if err != nil {
+		t.Error("should not be an error")
+	}
+
+	a, ok := agg2.(*TestAggregateOther)
+	if !ok {
+		t.Error("wrong aggregate type")
+	}
+
+	assert.Equal(t, 1, a.appliedEvents)
+}
+
 func TestAggregateStore_AggregateNotRegistered(t *testing.T) {
 	store, _ := createStore(t)
 
@@ -296,7 +338,8 @@ const TestAggregateOtherType eh.AggregateType = "TestAggregateOther"
 
 type TestAggregateOther struct {
 	*AggregateBase
-	err error
+	err           error
+	appliedEvents int
 }
 
 var _ = VersionedAggregate(&TestAggregateOther{})
@@ -316,5 +359,21 @@ func (a *TestAggregateOther) ApplyEvent(ctx context.Context, event eh.Event) err
 		return a.err
 	}
 
+	a.appliedEvents++
+
 	return nil
 }
+
+func (a *TestAggregateOther) CreateSnapshot() *eh.Snapshot {
+	return &eh.Snapshot{
+		Version:       a.AggregateVersion(),
+		Timestamp:     time.Now(),
+		AggregateType: TestAggregateType,
+		State:         a,
+	}
+}
+
+func (a *TestAggregateOther) ApplySnapshot(snapshot *eh.Snapshot) {
+	agg := snapshot.State.(*TestAggregateOther)
+	a.id = agg.id
+}
diff --git a/aggregatestore/events/snapshotstrategy.go b/aggregatestore/events/snapshotstrategy.go
new file mode 100644
index 0000000..0000359
--- /dev/null
+++ b/aggregatestore/events/snapshotstrategy.go
@@ -0,0 +1,65 @@
+// Copyright (c) 2021 - 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 events
+
+import (
+	"time"
+
+	eh "github.com/looplab/eventhorizon"
+)
+
+// NoSnapshotStrategy no snapshot should be taken.
+type NoSnapshotStrategy struct {
+}
+
+func (s *NoSnapshotStrategy) ShouldTakeSnapshot(_ int,
+	_ time.Time,
+	_ eh.Event) bool {
+	return false
+}
+
+// EveryNumberEventSnapshotStrategy use to take a snapshot every n number of events.
+type EveryNumberEventSnapshotStrategy struct {
+	snapshotThreshold int
+}
+
+func NewEveryNumberEventSnapshotStrategy(threshold int) *EveryNumberEventSnapshotStrategy {
+	return &EveryNumberEventSnapshotStrategy{
+		snapshotThreshold: threshold,
+	}
+}
+
+func (s *EveryNumberEventSnapshotStrategy) ShouldTakeSnapshot(lastSnapshotVersion int,
+	_ time.Time,
+	event eh.Event) bool {
+	return event.Version()-lastSnapshotVersion >= s.snapshotThreshold
+}
+
+// PeriodSnapshotStrategy use to take a snapshot every time a period has elapsed, for example every hour.
+type PeriodSnapshotStrategy struct {
+	snapshotThreshold time.Duration
+}
+
+func NewPeriodSnapshotStrategy(threshold time.Duration) *PeriodSnapshotStrategy {
+	return &PeriodSnapshotStrategy{
+		snapshotThreshold: threshold,
+	}
+}
+
+func (s *PeriodSnapshotStrategy) ShouldTakeSnapshot(_ int,
+	lastSnapshotTimestamp time.Time,
+	event eh.Event) bool {
+	return event.Timestamp().Sub(lastSnapshotTimestamp) >= s.snapshotThreshold
+}
diff --git a/aggregatestore/events/snapshotstrategy_test.go b/aggregatestore/events/snapshotstrategy_test.go
new file mode 100644
index 0000000..3637b72
--- /dev/null
+++ b/aggregatestore/events/snapshotstrategy_test.go
@@ -0,0 +1,70 @@
+// Copyright (c) 2021 - 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 events
+
+import (
+	"testing"
+	"time"
+
+	eh "github.com/looplab/eventhorizon"
+	"github.com/looplab/eventhorizon/mocks"
+	"github.com/looplab/eventhorizon/uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNoSnapshotStrategy_ShouldNotTakeSnapshot(t *testing.T) {
+	strategy := NoSnapshotStrategy{}
+
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+	result := strategy.ShouldTakeSnapshot(1, timestamp, eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event"}, timestamp))
+
+	assert.False(t, result)
+}
+
+func TestEveryNumberEventSnapshotStrategy_ShouldTakeSnapshot(t *testing.T) {
+	strategy := NewEveryNumberEventSnapshotStrategy(2)
+
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+	lastEvt := eh.NewEvent(mocks.EventType,
+		&mocks.EventData{Content: "event"}, timestamp,
+		eh.ForAggregate("testAgg", uuid.New(), 4))
+
+	result := strategy.ShouldTakeSnapshot(2, timestamp, lastEvt)
+
+	assert.True(t, result)
+
+	result = strategy.ShouldTakeSnapshot(3, timestamp, lastEvt)
+
+	assert.False(t, result)
+}
+
+func TestPeriodSnapshotStrategy_ShouldTakeSnapshot(t *testing.T) {
+	strategy := NewPeriodSnapshotStrategy(10 * time.Minute)
+
+	timestamp := time.Date(2009, time.November, 10, 23, 20, 0, 0, time.UTC)
+	lastEvt := eh.NewEvent(mocks.EventType,
+		&mocks.EventData{Content: "event"}, timestamp,
+		eh.ForAggregate("testAgg", uuid.New(), 4))
+
+	snapshotTimestamp := time.Date(2009, time.November, 10, 23, 10, 0, 0, time.UTC)
+	result := strategy.ShouldTakeSnapshot(10, snapshotTimestamp, lastEvt)
+
+	assert.True(t, result)
+
+	snapshotTimestamp = time.Date(2009, time.November, 10, 23, 15, 0, 0, time.UTC)
+	result = strategy.ShouldTakeSnapshot(10, snapshotTimestamp, lastEvt)
+
+	assert.False(t, result)
+}
diff --git a/commandhandler/aggregate/commandhandler_test.go b/commandhandler/aggregate/commandhandler_test.go
index 8a32fc7..a9fc61f 100644
--- a/commandhandler/aggregate/commandhandler_test.go
+++ b/commandhandler/aggregate/commandhandler_test.go
@@ -28,6 +28,7 @@ import (
 func TestNewCommandHandler(t *testing.T) {
 	store := &mocks.AggregateStore{
 		Aggregates: make(map[uuid.UUID]eh.Aggregate),
+		Snapshots:  make(map[uuid.UUID]eh.Snapshot),
 	}
 
 	h, err := NewCommandHandler(mocks.AggregateType, store)
@@ -76,6 +77,7 @@ func TestCommandHandler(t *testing.T) {
 func TestCommandHandler_AggregateNotFound(t *testing.T) {
 	store := &mocks.AggregateStore{
 		Aggregates: map[uuid.UUID]eh.Aggregate{},
+		Snapshots:  make(map[uuid.UUID]eh.Snapshot),
 	}
 
 	h, err := NewCommandHandler(mocks.AggregateType, store)
@@ -155,6 +157,7 @@ func BenchmarkCommandHandler(b *testing.B) {
 		Aggregates: map[uuid.UUID]eh.Aggregate{
 			a.EntityID(): a,
 		},
+		Snapshots: make(map[uuid.UUID]eh.Snapshot),
 	}
 
 	h, err := NewCommandHandler(mocks.AggregateType, store)
@@ -183,6 +186,7 @@ func createAggregateAndHandler(t *testing.T) (*mocks.Aggregate, *CommandHandler,
 		Aggregates: map[uuid.UUID]eh.Aggregate{
 			a.EntityID(): a,
 		},
+		Snapshots: make(map[uuid.UUID]eh.Snapshot),
 	}
 
 	h, err := NewCommandHandler(mocks.AggregateType, store)
diff --git a/eventbus/gcp/eventbus_test.go b/eventbus/gcp/eventbus_test.go
index 84ef32e..c7a78c6 100644
--- a/eventbus/gcp/eventbus_test.go
+++ b/eventbus/gcp/eventbus_test.go
@@ -22,7 +22,6 @@ import (
 	"testing"
 	"time"
 
-	"cloud.google.com/go/pubsub"
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventbus"
 )
@@ -60,26 +59,6 @@ func TestEventBusIntegration(t *testing.T) {
 	eventbus.AcceptanceTest(t, bus1, bus2, time.Second)
 }
 
-func TestEventBusWithConfigIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-
-	bus1, appID, err := newTestEventBusWithTopicConfig("")
-	if err != nil {
-		t.Fatal("there should be no error:", err)
-	}
-
-	bus2, _, err := newTestEventBusWithTopicConfig(appID)
-	if err != nil {
-		t.Fatal("there should be no error:", err)
-	}
-
-	t.Logf("using topic: %s_events", appID)
-
-	eventbus.AcceptanceTest(t, bus1, bus2, time.Second)
-}
-
 func TestEventBusLoadtest(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
@@ -129,31 +108,3 @@ func newTestEventBus(appID string) (eh.EventBus, string, error) {
 
 	return bus, appID, nil
 }
-
-func newTestEventBusWithTopicConfig(appID string) (eh.EventBus, string, error) {
-	// Connect to localhost if not running inside docker
-	if os.Getenv("PUBSUB_EMULATOR_HOST") == "" {
-		os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8793")
-	}
-
-	// Get a random app ID.
-	if appID == "" {
-		bts := make([]byte, 8)
-		if _, err := rand.Read(bts); err != nil {
-			return nil, "", fmt.Errorf("could not randomize app ID: %w", err)
-		}
-
-		appID = "app-" + hex.EncodeToString(bts)
-	}
-
-	// Create an empty config. The emulator doesn't support configurable message retention.
-	topicConfig := &pubsub.TopicConfig{
-		// RetentionDuration: 7 * 24 * time.Hour,
-	}
-	bus, err := NewEventBus("project_id", appID, WithTopicOptions(topicConfig))
-	if err != nil {
-		return nil, "", fmt.Errorf("could not create event bus: %w", err)
-	}
-
-	return bus, appID, nil
-}
diff --git a/eventbus/kafka/eventbus_test.go b/eventbus/kafka/eventbus_test.go
index 8d21a88..0186aeb 100644
--- a/eventbus/kafka/eventbus_test.go
+++ b/eventbus/kafka/eventbus_test.go
@@ -18,6 +18,7 @@ import (
 	"crypto/rand"
 	"encoding/hex"
 	"fmt"
+	"github.com/stretchr/testify/assert"
 	"os"
 	"testing"
 	"time"
@@ -25,6 +26,7 @@ import (
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventbus"
 	"github.com/segmentio/kafka-go"
+	"github.com/stretchr/testify/require"
 )
 
 func TestAddHandlerIntegration(t *testing.T) {
@@ -142,40 +144,12 @@ func TestWithStartOffset(t *testing.T) {
 		t.Run(desc, func(t *testing.T) {
 			eb, _, err := newTestEventBus("", WithStartOffset(tc.startOffset))
 			if err != nil {
-				t.Fatalf("expected no error, got: %s", err.Error())
+				t.Fatal("there should be no error:", err)
 			}
+			require.NoError(t, err)
 
 			underlyingEb := eb.(*EventBus)
-			if want, got := tc.startOffset, underlyingEb.startOffset; want != got {
-				t.Fatalf("expected topics to be equal, want %d, got: %d", want, got)
-			}
-		})
-	}
-}
-
-func TestWithTopic(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-
-	testCases := map[string]struct {
-		topic string
-	}{
-		"test1": {"test_test1"},
-		"test2": {"test_test2"},
-	}
-
-	for desc, tc := range testCases {
-		t.Run(desc, func(t *testing.T) {
-			eb, _, err := newTestEventBus("", WithTopic(tc.topic))
-			if err != nil {
-				t.Fatalf("expected no error, got: %s", err.Error())
-			}
-
-			underlyingEb := eb.(*EventBus)
-			if want, got := tc.topic, underlyingEb.topic; want != got {
-				t.Fatalf("expected topics to be equal, want %s, got: %s", want, got)
-			}
+			assert.Equal(t, tc.startOffset, underlyingEb.startOffset)
 		})
 	}
 }
diff --git a/eventstore/acceptanece_testing.go b/eventstore/acceptance_testing.go
similarity index 76%
rename from eventstore/acceptanece_testing.go
rename to eventstore/acceptance_testing.go
index 4147545..2c67cb8 100644
--- a/eventstore/acceptanece_testing.go
+++ b/eventstore/acceptance_testing.go
@@ -25,17 +25,17 @@ import (
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/mocks"
 	"github.com/looplab/eventhorizon/uuid"
+	"github.com/stretchr/testify/assert"
 )
 
 // AcceptanceTest is the acceptance test that all implementations of EventStore
 // should pass. It should manually be called from a test case in each
 // implementation:
 //
-//   func TestEventStore(t *testing.T) {
-//       store := NewEventStore()
-//       eventstore.AcceptanceTest(t, store, context.Background())
-//   }
-//
+//	func TestEventStore(t *testing.T) {
+//	    store := NewEventStore()
+//	    eventstore.AcceptanceTest(t, store, context.Background())
+//	}
 func AcceptanceTest(t *testing.T, store eh.EventStore, ctx context.Context) []eh.Event {
 	savedEvents := []eh.Event{}
 
@@ -214,6 +214,80 @@ func AcceptanceTest(t *testing.T, store eh.EventStore, ctx context.Context) []eh
 	return savedEvents
 }
 
+func SnapshotAcceptanceTest(t *testing.T, store eh.EventStore, ctx context.Context) {
+	snapshotStore, ok := store.(eh.SnapshotStore)
+	if !ok {
+		return
+	}
+
+	id := uuid.New()
+
+	eventStoreErr := &eh.EventStoreError{}
+
+	err := snapshotStore.SaveSnapshot(ctx, id, eh.Snapshot{})
+	if !errors.As(err, &eventStoreErr) {
+		t.Error("there should be a event store error:", err)
+	}
+
+	err = snapshotStore.SaveSnapshot(ctx, id, eh.Snapshot{
+		AggregateType: "test",
+	})
+	if !errors.As(err, &eventStoreErr) {
+		t.Error("there should be a event store error:", err)
+	}
+
+	type TestData struct {
+		Data string
+	}
+
+	snapshot := eh.Snapshot{
+		Version:       1,
+		AggregateType: "test",
+		Timestamp:     time.Now(),
+		State: &TestData{
+			Data: "this is incredible data",
+		},
+	}
+
+	eh.RegisterSnapshotData("test", func(uuid.UUID) eh.SnapshotData { return &TestData{} })
+
+	if err := snapshotStore.SaveSnapshot(ctx, id, snapshot); err != nil {
+		t.Error("there should not be an error")
+	}
+
+	loaded, err := snapshotStore.LoadSnapshot(ctx, uuid.New())
+	if loaded != nil {
+		t.Error("snapshot should be nil, it doesnt exists")
+	}
+
+	loaded, err = snapshotStore.LoadSnapshot(ctx, id)
+	if err != nil {
+		t.Error("there should  be an error")
+	}
+
+	assert.Equal(t, snapshot.Version, loaded.Version)
+	assert.Equal(t, snapshot.AggregateType, loaded.AggregateType)
+	assert.Equal(t, snapshot.State, loaded.State)
+
+	snapshot.Version += 1
+	snapshot.Timestamp = time.Now()
+	snapshot.State = &TestData{
+		Data: "this is new incredible data",
+	}
+	if err := snapshotStore.SaveSnapshot(ctx, id, snapshot); err != nil {
+		t.Error("there should not be a error")
+	}
+
+	loaded, err = snapshotStore.LoadSnapshot(ctx, id)
+	if err != nil {
+		t.Error("there should not be a error")
+	}
+
+	assert.Equal(t, snapshot.Version, loaded.Version)
+	assert.Equal(t, snapshot.AggregateType, loaded.AggregateType)
+	assert.Equal(t, snapshot.State, loaded.State)
+}
+
 func eventsToString(events []eh.Event) string {
 	parts := make([]string, len(events))
 	for i, e := range events {
diff --git a/eventstore/mongodb_v2/eventstore_test.go b/eventstore/mongodb_v2/eventstore_test.go
index a519b0a..0efcb9d 100644
--- a/eventstore/mongodb_v2/eventstore_test.go
+++ b/eventstore/mongodb_v2/eventstore_test.go
@@ -62,6 +62,8 @@ func TestEventStoreIntegration(t *testing.T) {
 
 	eventstore.AcceptanceTest(t, store, context.Background())
 
+	eventstore.SnapshotAcceptanceTest(t, store, context.Background())
+
 	if err := store.Close(); err != nil {
 		t.Error("there should be no error:", err)
 	}
diff --git a/middleware/commandhandler/scheduler/middleware_test.go b/middleware/commandhandler/scheduler/middleware_test.go
index 6a84df1..ec18397 100644
--- a/middleware/commandhandler/scheduler/middleware_test.go
+++ b/middleware/commandhandler/scheduler/middleware_test.go
@@ -128,7 +128,7 @@ func TestMiddleware_PersistedIntegration(t *testing.T) {
 		addr = "localhost:27017"
 	}
 
-	url := "mongodb://" + addr
+	url := "mongodb://" + addr + "/?readPreference=primary&directConnection=true&ssl=false"
 
 	// Get a random DB name.
 	b := make([]byte, 4)
diff --git a/middleware/eventhandler/async/middleware_test.go b/middleware/eventhandler/async/middleware_test.go
index 28dfc5e..ad74e99 100644
--- a/middleware/eventhandler/async/middleware_test.go
+++ b/middleware/eventhandler/async/middleware_test.go
@@ -37,11 +37,6 @@ func TestMiddleware(t *testing.T) {
 	m, errCh := NewMiddleware()
 	h := eh.UseEventHandlerMiddleware(inner, m)
 
-	_, ok := h.(eh.EventHandlerChain)
-	if !ok {
-		t.Error("handler is not an EventHandlerChain")
-	}
-
 	if err := h.HandleEvent(context.Background(), event); err != nil {
 		t.Error("there should never be an error:", err)
 	}
diff --git a/middleware/eventhandler/ephemeral/middleware_test.go b/middleware/eventhandler/ephemeral/middleware_test.go
deleted file mode 100644
index c37ba74..0000000
--- a/middleware/eventhandler/ephemeral/middleware_test.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package ephemeral
-
-import (
-	"testing"
-
-	eh "github.com/looplab/eventhorizon"
-	"github.com/looplab/eventhorizon/mocks"
-)
-
-func TestInnerHandler(t *testing.T) {
-	m := NewMiddleware()
-	h := m(mocks.NewEventHandler("test"))
-	_, ok := h.(eh.EventHandlerChain)
-	if !ok {
-		t.Error("handler is not an EventHandlerChain")
-	}
-}
diff --git a/middleware/eventhandler/observer/middleware_test.go b/middleware/eventhandler/observer/middleware_test.go
index 73b7e50..4929211 100644
--- a/middleware/eventhandler/observer/middleware_test.go
+++ b/middleware/eventhandler/observer/middleware_test.go
@@ -74,10 +74,5 @@ func TestMiddleware(t *testing.T) {
 		t.Error("the handler type should be correct:", h5.HandlerType())
 	}
 
-	_, ok := h5.(eh.EventHandlerChain)
-	if !ok {
-		t.Error("handler is not an EventHandlerChain")
-	}
-
 	t.Log(h5.HandlerType())
 }