Repository: looplab/eventhorizon
Pull request #323 (2021-06-28 06:49:03)

Feature / Add option for projecting with version gaps
### Description

Add option for projecting with version gaps. This can be useful if a projector only projects some events from an aggregate.

### Affected Components

- Projector event handler

### Related Issues

### Solution and Design

Adds an option which changes the internal checks for versions when loading and projecting entities.

### Steps to test and verify



Patch:
diff --git a/eventhandler/projector/eventhandler.go b/eventhandler/projector/eventhandler.go
index b1e969d..54b0803 100644
--- a/eventhandler/projector/eventhandler.go
+++ b/eventhandler/projector/eventhandler.go
@@ -23,16 +23,6 @@ import (
 	"github.com/looplab/eventhorizon/repo/version"
 )
 
-// EventHandler is a CQRS projection handler to run a Projector implementation.
-type EventHandler struct {
-	projector Projector
-	repo      eh.ReadWriteRepo
-	factoryFn func() eh.Entity
-	useWait   bool
-}
-
-var _ = eh.EventHandler(&EventHandler{})
-
 // Projector is a projector of events onto models.
 type Projector interface {
 	// ProjectorType returns the type of the projector.
@@ -82,6 +72,17 @@ func (e Error) Cause() error {
 // ErrModelNotSet is when a model factory is not set on the EventHandler.
 var ErrModelNotSet = errors.New("model not set")
 
+// EventHandler is a CQRS projection handler to run a Projector implementation.
+type EventHandler struct {
+	projector              Projector
+	repo                   eh.ReadWriteRepo
+	factoryFn              func() eh.Entity
+	useWait                bool
+	useIrregularVersioning bool
+}
+
+var _ = eh.EventHandler(&EventHandler{})
+
 // NewEventHandler creates a new EventHandler.
 func NewEventHandler(projector Projector, repo eh.ReadWriteRepo, options ...Option) *EventHandler {
 	h := &EventHandler{
@@ -104,6 +105,15 @@ func WithWait() Option {
 	}
 }
 
+// WithIrregularVersioning sets the option to allow gaps in the version numbers.
+// This can be useful for projectors that project only some events of a larger
+// aggregate, which will lead to gaps in the versions.
+func WithIrregularVersioning() Option {
+	return func(h *EventHandler) {
+		h.useIrregularVersioning = true
+	}
+}
+
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (h *EventHandler) HandlerType() eh.EventHandlerType {
 	return eh.EventHandlerType("projector_" + h.projector.ProjectorType())
@@ -113,14 +123,20 @@ func (h *EventHandler) HandlerType() eh.EventHandlerType {
 // It will try to find the correct version of the model, waiting for it the projector
 // has the WithWait option set.
 func (h *EventHandler) HandleEvent(ctx context.Context, event eh.Event) error {
-	// Get or create the model, trying to find it with a min version (and optional
-	// retry) if the underlying repo supports it.
-	findCtx := version.NewContextWithMinVersion(ctx, event.Version()-1)
-	if h.useWait {
-		var cancel func()
-		findCtx, cancel = version.NewContextWithMinVersionWait(ctx, event.Version()-1)
-		defer cancel()
+	findCtx := ctx
+	// Irregular versioning skips min version loading.
+	if !h.useIrregularVersioning {
+		// Try to find it with a min version (and optional retry) if the
+		// underlying repo supports it.
+		findCtx = version.NewContextWithMinVersion(ctx, event.Version()-1)
+		if h.useWait {
+			var cancel func()
+			findCtx, cancel = version.NewContextWithMinVersionWait(ctx, event.Version()-1)
+			defer cancel()
+		}
 	}
+
+	// Get or create the model.
 	entity, err := h.repo.Find(findCtx, event.AggregateID())
 	if rrErr, ok := err.(eh.RepoError); ok && rrErr.Err == eh.ErrEntityNotFound {
 		if h.factoryFn == nil {
@@ -149,12 +165,24 @@ func (h *EventHandler) HandleEvent(ctx context.Context, event eh.Event) error {
 			return nil
 		}
 
-		if entity.AggregateVersion()+1 != event.Version() {
-			return Error{
-				Err:           eh.ErrIncorrectEntityVersion,
-				Projector:     h.projector.ProjectorType().String(),
-				EventVersion:  event.Version(),
-				EntityVersion: entityVersion,
+		// Irregular versioning has looser checks on the version.
+		if h.useIrregularVersioning {
+			if entity.AggregateVersion() >= event.Version() {
+				return Error{
+					Err:           eh.ErrIncorrectEntityVersion,
+					Projector:     h.projector.ProjectorType().String(),
+					EventVersion:  event.Version(),
+					EntityVersion: entityVersion,
+				}
+			}
+		} else {
+			if entity.AggregateVersion()+1 != event.Version() {
+				return Error{
+					Err:           eh.ErrIncorrectEntityVersion,
+					Projector:     h.projector.ProjectorType().String(),
+					EventVersion:  event.Version(),
+					EntityVersion: entityVersion,
+				}
 			}
 		}
 	}


Test patch:
diff --git a/eventhandler/projector/eventhandler_test.go b/eventhandler/projector/eventhandler_test.go
index 814c73b..196d120 100644
--- a/eventhandler/projector/eventhandler_test.go
+++ b/eventhandler/projector/eventhandler_test.go
@@ -147,6 +147,55 @@ func TestEventHandler_UpdateModelWithVersion(t *testing.T) {
 	if err := handler.HandleEvent(ctx, event); err != nil {
 		t.Error("there should be no error:", err)
 	}
+
+	// Handling a future event with a gap in versions should produce an error.
+	futureEvent := eh.NewEvent(mocks.EventType, eventData, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 8))
+	expectedErr := Error{
+		Err:           eh.ErrIncorrectEntityVersion,
+		Projector:     TestProjectorType.String(),
+		EventVersion:  8,
+		EntityVersion: 1,
+	}
+	err := handler.HandleEvent(ctx, futureEvent)
+	if !errors.Is(err, expectedErr) {
+		t.Error("there should be an error:", err)
+	}
+
+	// Handling a "bad projector" which sets the version incorrectly.
+	nextEvent := eh.NewEvent(mocks.EventType, eventData, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 2))
+	projector.newEntity = &mocks.Model{
+		ID:      id,
+		Version: 3,
+		Content: "version 1",
+	}
+	expectedErr = Error{
+		Err:           eh.ErrIncorrectEntityVersion,
+		Projector:     TestProjectorType.String(),
+		EventVersion:  2,
+		EntityVersion: 3,
+	}
+	err = handler.HandleEvent(ctx, nextEvent)
+	if !errors.Is(err, expectedErr) {
+		t.Error("there should be an error:", err)
+	}
+
+	// Handling a future event with a gap in versions should not produce
+	// an error when the irregular versioning option is set.
+	handler = NewEventHandler(projector, version.NewRepo(repo), WithIrregularVersioning())
+	handler.SetEntityFactory(func() eh.Entity {
+		return &mocks.Model{}
+	})
+	// The projector is allowed to bump the version from 3 straight to 8.
+	projector.newEntity = &mocks.Model{
+		ID:      id,
+		Version: 8,
+		Content: "version 1",
+	}
+	if err := handler.HandleEvent(ctx, futureEvent); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func TestEventHandler_UpdateModelWithEventsOutOfOrder(t *testing.T) {