fix: Add back errors for event store validation

looplab · Oct 25, 2021 · 7af84f8 · 7af84f8
1 parent d22a0e0
commit 7af84f8
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 12 deletions.
diff --git a/eventstore.go b/eventstore.go
@@ -16,6 +16,7 @@ package eventhorizon
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 
@@ -34,6 +35,17 @@ type EventStore interface {
 	Close() error
 }
 
+var (
+	// Events in the same save operation is for different aggregate IDs.
+	ErrMismatchedEventAggregateIDs = errors.New("mismatched event aggregate IDs")
+	// Events in the same save operation is for different aggregate types.
+	ErrMismatchedEventAggregateTypes = errors.New("mismatched event aggregate types")
+	// Events in the same save operation have non-serial versions.
+	ErrNonSerialEventVersions = errors.New("non-serial event versions")
+	// Other events has been saved for this aggregate since the operation started.
+	ErrEventConflictFromOtherSave = errors.New("event conflict from other save")
+)
+
 // EventStoreOperation is the operation done when an error happened.
 type EventStoreOperation string
 

diff --git a/eventstore/memory/eventstore.go b/eventstore/memory/eventstore.go
@@ -105,7 +105,7 @@ func (s *EventStore) save(ctx context.Context, events []eh.Event, originalVersio
 		// Only accept events belonging to the same aggregate.
 		if event.AggregateID() != id {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("event has different aggregate ID"),
+				Err:              eh.ErrMismatchedEventAggregateIDs,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -116,7 +116,7 @@ func (s *EventStore) save(ctx context.Context, events []eh.Event, originalVersio
 
 		if event.AggregateType() != at {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("event has different aggregate type"),
+				Err:              eh.ErrMismatchedEventAggregateTypes,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -128,7 +128,7 @@ func (s *EventStore) save(ctx context.Context, events []eh.Event, originalVersio
 		// Only accept events that apply to the correct aggregate version.
 		if event.Version() != originalVersion+i+1 {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("invalid event version"),
+				Err:              eh.ErrNonSerialEventVersions,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -169,7 +169,7 @@ func (s *EventStore) save(ctx context.Context, events []eh.Event, originalVersio
 		if aggregate, ok := s.db[id]; ok {
 			if aggregate.Version != originalVersion {
 				return &eh.EventStoreError{
-					Err:              fmt.Errorf("invalid original aggregate version, new version: %d", aggregate.Version),
+					Err:              eh.ErrEventConflictFromOtherSave,
 					Op:               eh.EventStoreOpSave,
 					AggregateType:    at,
 					AggregateID:      id,

diff --git a/eventstore/mongodb/eventstore.go b/eventstore/mongodb/eventstore.go
@@ -113,7 +113,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		// Only accept events belonging to the same aggregate.
 		if event.AggregateID() != id {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("event has different aggregate ID"),
+				Err:              eh.ErrMismatchedEventAggregateIDs,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -124,7 +124,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 
 		if event.AggregateType() != at {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("event has different aggregate type"),
+				Err:              eh.ErrMismatchedEventAggregateTypes,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -136,7 +136,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		// Only accept events that apply to the correct aggregate version.
 		if event.Version() != originalVersion+i+1 {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("invalid event version"),
+				Err:              eh.ErrNonSerialEventVersions,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -203,7 +203,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 			}
 		} else if r.MatchedCount == 0 {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("invalid original aggregate version, new version %d", originalVersion),
+				Err:              eh.ErrEventConflictFromOtherSave,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,

diff --git a/eventstore/mongodb_v2/eventstore.go b/eventstore/mongodb_v2/eventstore.go
@@ -138,7 +138,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		// Only accept events belonging to the same aggregate.
 		if event.AggregateID() != id {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("event has different aggregate ID"),
+				Err:              eh.ErrMismatchedEventAggregateIDs,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -149,7 +149,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 
 		if event.AggregateType() != at {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("event has different aggregate type"),
+				Err:              eh.ErrMismatchedEventAggregateTypes,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -161,7 +161,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		// Only accept events that apply to the correct aggregate version.
 		if event.Version() != originalVersion+i+1 {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("invalid event version"),
+				Err:              eh.ErrNonSerialEventVersions,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -285,7 +285,7 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 			); err != nil {
 				return nil, fmt.Errorf("could not update stream: %w", err)
 			} else if r.MatchedCount == 0 {
-				return nil, fmt.Errorf("invalid original aggregate version, new version %d", originalVersion)
+				return nil, eh.ErrEventConflictFromOtherSave
 			}
 		}