feat(eventstore): Precondition (#69)

* start org

* refactor(eventstore): filter in sql for querier

* feat(eventstore): Aggregate precondition

preconditions are checked right before insert. Insert is still transaction save

* feat(eventstore): check preconditions in repository

* test(eventstore): test precondition in models

* test(eventstore): precondition-tests

* refactor(eventstore): querier as type

* fix(precondition): rename validation from precondition to validation

* test(eventstore): isErr func instead of wantErr bool

* fix: delete org files

* remove comment

Co-authored-by: Livio Amstutz <[email protected]>

internal/eventstore/internal/repository/sql/filter.go

@@ -2,6 +2,7 @@ package sql
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"strconv"
 	"strings"
@@ -30,23 +31,18 @@ const (
 		" FROM eventstore.events"
 )
 
+type Querier interface {
+	Query(query string, args ...interface{}) (*sql.Rows, error)
+}
+
 func (db *SQL) Filter(ctx context.Context, searchQuery *es_models.SearchQuery) (events []*models.Event, err error) {
-	where, values := prepareWhere(searchQuery)
-	query := selectStmt + where
+	return filter(db.client, searchQuery)
+}
 
-	query += " ORDER BY event_sequence"
-	if searchQuery.Desc {
-		query += " DESC"
-	}
+func filter(querier Querier, searchQuery *es_models.SearchQuery) (events []*es_models.Event, err error) {
+	query, values := prepareQuery(searchQuery)
 
-	if searchQuery.Limit > 0 {
-		values = append(values, searchQuery.Limit)
-		query += " LIMIT ?"
-	}
-
-	query = numberPlaceholder(query, "?", "$")
-
-	rows, err := db.client.Query(query, values...)
+	rows, err := querier.Query(query, values...)
 	if err != nil {
 		logging.Log("SQL-HP3Uk").WithError(err).Info("query failed")
 		return nil, errors.ThrowInternal(err, "SQL-IJuyR", "unable to filter events")
@@ -86,6 +82,25 @@ func (db *SQL) Filter(ctx context.Context, searchQuery *es_models.SearchQuery) (
 	return events, nil
 }
 
+func prepareQuery(searchQuery *es_models.SearchQuery) (query string, values []interface{}) {
+	where, values := prepareWhere(searchQuery)
+	query = selectStmt + where
+
+	query += " ORDER BY event_sequence"
+	if searchQuery.Desc {
+		query += " DESC"
+	}
+
+	if searchQuery.Limit > 0 {
+		values = append(values, searchQuery.Limit)
+		query += " LIMIT ?"
+	}
+
+	query = numberPlaceholder(query, "?", "$")
+
+	return query, values
+}
+
 func numberPlaceholder(query, old, new string) string {
 	for i, hasChanged := 1, true; hasChanged; i++ {
 		newQuery := strings.Replace(query, old, new+strconv.Itoa(i), 1)

internal/eventstore/internal/repository/sql/push.go

@@ -3,9 +3,10 @@ package sql
 import (
 	"context"
 	"database/sql"
+	"errors"
 
 	"github.com/caos/logging"
-	"github.com/caos/zitadel/internal/errors"
+	caos_errs "github.com/caos/zitadel/internal/errors"
 	"github.com/caos/zitadel/internal/eventstore/models"
 	"github.com/cockroachdb/cockroach-go/crdb"
 )
@@ -31,11 +32,16 @@ func (db *SQL) PushAggregates(ctx context.Context, aggregates ...*models.Aggrega
 		if err != nil {
 			tx.Rollback()
 			logging.Log("SQL-9ctx5").WithError(err).Warn("prepare failed")
-			return errors.ThrowInternal(err, "SQL-juCgA", "prepare failed")
+			return caos_errs.ThrowInternal(err, "SQL-juCgA", "prepare failed")
 		}
 
 		for _, aggregate := range aggregates {
-			err = insertEvents(stmt, aggregate.Events)
+			err = precondtion(tx, aggregate)
+			if err != nil {
+				tx.Rollback()
+				return err
+			}
+			err = insertEvents(stmt, Sequence(aggregate.PreviousSequence), aggregate.Events)
 			if err != nil {
 				tx.Rollback()
 				return err
@@ -44,15 +50,29 @@ func (db *SQL) PushAggregates(ctx context.Context, aggregates ...*models.Aggrega
 		return nil
 	})
 
-	if _, ok := err.(*errors.CaosError); !ok && err != nil {
-		err = errors.ThrowInternal(err, "SQL-DjgtG", "unable to store events")
+	if err != nil && !errors.Is(err, &caos_errs.CaosError{}) {
+		err = caos_errs.ThrowInternal(err, "SQL-DjgtG", "unable to store events")
 	}
 
 	return err
 }
 
-func insertEvents(stmt *sql.Stmt, events []*models.Event) error {
-	currentSequence := Sequence(events[0].PreviousSequence)
+func precondtion(tx *sql.Tx, aggregate *models.Aggregate) error {
+	if aggregate.Precondition == nil {
+		return nil
+	}
+	events, err := filter(tx, aggregate.Precondition.Query)
+	if err != nil {
+		return caos_errs.ThrowPreconditionFailed(err, "SQL-oBPxB", "filter failed")
+	}
+	err = aggregate.Precondition.Validation(events...)
+	if err != nil {
+		return caos_errs.ThrowPreconditionFailed(err, "SQL-s6hqU", "validation failed")
+	}
+	return nil
+}
+
+func insertEvents(stmt *sql.Stmt, previousSequence Sequence, events []*models.Event) error {
 	for _, event := range events {
 		if event.Data == nil || len(event.Data) == 0 {
 			//json decoder failes with EOF if json text is empty
@@ -62,27 +82,27 @@ func insertEvents(stmt *sql.Stmt, events []*models.Event) error {
 		rows, err := stmt.Query(event.Type, event.AggregateType, event.AggregateID, event.AggregateVersion, event.CreationDate, event.Data, event.EditorUser, event.EditorService, event.ResourceOwner,
 			event.AggregateType, event.AggregateID,
 			event.AggregateType, event.AggregateID,
-			currentSequence, event.AggregateType, event.AggregateID,
-			event.AggregateType, event.AggregateID, currentSequence)
+			previousSequence, event.AggregateType, event.AggregateID,
+			event.AggregateType, event.AggregateID, previousSequence)
 
 		if err != nil {
 			logging.Log("SQL-EXA0q").WithError(err).Info("query failed")
-			return errors.ThrowInternal(err, "SQL-SBP37", "unable to create event")
+			return caos_errs.ThrowInternal(err, "SQL-SBP37", "unable to create event")
 		}
 		defer rows.Close()
 
 		rowInserted := false
 		for rows.Next() {
 			rowInserted = true
-			err = rows.Scan(&event.ID, &currentSequence, &event.CreationDate)
+			err = rows.Scan(&event.ID, &previousSequence, &event.CreationDate)
 			logging.Log("SQL-rAvLD").OnError(err).Info("unable to scan result into event")
 		}
 
 		if !rowInserted {
-			return errors.ThrowAlreadyExists(nil, "SQL-GKcAa", "wrong sequence")
+			return caos_errs.ThrowAlreadyExists(nil, "SQL-GKcAa", "wrong sequence")
 		}
 
-		event.Sequence = uint64(currentSequence)
+		event.Sequence = uint64(previousSequence)
 	}
 
 	return nil

internal/eventstore/internal/repository/sql/push_test.go

@@ -59,6 +59,21 @@ func TestSQL_PushAggregates(t *testing.T) {
 			isError:           errors.IsInternal,
 			shouldCheckEvents: false,
 		},
+		{
+			name: "aggregate precondtion fails",
+			fields: fields{
+				client: mockDB(t).
+					expectBegin(nil).
+					expectSavepoint().
+					expectPrepareInsert().
+					expectFilterEventsError(errors.CreateCaosError(nil, "SQL-IzJOf", "err")).
+					expectRollback(nil),
+			},
+
+			args:              args{aggregates: []*models.Aggregate{aggregateWithPrecondition(&models.Aggregate{}, models.NewSearchQuery().SetLimit(1), nil)}},
+			isError:           errors.IsPreconditionFailed,
+			shouldCheckEvents: false,
+		},
 		{
 			name: "one aggregate two events success",
 			fields: fields{
@@ -94,9 +109,10 @@ func TestSQL_PushAggregates(t *testing.T) {
 			},
 			args: args{
 				aggregates: []*models.Aggregate{
-					&models.Aggregate{
+					{
+						PreviousSequence: 34,
 						Events: []*models.Event{
-							&models.Event{
+							{
 								AggregateID:      "aggID",
 								AggregateType:    "aggType",
 								AggregateVersion: "v0.0.1",
@@ -104,9 +120,8 @@ func TestSQL_PushAggregates(t *testing.T) {
 								EditorUser:       "usr",
 								ResourceOwner:    "ro",
 								Type:             "eventTyp",
-								PreviousSequence: 34,
 							},
-							&models.Event{
+							{
 								AggregateID:      "aggID",
 								AggregateType:    "aggType",
 								AggregateVersion: "v0.0.1",
@@ -114,7 +129,6 @@ func TestSQL_PushAggregates(t *testing.T) {
 								EditorUser:       "usr2",
 								ResourceOwner:    "ro2",
 								Type:             "eventTyp",
-								PreviousSequence: 0,
 							},
 						},
 					},
@@ -157,9 +171,10 @@ func TestSQL_PushAggregates(t *testing.T) {
 			},
 			args: args{
 				aggregates: []*models.Aggregate{
-					&models.Aggregate{
+					{
+						PreviousSequence: 34,
 						Events: []*models.Event{
-							&models.Event{
+							{
 								AggregateID:      "aggID",
 								AggregateType:    "aggType",
 								AggregateVersion: "v0.0.1",
@@ -167,13 +182,13 @@ func TestSQL_PushAggregates(t *testing.T) {
 								EditorUser:       "usr",
 								ResourceOwner:    "ro",
 								Type:             "eventTyp",
-								PreviousSequence: 34,
 							},
 						},
 					},
-					&models.Aggregate{
+					{
+						PreviousSequence: 40,
 						Events: []*models.Event{
-							&models.Event{
+							{
 								AggregateID:      "aggID2",
 								AggregateType:    "aggType2",
 								AggregateVersion: "v0.0.1",
@@ -181,7 +196,6 @@ func TestSQL_PushAggregates(t *testing.T) {
 								EditorUser:       "usr",
 								ResourceOwner:    "ro",
 								Type:             "eventTyp",
-								PreviousSequence: 40,
 							},
 						},
 					},
@@ -212,9 +226,9 @@ func TestSQL_PushAggregates(t *testing.T) {
 			},
 			args: args{
 				aggregates: []*models.Aggregate{
-					&models.Aggregate{
+					{
 						Events: []*models.Event{
-							&models.Event{
+							{
 								AggregateID:      "aggID",
 								AggregateType:    "aggType",
 								AggregateVersion: "v0.0.1",
@@ -224,7 +238,7 @@ func TestSQL_PushAggregates(t *testing.T) {
 								Type:             "eventTyp",
 								PreviousSequence: 34,
 							},
-							&models.Event{
+							{
 								AggregateID:      "aggID",
 								AggregateType:    "aggType",
 								AggregateVersion: "v0.0.1",
@@ -265,9 +279,9 @@ func TestSQL_PushAggregates(t *testing.T) {
 			},
 			args: args{
 				aggregates: []*models.Aggregate{
-					&models.Aggregate{
+					{
 						Events: []*models.Event{
-							&models.Event{
+							{
 								AggregateID:      "aggID",
 								AggregateType:    "aggType",
 								AggregateVersion: "v0.0.1",
@@ -321,3 +335,95 @@ func noErr(err error) bool {
 func functionName(i interface{}) string {
 	return runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
 }
+
+func Test_precondtion(t *testing.T) {
+	type fields struct {
+		client *dbMock
+	}
+	type args struct {
+		aggregate *models.Aggregate
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		isErr  func(error) bool
+	}{
+		{
+			name: "no precondition",
+			fields: fields{
+				client: mockDB(t).
+					expectBegin(nil),
+			},
+			args: args{
+				aggregate: &models.Aggregate{},
+			},
+		},
+		{
+			name: "precondition fails",
+			fields: fields{
+				client: mockDB(t).
+					expectBegin(nil).expectFilterEventsLimit(5, 0),
+			},
+			args: args{
+				aggregate: aggregateWithPrecondition(&models.Aggregate{}, models.NewSearchQuery().SetLimit(5), validationFunc(errors.CreateCaosError(nil, "SQL-LBIKm", "err"))),
+			},
+			isErr: errors.IsPreconditionFailed,
+		},
+		{
+			name: "precondition with filter error",
+			fields: fields{
+				client: mockDB(t).
+					expectBegin(nil).expectFilterEventsError(errors.ThrowInternal(nil, "SQL-ac9EW", "err")),
+			},
+			args: args{
+				aggregate: aggregateWithPrecondition(&models.Aggregate{}, models.NewSearchQuery().SetLimit(5), validationFunc(errors.CreateCaosError(nil, "SQL-LBIKm", "err"))),
+			},
+			isErr: errors.IsPreconditionFailed,
+		},
+		{
+			name: "precondition no events",
+			fields: fields{
+				client: mockDB(t).
+					expectBegin(nil).expectFilterEventsLimit(5, 0),
+			},
+			args: args{
+				aggregate: aggregateWithPrecondition(&models.Aggregate{}, models.NewSearchQuery().SetLimit(5), validationFunc(nil)),
+			},
+		},
+		{
+			name: "precondition with events",
+			fields: fields{
+				client: mockDB(t).
+					expectBegin(nil).expectFilterEventsLimit(5, 3),
+			},
+			args: args{
+				aggregate: aggregateWithPrecondition(&models.Aggregate{}, models.NewSearchQuery().SetLimit(5), validationFunc(nil)),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tx, err := tt.fields.client.sqlClient.Begin()
+			if err != nil {
+				t.Errorf("unable to start tx %v", err)
+				t.FailNow()
+			}
+			err = precondtion(tx, tt.args.aggregate)
+			if (tt.isErr != nil && err == nil) || (tt.isErr != nil && !tt.isErr(err)) {
+				t.Error("precondtion() wrong error", err)
+			}
+		})
+	}
+}
+
+func aggregateWithPrecondition(aggregate *models.Aggregate, query *models.SearchQuery, precondition func(...*models.Event) error) *models.Aggregate {
+	aggregate.SetPrecondition(query, precondition)
+	return aggregate
+}
+
+func validationFunc(err error) func(events ...*models.Event) error {
+	return func(events ...*models.Event) error {
+		return err
+	}
+}