Repository: looplab/eventhorizon
Pull request #324 (2021-06-28 12:31:25)

Feature / Add option for entity lookup on projections
### Description

Add option for entity lookup on projections, useful for projections where the aggregate ID is not used as the entity ID, or where static entity IDs are used (singletons etc).

### Affected Components

- Projector event handler

### Related Issues

### Solution and Design

Adds an optional lookup function, a default function is provided that uses the aggregate ID as entity ID (same behaviour as before).

### Steps to test and verify



Patch:
diff --git a/eventhandler/projector/eventhandler.go b/eventhandler/projector/eventhandler.go
index 54b0803..ee2b513 100644
--- a/eventhandler/projector/eventhandler.go
+++ b/eventhandler/projector/eventhandler.go
@@ -21,6 +21,7 @@ import (
 
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/repo/version"
+	"github.com/looplab/eventhorizon/uuid"
 )
 
 // Projector is a projector of events onto models.
@@ -79,6 +80,7 @@ type EventHandler struct {
 	factoryFn              func() eh.Entity
 	useWait                bool
 	useIrregularVersioning bool
+	entityLookupFn         func(eh.Event) uuid.UUID
 }
 
 var _ = eh.EventHandler(&EventHandler{})
@@ -86,8 +88,9 @@ var _ = eh.EventHandler(&EventHandler{})
 // NewEventHandler creates a new EventHandler.
 func NewEventHandler(projector Projector, repo eh.ReadWriteRepo, options ...Option) *EventHandler {
 	h := &EventHandler{
-		projector: projector,
-		repo:      repo,
+		projector:      projector,
+		repo:           repo,
+		entityLookupFn: defaultEntityLookupFn,
 	}
 	for _, option := range options {
 		option(h)
@@ -114,6 +117,20 @@ func WithIrregularVersioning() Option {
 	}
 }
 
+// WithEntityLookup can be used to provide an alternative ID (from the aggregate ID)
+// for fetching the projected entity. The lookup func can for example extract
+// another field from the event or use a static ID for some singleton-like projections.
+func WithEntityLookup(f func(eh.Event) uuid.UUID) Option {
+	return func(h *EventHandler) {
+		h.entityLookupFn = f
+	}
+}
+
+// defaultEntitypLookupFn does a lookup by the aggregate ID of the event.
+func defaultEntityLookupFn(event eh.Event) uuid.UUID {
+	return event.AggregateID()
+}
+
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (h *EventHandler) HandlerType() eh.EventHandlerType {
 	return eh.EventHandlerType("projector_" + h.projector.ProjectorType())
@@ -137,7 +154,7 @@ func (h *EventHandler) HandleEvent(ctx context.Context, event eh.Event) error {
 	}
 
 	// Get or create the model.
-	entity, err := h.repo.Find(findCtx, event.AggregateID())
+	entity, err := h.repo.Find(findCtx, h.entityLookupFn(event))
 	if rrErr, ok := err.(eh.RepoError); ok && rrErr.Err == eh.ErrEntityNotFound {
 		if h.factoryFn == nil {
 			return Error{
@@ -213,6 +230,14 @@ func (h *EventHandler) HandleEvent(ctx context.Context, event eh.Event) error {
 
 	// Update or remove the model.
 	if newEntity != nil {
+		if newEntity.EntityID() != h.entityLookupFn(event) {
+			return Error{
+				Err:           eh.ErrMissingEntityID,
+				Projector:     h.projector.ProjectorType().String(),
+				EventVersion:  event.Version(),
+				EntityVersion: entityVersion,
+			}
+		}
 		if err := h.repo.Save(ctx, newEntity); err != nil {
 			return Error{
 				Err:           err,
@@ -222,7 +247,7 @@ func (h *EventHandler) HandleEvent(ctx context.Context, event eh.Event) error {
 			}
 		}
 	} else {
-		if err := h.repo.Remove(ctx, event.AggregateID()); err != nil {
+		if err := h.repo.Remove(ctx, h.entityLookupFn(event)); err != nil {
 			return Error{
 				Err:           err,
 				Projector:     h.projector.ProjectorType().String(),


Test patch:
diff --git a/eventhandler/projector/eventhandler_test.go b/eventhandler/projector/eventhandler_test.go
index 196d120..1d14981 100644
--- a/eventhandler/projector/eventhandler_test.go
+++ b/eventhandler/projector/eventhandler_test.go
@@ -372,6 +372,49 @@ func TestEventHandler_ProjectError(t *testing.T) {
 	}
 }
 
+func TestEventHandler_EntityLookup(t *testing.T) {
+	repo := &mocks.Repo{}
+	projector := &TestProjector{}
+	handler := NewEventHandler(projector, repo,
+		WithEntityLookup(func(event eh.Event) uuid.UUID {
+			eventData := event.Data().(*mocks.EventData)
+			return uuid.MustParse(eventData.Content)
+		}),
+	)
+	handler.SetEntityFactory(func() eh.Entity {
+		return &mocks.SimpleModel{}
+	})
+
+	ctx := context.Background()
+
+	aggregateID := uuid.New()
+	entityID := uuid.New()
+	eventData := &mocks.EventData{Content: entityID.String()}
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+	event := eh.NewEvent(mocks.EventType, eventData, timestamp,
+		eh.ForAggregate(mocks.AggregateType, aggregateID, 1))
+	entity := &mocks.SimpleModel{
+		ID: entityID,
+	}
+	repo.Entity = entity
+	projector.newEntity = &mocks.SimpleModel{
+		ID:      entityID,
+		Content: "updated",
+	}
+	if err := handler.HandleEvent(ctx, event); err != nil {
+		t.Error("there should be no error:", err)
+	}
+	if projector.event != event {
+		t.Error("the handled event should be correct:", projector.event)
+	}
+	if projector.entity != entity {
+		t.Error("the entity should be correct:", projector.entity)
+	}
+	if repo.Entity != projector.newEntity {
+		t.Error("the new entity should be correct:", repo.Entity)
+	}
+}
+
 const (
 	TestProjectorType Type = "TestProjector"
 )