TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-01-13 02:13:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: eventstore v2

Browse Source
This commit is contained in:
adlerhurst 2020-10-06 21:28:09 +02:00
parent 46a68c15bf
commit 71fd4bf9f9
13 changed files with 251 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
internal/eventstore/v2/eventstore.go
View File

@ -10,6 +10,7 @@ import (
"github.com/caos/zitadel/internal/eventstore/v2/repository"
)
//Event is the representation of a state change
type Event interface {
//CheckPrevious ensures the event order if true
// if false the previous sequence is not checked on push
@ -61,7 +62,7 @@ type aggregater interface {
//PreviouseSequence should return the sequence of the latest event of this aggregate
// stored in the eventstore
// it's set to the first event of this push transaction,
// later events consume the sequence of the previously pushed event
// later events consume the sequence of the previously pushed event of the aggregate
PreviousSequence() uint64
}
@ -111,7 +112,7 @@ func (es *Eventstore) aggregatesToEvents(aggregates []aggregater) ([]*repository
//FilterEvents filters the stored events based on the searchQuery
// and maps the events to the defined event structs
func (es *Eventstore) FilterEvents(ctx context.Context, queryFactory *SearchQueryFactory) ([]Event, error) {
query, err := queryFactory.Build()
query, err := queryFactory.build()
if err != nil {
return nil, err
}
@ -166,7 +167,7 @@ func (es *Eventstore) FilterToReducer(ctx context.Context, searchQuery *SearchQu
//LatestSequence filters the latest sequence for the given search query
func (es *Eventstore) LatestSequence(ctx context.Context, queryFactory *SearchQueryFactory) (uint64, error) {
query, err := queryFactory.Build()
query, err := queryFactory.build()
if err != nil {
return 0, err
}

3
internal/eventstore/v2/repository/event.go
View File

@ -1,7 +1,6 @@
package repository
import (
"database/sql"
"time"
)
@ -60,8 +59,6 @@ type Event struct {
// an aggregate can only be managed by one organisation
// use the ID of the org
ResourceOwner string
stmt *sql.Stmt
}
//EventType is the description of the change

2
internal/eventstore/v2/repository/repository.go
View File

@ -4,7 +4,9 @@ import (
"context"
)
//Repository pushes and filters events
type Repository interface {
//Health checks if the connection to the storage is available
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

74
internal/eventstore/v2/repository/search_query.go
View File

@ -2,6 +2,7 @@ package repository
import "github.com/caos/zitadel/internal/errors"
//SearchQuery defines the which and how data are queried
type SearchQuery struct {
Columns Columns
Limit uint64
@ -9,40 +10,68 @@ type SearchQuery struct {
Filters []*Filter
}
//Columns defines which fields of the event are needed for the query
type Columns int32
const (
Columns_Event = iota
Columns_Max_Sequence
//insert new columns-types above this ColumnsCount because count is needed for validation
ColumnsCount
//ColumnsEvent represents all fields of an event
ColumnsEvent = iota + 1
//ColumnsMaxSequence represents the latest sequence of the filtered events
ColumnsMaxSequence
columnsCount
)
func (c Columns) Validate() error {
if c <= 0 || c >= columnsCount {
return errors.ThrowPreconditionFailed(nil, "REPOS-x8R35", "column out of range")
}
return nil
}
//Filter represents all fields needed to compare a field of an event with a value
type Filter struct {
Field Field
Value interface{}
Operation Operation
}
//Operation defines how fields are compared
type Operation int32
const (
Operation_Equals Operation = 1 + iota
Operation_Greater
Operation_Less
Operation_In
// OperationEquals compares two values for equality
OperationEquals Operation = iota + 1
// OperationGreater compares if the given values is greater than the stored one
OperationGreater
// OperationLess compares if the given values is less than the stored one
OperationLess
//OperationIn checks if a stored value matches one of the passed value list
OperationIn
operationCount
)
//Field is the representation of a field from the event
type Field int32
const (
Field_AggregateType Field = 1 + iota
Field_AggregateID
Field_LatestSequence
Field_ResourceOwner
Field_EditorService
Field_EditorUser
Field_EventType
//FieldAggregateType represents the aggregate type field
FieldAggregateType Field = iota + 1
//FieldAggregateID represents the aggregate id field
FieldAggregateID
//FieldSequence represents the sequence field
FieldSequence
//FieldResourceOwner represents the resource owner field
FieldResourceOwner
//FieldEditorService represents the editor service field
FieldEditorService
//FieldEditorUser represents the editor user field
FieldEditorUser
//FieldEventType represents the event type field
FieldEventType
fieldCount
)
//NewFilter is used in tests. Use searchQuery.*Filter() instead
@ -54,27 +83,18 @@ func NewFilter(field Field, value interface{}, operation Operation) *Filter {
}
}
// func (f *Filter) Field() Field {
// return f.field
// }
// func (f *Filter) Operation() Operation {
// return f.operation
// }
// func (f *Filter) Value() interface{} {
// return f.value
// }
//Validate checks if the fields of the filter have valid values
func (f *Filter) Validate() error {
if f == nil {
return errors.ThrowPreconditionFailed(nil, "REPO-z6KcG", "filter is nil")
}
if f.Field <= 0 {
if f.Field <= 0 || f.Field >= fieldCount {
return errors.ThrowPreconditionFailed(nil, "REPO-zw62U", "field not definded")
}
if f.Value == nil {
return errors.ThrowPreconditionFailed(nil, "REPO-GJ9ct", "no value definded")
}
if f.Operation <= 0 {
if f.Operation <= 0 || f.Operation >= operationCount {
return errors.ThrowPreconditionFailed(nil, "REPO-RrQTy", "operation not definded")
}
return nil

58
internal/eventstore/v2/repository/search_query_test.go
View File

@ -19,11 +19,11 @@ func TestNewFilter(t *testing.T) {
{
name: "aggregateID equals",
args: args{
field: Field_AggregateID,
field: FieldAggregateID,
value: "hodor",
operation: Operation_Equals,
operation: OperationEquals,
},
want: &Filter{Field: Field_AggregateID, Operation: Operation_Equals, Value: "hodor"},
want: &Filter{Field: FieldAggregateID, Operation: OperationEquals, Value: "hodor"},
},
}
for _, tt := range tests {
@ -50,8 +50,8 @@ func TestFilter_Validate(t *testing.T) {
{
name: "correct filter",
fields: fields{
field: Field_LatestSequence,
operation: Operation_Greater,
field: FieldSequence,
operation: OperationGreater,
value: uint64(235),
},
wantErr: false,
@ -64,7 +64,7 @@ func TestFilter_Validate(t *testing.T) {
{
name: "no field error",
fields: fields{
operation: Operation_Greater,
operation: OperationGreater,
value: uint64(235),
},
wantErr: true,
@ -72,15 +72,15 @@ func TestFilter_Validate(t *testing.T) {
{
name: "no value error",
fields: fields{
field: Field_LatestSequence,
operation: Operation_Greater,
field: FieldSequence,
operation: OperationGreater,
},
wantErr: true,
},
{
name: "no operation error",
fields: fields{
field: Field_LatestSequence,
field: FieldSequence,
value: uint64(235),
},
wantErr: true,
@ -102,3 +102,43 @@ func TestFilter_Validate(t *testing.T) {
})
}
}
func TestColumns_Validate(t *testing.T) {
type fields struct {
columns Columns
}
tests := []struct {
name string
fields fields
wantErr bool
}{
{
name: "correct filter",
fields: fields{
columns: ColumnsEvent,
},
wantErr: false,
},
{
name: "columns too low",
fields: fields{
columns: 0,
},
wantErr: true,
},
{
name: "columns too high",
fields: fields{
columns: columnsCount,
},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if err := tt.fields.columns.Validate(); (err != nil) != tt.wantErr {
t.Errorf("Columns.Validate() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}

22
internal/eventstore/v2/repository/sql/crdb.go
View File

@ -222,19 +222,19 @@ func (db *CRDB) maxSequenceQuery() string {
func (db *CRDB) columnName(col repository.Field) string {
switch col {
case repository.Field_AggregateID:
case repository.FieldAggregateID:
return "aggregate_id"
case repository.Field_AggregateType:
case repository.FieldAggregateType:
return "aggregate_type"
case repository.Field_LatestSequence:
case repository.FieldSequence:
return "event_sequence"
case repository.Field_ResourceOwner:
case repository.FieldResourceOwner:
return "resource_owner"
case repository.Field_EditorService:
case repository.FieldEditorService:
return "editor_service"
case repository.Field_EditorUser:
case repository.FieldEditorUser:
return "editor_user"
case repository.Field_EventType:
case repository.FieldEventType:
return "event_type"
default:
return ""
@ -242,7 +242,7 @@ func (db *CRDB) columnName(col repository.Field) string {
}
func (db *CRDB) conditionFormat(operation repository.Operation) string {
if operation == repository.Operation_In {
if operation == repository.OperationIn {
return "%s %s ANY(?)"
}
return "%s %s ?"
@ -250,11 +250,11 @@ func (db *CRDB) conditionFormat(operation repository.Operation) string {
func (db *CRDB) operation(operation repository.Operation) string {
switch operation {
case repository.Operation_Equals, repository.Operation_In:
case repository.OperationEquals, repository.OperationIn:
return "="
case repository.Operation_Greater:
case repository.OperationGreater:
return ">"
case repository.Operation_Less:
case repository.OperationLess:
return "<"
}
return ""

26
internal/eventstore/v2/repository/sql/crdb_test.go
View File

@ -81,7 +81,7 @@ func TestCRDB_operation(t *testing.T) {
{
name: "greater",
args: args{
operation: repository.Operation_Greater,
operation: repository.OperationGreater,
},
res: res{
op: ">",
@ -90,7 +90,7 @@ func TestCRDB_operation(t *testing.T) {
{
name: "less",
args: args{
operation: repository.Operation_Less,
operation: repository.OperationLess,
},
res: res{
op: "<",
@ -99,7 +99,7 @@ func TestCRDB_operation(t *testing.T) {
{
name: "equals",
args: args{
operation: repository.Operation_Equals,
operation: repository.OperationEquals,
},
res: res{
op: "=",
@ -108,7 +108,7 @@ func TestCRDB_operation(t *testing.T) {
{
name: "in",
args: args{
operation: repository.Operation_In,
operation: repository.OperationIn,
},
res: res{
op: "=",
@ -140,7 +140,7 @@ func TestCRDB_conditionFormat(t *testing.T) {
{
name: "default",
args: args{
operation: repository.Operation_Equals,
operation: repository.OperationEquals,
},
res: res{
format: "%s %s ?",
@ -149,7 +149,7 @@ func TestCRDB_conditionFormat(t *testing.T) {
{
name: "in",
args: args{
operation: repository.Operation_In,
operation: repository.OperationIn,
},
res: res{
format: "%s %s ANY(?)",
@ -190,7 +190,7 @@ func TestCRDB_columnName(t *testing.T) {
{
name: "aggregate id",
args: args{
field: repository.Field_AggregateID,
field: repository.FieldAggregateID,
},
res: res{
name: "aggregate_id",
@ -199,7 +199,7 @@ func TestCRDB_columnName(t *testing.T) {
{
name: "aggregate type",
args: args{
field: repository.Field_AggregateType,
field: repository.FieldAggregateType,
},
res: res{
name: "aggregate_type",
@ -208,7 +208,7 @@ func TestCRDB_columnName(t *testing.T) {
{
name: "editor service",
args: args{
field: repository.Field_EditorService,
field: repository.FieldEditorService,
},
res: res{
name: "editor_service",
@ -217,7 +217,7 @@ func TestCRDB_columnName(t *testing.T) {
{
name: "editor user",
args: args{
field: repository.Field_EditorUser,
field: repository.FieldEditorUser,
},
res: res{
name: "editor_user",
@ -226,7 +226,7 @@ func TestCRDB_columnName(t *testing.T) {
{
name: "event type",
args: args{
field: repository.Field_EventType,
field: repository.FieldEventType,
},
res: res{
name: "event_type",
@ -235,7 +235,7 @@ func TestCRDB_columnName(t *testing.T) {
{
name: "latest sequence",
args: args{
field: repository.Field_LatestSequence,
field: repository.FieldSequence,
},
res: res{
name: "event_sequence",
@ -244,7 +244,7 @@ func TestCRDB_columnName(t *testing.T) {
{
name: "resource owner",
args: args{
field: repository.Field_ResourceOwner,
field: repository.FieldResourceOwner,
},
res: res{
name: "resource_owner",

6
internal/eventstore/v2/repository/sql/local_crdb_test.go
View File

@ -169,11 +169,11 @@ func TestInsert(t *testing.T) {
}
fmt.Println("====================")
filtededEvents, err := crdb.Filter(context.Background(), &repository.SearchQuery{
Columns: repository.Columns_Event,
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
{
Field: repository.Field_AggregateType,
Operation: repository.Operation_Equals,
Field: repository.FieldAggregateType,
Operation: repository.OperationEquals,
Value: repository.AggregateType("agg.type"),
},
},

16
internal/eventstore/v2/repository/sql/query.go
View File

@ -12,7 +12,7 @@ import (
"github.com/lib/pq"
)
type criteriaer interface {
type querier interface {
columnName(repository.Field) string
operation(repository.Operation) string
conditionFormat(repository.Operation) string
@ -24,7 +24,7 @@ type criteriaer interface {
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) {
func buildQuery(criteria querier, searchQuery *repository.SearchQuery) (query string, values []interface{}, rowScanner rowScan) {
query, rowScanner = prepareColumns(criteria, searchQuery.Columns)
where, values := prepareCondition(criteria, searchQuery.Filters)
if where == "" || query == "" {
@ -32,7 +32,7 @@ func buildQuery(criteria criteriaer, searchQuery *repository.SearchQuery) (query
}
query += where
if searchQuery.Columns != repository.Columns_Max_Sequence {
if searchQuery.Columns != repository.ColumnsMaxSequence {
query += " ORDER BY event_sequence"
if searchQuery.Desc {
query += " DESC"
@ -49,11 +49,11 @@ func buildQuery(criteria criteriaer, searchQuery *repository.SearchQuery) (query
return query, values, rowScanner
}
func prepareColumns(criteria criteriaer, columns repository.Columns) (string, func(s scan, dest interface{}) error) {
func prepareColumns(criteria querier, columns repository.Columns) (string, func(s scan, dest interface{}) error) {
switch columns {
case repository.Columns_Max_Sequence:
case repository.ColumnsMaxSequence:
return criteria.maxSequenceQuery(), maxSequenceScanner
case repository.Columns_Event:
case repository.ColumnsEvent:
return criteria.eventQuery(), eventsScanner
default:
return "", nil
@ -110,7 +110,7 @@ func eventsScanner(scanner scan, dest interface{}) (err error) {
return nil
}
func prepareCondition(criteria criteriaer, filters []*repository.Filter) (clause string, values []interface{}) {
func prepareCondition(criteria querier, filters []*repository.Filter) (clause string, values []interface{}) {
values = make([]interface{}, len(filters))
clauses := make([]string, len(filters))
@ -133,7 +133,7 @@ func prepareCondition(criteria criteriaer, filters []*repository.Filter) (clause
return " WHERE " + strings.Join(clauses, " AND "), values
}
func getCondition(cond criteriaer, filter *repository.Filter) (condition string) {
func getCondition(cond querier, filter *repository.Filter) (condition string) {
field := cond.columnName(filter.Field)
operation := cond.operation(filter.Operation)
if field == "" || operation == "" {

52
internal/eventstore/v2/repository/sql/query_test.go
View File

@ -22,32 +22,32 @@ func Test_getCondition(t *testing.T) {
}{
{
name: "equals",
args: args{filter: repository.NewFilter(repository.Field_AggregateID, "", repository.Operation_Equals)},
args: args{filter: repository.NewFilter(repository.FieldAggregateID, "", repository.OperationEquals)},
want: "aggregate_id = ?",
},
{
name: "greater",
args: args{filter: repository.NewFilter(repository.Field_LatestSequence, 0, repository.Operation_Greater)},
args: args{filter: repository.NewFilter(repository.FieldSequence, 0, repository.OperationGreater)},
want: "event_sequence > ?",
},
{
name: "less",
args: args{filter: repository.NewFilter(repository.Field_LatestSequence, 5000, repository.Operation_Less)},
args: args{filter: repository.NewFilter(repository.FieldSequence, 5000, repository.OperationLess)},
want: "event_sequence < ?",
},
{
name: "in list",
args: args{filter: repository.NewFilter(repository.Field_AggregateType, []repository.AggregateType{"movies", "actors"}, repository.Operation_In)},
args: args{filter: repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"movies", "actors"}, repository.OperationIn)},
want: "aggregate_type = ANY(?)",
},
{
name: "invalid operation",
args: args{filter: repository.NewFilter(repository.Field_AggregateType, []repository.AggregateType{"movies", "actors"}, repository.Operation(-1))},
args: args{filter: repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"movies", "actors"}, repository.Operation(-1))},
want: "",
},
{
name: "invalid field",
args: args{filter: repository.NewFilter(repository.Field(-1), []repository.AggregateType{"movies", "actors"}, repository.Operation_Equals)},
args: args{filter: repository.NewFilter(repository.Field(-1), []repository.AggregateType{"movies", "actors"}, repository.OperationEquals)},
want: "",
},
{
@ -97,7 +97,7 @@ func Test_prepareColumns(t *testing.T) {
{
name: "max column",
args: args{
columns: repository.Columns_Max_Sequence,
columns: repository.ColumnsMaxSequence,
dest: new(Sequence),
},
res: res{
@ -111,7 +111,7 @@ func Test_prepareColumns(t *testing.T) {
{
name: "max sequence wrong dest type",
args: args{
columns: repository.Columns_Max_Sequence,
columns: repository.ColumnsMaxSequence,
dest: new(uint64),
},
res: res{
@ -122,7 +122,7 @@ func Test_prepareColumns(t *testing.T) {
{
name: "events",
args: args{
columns: repository.Columns_Event,
columns: repository.ColumnsEvent,
dest: &[]*repository.Event{},
},
res: res{
@ -138,7 +138,7 @@ func Test_prepareColumns(t *testing.T) {
{
name: "events wrong dest type",
args: args{
columns: repository.Columns_Event,
columns: repository.ColumnsEvent,
dest: []*repository.Event{},
},
res: res{
@ -149,7 +149,7 @@ func Test_prepareColumns(t *testing.T) {
{
name: "event query error",
args: args{
columns: repository.Columns_Event,
columns: repository.ColumnsEvent,
dest: &[]*repository.Event{},
dbErr: sql.ErrConnDone,
},
@ -240,7 +240,7 @@ func Test_prepareCondition(t *testing.T) {
name: "invalid condition",
args: args{
filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateID, "wrong", repository.Operation(-1)),
repository.NewFilter(repository.FieldAggregateID, "wrong", repository.Operation(-1)),
},
},
res: res{
@ -252,7 +252,7 @@ func Test_prepareCondition(t *testing.T) {
name: "array as condition value",
args: args{
filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, []repository.AggregateType{"user", "org"}, repository.Operation_In),
repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"user", "org"}, repository.OperationIn),
},
},
res: res{
@ -264,9 +264,9 @@ func Test_prepareCondition(t *testing.T) {
name: "multiple filters",
args: args{
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),
repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"user", "org"}, repository.OperationIn),
repository.NewFilter(repository.FieldAggregateID, "1234", repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, []repository.EventType{"user.created", "org.created"}, repository.OperationIn),
},
},
res: res{
@ -314,13 +314,13 @@ func Test_buildQuery(t *testing.T) {
args: args{
// NewSearchQueryFactory("user").OrderDesc()
query: &repository.SearchQuery{
Columns: repository.Columns_Event,
Columns: repository.ColumnsEvent,
Desc: true,
Filters: []*repository.Filter{
{
Field: repository.Field_AggregateType,
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.Operation_Equals,
Operation: repository.OperationEquals,
},
},
},
@ -335,14 +335,14 @@ func Test_buildQuery(t *testing.T) {
name: "with limit",
args: args{
query: &repository.SearchQuery{
Columns: repository.Columns_Event,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 5,
Filters: []*repository.Filter{
{
Field: repository.Field_AggregateType,
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.Operation_Equals,
Operation: repository.OperationEquals,
},
},
},
@ -357,14 +357,14 @@ func Test_buildQuery(t *testing.T) {
name: "with limit and order by desc",
args: args{
query: &repository.SearchQuery{
Columns: repository.Columns_Event,
Columns: repository.ColumnsEvent,
Desc: true,
Limit: 5,
Filters: []*repository.Filter{
{
Field: repository.Field_AggregateType,
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.Operation_Equals,
Operation: repository.OperationEquals,
},
},
},
@ -392,7 +392,7 @@ func Test_buildQuery(t *testing.T) {
name: "invalid condition",
args: args{
query: &repository.SearchQuery{
Columns: repository.Columns_Event,
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
{},
},

2
internal/eventstore/v2/repository/version.go
View File

@ -8,8 +8,10 @@ import (
var versionRegexp = regexp.MustCompile(`^v[0-9]+(\.[0-9]+){0,2}$`)
//Version represents the semver of an aggregate
type Version string
//Validate checks if the v is semver
func (v Version) Validate() error {
if !versionRegexp.MatchString(string(v)) {
return errors.ThrowPreconditionFailed(nil, "MODEL-luDuS", "version is not semver")

44
internal/eventstore/v2/search_query.go
View File

@ -5,6 +5,8 @@ import (
"github.com/caos/zitadel/internal/eventstore/v2/repository"
)
//SearchQueryFactory represents the builder for your filter
// if invalid data are set the filter will fail
type SearchQueryFactory struct {
columns repository.Columns
limit uint64
@ -16,18 +18,27 @@ type SearchQueryFactory struct {
resourceOwner string
}
// Columns defines which fields of the event are needed for the query
type Columns repository.Columns
const (
Columns_Event Columns = repository.Columns_Event
Columns_Max_Sequence Columns = repository.Columns_Max_Sequence
//ColumnsEvent represents all fields of an event
ColumnsEvent Columns = repository.ColumnsEvent
// ColumnsMaxSequence represents the latest sequence of the filtered events
ColumnsMaxSequence Columns = repository.ColumnsMaxSequence
)
// AggregateType is the object name
type AggregateType repository.AggregateType
// EventType is the description of the change
type EventType repository.EventType
func NewSearchQueryFactory(aggregateTypes ...AggregateType) *SearchQueryFactory {
// NewSearchQueryFactory creates a new factory for event filters
// aggregateTypes must contain at least one aggregate type
func NewSearchQueryFactory(columns Columns, aggregateTypes ...AggregateType) *SearchQueryFactory {
return &SearchQueryFactory{
columns: repository.Columns(columns),
aggregateTypes: aggregateTypes,
}
}
@ -72,10 +83,10 @@ func (factory *SearchQueryFactory) OrderAsc() *SearchQueryFactory {
return factory
}
func (factory *SearchQueryFactory) Build() (*repository.SearchQuery, error) {
func (factory *SearchQueryFactory) build() (*repository.SearchQuery, error) {
if factory == nil ||
len(factory.aggregateTypes) < 1 ||
(factory.columns < 0 || factory.columns >= repository.ColumnsCount) {
factory.columns.Validate() != nil {
return nil, errors.ThrowPreconditionFailed(nil, "MODEL-tGAD3", "factory invalid")
}
filters := []*repository.Filter{
@ -89,6 +100,9 @@ func (factory *SearchQueryFactory) Build() (*repository.SearchQuery, error) {
factory.resourceOwnerFilter,
} {
if filter := f(); filter != nil {
if err := filter.Validate(); err != nil {
return nil, err
}
filters = append(filters, filter)
}
}
@ -106,9 +120,9 @@ func (factory *SearchQueryFactory) aggregateIDFilter() *repository.Filter {
return nil
}
if len(factory.aggregateIDs) == 1 {
return repository.NewFilter(repository.Field_AggregateID, factory.aggregateIDs[0], repository.Operation_Equals)
return repository.NewFilter(repository.FieldAggregateID, factory.aggregateIDs[0], repository.OperationEquals)
}
return repository.NewFilter(repository.Field_AggregateID, factory.aggregateIDs, repository.Operation_In)
return repository.NewFilter(repository.FieldAggregateID, factory.aggregateIDs, repository.OperationIn)
}
func (factory *SearchQueryFactory) eventTypeFilter() *repository.Filter {
@ -116,32 +130,32 @@ func (factory *SearchQueryFactory) eventTypeFilter() *repository.Filter {
return nil
}
if len(factory.eventTypes) == 1 {
return repository.NewFilter(repository.Field_EventType, factory.eventTypes[0], repository.Operation_Equals)
return repository.NewFilter(repository.FieldEventType, factory.eventTypes[0], repository.OperationEquals)
}
return repository.NewFilter(repository.Field_EventType, factory.eventTypes, repository.Operation_In)
return repository.NewFilter(repository.FieldEventType, factory.eventTypes, repository.OperationIn)
}
func (factory *SearchQueryFactory) aggregateTypeFilter() *repository.Filter {
if len(factory.aggregateTypes) == 1 {
return repository.NewFilter(repository.Field_AggregateType, factory.aggregateTypes[0], repository.Operation_Equals)
return repository.NewFilter(repository.FieldAggregateType, factory.aggregateTypes[0], repository.OperationEquals)
}
return repository.NewFilter(repository.Field_AggregateType, factory.aggregateTypes, repository.Operation_In)
return repository.NewFilter(repository.FieldAggregateType, factory.aggregateTypes, repository.OperationIn)
}
func (factory *SearchQueryFactory) eventSequenceFilter() *repository.Filter {
if factory.eventSequence == 0 {
return nil
}
sortOrder := repository.Operation_Greater
sortOrder := repository.OperationGreater
if factory.desc {
sortOrder = repository.Operation_Less
sortOrder = repository.OperationLess
}
return repository.NewFilter(repository.Field_LatestSequence, factory.eventSequence, sortOrder)
return repository.NewFilter(repository.FieldSequence, factory.eventSequence, sortOrder)
}
func (factory *SearchQueryFactory) resourceOwnerFilter() *repository.Filter {
if factory.resourceOwner == "" {
return nil
}
return repository.NewFilter(repository.Field_ResourceOwner, factory.resourceOwner, repository.Operation_Equals)
return repository.NewFilter(repository.FieldResourceOwner, factory.resourceOwner, repository.OperationEquals)
}

95
internal/eventstore/v2/search_query_test.go
View File

@ -1,6 +1,7 @@
package eventstore
import (
"math"
"reflect"
"testing"
@ -63,6 +64,7 @@ func testSetSortOrder(asc bool) func(factory *SearchQueryFactory) *SearchQueryFa
func TestSearchQueryFactorySetters(t *testing.T) {
type args struct {
columns Columns
aggregateTypes []AggregateType
setters []func(*SearchQueryFactory) *SearchQueryFactory
}
@ -74,19 +76,21 @@ func TestSearchQueryFactorySetters(t *testing.T) {
{
name: "New factory",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user", "org"},
},
res: &SearchQueryFactory{
columns: repository.Columns(ColumnsEvent),
aggregateTypes: []AggregateType{"user", "org"},
},
},
{
name: "set columns",
args: args{
setters: []func(*SearchQueryFactory) *SearchQueryFactory{testSetColumns(repository.Columns_Max_Sequence)},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{testSetColumns(repository.ColumnsMaxSequence)},
},
res: &SearchQueryFactory{
columns: repository.Columns_Max_Sequence,
columns: repository.ColumnsMaxSequence,
},
},
{
@ -149,7 +153,7 @@ func TestSearchQueryFactorySetters(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
factory := NewSearchQueryFactory(tt.args.aggregateTypes...)
factory := NewSearchQueryFactory(tt.args.columns, tt.args.aggregateTypes...)
for _, setter := range tt.args.setters {
factory = setter(factory)
}
@ -162,6 +166,7 @@ func TestSearchQueryFactorySetters(t *testing.T) {
func TestSearchQueryFactoryBuild(t *testing.T) {
type args struct {
columns Columns
aggregateTypes []AggregateType
setters []func(*SearchQueryFactory) *SearchQueryFactory
}
@ -177,6 +182,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "no aggregate types",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{},
},
@ -188,6 +194,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "invalid column (too low)",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetColumns(Columns(-1)),
@ -200,9 +207,10 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "invalid column (too high)",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetColumns(repository.ColumnsCount),
testSetColumns(math.MaxInt32),
},
},
res: res{
@ -212,17 +220,18 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{},
},
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
},
},
},
@ -230,17 +239,18 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate types",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user", "org"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{},
},
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, []AggregateType{"user", "org"}, repository.Operation_In),
repository.NewFilter(repository.FieldAggregateType, []AggregateType{"user", "org"}, repository.OperationIn),
},
},
},
@ -248,6 +258,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type, limit, desc",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetLimit(5),
@ -258,12 +269,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: true,
Limit: 5,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_LatestSequence, uint64(100), repository.Operation_Less),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationLess),
},
},
},
@ -271,6 +282,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type, limit, asc",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetLimit(5),
@ -281,12 +293,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 5,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_LatestSequence, uint64(100), repository.Operation_Greater),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationGreater),
},
},
},
@ -294,23 +306,24 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type, limit, desc, max event sequence cols",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetLimit(5),
testSetSortOrder(false),
testSetSequence(100),
testSetColumns(repository.Columns_Max_Sequence),
testSetColumns(repository.ColumnsMaxSequence),
},
},
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: repository.Columns_Max_Sequence,
Columns: repository.ColumnsMaxSequence,
Desc: true,
Limit: 5,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_LatestSequence, uint64(100), repository.Operation_Less),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationLess),
},
},
},
@ -318,6 +331,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and aggregate id",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetAggregateIDs("1234"),
@ -326,12 +340,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_AggregateID, "1234", repository.Operation_Equals),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "1234", repository.OperationEquals),
},
},
},
@ -339,6 +353,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and aggregate ids",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetAggregateIDs("1234", "0815"),
@ -347,12 +362,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_AggregateID, []string{"1234", "0815"}, repository.Operation_In),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, []string{"1234", "0815"}, repository.OperationIn),
},
},
},
@ -360,6 +375,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and sequence greater",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetSequence(8),
@ -368,12 +384,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_LatestSequence, uint64(8), repository.Operation_Greater),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(8), repository.OperationGreater),
},
},
},
@ -381,6 +397,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and event type",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetEventTypes("user.created"),
@ -389,12 +406,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_EventType, EventType("user.created"), repository.Operation_Equals),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, EventType("user.created"), repository.OperationEquals),
},
},
},
@ -402,6 +419,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and event types",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetEventTypes("user.created", "user.changed"),
@ -410,12 +428,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_EventType, []EventType{"user.created", "user.changed"}, repository.Operation_In),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, []EventType{"user.created", "user.changed"}, repository.OperationIn),
},
},
},
@ -423,6 +441,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type resource owner",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryFactory) *SearchQueryFactory{
testSetResourceOwner("hodor"),
@ -431,12 +450,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: 0,
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.Field_AggregateType, AggregateType("user"), repository.Operation_Equals),
repository.NewFilter(repository.Field_ResourceOwner, "hodor", repository.Operation_Equals),
repository.NewFilter(repository.FieldAggregateType, AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldResourceOwner, "hodor", repository.OperationEquals),
},
},
},
@ -444,11 +463,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
factory := NewSearchQueryFactory(tt.args.aggregateTypes...)
factory := NewSearchQueryFactory(tt.args.columns, tt.args.aggregateTypes...)
for _, f := range tt.args.setters {
factory = f(factory)
}
query, err := factory.Build()
query, err := factory.build()
if tt.res.isErr != nil && !tt.res.isErr(err) {
t.Errorf("wrong error: %v", err)
return