feat: uniqueness (#1190)

* feat: uniqueness on events

* fix: some tests

* fix: add unique username

* fix: nice error message

* fix: add unique test

* fix: add unique test

* fix: add unique constraint to events

* fix: correct unique constraint on user events

* fix: remove user constraint

* fix: add unique constraints

* fix: add unique constraints

* fix: add unique constraints

* fix: unnique constraints without interface

* fix: unique idp config

* fix: unique constraint comments

* fix: unique constraints in one table

* fix: unique constraints error

* fix: fix unique constraint on create user

* fix: fix unique constraint on create project

* fix: fix unique constraint on idp configs

internal/eventstore/v2/repository/sql/crdb.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"github.com/lib/pq"
 	"regexp"
 	"strconv"
 
@@ -88,6 +89,18 @@ const (
 		"		FROM data " +
 		"	) " +
 		"RETURNING id, event_sequence, previous_sequence, creation_date, resource_owner"
+	uniqueInsert = `INSERT INTO eventstore.unique_constraints
+					(
+						unique_type,
+						unique_field
+					) 
+					VALUES (  
+						$1,
+						$2
+					)`
+
+	uniqueDelete = `DELETE FROM eventstore.unique_constraints
+					WHERE unique_type = $1 and unique_field = $2`
 )
 
 type CRDB struct {
@@ -102,7 +115,7 @@ func (db *CRDB) Health(ctx context.Context) error { return db.client.Ping() }
 
 // Push adds all events to the eventstreams of the aggregates.
 // This call is transaction save. The transaction will be rolled back if one event fails
-func (db *CRDB) Push(ctx context.Context, events ...*repository.Event) error {
+func (db *CRDB) Push(ctx context.Context, events []*repository.Event, uniqueConstraints ...*repository.UniqueConstraint) error {
 	err := crdb.ExecuteTx(ctx, db.client, nil, func(tx *sql.Tx) error {
 		stmt, err := tx.PrepareContext(ctx, crdbInsert)
 		if err != nil {
@@ -136,6 +149,10 @@ func (db *CRDB) Push(ctx context.Context, events ...*repository.Event) error {
 			}
 		}
 
+		err = db.handleUniqueConstraints(ctx, tx, uniqueConstraints...)
+		if err != nil {
+			return err
+		}
 		return nil
 	})
 	if err != nil && !errors.Is(err, &caos_errs.CaosError{}) {
@@ -145,6 +162,39 @@ func (db *CRDB) Push(ctx context.Context, events ...*repository.Event) error {
 	return err
 }
 
+// handleUniqueConstraints adds or removes unique constraints
+func (db *CRDB) handleUniqueConstraints(ctx context.Context, tx *sql.Tx, uniqueConstraints ...*repository.UniqueConstraint) (err error) {
+	if uniqueConstraints == nil || len(uniqueConstraints) == 0 || (len(uniqueConstraints) == 1 && uniqueConstraints[0] == nil) {
+		return nil
+	}
+
+	for _, uniqueConstraint := range uniqueConstraints {
+		if uniqueConstraint.Action == repository.UniqueConstraintAdd {
+			_, err := tx.ExecContext(ctx, uniqueInsert, uniqueConstraint.UniqueType, uniqueConstraint.UniqueField)
+			if err != nil {
+				logging.LogWithFields("SQL-IP3js",
+					"unique_type", uniqueConstraint.UniqueType,
+					"unique_field", uniqueConstraint.UniqueField).WithError(err).Info("insert unique constraint failed")
+
+				if db.isUniqueViolationError(err) {
+					return caos_errs.ThrowAlreadyExists(err, "SQL-M0dsf", uniqueConstraint.ErrorMessage)
+				}
+
+				return caos_errs.ThrowInternal(err, "SQL-dM9ds", "unable to create unique constraint ")
+			}
+		} else if uniqueConstraint.Action == repository.UniqueConstraintRemoved {
+			_, err := tx.ExecContext(ctx, uniqueDelete, uniqueConstraint.UniqueType, uniqueConstraint.UniqueField)
+			if err != nil {
+				logging.LogWithFields("SQL-M0vsf",
+					"unique_type", uniqueConstraint.UniqueType,
+					"unique_field", uniqueConstraint.UniqueField).WithError(err).Info("delete unique constraint failed")
+				return caos_errs.ThrowInternal(err, "SQL-2M9fs", "unable to remove unique constraint ")
+			}
+		}
+	}
+	return nil
+}
+
 // Filter returns all events matching the given search query
 func (db *CRDB) Filter(ctx context.Context, searchQuery *repository.SearchQuery) (events []*repository.Event, err error) {
 	events = []*repository.Event{}
@@ -262,3 +312,12 @@ func (db *CRDB) placeholder(query string) string {
 	}
 	return replaced
 }
+
+func (db *CRDB) isUniqueViolationError(err error) bool {
+	if pqErr, ok := err.(*pq.Error); ok {
+		if pqErr.Code == "23505" {
+			return true
+		}
+	}
+	return false
+}

internal/eventstore/v2/repository/sql/crdb_test.go

@@ -266,11 +266,15 @@ func TestCRDB_columnName(t *testing.T) {
 
 func TestCRDB_Push_OneAggregate(t *testing.T) {
 	type args struct {
-		ctx    context.Context
-		events []*repository.Event
+		ctx               context.Context
+		events            []*repository.Event
+		uniqueConstraints *repository.UniqueConstraint
+		uniqueDataType    string
+		uniqueDataField   string
 	}
 	type eventsRes struct {
 		pushedEventsCount int
+		uniqueCount       int
 		aggType           repository.AggregateType
 		aggID             []string
 	}
@@ -334,26 +338,84 @@ func TestCRDB_Push_OneAggregate(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "push 1 event and add unique constraint",
+			args: args{
+				ctx: context.Background(),
+				events: []*repository.Event{
+					generateEvent(t, "10"),
+				},
+				uniqueConstraints: generateAddUniqueConstraint(t, "usernames", "field"),
+			},
+			res: res{
+				wantErr: false,
+				eventsRes: eventsRes{
+					pushedEventsCount: 1,
+					uniqueCount:       1,
+					aggID:             []string{"10"},
+					aggType:           repository.AggregateType(t.Name()),
+				}},
+		},
+		{
+			name: "push 1 event and remove unique constraint",
+			args: args{
+				ctx: context.Background(),
+				events: []*repository.Event{
+					generateEvent(t, "11"),
+				},
+				uniqueConstraints: generateRemoveUniqueConstraint(t, "usernames", "testremove"),
+				uniqueDataType:    "usernames",
+				uniqueDataField:   "testremove",
+			},
+			res: res{
+				wantErr: false,
+				eventsRes: eventsRes{
+					pushedEventsCount: 1,
+					uniqueCount:       0,
+					aggID:             []string{"11"},
+					aggType:           repository.AggregateType(t.Name()),
+				}},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			db := &CRDB{
 				client: testCRDBClient,
 			}
-			if err := db.Push(tt.args.ctx, tt.args.events...); (err != nil) != tt.res.wantErr {
+			if tt.args.uniqueDataType != "" && tt.args.uniqueDataField != "" {
+				err := fillUniqueData(tt.args.uniqueDataType, tt.args.uniqueDataField)
+				if err != nil {
+					t.Error("unable to prefill insert unique data: ", err)
+					return
+				}
+			}
+			if err := db.Push(tt.args.ctx, tt.args.events, tt.args.uniqueConstraints); (err != nil) != tt.res.wantErr {
 				t.Errorf("CRDB.Push() error = %v, wantErr %v", err, tt.res.wantErr)
 			}
 
-			countRow := testCRDBClient.QueryRow("SELECT COUNT(*) FROM eventstore.events where aggregate_type = $1 AND aggregate_id = ANY($2)", tt.res.eventsRes.aggType, pq.Array(tt.res.eventsRes.aggID))
-			var count int
-			err := countRow.Scan(&count)
+			countEventRow := testCRDBClient.QueryRow("SELECT COUNT(*) FROM eventstore.events where aggregate_type = $1 AND aggregate_id = ANY($2)", tt.res.eventsRes.aggType, pq.Array(tt.res.eventsRes.aggID))
+			var eventCount int
+			err := countEventRow.Scan(&eventCount)
 			if err != nil {
 				t.Error("unable to query inserted rows: ", err)
 				return
 			}
-			if count != tt.res.eventsRes.pushedEventsCount {
-				t.Errorf("expected push count %d got %d", tt.res.eventsRes.pushedEventsCount, count)
+			if eventCount != tt.res.eventsRes.pushedEventsCount {
+				t.Errorf("expected push count %d got %d", tt.res.eventsRes.pushedEventsCount, eventCount)
 			}
+			if tt.args.uniqueConstraints != nil {
+				countUniqueRow := testCRDBClient.QueryRow("SELECT COUNT(*) FROM eventstore.unique_constraints where unique_type = $1 AND unique_field = $2", tt.args.uniqueConstraints.UniqueType, tt.args.uniqueConstraints.UniqueField)
+				var uniqueCount int
+				err := countUniqueRow.Scan(&uniqueCount)
+				if err != nil {
+					t.Error("unable to query inserted rows: ", err)
+					return
+				}
+				if uniqueCount != tt.res.eventsRes.uniqueCount {
+					t.Errorf("expected unique count %d got %d", tt.res.eventsRes.uniqueCount, uniqueCount)
+				}
+			}
+
 		})
 	}
 }
@@ -445,7 +507,7 @@ func TestCRDB_Push_MultipleAggregate(t *testing.T) {
 			db := &CRDB{
 				client: testCRDBClient,
 			}
-			if err := db.Push(context.Background(), tt.args.events...); (err != nil) != tt.res.wantErr {
+			if err := db.Push(context.Background(), tt.args.events); (err != nil) != tt.res.wantErr {
 				t.Errorf("CRDB.Push() error = %v, wantErr %v", err, tt.res.wantErr)
 			}
 
@@ -616,7 +678,7 @@ func TestCRDB_Push_Parallel(t *testing.T) {
 			for _, events := range tt.args.events {
 				wg.Add(1)
 				go func(events []*repository.Event) {
-					err := db.Push(context.Background(), events...)
+					err := db.Push(context.Background(), events)
 					if err != nil {
 						errsMu.Lock()
 						errs = append(errs, err)
@@ -728,7 +790,7 @@ func TestCRDB_Filter(t *testing.T) {
 			}
 
 			// setup initial data for query
-			if err := db.Push(context.Background(), tt.fields.existingEvents...); err != nil {
+			if err := db.Push(context.Background(), tt.fields.existingEvents); err != nil {
 				t.Errorf("error in setup = %v", err)
 				return
 			}
@@ -814,7 +876,7 @@ func TestCRDB_LatestSequence(t *testing.T) {
 			}
 
 			// setup initial data for query
-			if err := db.Push(context.Background(), tt.fields.existingEvents...); err != nil {
+			if err := db.Push(context.Background(), tt.fields.existingEvents); err != nil {
 				t.Errorf("error in setup = %v", err)
 				return
 			}
@@ -956,7 +1018,7 @@ func TestCRDB_Push_ResourceOwner(t *testing.T) {
 			db := &CRDB{
 				client: testCRDBClient,
 			}
-			if err := db.Push(context.Background(), tt.args.events...); err != nil {
+			if err := db.Push(context.Background(), tt.args.events); err != nil {
 				t.Errorf("CRDB.Push() error = %v", err)
 			}
 
@@ -1036,3 +1098,25 @@ func generateEventWithData(t *testing.T, aggregateID string, data []byte) *repos
 		Data:          data,
 	}
 }
+
+func generateAddUniqueConstraint(t *testing.T, table, uniqueField string) *repository.UniqueConstraint {
+	t.Helper()
+	e := &repository.UniqueConstraint{
+		UniqueType:  table,
+		UniqueField: uniqueField,
+		Action:      repository.UniqueConstraintAdd,
+	}
+
+	return e
+}
+
+func generateRemoveUniqueConstraint(t *testing.T, table, uniqueField string) *repository.UniqueConstraint {
+	t.Helper()
+	e := &repository.UniqueConstraint{
+		UniqueType:  table,
+		UniqueField: uniqueField,
+		Action:      repository.UniqueConstraintRemoved,
+	}
+
+	return e
+}

internal/eventstore/v2/repository/sql/local_crdb_test.go

@@ -80,6 +80,11 @@ func executeMigrations() error {
 	return nil
 }
 
+func fillUniqueData(unique_type, field string) error {
+	_, err := testCRDBClient.Exec("INSERT INTO eventstore.unique_constraints (unique_type, unique_field) VALUES ($1, $2)", unique_type, field)
+	return err
+}
+
 type migrationPaths []string
 
 type version struct {

internal/eventstore/v2/repository/sql/query_test.go

@@ -521,7 +521,7 @@ func Test_query_events_with_crdb(t *testing.T) {
 			}
 
 			// setup initial data for query
-			if err := db.Push(context.Background(), tt.fields.existingEvents...); err != nil {
+			if err := db.Push(context.Background(), tt.fields.existingEvents); err != nil {
 				t.Errorf("error in setup = %v", err)
 				return
 			}