diff --git a/internal/eventstore/v2/eventstore_test.go b/internal/eventstore/v2/eventstore_test.go index b92084aa0..624fe41c4 100644 --- a/internal/eventstore/v2/eventstore_test.go +++ b/internal/eventstore/v2/eventstore_test.go @@ -8,6 +8,7 @@ import ( "github.com/caos/zitadel/internal/eventstore/v2/repository" ) +// testEvent implements the Event interface type testEvent struct { description string shouldCheckPrevious bool @@ -17,6 +18,23 @@ func (e *testEvent) CheckPrevious() bool { return e.shouldCheckPrevious } +func (e *testEvent) EditorService() string { + return "editorService" +} +func (e *testEvent) EditorUser() string { + return "editorUser" +} +func (e *testEvent) Type() EventType { + return "test.event" +} +func (e *testEvent) Data() interface{} { + return nil +} + +func (e *testEvent) PreviousSequence() uint64 { + return 0 +} + func testPushMapper(Event) (*repository.Event, error) { return &repository.Event{AggregateID: "aggregateID"}, nil } diff --git a/internal/eventstore/v2/repository/aggregate.go b/internal/eventstore/v2/repository/aggregate.go deleted file mode 100644 index edbe89970..000000000 --- a/internal/eventstore/v2/repository/aggregate.go +++ /dev/null @@ -1,22 +0,0 @@ -package repository - -//AggregateType is the object name -type AggregateType string - -// //Aggregate represents an object -// type Aggregate struct { -// //ID id is the unique identifier of the aggregate -// // the client must generate it by it's own -// ID string -// //Type describes the meaning of this aggregate -// // it could an object like user -// Type AggregateType - -// //ResourceOwner is the organisation which owns this aggregate -// // an aggregate can only be managed by one organisation -// // use the ID of the org -// ResourceOwner string - -// //Events describe all the changes made on an aggregate -// Events []*Event -// } diff --git a/internal/eventstore/v2/repository/event.go b/internal/eventstore/v2/repository/event.go index 5f00a5dff..66d6a0f2b 100644 --- a/internal/eventstore/v2/repository/event.go +++ b/internal/eventstore/v2/repository/event.go @@ -66,3 +66,6 @@ type Event struct { //EventType is the description of the change type EventType string + +//AggregateType is the object name +type AggregateType string diff --git a/internal/eventstore/v2/repository/filter.go b/internal/eventstore/v2/repository/filter.go deleted file mode 100644 index 3b033916d..000000000 --- a/internal/eventstore/v2/repository/filter.go +++ /dev/null @@ -1,62 +0,0 @@ -package repository - -import ( - "context" - "database/sql" - - "github.com/caos/logging" - "github.com/caos/zitadel/internal/errors" - es_models "github.com/caos/zitadel/internal/eventstore/models" -) - -type Querier interface { - Query(query string, args ...interface{}) (*sql.Rows, error) -} - -// Filter returns all events matching the given search query -func (db *CRDB) Filter(ctx context.Context, searchQuery *es_models.SearchQueryFactory) (events []*Event, err error) { - return filter(db.db, searchQuery) -} - -func filter(querier Querier, searchQuery *es_models.SearchQueryFactory) (events []*Event, err error) { - query, limit, values, rowScanner := buildQuery(searchQuery) - if query == "" { - return nil, errors.ThrowInvalidArgument(nil, "SQL-rWeBw", "invalid query factory") - } - - 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") - } - defer rows.Close() - - events = make([]*Event, 0, limit) - - for rows.Next() { - event := new(Event) - err := rowScanner(rows.Scan, event) - if err != nil { - return nil, err - } - - events = append(events, event) - } - - return events, nil -} - -// func (db *SQL) LatestSequence(ctx context.Context, queryFactory *es_models.SearchQueryFactory) (uint64, error) { -// query, _, values, rowScanner := buildQuery(queryFactory) -// if query == "" { -// return 0, errors.ThrowInvalidArgument(nil, "SQL-rWeBw", "invalid query factory") -// } -// row := db.client.QueryRow(query, values...) -// sequence := new(Sequence) -// err := rowScanner(row.Scan, sequence) -// if err != nil { -// logging.Log("SQL-WsxTg").WithError(err).Info("query failed") -// return 0, errors.ThrowInternal(err, "SQL-Yczyx", "unable to filter latest sequence") -// } -// return uint64(*sequence), nil -// } diff --git a/internal/eventstore/v2/repository/query.go b/internal/eventstore/v2/repository/query.go deleted file mode 100644 index 0d496d2c9..000000000 --- a/internal/eventstore/v2/repository/query.go +++ /dev/null @@ -1,199 +0,0 @@ -package repository - -import ( - "database/sql" - "errors" - "fmt" - "strconv" - "strings" - - "github.com/caos/logging" - z_errors "github.com/caos/zitadel/internal/errors" - "github.com/caos/zitadel/internal/eventstore/models" - es_models "github.com/caos/zitadel/internal/eventstore/models" - "github.com/lib/pq" -) - -const ( - selectStmt = "SELECT" + - " creation_date" + - ", event_type" + - ", event_sequence" + - ", previous_sequence" + - ", event_data" + - ", editor_service" + - ", editor_user" + - ", resource_owner" + - ", aggregate_type" + - ", aggregate_id" + - ", aggregate_version" + - " FROM eventstore.events" -) - -func buildQuery(queryFactory *models.SearchQueryFactory) (query string, limit uint64, values []interface{}, rowScanner func(s scanner, dest interface{}) error) { - searchQuery, err := queryFactory.Build() - if err != nil { - logging.Log("SQL-cshKu").WithError(err).Warn("search query factory invalid") - return "", 0, nil, nil - } - query, rowScanner = prepareColumns(searchQuery.Columns) - where, values := prepareCondition(searchQuery.Filters) - if where == "" || query == "" { - return "", 0, nil, nil - } - query += where - - if searchQuery.Columns != models.Columns_Max_Sequence { - 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, searchQuery.Limit, values, rowScanner -} - -func prepareCondition(filters []*models.Filter) (clause string, values []interface{}) { - values = make([]interface{}, len(filters)) - clauses := make([]string, len(filters)) - - if len(filters) == 0 { - return clause, values - } - for i, filter := range filters { - value := filter.GetValue() - switch value.(type) { - case []bool, []float64, []int64, []string, []models.AggregateType, []models.EventType, *[]bool, *[]float64, *[]int64, *[]string, *[]models.AggregateType, *[]models.EventType: - value = pq.Array(value) - } - - clauses[i] = getCondition(filter) - if clauses[i] == "" { - return "", nil - } - values[i] = value - } - return " WHERE " + strings.Join(clauses, " AND "), values -} - -type scanner func(dest ...interface{}) error - -func prepareColumns(columns models.Columns) (string, func(s scanner, dest interface{}) error) { - switch columns { - case models.Columns_Max_Sequence: - return "SELECT MAX(event_sequence) FROM eventstore.events", func(row scanner, dest interface{}) (err error) { - sequence, ok := dest.(*Sequence) - if !ok { - return z_errors.ThrowInvalidArgument(nil, "SQL-NBjA9", "type must be sequence") - } - err = row(sequence) - if err == nil || errors.Is(err, sql.ErrNoRows) { - return nil - } - return z_errors.ThrowInternal(err, "SQL-bN5xg", "something went wrong") - } - case models.Columns_Event: - return selectStmt, func(row scanner, dest interface{}) (err error) { - event, ok := dest.(*models.Event) - if !ok { - return z_errors.ThrowInvalidArgument(nil, "SQL-4GP6F", "type must be event") - } - var previousSequence Sequence - data := make(Data, 0) - - err = row( - &event.CreationDate, - &event.Type, - &event.Sequence, - &previousSequence, - &data, - &event.EditorService, - &event.EditorUser, - &event.ResourceOwner, - &event.AggregateType, - &event.AggregateID, - &event.AggregateVersion, - ) - - if err != nil { - logging.Log("SQL-kn1Sw").WithError(err).Warn("unable to scan row") - return z_errors.ThrowInternal(err, "SQL-J0hFS", "unable to scan row") - } - - event.PreviousSequence = uint64(previousSequence) - - event.Data = make([]byte, len(data)) - copy(event.Data, data) - - return nil - } - default: - return "", nil - } -} - -func numberPlaceholder(query, old, new string) string { - for i, hasChanged := 1, true; hasChanged; i++ { - newQuery := strings.Replace(query, old, new+strconv.Itoa(i), 1) - hasChanged = query != newQuery - query = newQuery - } - return query -} - -func getCondition(filter *es_models.Filter) (condition string) { - field := getField(filter.GetField()) - operation := getOperation(filter.GetOperation()) - if field == "" || operation == "" { - return "" - } - format := getConditionFormat(filter.GetOperation()) - - return fmt.Sprintf(format, field, operation) -} - -func getConditionFormat(operation es_models.Operation) string { - if operation == es_models.Operation_In { - return "%s %s ANY(?)" - } - return "%s %s ?" -} - -func getField(field es_models.Field) string { - switch field { - case es_models.Field_AggregateID: - return "aggregate_id" - case es_models.Field_AggregateType: - return "aggregate_type" - case es_models.Field_LatestSequence: - return "event_sequence" - case es_models.Field_ResourceOwner: - return "resource_owner" - case es_models.Field_EditorService: - return "editor_service" - case es_models.Field_EditorUser: - return "editor_user" - case es_models.Field_EventType: - return "event_type" - } - return "" -} - -func getOperation(operation es_models.Operation) string { - switch operation { - case es_models.Operation_Equals, es_models.Operation_In: - return "=" - case es_models.Operation_Greater: - return ">" - case es_models.Operation_Less: - return "<" - } - return "" -} diff --git a/internal/eventstore/v2/repository/repository.go b/internal/eventstore/v2/repository/repository.go index ed3ec67d6..108c6ceba 100644 --- a/internal/eventstore/v2/repository/repository.go +++ b/internal/eventstore/v2/repository/repository.go @@ -6,7 +6,6 @@ import ( type Repository interface { Health(ctx context.Context) error - // PushEvents adds all events of the given aggregates to the eventstreams of the aggregates. // This call is transaction save. The transaction will be rolled back if one event fails Push(ctx context.Context, events ...*Event) error diff --git a/internal/eventstore/v2/repository/crdb.go b/internal/eventstore/v2/repository/sql/crdb.go similarity index 52% rename from internal/eventstore/v2/repository/crdb.go rename to internal/eventstore/v2/repository/sql/crdb.go index f0cb495b1..3a7582578 100644 --- a/internal/eventstore/v2/repository/crdb.go +++ b/internal/eventstore/v2/repository/sql/crdb.go @@ -1,12 +1,15 @@ -package repository +package sql import ( "context" "database/sql" "errors" + "regexp" + "strconv" "github.com/caos/logging" caos_errs "github.com/caos/zitadel/internal/errors" + "github.com/caos/zitadel/internal/eventstore/v2/repository" "github.com/cockroachdb/cockroach-go/v2/crdb" //sql import for cockroach @@ -99,15 +102,15 @@ const ( ) type CRDB struct { - db *sql.DB + client *sql.DB } -func (db *CRDB) Health(ctx context.Context) error { return db.db.Ping() } +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 ...*Event) error { - err := crdb.ExecuteTx(ctx, db.db, nil, func(tx *sql.Tx) error { +func (db *CRDB) Push(ctx context.Context, events ...*repository.Event) error { + err := crdb.ExecuteTx(ctx, db.client, nil, func(tx *sql.Tx) error { stmt, err := tx.PrepareContext(ctx, crdbInsert) if err != nil { tx.Rollback() @@ -154,74 +157,138 @@ func (db *CRDB) Push(ctx context.Context, events ...*Event) error { } // Filter returns all events matching the given search query -// func (db *CRDB) Filter(ctx context.Context, searchQuery *SearchQuery) (events []*Event, err error) { +func (db *CRDB) Filter(ctx context.Context, searchQuery *repository.SearchQuery) (events []*repository.Event, err error) { + rows, rowScanner, err := db.query(searchQuery) + if err != nil { + return nil, err + } + defer rows.Close() -// return events, nil -// } + for rows.Next() { + event := new(repository.Event) + err := rowScanner(rows.Scan, event) + if err != nil { + return nil, err + } -//LatestSequence returns the latests sequence found by the the search query -func (db *CRDB) LatestSequence(ctx context.Context, queryFactory *SearchQuery) (uint64, error) { - return 0, nil + events = append(events, event) + } + + return events, nil } -func (db *CRDB) prepareQuery(columns Columns) (string, func(s scanner, dest interface{}) error) { - switch columns { - case Columns_Max_Sequence: - return "SELECT MAX(event_sequence) FROM eventstore.events", func(scan scanner, dest interface{}) (err error) { - sequence, ok := dest.(*Sequence) - if !ok { - return caos_errs.ThrowInvalidArgument(nil, "SQL-NBjA9", "type must be sequence") - } - err = scan(sequence) - if err == nil || errors.Is(err, sql.ErrNoRows) { - return nil - } - return caos_errs.ThrowInternal(err, "SQL-bN5xg", "something went wrong") - } - case Columns_Event: - return selectStmt, func(row scanner, dest interface{}) (err error) { - event, ok := dest.(*Event) - if !ok { - return caos_errs.ThrowInvalidArgument(nil, "SQL-4GP6F", "type must be event") - } - var previousSequence Sequence - data := make(Data, 0) +//LatestSequence returns the latests sequence found by the the search query +func (db *CRDB) LatestSequence(ctx context.Context, searchQuery *repository.SearchQuery) (uint64, error) { + rows, rowScanner, err := db.query(searchQuery) + if err != nil { + return 0, err + } + defer rows.Close() - err = row( - &event.CreationDate, - &event.Type, - &event.Sequence, - &previousSequence, - &data, - &event.EditorService, - &event.EditorUser, - &event.ResourceOwner, - &event.AggregateType, - &event.AggregateID, - &event.Version, - ) + if !rows.Next() { + return 0, caos_errs.ThrowNotFound(nil, "SQL-cAEzS", "latest sequence not found") + } - if err != nil { - logging.Log("SQL-kn1Sw").WithError(err).Warn("unable to scan row") - return caos_errs.ThrowInternal(err, "SQL-J0hFS", "unable to scan row") - } + var seq Sequence + err = rowScanner(rows.Scan, &seq) + if err != nil { + return 0, err + } - event.PreviousSequence = uint64(previousSequence) + return uint64(seq), nil +} - event.Data = make([]byte, len(data)) - copy(event.Data, data) +func (db *CRDB) query(searchQuery *repository.SearchQuery) (*sql.Rows, rowScan, error) { + query, values, rowScanner := buildQuery(db, searchQuery) + if query == "" { + return nil, nil, caos_errs.ThrowInvalidArgument(nil, "SQL-rWeBw", "invalid query factory") + } - return nil - } + rows, err := db.client.Query(query, values...) + if err != nil { + logging.Log("SQL-HP3Uk").WithError(err).Info("query failed") + return nil, nil, caos_errs.ThrowInternal(err, "SQL-IJuyR", "unable to filter events") + } + return rows, rowScanner, nil +} + +func (db *CRDB) eventQuery() string { + return "SELECT" + + " creation_date" + + ", event_type" + + ", event_sequence" + + ", previous_sequence" + + ", event_data" + + ", editor_service" + + ", editor_user" + + ", resource_owner" + + ", aggregate_type" + + ", aggregate_id" + + ", aggregate_version" + + " FROM eventstore.events" +} +func (db *CRDB) maxSequenceQuery() string { + return "SELECT MAX(event_sequence) FROM eventstore.events" +} + +func (db *CRDB) columnName(col repository.Field) string { + switch col { + case repository.Field_AggregateID: + return "aggregate_id" + case repository.Field_AggregateType: + return "aggregate_type" + case repository.Field_LatestSequence: + return "event_sequence" + case repository.Field_ResourceOwner: + return "resource_owner" + case repository.Field_EditorService: + return "editor_service" + case repository.Field_EditorUser: + return "editor_user" + case repository.Field_EventType: + return "event_type" default: - return "", nil + return "" } } -func (db *CRDB) prepareFilter(filters []*Filter) string { - filter := "" - // for _, f := range filters{ - // f. - // } - return filter +func (db *CRDB) conditionFormat(operation repository.Operation) string { + if operation == repository.Operation_In { + return "%s %s ANY(?)" + } + return "%s %s ?" +} + +func (db *CRDB) operation(operation repository.Operation) string { + switch operation { + case repository.Operation_Equals, repository.Operation_In: + return "=" + case repository.Operation_Greater: + return ">" + case repository.Operation_Less: + return "<" + } + return "" +} + +var ( + placeholder = regexp.MustCompile(`\?`) +) + +//placeholder replaces all "?" with postgres placeholders ($) +func (db *CRDB) placeholder(query string) string { + occurances := placeholder.FindAllStringIndex(query, -1) + if len(occurances) == 0 { + return query + } + replaced := query[:occurances[0][0]] + + for i, l := range occurances { + nextIDX := len(query) + if i < len(occurances)-1 { + nextIDX = occurances[i+1][0] + } + replaced = replaced + "$" + strconv.Itoa(i+1) + query[l[1]:nextIDX] + } + return replaced } diff --git a/internal/eventstore/v2/repository/local_crdb_test.go b/internal/eventstore/v2/repository/sql/local_crdb_test.go similarity index 99% rename from internal/eventstore/v2/repository/local_crdb_test.go rename to internal/eventstore/v2/repository/sql/local_crdb_test.go index 19490e596..33bfd82a3 100644 --- a/internal/eventstore/v2/repository/local_crdb_test.go +++ b/internal/eventstore/v2/repository/sql/local_crdb_test.go @@ -1,4 +1,4 @@ -package repository +package sql import ( "database/sql" diff --git a/internal/eventstore/v2/repository/sql/query.go b/internal/eventstore/v2/repository/sql/query.go new file mode 100644 index 000000000..e362a4adb --- /dev/null +++ b/internal/eventstore/v2/repository/sql/query.go @@ -0,0 +1,142 @@ +package sql + +import ( + "database/sql" + "errors" + "fmt" + "strings" + + "github.com/caos/logging" + z_errors "github.com/caos/zitadel/internal/errors" + "github.com/caos/zitadel/internal/eventstore/v2/repository" + "github.com/lib/pq" +) + +type criteriaer interface { + columnName(repository.Field) string + operation(repository.Operation) string + conditionFormat(repository.Operation) string + placeholder(query string) string + eventQuery() string + maxSequenceQuery() string +} + +type rowScan func(scan, interface{}) error +type scan func(dest ...interface{}) error + +func buildQuery(criteria criteriaer, searchQuery *repository.SearchQuery) (query string, values []interface{}, rowScanner rowScan) { + query, rowScanner = prepareColumns(criteria, searchQuery.Columns) + where, values := prepareCondition(criteria, searchQuery.Filters) + if where == "" || query == "" { + return "", nil, nil + } + query += where + + if searchQuery.Columns != repository.Columns_Max_Sequence { + query += " ORDER BY event_sequence" + if searchQuery.Desc { + query += " DESC" + } + } + + if searchQuery.Limit > 0 { + values = append(values, searchQuery.Limit) + query += " LIMIT ?" + } + + query = criteria.placeholder(query) + + return query, values, rowScanner +} + +func prepareColumns(criteria criteriaer, columns repository.Columns) (string, func(s scan, dest interface{}) error) { + switch columns { + case repository.Columns_Max_Sequence: + return criteria.maxSequenceQuery(), maxSequenceRowScanner + case repository.Columns_Event: + return criteria.eventQuery(), eventRowScanner + default: + return "", nil + } +} + +func maxSequenceRowScanner(row scan, dest interface{}) (err error) { + sequence, ok := dest.(*Sequence) + if !ok { + return z_errors.ThrowInvalidArgument(nil, "SQL-NBjA9", "type must be sequence") + } + err = row(sequence) + if err == nil || errors.Is(err, sql.ErrNoRows) { + return nil + } + return z_errors.ThrowInternal(err, "SQL-bN5xg", "something went wrong") +} + +func eventRowScanner(row scan, dest interface{}) (err error) { + event, ok := dest.(*repository.Event) + if !ok { + return z_errors.ThrowInvalidArgument(nil, "SQL-4GP6F", "type must be event") + } + var previousSequence Sequence + data := make(Data, 0) + + err = row( + &event.CreationDate, + &event.Type, + &event.Sequence, + &previousSequence, + &data, + &event.EditorService, + &event.EditorUser, + &event.ResourceOwner, + &event.AggregateType, + &event.AggregateID, + &event.Version, + ) + + if err != nil { + logging.Log("SQL-kn1Sw").WithError(err).Warn("unable to scan row") + return z_errors.ThrowInternal(err, "SQL-J0hFS", "unable to scan row") + } + + event.PreviousSequence = uint64(previousSequence) + + event.Data = make([]byte, len(data)) + copy(event.Data, data) + + return nil +} + +func prepareCondition(criteria criteriaer, filters []*repository.Filter) (clause string, values []interface{}) { + values = make([]interface{}, len(filters)) + clauses := make([]string, len(filters)) + + if len(filters) == 0 { + return clause, values + } + for i, filter := range filters { + value := filter.Value() + switch value.(type) { + case []bool, []float64, []int64, []string, []repository.AggregateType, []repository.EventType, *[]bool, *[]float64, *[]int64, *[]string, *[]repository.AggregateType, *[]repository.EventType: + value = pq.Array(value) + } + + clauses[i] = getCondition(criteria, filter) + if clauses[i] == "" { + return "", nil + } + values[i] = value + } + return " WHERE " + strings.Join(clauses, " AND "), values +} + +func getCondition(cond criteriaer, filter *repository.Filter) (condition string) { + field := cond.columnName(filter.Field()) + operation := cond.operation(filter.Operation()) + if field == "" || operation == "" { + return "" + } + format := cond.conditionFormat(filter.Operation()) + + return fmt.Sprintf(format, field, operation) +} diff --git a/internal/eventstore/v2/repository/query_test.go b/internal/eventstore/v2/repository/sql/query_test.go similarity index 70% rename from internal/eventstore/v2/repository/query_test.go rename to internal/eventstore/v2/repository/sql/query_test.go index 4537c6088..ba01e66ac 100644 --- a/internal/eventstore/v2/repository/query_test.go +++ b/internal/eventstore/v2/repository/sql/query_test.go @@ -1,4 +1,4 @@ -package repository +package sql import ( "database/sql" @@ -7,8 +7,7 @@ import ( "time" "github.com/caos/zitadel/internal/errors" - "github.com/caos/zitadel/internal/eventstore/models" - es_models "github.com/caos/zitadel/internal/eventstore/models" + "github.com/caos/zitadel/internal/eventstore/v2/repository" "github.com/lib/pq" ) @@ -60,12 +59,12 @@ func Test_numberPlaceholder(t *testing.T) { func Test_getOperation(t *testing.T) { t.Run("all ops", func(t *testing.T) { - for op, expected := range map[es_models.Operation]string{ - es_models.Operation_Equals: "=", - es_models.Operation_In: "=", - es_models.Operation_Greater: ">", - es_models.Operation_Less: "<", - es_models.Operation(-1): "", + for op, expected := range map[repository.Operation]string{ + repository.Operation_Equals: "=", + repository.Operation_In: "=", + repository.Operation_Greater: ">", + repository.Operation_Less: "<", + repository.Operation(-1): "", } { if got := getOperation(op); got != expected { t.Errorf("getOperation() = %v, want %v", got, expected) @@ -76,15 +75,15 @@ func Test_getOperation(t *testing.T) { func Test_getField(t *testing.T) { t.Run("all fields", func(t *testing.T) { - for field, expected := range map[es_models.Field]string{ - es_models.Field_AggregateType: "aggregate_type", - es_models.Field_AggregateID: "aggregate_id", - es_models.Field_LatestSequence: "event_sequence", - es_models.Field_ResourceOwner: "resource_owner", - es_models.Field_EditorService: "editor_service", - es_models.Field_EditorUser: "editor_user", - es_models.Field_EventType: "event_type", - es_models.Field(-1): "", + for field, expected := range map[repository.Field]string{ + repository.Field_AggregateType: "aggregate_type", + repository.Field_AggregateID: "aggregate_id", + repository.Field_LatestSequence: "event_sequence", + repository.Field_ResourceOwner: "resource_owner", + repository.Field_EditorService: "editor_service", + repository.Field_EditorUser: "editor_user", + repository.Field_EventType: "event_type", + repository.Field(-1): "", } { if got := getField(field); got != expected { t.Errorf("getField() = %v, want %v", got, expected) @@ -95,7 +94,7 @@ func Test_getField(t *testing.T) { func Test_getConditionFormat(t *testing.T) { type args struct { - operation es_models.Operation + operation repository.Operation } tests := []struct { name string @@ -105,14 +104,14 @@ func Test_getConditionFormat(t *testing.T) { { name: "no in operation", args: args{ - operation: es_models.Operation_Equals, + operation: repository.Operation_Equals, }, want: "%s %s ?", }, { name: "in operation", args: args{ - operation: es_models.Operation_In, + operation: repository.Operation_In, }, want: "%s %s ANY(?)", }, @@ -128,7 +127,7 @@ func Test_getConditionFormat(t *testing.T) { func Test_getCondition(t *testing.T) { type args struct { - filter *es_models.Filter + filter *repository.Filter } tests := []struct { name string @@ -137,37 +136,37 @@ func Test_getCondition(t *testing.T) { }{ { name: "equals", - args: args{filter: es_models.NewFilter(es_models.Field_AggregateID, "", es_models.Operation_Equals)}, + args: args{filter: repository.NewFilter(repository.Field_AggregateID, "", repository.Operation_Equals)}, want: "aggregate_id = ?", }, { name: "greater", - args: args{filter: es_models.NewFilter(es_models.Field_LatestSequence, 0, es_models.Operation_Greater)}, + args: args{filter: repository.NewFilter(repository.Field_LatestSequence, 0, repository.Operation_Greater)}, want: "event_sequence > ?", }, { name: "less", - args: args{filter: es_models.NewFilter(es_models.Field_LatestSequence, 5000, es_models.Operation_Less)}, + args: args{filter: repository.NewFilter(repository.Field_LatestSequence, 5000, repository.Operation_Less)}, want: "event_sequence < ?", }, { name: "in list", - args: args{filter: es_models.NewFilter(es_models.Field_AggregateType, []es_models.AggregateType{"movies", "actors"}, es_models.Operation_In)}, + args: args{filter: repository.NewFilter(repository.Field_AggregateType, []repository.AggregateType{"movies", "actors"}, repository.Operation_In)}, want: "aggregate_type = ANY(?)", }, { name: "invalid operation", - args: args{filter: es_models.NewFilter(es_models.Field_AggregateType, []es_models.AggregateType{"movies", "actors"}, es_models.Operation(-1))}, + args: args{filter: repository.NewFilter(repository.Field_AggregateType, []repository.AggregateType{"movies", "actors"}, repository.Operation(-1))}, want: "", }, { name: "invalid field", - args: args{filter: es_models.NewFilter(es_models.Field(-1), []es_models.AggregateType{"movies", "actors"}, es_models.Operation_Equals)}, + args: args{filter: repository.NewFilter(repository.Field(-1), []repository.AggregateType{"movies", "actors"}, repository.Operation_Equals)}, want: "", }, { name: "invalid field and operation", - args: args{filter: es_models.NewFilter(es_models.Field(-1), []es_models.AggregateType{"movies", "actors"}, es_models.Operation(-1))}, + args: args{filter: repository.NewFilter(repository.Field(-1), []repository.AggregateType{"movies", "actors"}, repository.Operation(-1))}, want: "", }, } @@ -182,7 +181,7 @@ func Test_getCondition(t *testing.T) { func Test_prepareColumns(t *testing.T) { type args struct { - columns models.Columns + columns repository.Columns dest interface{} dbErr error } @@ -199,7 +198,7 @@ func Test_prepareColumns(t *testing.T) { }{ { name: "invalid columns", - args: args{columns: es_models.Columns(-1)}, + args: args{columns: repository.Columns(-1)}, res: res{ query: "", dbErr: func(err error) bool { return err == nil }, @@ -208,7 +207,7 @@ func Test_prepareColumns(t *testing.T) { { name: "max column", args: args{ - columns: es_models.Columns_Max_Sequence, + columns: repository.Columns_Max_Sequence, dest: new(Sequence), }, res: res{ @@ -220,7 +219,7 @@ func Test_prepareColumns(t *testing.T) { { name: "max sequence wrong dest type", args: args{ - columns: es_models.Columns_Max_Sequence, + columns: repository.Columns_Max_Sequence, dest: new(uint64), }, res: res{ @@ -231,19 +230,19 @@ func Test_prepareColumns(t *testing.T) { { name: "event", args: args{ - columns: es_models.Columns_Event, - dest: new(models.Event), + columns: repository.Columns_Event, + dest: new(repository.Event), }, res: res{ query: "SELECT creation_date, event_type, event_sequence, previous_sequence, event_data, editor_service, editor_user, resource_owner, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events", - dbRow: []interface{}{time.Time{}, models.EventType(""), uint64(5), Sequence(0), Data(nil), "", "", "", models.AggregateType("user"), "hodor", models.Version("")}, - expected: models.Event{AggregateID: "hodor", AggregateType: "user", Sequence: 5, Data: make(Data, 0)}, + dbRow: []interface{}{time.Time{}, repository.EventType(""), uint64(5), Sequence(0), Data(nil), "", "", "", repository.AggregateType("user"), "hodor", repository.Version("")}, + expected: repository.Event{AggregateID: "hodor", AggregateType: "user", Sequence: 5, Data: make(Data, 0)}, }, }, { name: "event wrong dest type", args: args{ - columns: es_models.Columns_Event, + columns: repository.Columns_Event, dest: new(uint64), }, res: res{ @@ -254,8 +253,8 @@ func Test_prepareColumns(t *testing.T) { { name: "event query error", args: args{ - columns: es_models.Columns_Event, - dest: new(models.Event), + columns: repository.Columns_Event, + dest: new(repository.Event), dbErr: sql.ErrConnDone, }, res: res{ @@ -290,7 +289,7 @@ func Test_prepareColumns(t *testing.T) { } } -func prepareTestScan(err error, res []interface{}) scanner { +func prepareTestScan(err error, res []interface{}) scan { return func(dests ...interface{}) error { if err != nil { return err @@ -308,7 +307,7 @@ func prepareTestScan(err error, res []interface{}) scanner { func Test_prepareCondition(t *testing.T) { type args struct { - filters []*models.Filter + filters []*repository.Filter } type res struct { clause string @@ -332,7 +331,7 @@ func Test_prepareCondition(t *testing.T) { { name: "empty filters", args: args{ - filters: []*es_models.Filter{}, + filters: []*repository.Filter{}, }, res: res{ clause: "", @@ -342,8 +341,8 @@ func Test_prepareCondition(t *testing.T) { { name: "invalid condition", args: args{ - filters: []*es_models.Filter{ - es_models.NewFilter(es_models.Field_AggregateID, "wrong", es_models.Operation(-1)), + filters: []*repository.Filter{ + repository.NewFilter(repository.Field_AggregateID, "wrong", repository.Operation(-1)), }, }, res: res{ @@ -354,27 +353,27 @@ func Test_prepareCondition(t *testing.T) { { name: "array as condition value", args: args{ - filters: []*es_models.Filter{ - es_models.NewFilter(es_models.Field_AggregateType, []es_models.AggregateType{"user", "org"}, es_models.Operation_In), + filters: []*repository.Filter{ + repository.NewFilter(repository.Field_AggregateType, []repository.AggregateType{"user", "org"}, repository.Operation_In), }, }, res: res{ clause: " WHERE aggregate_type = ANY(?)", - values: []interface{}{pq.Array([]es_models.AggregateType{"user", "org"})}, + values: []interface{}{pq.Array([]repository.AggregateType{"user", "org"})}, }, }, { name: "multiple filters", args: args{ - filters: []*es_models.Filter{ - es_models.NewFilter(es_models.Field_AggregateType, []es_models.AggregateType{"user", "org"}, es_models.Operation_In), - es_models.NewFilter(es_models.Field_AggregateID, "1234", es_models.Operation_Equals), - es_models.NewFilter(es_models.Field_EventType, []es_models.EventType{"user.created", "org.created"}, es_models.Operation_In), + filters: []*repository.Filter{ + repository.NewFilter(repository.Field_AggregateType, []repository.AggregateType{"user", "org"}, repository.Operation_In), + repository.NewFilter(repository.Field_AggregateID, "1234", repository.Operation_Equals), + repository.NewFilter(repository.Field_EventType, []repository.EventType{"user.created", "org.created"}, repository.Operation_In), }, }, res: res{ clause: " WHERE aggregate_type = ANY(?) AND aggregate_id = ? AND event_type = ANY(?)", - values: []interface{}{pq.Array([]es_models.AggregateType{"user", "org"}), "1234", pq.Array([]es_models.EventType{"user.created", "org.created"})}, + values: []interface{}{pq.Array([]repository.AggregateType{"user", "org"}), "1234", pq.Array([]repository.EventType{"user.created", "org.created"})}, }, }, } @@ -399,7 +398,7 @@ func Test_prepareCondition(t *testing.T) { func Test_buildQuery(t *testing.T) { type args struct { - queryFactory *models.SearchQueryFactory + queryFactory *repository.SearchQuery } type res struct { query string @@ -427,35 +426,36 @@ func Test_buildQuery(t *testing.T) { { name: "with order by desc", args: args{ - queryFactory: es_models.NewSearchQueryFactory("user").OrderDesc(), + // NewSearchQueryFactory("user").OrderDesc() + queryFactory: &repository.SearchQuery{Desc: true}, }, res: res{ query: "SELECT creation_date, event_type, event_sequence, previous_sequence, event_data, editor_service, editor_user, resource_owner, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = $1 ORDER BY event_sequence DESC", rowScanner: true, - values: []interface{}{es_models.AggregateType("user")}, + values: []interface{}{repository.AggregateType("user")}, }, }, { name: "with limit", args: args{ - queryFactory: es_models.NewSearchQueryFactory("user").Limit(5), + queryFactory: repository.NewSearchQueryFactory("user").Limit(5), }, res: res{ query: "SELECT creation_date, event_type, event_sequence, previous_sequence, event_data, editor_service, editor_user, resource_owner, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = $1 ORDER BY event_sequence LIMIT $2", rowScanner: true, - values: []interface{}{es_models.AggregateType("user"), uint64(5)}, + values: []interface{}{repository.AggregateType("user"), uint64(5)}, limit: 5, }, }, { name: "with limit and order by desc", args: args{ - queryFactory: es_models.NewSearchQueryFactory("user").Limit(5).OrderDesc(), + queryFactory: repository.NewSearchQueryFactory("user").Limit(5).OrderDesc(), }, res: res{ query: "SELECT creation_date, event_type, event_sequence, previous_sequence, event_data, editor_service, editor_user, resource_owner, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = $1 ORDER BY event_sequence DESC LIMIT $2", rowScanner: true, - values: []interface{}{es_models.AggregateType("user"), uint64(5)}, + values: []interface{}{repository.AggregateType("user"), uint64(5)}, limit: 5, }, }, @@ -484,3 +484,5 @@ func Test_buildQuery(t *testing.T) { }) } } + +// func buildQuery(t *testing.T, factory *reposear) diff --git a/internal/eventstore/v2/repository/sql_types.go b/internal/eventstore/v2/repository/sql/sql_types.go similarity index 95% rename from internal/eventstore/v2/repository/sql_types.go rename to internal/eventstore/v2/repository/sql/sql_types.go index 9625b1bff..78ed8f7a6 100644 --- a/internal/eventstore/v2/repository/sql_types.go +++ b/internal/eventstore/v2/repository/sql/sql_types.go @@ -1,6 +1,8 @@ -package repository +package sql -import "database/sql/driver" +import ( + "database/sql/driver" +) // Data represents a byte array that may be null. // Data implements the sql.Scanner interface