TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-11 19:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(eventstore): sub queries (#1805)

Browse Source 
* sub queries

* fix: tests

* add builder to tests

* new search query

* rename searchquerybuilder to builder

* remove comment from code

* test with multiple queries

* add filters test

* fix(contibute): listing

* add validate module

* fix: search queries

* remove unused event type in query

* ignore query if error in marshal

* go mod tidy

* update privacy policy query

* update queries

Co-authored-by: Livio Amstutz <livio.a@gmail.com>
This commit is contained in:
Silvan
2021-07-06 13:55:57 +02:00
committed by GitHub
parent 0e472a347f
commit 5349d96ce4
85 changed files with 1258 additions and 686 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/eventstore/asset.go
View File

@@ -14,8 +14,6 @@ type AssetAction int32
const (
AssetAdd AssetAction = iota
AssetRemove
assetActionCount
)
func NewAddAsset(

23
internal/eventstore/eventstore.go
View File

@@ -95,18 +95,6 @@ func uniqueConstraintsToRepository(constraints []*EventUniqueConstraint) (unique
return uniqueConstraints
}
func assetsToRepository(assets []*Asset) (result []*repository.Asset) {
result = make([]*repository.Asset, len(assets))
for i, asset := range assets {
result[i] = &repository.Asset{
ID: asset.ID,
Asset: asset.Asset,
Action: assetActionToRepository(asset.Action),
}
}
return result
}
//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 *SearchQueryBuilder) ([]EventReader, error) {
@@ -241,14 +229,3 @@ func uniqueConstraintActionToRepository(action UniqueConstraintAction) repositor
return repository.UniqueConstraintAdd
}
}
func assetActionToRepository(action AssetAction) repository.AssetAction {
switch action {
case AssetAdd:
return repository.AssetAdded
case AssetRemove:
return repository.AssetRemoved
default:
return repository.AssetAdded
}
}

126
internal/eventstore/eventstore_test.go
View File

@@ -13,31 +13,6 @@ import (
"github.com/caos/zitadel/internal/eventstore/repository"
)
type testAggregate struct {
id string
events []EventPusher
}
func (a *testAggregate) ID() string {
return a.id
}
func (a *testAggregate) Type() AggregateType {
return "test.aggregate"
}
func (a *testAggregate) Events() []EventPusher {
return a.events
}
func (a *testAggregate) ResourceOwner() string {
return "caos"
}
func (a *testAggregate) Version() Version {
return "v1"
}
// testEvent implements the Event interface
type testEvent struct {
BaseEvent
@@ -368,8 +343,7 @@ func Test_eventData(t *testing.T) {
func TestEventstore_aggregatesToEvents(t *testing.T) {
type args struct {
aggregates []Aggregate
events []EventPusher
events []EventPusher
}
type res struct {
wantErr bool
@@ -892,8 +866,13 @@ func TestEventstore_FilterEvents(t *testing.T) {
name: "no events",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"no.aggregates"},
columns: repository.ColumnsEvent,
columns: repository.ColumnsEvent,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"no.aggregates"},
},
},
},
},
fields: fields{
@@ -915,8 +894,13 @@ func TestEventstore_FilterEvents(t *testing.T) {
name: "repo error",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"no.aggregates"},
columns: repository.ColumnsEvent,
columns: repository.ColumnsEvent,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"no.aggregates"},
},
},
},
},
fields: fields{
@@ -938,8 +922,13 @@ func TestEventstore_FilterEvents(t *testing.T) {
name: "found events",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"test.aggregate"},
columns: repository.ColumnsEvent,
columns: repository.ColumnsEvent,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"test.aggregate"},
},
},
},
},
fields: fields{
@@ -1016,8 +1005,13 @@ func TestEventstore_LatestSequence(t *testing.T) {
name: "no events",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"no.aggregates"},
columns: repository.ColumnsMaxSequence,
columns: repository.ColumnsMaxSequence,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"no.aggregates"},
},
},
},
},
fields: fields{
@@ -1034,8 +1028,13 @@ func TestEventstore_LatestSequence(t *testing.T) {
name: "repo error",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"no.aggregates"},
columns: repository.ColumnsMaxSequence,
columns: repository.ColumnsMaxSequence,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"no.aggregates"},
},
},
},
},
fields: fields{
@@ -1052,8 +1051,13 @@ func TestEventstore_LatestSequence(t *testing.T) {
name: "found events",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"test.aggregate"},
columns: repository.ColumnsMaxSequence,
columns: repository.ColumnsMaxSequence,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"test.aggregate"},
},
},
},
},
fields: fields{
@@ -1134,8 +1138,13 @@ func TestEventstore_FilterToReducer(t *testing.T) {
name: "no events",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"no.aggregates"},
columns: repository.ColumnsEvent,
columns: repository.ColumnsEvent,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"no.aggregates"},
},
},
},
readModel: &testReducer{
t: t,
@@ -1161,8 +1170,13 @@ func TestEventstore_FilterToReducer(t *testing.T) {
name: "repo error",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"no.aggregates"},
columns: repository.ColumnsEvent,
columns: repository.ColumnsEvent,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"no.aggregates"},
},
},
},
readModel: &testReducer{
t: t,
@@ -1187,7 +1201,15 @@ func TestEventstore_FilterToReducer(t *testing.T) {
{
name: "found events",
args: args{
query: NewSearchQueryBuilder(repository.ColumnsEvent, "test.aggregate"),
query: &SearchQueryBuilder{
columns: repository.ColumnsEvent,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"test.aggregate"},
},
},
},
readModel: &testReducer{
t: t,
expectedLength: 1,
@@ -1214,8 +1236,13 @@ func TestEventstore_FilterToReducer(t *testing.T) {
name: "append in reducer fails",
args: args{
query: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"test.aggregate"},
columns: repository.ColumnsEvent,
columns: repository.ColumnsEvent,
queries: []*SearchQuery{
{
builder: &SearchQueryBuilder{},
aggregateTypes: []AggregateType{"test.aggregate"},
},
},
},
readModel: &testReducer{
t: t,
@@ -1275,13 +1302,6 @@ func combineEventLists(lists ...[]*repository.Event) []*repository.Event {
return events
}
func linkEvents(events ...*repository.Event) []*repository.Event {
for i := 1; i < len(events); i++ {
// events[i].PreviousEvent = events[i-1]
}
return events
}
func compareEvents(t *testing.T, want, got *repository.Event) {
t.Helper()

3
internal/eventstore/example_test.go
View File

@@ -311,7 +311,8 @@ func TestUserReadModel(t *testing.T) {
fmt.Printf("%+v\n", events)
users := UsersReadModel{}
err = es.FilterToReducer(context.Background(), eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent, "test.user"), &users)
err = es.FilterToReducer(context.Background(), eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).AddQuery().AggregateTypes("test.user").Builder(), &users)
if err != nil {
t.Errorf("unexpected error on filter to reducer: %v", err)
}

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

@@ -7,7 +7,7 @@ type SearchQuery struct {
Columns Columns
Limit uint64
Desc bool
Filters []*Filter
Filters [][]*Filter
}
//Columns defines which fields of the event are needed for the query

5
internal/eventstore/repository/sql/crdb.go
View File

@@ -13,9 +13,6 @@ import (
"github.com/caos/zitadel/internal/eventstore/repository"
"github.com/cockroachdb/cockroach-go/v2/crdb"
"github.com/lib/pq"
//sql import for cockroach
_ "github.com/lib/pq"
)
const (
@@ -163,7 +160,7 @@ func (db *CRDB) Push(ctx context.Context, events []*repository.Event, uniqueCons
// 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) {
if len(uniqueConstraints) == 0 || (len(uniqueConstraints) == 1 && uniqueConstraints[0] == nil) {
return nil
}

63
internal/eventstore/repository/sql/crdb_test.go
View File

@@ -5,10 +5,8 @@ import (
"sync"
"testing"
"github.com/lib/pq"
_ "github.com/lib/pq"
"github.com/caos/zitadel/internal/eventstore/repository"
"github.com/lib/pq"
)
func TestCRDB_placeholder(t *testing.T) {
@@ -749,8 +747,10 @@ func TestCRDB_Filter(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, "not found", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, "not found", repository.OperationEquals),
},
},
},
},
@@ -771,9 +771,11 @@ func TestCRDB_Filter(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "303", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "303", repository.OperationEquals),
},
},
},
},
@@ -837,8 +839,10 @@ func TestCRDB_LatestSequence(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsMaxSequence,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, "not found", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, "not found", repository.OperationEquals),
},
},
},
},
@@ -859,8 +863,10 @@ func TestCRDB_LatestSequence(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsMaxSequence,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
},
},
},
},
@@ -1093,20 +1099,6 @@ func generateEvent(t *testing.T, aggregateID string, opts ...func(*repository.Ev
return e
}
func generateEventWithData(t *testing.T, aggregateID string, data []byte) *repository.Event {
t.Helper()
return &repository.Event{
AggregateID: aggregateID,
AggregateType: repository.AggregateType(t.Name()),
EditorService: "svc",
EditorUser: "user",
ResourceOwner: "ro",
Type: "test.created",
Version: "v1",
Data: data,
}
}
func generateAddUniqueConstraint(t *testing.T, table, uniqueField string) *repository.UniqueConstraint {
t.Helper()
e := &repository.UniqueConstraint{
@@ -1128,22 +1120,3 @@ func generateRemoveUniqueConstraint(t *testing.T, table, uniqueField string) *re
return e
}
func generateAddAsset(t *testing.T, id string, asset []byte) *repository.Asset {
t.Helper()
e := &repository.Asset{
ID: id,
Asset: asset,
Action: repository.AssetAdded,
}
return e
}
func generateRemoveAsset(t *testing.T, id string) *repository.Asset {
t.Helper()
e := &repository.Asset{
ID: id,
Action: repository.AssetRemoved,
}
return e
}

5
internal/eventstore/repository/sql/local_crdb_test.go
View File

@@ -100,11 +100,6 @@ func fillUniqueData(unique_type, field string) error {
return err
}
func fillAssets(id string, asset []byte) error {
_, err := testCRDBClient.Exec("INSERT INTO eventstore.assets (id, asset) VALUES ($1, $2)", id, asset)
return err
}
type migrationPaths []string
type version struct {

48
internal/eventstore/repository/sql/query.go
View File

@@ -9,7 +9,6 @@ import (
"strings"
"github.com/caos/logging"
caos_errs "github.com/caos/zitadel/internal/errors"
z_errors "github.com/caos/zitadel/internal/errors"
"github.com/caos/zitadel/internal/eventstore/repository"
"github.com/lib/pq"
@@ -26,14 +25,13 @@ type querier interface {
orderByEventSequence(desc bool) string
}
type rowScan func(scan, interface{}) error
type scan func(dest ...interface{}) error
func query(ctx context.Context, criteria querier, searchQuery *repository.SearchQuery, dest interface{}) error {
query, rowScanner := prepareColumns(criteria, searchQuery.Columns)
where, values := prepareCondition(criteria, searchQuery.Filters)
if where == "" || query == "" {
return caos_errs.ThrowInvalidArgument(nil, "SQL-rWeBw", "invalid query factory")
return z_errors.ThrowInvalidArgument(nil, "SQL-rWeBw", "invalid query factory")
}
query += where
@@ -51,7 +49,7 @@ func query(ctx context.Context, criteria querier, searchQuery *repository.Search
rows, err := criteria.db().QueryContext(ctx, query, values...)
if err != nil {
logging.Log("SQL-HP3Uk").WithError(err).Info("query failed")
return caos_errs.ThrowInternal(err, "SQL-IJuyR", "unable to filter events")
return z_errors.ThrowInternal(err, "SQL-IJuyR", "unable to filter events")
}
defer rows.Close()
@@ -125,31 +123,37 @@ func eventsScanner(scanner scan, dest interface{}) (err error) {
return nil
}
func prepareCondition(criteria querier, filters []*repository.Filter) (clause string, values []interface{}) {
values = make([]interface{}, len(filters))
clauses := make([]string, len(filters))
func prepareCondition(criteria querier, filters [][]*repository.Filter) (clause string, values []interface{}) {
values = make([]interface{}, 0, 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)
case map[string]interface{}:
var err error
value, err = json.Marshal(value)
logging.Log("SQL-BSsNy").OnError(err).Warn("unable to marshal search value")
}
clauses[i] = getCondition(criteria, filter)
if clauses[i] == "" {
return "", nil
clauses := make([]string, len(filters))
for idx, filter := range filters {
subClauses := make([]string, 0, len(filter))
for _, f := range filter {
value := f.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)
case map[string]interface{}:
var err error
value, err = json.Marshal(value)
logging.Log("SQL-BSsNy").OnError(err).Warn("unable to marshal search value")
continue
}
subClauses = append(subClauses, getCondition(criteria, f))
if subClauses[len(subClauses)-1] == "" {
return "", nil
}
values = append(values, value)
}
values[i] = value
clauses[idx] = "( " + strings.Join(subClauses, " AND ") + " )"
}
return " WHERE " + strings.Join(clauses, " AND "), values
return " WHERE " + strings.Join(clauses, " OR "), values
}
func getCondition(cond querier, filter *repository.Filter) (condition string) {

190
internal/eventstore/repository/sql/query_test.go
View File

@@ -208,7 +208,7 @@ func prepareTestScan(err error, res []interface{}) scan {
func Test_prepareCondition(t *testing.T) {
type args struct {
filters []*repository.Filter
filters [][]*repository.Filter
}
type res struct {
clause string
@@ -232,7 +232,7 @@ func Test_prepareCondition(t *testing.T) {
{
name: "empty filters",
args: args{
filters: []*repository.Filter{},
filters: [][]*repository.Filter{},
},
res: res{
clause: "",
@@ -242,8 +242,10 @@ func Test_prepareCondition(t *testing.T) {
{
name: "invalid condition",
args: args{
filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateID, "wrong", repository.Operation(-1)),
filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateID, "wrong", repository.Operation(-1)),
},
},
},
res: res{
@@ -254,26 +256,30 @@ func Test_prepareCondition(t *testing.T) {
{
name: "array as condition value",
args: args{
filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"user", "org"}, repository.OperationIn),
filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"user", "org"}, repository.OperationIn),
},
},
},
res: res{
clause: " WHERE aggregate_type = ANY(?)",
clause: " WHERE ( aggregate_type = ANY(?) )",
values: []interface{}{pq.Array([]repository.AggregateType{"user", "org"})},
},
},
{
name: "multiple filters",
args: args{
filters: []*repository.Filter{
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),
filters: [][]*repository.Filter{
{
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{
clause: " WHERE aggregate_type = ANY(?) AND aggregate_id = ? AND event_type = ANY(?)",
clause: " WHERE ( aggregate_type = ANY(?) AND aggregate_id = ? AND event_type = ANY(?) )",
values: []interface{}{pq.Array([]repository.AggregateType{"user", "org"}), "1234", pq.Array([]repository.EventType{"user.created", "org.created"})},
},
},
@@ -304,7 +310,6 @@ func Test_query_events_with_crdb(t *testing.T) {
}
type fields struct {
existingEvents []*repository.Event
existingAssets []*repository.Asset
client *sql.DB
}
type res struct {
@@ -322,8 +327,10 @@ func Test_query_events_with_crdb(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, "not found", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, "not found", repository.OperationEquals),
},
},
},
},
@@ -345,8 +352,10 @@ func Test_query_events_with_crdb(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
},
},
},
},
@@ -369,9 +378,11 @@ func Test_query_events_with_crdb(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "303", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, t.Name(), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "303", repository.OperationEquals),
},
},
},
},
@@ -395,8 +406,10 @@ func Test_query_events_with_crdb(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldResourceOwner, "caos", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldResourceOwner, "caos", repository.OperationEquals),
},
},
},
},
@@ -420,9 +433,11 @@ func Test_query_events_with_crdb(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldEditorService, "MANAGEMENT-API", repository.OperationEquals),
repository.NewFilter(repository.FieldEditorService, "ADMIN-API", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldEditorService, "MANAGEMENT-API", repository.OperationEquals),
repository.NewFilter(repository.FieldEditorService, "ADMIN-API", repository.OperationEquals),
},
},
},
},
@@ -446,10 +461,12 @@ func Test_query_events_with_crdb(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldEditorUser, "adlerhurst", repository.OperationEquals),
repository.NewFilter(repository.FieldEditorUser, "nobody", repository.OperationEquals),
repository.NewFilter(repository.FieldEditorUser, "", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldEditorUser, "adlerhurst", repository.OperationEquals),
repository.NewFilter(repository.FieldEditorUser, "nobody", repository.OperationEquals),
repository.NewFilter(repository.FieldEditorUser, "", repository.OperationEquals),
},
},
},
},
@@ -475,9 +492,11 @@ func Test_query_events_with_crdb(t *testing.T) {
args: args{
searchQuery: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldEventType, repository.EventType("user.created"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, repository.EventType("user.updated"), repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldEventType, repository.EventType("user.created"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, repository.EventType("user.updated"), repository.OperationEquals),
},
},
},
},
@@ -559,18 +578,20 @@ func Test_query_events_mocked(t *testing.T) {
query: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Desc: true,
Filters: []*repository.Filter{
Filters: [][]*repository.Filter{
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
},
},
},
},
},
fields: fields{
mock: newMockClient(t).expectQuery(t,
`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`,
`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`,
[]driver.Value{repository.AggregateType("user")},
),
},
@@ -586,18 +607,20 @@ func Test_query_events_mocked(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 5,
Filters: []*repository.Filter{
Filters: [][]*repository.Filter{
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
},
},
},
},
},
fields: fields{
mock: newMockClient(t).expectQuery(t,
`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`,
`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`,
[]driver.Value{repository.AggregateType("user"), uint64(5)},
),
},
@@ -613,18 +636,20 @@ func Test_query_events_mocked(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: true,
Limit: 5,
Filters: []*repository.Filter{
Filters: [][]*repository.Filter{
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
},
},
},
},
},
fields: fields{
mock: newMockClient(t).expectQuery(t,
`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`,
`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`,
[]driver.Value{repository.AggregateType("user"), uint64(5)},
),
},
@@ -640,18 +665,20 @@ func Test_query_events_mocked(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: true,
Limit: 0,
Filters: []*repository.Filter{
Filters: [][]*repository.Filter{
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
},
},
},
},
},
fields: fields{
mock: newMockClient(t).expectQueryErr(t,
`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`,
`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`,
[]driver.Value{repository.AggregateType("user")},
sql.ErrConnDone),
},
@@ -667,18 +694,20 @@ func Test_query_events_mocked(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: true,
Limit: 0,
Filters: []*repository.Filter{
Filters: [][]*repository.Filter{
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
},
},
},
},
},
fields: fields{
mock: newMockClient(t).expectQuery(t,
`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`,
`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`,
[]driver.Value{repository.AggregateType("user")},
&repository.Event{Sequence: 100}),
},
@@ -702,8 +731,10 @@ func Test_query_events_mocked(t *testing.T) {
args: args{
query: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Filters: []*repository.Filter{
{},
Filters: [][]*repository.Filter{
{
{},
},
},
},
},
@@ -711,6 +742,47 @@ func Test_query_events_mocked(t *testing.T) {
wantErr: true,
},
},
{
name: "with subqueries",
args: args{
dest: &[]*repository.Event{},
query: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Desc: true,
Limit: 5,
Filters: [][]*repository.Filter{
{
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("user"),
Operation: repository.OperationEquals,
},
},
{
{
Field: repository.FieldAggregateType,
Value: repository.AggregateType("org"),
Operation: repository.OperationEquals,
},
{
Field: repository.FieldAggregateID,
Value: "asdf42",
Operation: repository.OperationEquals,
},
},
},
},
},
fields: fields{
mock: newMockClient(t).expectQuery(t,
`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 \) OR \( aggregate_type = \$2 AND aggregate_id = \$3 \) ORDER BY event_sequence DESC LIMIT \$4`,
[]driver.Value{repository.AggregateType("user"), repository.AggregateType("org"), "asdf42", uint64(5)},
),
},
res: res{
wantErr: false,
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {

170
internal/eventstore/search_query.go
View File

@@ -8,15 +8,20 @@ import (
//SearchQueryBuilder represents the builder for your filter
// if invalid data are set the filter will fail
type SearchQueryBuilder struct {
columns repository.Columns
limit uint64
desc bool
columns repository.Columns
limit uint64
desc bool
resourceOwner string
queries []*SearchQuery
}
type SearchQuery struct {
builder *SearchQueryBuilder
aggregateTypes []AggregateType
aggregateIDs []string
eventSequence uint64
eventTypes []EventType
eventData map[string]interface{}
resourceOwner string
}
// Columns defines which fields of the event are needed for the query
@@ -37,81 +42,120 @@ type EventType repository.EventType
// NewSearchQueryBuilder creates a new factory for event filters
// aggregateTypes must contain at least one aggregate type
func NewSearchQueryBuilder(columns Columns, aggregateTypes ...AggregateType) *SearchQueryBuilder {
func NewSearchQueryBuilder(columns Columns) *SearchQueryBuilder {
return &SearchQueryBuilder{
columns: repository.Columns(columns),
aggregateTypes: aggregateTypes,
columns: repository.Columns(columns),
}
}
//Columns defines which fields are set
func (factory *SearchQueryBuilder) Columns(columns Columns) *SearchQueryBuilder {
factory.columns = repository.Columns(columns)
return factory
}
//Limit defines how many events are returned maximally.
func (factory *SearchQueryBuilder) Limit(limit uint64) *SearchQueryBuilder {
factory.limit = limit
return factory
}
func (factory *SearchQueryBuilder) SequenceGreater(sequence uint64) *SearchQueryBuilder {
factory.eventSequence = sequence
return factory
}
func (factory *SearchQueryBuilder) AggregateIDs(ids ...string) *SearchQueryBuilder {
factory.aggregateIDs = ids
return factory
}
func (factory *SearchQueryBuilder) EventTypes(types ...EventType) *SearchQueryBuilder {
factory.eventTypes = types
return factory
}
//ResourceOwner defines the resource owner (org) of the events
func (factory *SearchQueryBuilder) ResourceOwner(resourceOwner string) *SearchQueryBuilder {
factory.resourceOwner = resourceOwner
return factory
}
//OrderDesc changes the sorting order of the returned events to descending
func (factory *SearchQueryBuilder) OrderDesc() *SearchQueryBuilder {
factory.desc = true
return factory
}
//OrderAsc changes the sorting order of the returned events to ascending
func (factory *SearchQueryBuilder) OrderAsc() *SearchQueryBuilder {
factory.desc = false
return factory
}
func (factory *SearchQueryBuilder) EventData(query map[string]interface{}) *SearchQueryBuilder {
factory.eventData = query
return factory
//AddQuery creates a new sub query.
//All fields in the sub query are AND-connected in the storage request.
//Multiple sub queries are OR-connected in the storage request.
func (factory *SearchQueryBuilder) AddQuery() *SearchQuery {
query := &SearchQuery{
builder: factory,
}
factory.queries = append(factory.queries, query)
return query
}
//Or creates a new sub query on the search query builder
func (query SearchQuery) Or() *SearchQuery {
return query.builder.AddQuery()
}
//AggregateTypes filters for events with the given aggregate types
func (query *SearchQuery) AggregateTypes(types ...AggregateType) *SearchQuery {
query.aggregateTypes = types
return query
}
//SequenceGreater filters for events with sequence greater the requested sequence
func (query *SearchQuery) SequenceGreater(sequence uint64) *SearchQuery {
query.eventSequence = sequence
return query
}
//AggregateIDs filters for events with the given aggregate id's
func (query *SearchQuery) AggregateIDs(ids ...string) *SearchQuery {
query.aggregateIDs = ids
return query
}
//EventTypes filters for events with the given event types
func (query *SearchQuery) EventTypes(types ...EventType) *SearchQuery {
query.eventTypes = types
return query
}
//EventData filters for events with the given event data.
//Use this call with care as it will be slower than the other filters.
func (query *SearchQuery) EventData(data map[string]interface{}) *SearchQuery {
query.eventData = data
return query
}
//Builder returns the SearchQueryBuilder of the sub query
func (query *SearchQuery) Builder() *SearchQueryBuilder {
return query.builder
}
func (factory *SearchQueryBuilder) build() (*repository.SearchQuery, error) {
if factory == nil ||
len(factory.aggregateTypes) < 1 ||
len(factory.queries) < 1 ||
factory.columns.Validate() != nil {
return nil, errors.ThrowPreconditionFailed(nil, "MODEL-4m9gs", "factory invalid")
}
filters := []*repository.Filter{
factory.aggregateTypeFilter(),
}
filters := make([][]*repository.Filter, len(factory.queries))
for _, f := range []func() *repository.Filter{
factory.aggregateIDFilter,
factory.eventSequenceFilter,
factory.eventTypeFilter,
factory.resourceOwnerFilter,
factory.eventDataFilter,
} {
if filter := f(); filter != nil {
if err := filter.Validate(); err != nil {
return nil, err
for i, query := range factory.queries {
for _, f := range []func() *repository.Filter{
query.aggregateTypeFilter,
query.aggregateIDFilter,
query.eventSequenceFilter,
query.eventTypeFilter,
query.eventDataFilter,
query.builder.resourceOwnerFilter,
} {
if filter := f(); filter != nil {
if err := filter.Validate(); err != nil {
return nil, err
}
filters[i] = append(filters[i], filter)
}
filters = append(filters, filter)
}
}
return &repository.SearchQuery{
@@ -122,50 +166,50 @@ func (factory *SearchQueryBuilder) build() (*repository.SearchQuery, error) {
}, nil
}
func (factory *SearchQueryBuilder) aggregateIDFilter() *repository.Filter {
if len(factory.aggregateIDs) < 1 {
func (query *SearchQuery) aggregateIDFilter() *repository.Filter {
if len(query.aggregateIDs) < 1 {
return nil
}
if len(factory.aggregateIDs) == 1 {
return repository.NewFilter(repository.FieldAggregateID, factory.aggregateIDs[0], repository.OperationEquals)
if len(query.aggregateIDs) == 1 {
return repository.NewFilter(repository.FieldAggregateID, query.aggregateIDs[0], repository.OperationEquals)
}
return repository.NewFilter(repository.FieldAggregateID, factory.aggregateIDs, repository.OperationIn)
return repository.NewFilter(repository.FieldAggregateID, query.aggregateIDs, repository.OperationIn)
}
func (factory *SearchQueryBuilder) eventTypeFilter() *repository.Filter {
if len(factory.eventTypes) < 1 {
func (query *SearchQuery) eventTypeFilter() *repository.Filter {
if len(query.eventTypes) < 1 {
return nil
}
if len(factory.eventTypes) == 1 {
return repository.NewFilter(repository.FieldEventType, repository.EventType(factory.eventTypes[0]), repository.OperationEquals)
if len(query.eventTypes) == 1 {
return repository.NewFilter(repository.FieldEventType, repository.EventType(query.eventTypes[0]), repository.OperationEquals)
}
eventTypes := make([]repository.EventType, len(factory.eventTypes))
for i, eventType := range factory.eventTypes {
eventTypes := make([]repository.EventType, len(query.eventTypes))
for i, eventType := range query.eventTypes {
eventTypes[i] = repository.EventType(eventType)
}
return repository.NewFilter(repository.FieldEventType, eventTypes, repository.OperationIn)
}
func (factory *SearchQueryBuilder) aggregateTypeFilter() *repository.Filter {
if len(factory.aggregateTypes) == 1 {
return repository.NewFilter(repository.FieldAggregateType, repository.AggregateType(factory.aggregateTypes[0]), repository.OperationEquals)
func (query *SearchQuery) aggregateTypeFilter() *repository.Filter {
if len(query.aggregateTypes) == 1 {
return repository.NewFilter(repository.FieldAggregateType, repository.AggregateType(query.aggregateTypes[0]), repository.OperationEquals)
}
aggregateTypes := make([]repository.AggregateType, len(factory.aggregateTypes))
for i, aggregateType := range factory.aggregateTypes {
aggregateTypes := make([]repository.AggregateType, len(query.aggregateTypes))
for i, aggregateType := range query.aggregateTypes {
aggregateTypes[i] = repository.AggregateType(aggregateType)
}
return repository.NewFilter(repository.FieldAggregateType, aggregateTypes, repository.OperationIn)
}
func (factory *SearchQueryBuilder) eventSequenceFilter() *repository.Filter {
if factory.eventSequence == 0 {
func (query *SearchQuery) eventSequenceFilter() *repository.Filter {
if query.eventSequence == 0 {
return nil
}
sortOrder := repository.OperationGreater
if factory.desc {
if query.builder.desc {
sortOrder = repository.OperationLess
}
return repository.NewFilter(repository.FieldSequence, factory.eventSequence, sortOrder)
return repository.NewFilter(repository.FieldSequence, query.eventSequence, sortOrder)
}
func (factory *SearchQueryBuilder) resourceOwnerFilter() *repository.Filter {
@@ -175,9 +219,9 @@ func (factory *SearchQueryBuilder) resourceOwnerFilter() *repository.Filter {
return repository.NewFilter(repository.FieldResourceOwner, factory.resourceOwner, repository.OperationEquals)
}
func (factory *SearchQueryBuilder) eventDataFilter() *repository.Filter {
if len(factory.eventData) == 0 {
func (query *SearchQuery) eventDataFilter() *repository.Filter {
if len(query.eventData) == 0 {
return nil
}
return repository.NewFilter(repository.FieldEventData, factory.eventData, repository.OperationJSONContains)
return repository.NewFilter(repository.FieldEventData, query.eventData, repository.OperationJSONContains)
}

477
internal/eventstore/search_query_test.go
View File

@@ -9,6 +9,16 @@ import (
"github.com/caos/zitadel/internal/eventstore/repository"
)
func testAddQuery(queryFuncs ...func(*SearchQuery) *SearchQuery) func(*SearchQueryBuilder) *SearchQueryBuilder {
return func(builder *SearchQueryBuilder) *SearchQueryBuilder {
query := builder.AddQuery()
for _, queryFunc := range queryFuncs {
queryFunc(query)
}
return query.Builder()
}
}
func testSetColumns(columns Columns) func(factory *SearchQueryBuilder) *SearchQueryBuilder {
return func(factory *SearchQueryBuilder) *SearchQueryBuilder {
factory = factory.Columns(columns)
@@ -23,50 +33,66 @@ func testSetLimit(limit uint64) func(factory *SearchQueryBuilder) *SearchQueryBu
}
}
func testSetSequence(sequence uint64) func(factory *SearchQueryBuilder) *SearchQueryBuilder {
return func(factory *SearchQueryBuilder) *SearchQueryBuilder {
factory = factory.SequenceGreater(sequence)
return factory
}
}
func testSetAggregateIDs(aggregateIDs ...string) func(factory *SearchQueryBuilder) *SearchQueryBuilder {
return func(factory *SearchQueryBuilder) *SearchQueryBuilder {
factory = factory.AggregateIDs(aggregateIDs...)
return factory
}
}
func testSetEventTypes(eventTypes ...EventType) func(factory *SearchQueryBuilder) *SearchQueryBuilder {
return func(factory *SearchQueryBuilder) *SearchQueryBuilder {
factory = factory.EventTypes(eventTypes...)
return factory
}
}
func testSetResourceOwner(resourceOwner string) func(factory *SearchQueryBuilder) *SearchQueryBuilder {
return func(factory *SearchQueryBuilder) *SearchQueryBuilder {
factory = factory.ResourceOwner(resourceOwner)
return factory
}
}
func testSetSortOrder(asc bool) func(factory *SearchQueryBuilder) *SearchQueryBuilder {
return func(factory *SearchQueryBuilder) *SearchQueryBuilder {
if asc {
factory = factory.OrderAsc()
} else {
factory = factory.OrderDesc()
func testOr(queryFuncs ...func(*SearchQuery) *SearchQuery) func(*SearchQuery) *SearchQuery {
return func(query *SearchQuery) *SearchQuery {
subQuery := query.Or()
for _, queryFunc := range queryFuncs {
queryFunc(subQuery)
}
return factory
return subQuery
}
}
func testSetAggregateTypes(types ...AggregateType) func(*SearchQuery) *SearchQuery {
return func(query *SearchQuery) *SearchQuery {
query = query.AggregateTypes(types...)
return query
}
}
func testSetSequence(sequence uint64) func(*SearchQuery) *SearchQuery {
return func(query *SearchQuery) *SearchQuery {
query = query.SequenceGreater(sequence)
return query
}
}
func testSetAggregateIDs(aggregateIDs ...string) func(*SearchQuery) *SearchQuery {
return func(query *SearchQuery) *SearchQuery {
query = query.AggregateIDs(aggregateIDs...)
return query
}
}
func testSetEventTypes(eventTypes ...EventType) func(*SearchQuery) *SearchQuery {
return func(query *SearchQuery) *SearchQuery {
query = query.EventTypes(eventTypes...)
return query
}
}
func testSetResourceOwner(resourceOwner string) func(*SearchQueryBuilder) *SearchQueryBuilder {
return func(builder *SearchQueryBuilder) *SearchQueryBuilder {
builder = builder.ResourceOwner(resourceOwner)
return builder
}
}
func testSetSortOrder(asc bool) func(*SearchQueryBuilder) *SearchQueryBuilder {
return func(query *SearchQueryBuilder) *SearchQueryBuilder {
if asc {
query = query.OrderAsc()
} else {
query = query.OrderDesc()
}
return query
}
}
func TestSearchQueryFactorySetters(t *testing.T) {
type args struct {
columns Columns
aggregateTypes []AggregateType
setters []func(*SearchQueryBuilder) *SearchQueryBuilder
columns Columns
setters []func(*SearchQueryBuilder) *SearchQueryBuilder
}
tests := []struct {
name string
@@ -76,12 +102,10 @@ func TestSearchQueryFactorySetters(t *testing.T) {
{
name: "New factory",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user", "org"},
columns: ColumnsEvent,
},
res: &SearchQueryBuilder{
columns: repository.Columns(ColumnsEvent),
aggregateTypes: []AggregateType{"user", "org"},
columns: repository.Columns(ColumnsEvent),
},
},
{
@@ -105,28 +129,40 @@ func TestSearchQueryFactorySetters(t *testing.T) {
{
name: "set sequence",
args: args{
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testSetSequence(90)},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testAddQuery(testSetSequence(90))},
},
res: &SearchQueryBuilder{
eventSequence: 90,
queries: []*SearchQuery{
{
eventSequence: 90,
},
},
},
},
{
name: "set aggregateIDs",
args: args{
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testSetAggregateIDs("1235", "09824")},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testAddQuery(testSetAggregateIDs("1235", "09824"))},
},
res: &SearchQueryBuilder{
aggregateIDs: []string{"1235", "09824"},
queries: []*SearchQuery{
{
aggregateIDs: []string{"1235", "09824"},
},
},
},
},
{
name: "set eventTypes",
args: args{
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testSetEventTypes("user.created", "user.updated")},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testAddQuery(testSetEventTypes("user.created", "user.updated"))},
},
res: &SearchQueryBuilder{
eventTypes: []EventType{"user.created", "user.updated"},
queries: []*SearchQuery{
{
eventTypes: []EventType{"user.created", "user.updated"},
},
},
},
},
{
@@ -141,34 +177,35 @@ func TestSearchQueryFactorySetters(t *testing.T) {
{
name: "default search query",
args: args{
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testSetAggregateIDs("1235", "024"), testSetSortOrder(false)},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{testAddQuery(testSetAggregateTypes("user"), testSetAggregateIDs("1235", "024")), testSetSortOrder(false)},
},
res: &SearchQueryBuilder{
aggregateTypes: []AggregateType{"user"},
aggregateIDs: []string{"1235", "024"},
desc: true,
desc: true,
queries: []*SearchQuery{
{
aggregateTypes: []AggregateType{"user"},
aggregateIDs: []string{"1235", "024"},
},
},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
factory := NewSearchQueryBuilder(tt.args.columns, tt.args.aggregateTypes...)
builder := NewSearchQueryBuilder(tt.args.columns)
for _, setter := range tt.args.setters {
factory = setter(factory)
}
if !reflect.DeepEqual(factory, tt.res) {
t.Errorf("NewSearchQueryFactory() = %v, want %v", factory, tt.res)
builder = setter(builder)
}
assertBuilder(t, tt.res, builder)
})
}
}
func TestSearchQueryFactoryBuild(t *testing.T) {
type args struct {
columns Columns
aggregateTypes []AggregateType
setters []func(*SearchQueryBuilder) *SearchQueryBuilder
columns Columns
setters []func(*SearchQueryBuilder) *SearchQueryBuilder
}
type res struct {
isErr func(err error) bool
@@ -182,9 +219,8 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "no aggregate types",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{},
},
res: res{
isErr: errors.IsPreconditionFailed,
@@ -194,10 +230,10 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "invalid column (too low)",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetColumns(Columns(-1)),
testAddQuery(testSetAggregateTypes("user")),
},
},
res: res{
@@ -207,10 +243,10 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "invalid column (too high)",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetColumns(math.MaxInt32),
testAddQuery(testSetAggregateTypes("uesr")),
},
},
res: res{
@@ -220,9 +256,10 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testAddQuery(testSetAggregateTypes("user")),
},
},
res: res{
isErr: nil,
@@ -230,8 +267,10 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
},
},
},
},
@@ -239,9 +278,10 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate types",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user", "org"},
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testAddQuery(testSetAggregateTypes("user", "org")),
},
},
res: res{
isErr: nil,
@@ -249,8 +289,10 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"user", "org"}, repository.OperationIn),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, []repository.AggregateType{"user", "org"}, repository.OperationIn),
},
},
},
},
@@ -258,12 +300,14 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type, limit, desc",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetLimit(5),
testSetSortOrder(false),
testSetSequence(100),
testAddQuery(
testSetSequence(100),
testSetAggregateTypes("user"),
),
},
},
res: res{
@@ -272,9 +316,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: true,
Limit: 5,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationLess),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationLess),
},
},
},
},
@@ -282,12 +328,14 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type, limit, asc",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetLimit(5),
testSetSortOrder(true),
testSetSequence(100),
testAddQuery(
testSetSequence(100),
testSetAggregateTypes("user"),
),
},
},
res: res{
@@ -296,9 +344,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 5,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationGreater),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationGreater),
},
},
},
},
@@ -306,13 +356,15 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type, limit, desc, max event sequence cols",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetLimit(5),
testSetSortOrder(false),
testSetSequence(100),
testSetColumns(repository.ColumnsMaxSequence),
testAddQuery(
testSetSequence(100),
testSetAggregateTypes("user"),
),
},
},
res: res{
@@ -321,9 +373,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsMaxSequence,
Desc: true,
Limit: 5,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationLess),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(100), repository.OperationLess),
},
},
},
},
@@ -331,10 +385,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and aggregate id",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetAggregateIDs("1234"),
testAddQuery(
testSetAggregateTypes("user"),
testSetAggregateIDs("1234"),
),
},
},
res: res{
@@ -343,9 +399,45 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "1234", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "1234", repository.OperationEquals),
},
},
},
},
},
{
name: "filter multiple aggregate type and aggregate id",
args: args{
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testAddQuery(
testSetAggregateTypes("user"),
testSetAggregateIDs("1234"),
testOr(
testSetAggregateTypes("org"),
testSetAggregateIDs("izu"),
),
),
},
},
res: res{
isErr: nil,
query: &repository.SearchQuery{
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "1234", repository.OperationEquals),
},
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("org"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, "izu", repository.OperationEquals),
},
},
},
},
@@ -353,10 +445,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and aggregate ids",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetAggregateIDs("1234", "0815"),
testAddQuery(
testSetAggregateTypes("user"),
testSetAggregateIDs("1234", "0815"),
),
},
},
res: res{
@@ -365,9 +459,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, []string{"1234", "0815"}, repository.OperationIn),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldAggregateID, []string{"1234", "0815"}, repository.OperationIn),
},
},
},
},
@@ -375,10 +471,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and sequence greater",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetSequence(8),
testAddQuery(
testSetAggregateTypes("user"),
testSetSequence(8),
),
},
},
res: res{
@@ -387,9 +485,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(8), repository.OperationGreater),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldSequence, uint64(8), repository.OperationGreater),
},
},
},
},
@@ -397,10 +497,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and event type",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetEventTypes("user.created"),
testAddQuery(
testSetAggregateTypes("user"),
testSetEventTypes("user.created"),
),
},
},
res: res{
@@ -409,9 +511,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, repository.EventType("user.created"), repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, repository.EventType("user.created"), repository.OperationEquals),
},
},
},
},
@@ -419,10 +523,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type and event types",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetEventTypes("user.created", "user.changed"),
testAddQuery(
testSetAggregateTypes("user"),
testSetEventTypes("user.created", "user.changed"),
),
},
},
res: res{
@@ -431,9 +537,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, []repository.EventType{"user.created", "user.changed"}, repository.OperationIn),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldEventType, []repository.EventType{"user.created", "user.changed"}, repository.OperationIn),
},
},
},
},
@@ -441,10 +549,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "filter aggregate type resource owner",
args: args{
columns: ColumnsEvent,
aggregateTypes: []AggregateType{"user"},
columns: ColumnsEvent,
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testSetResourceOwner("hodor"),
testAddQuery(
testSetAggregateTypes("user"),
),
},
},
res: res{
@@ -453,9 +563,11 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
Columns: repository.ColumnsEvent,
Desc: false,
Limit: 0,
Filters: []*repository.Filter{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldResourceOwner, "hodor", repository.OperationEquals),
Filters: [][]*repository.Filter{
{
repository.NewFilter(repository.FieldAggregateType, repository.AggregateType("user"), repository.OperationEquals),
repository.NewFilter(repository.FieldResourceOwner, "hodor", repository.OperationEquals),
},
},
},
},
@@ -463,8 +575,12 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
{
name: "column invalid",
args: args{
columns: Columns(-1),
aggregateTypes: []AggregateType{"user"},
columns: Columns(-1),
setters: []func(*SearchQueryBuilder) *SearchQueryBuilder{
testAddQuery(
testSetAggregateTypes("user"),
),
},
},
res: res{
isErr: errors.IsPreconditionFailed,
@@ -473,7 +589,7 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
factory := NewSearchQueryBuilder(tt.args.columns, tt.args.aggregateTypes...)
factory := NewSearchQueryBuilder(tt.args.columns)
for _, f := range tt.args.setters {
factory = f(factory)
}
@@ -487,9 +603,96 @@ func TestSearchQueryFactoryBuild(t *testing.T) {
return
}
if !reflect.DeepEqual(query, tt.res.query) {
t.Errorf("NewSearchQueryFactory() = %+v, want %+v", factory, tt.res.query)
}
assertRepoQuery(t, tt.res.query, query)
})
}
}
func assertBuilder(t *testing.T, want, got *SearchQueryBuilder) {
t.Helper()
if got.columns != want.columns {
t.Errorf("wrong column: got: %v want: %v", got.columns, want.columns)
}
if got.desc != want.desc {
t.Errorf("wrong desc: got: %v want: %v", got.desc, want.desc)
}
if got.limit != want.limit {
t.Errorf("wrong limit: got: %v want: %v", got.limit, want.limit)
}
if got.resourceOwner != want.resourceOwner {
t.Errorf("wrong : got: %v want: %v", got.resourceOwner, want.resourceOwner)
}
if len(got.queries) != len(want.queries) {
t.Errorf("wrong length of queries: got: %v want: %v", len(got.queries), len(want.queries))
}
for i, query := range got.queries {
assertQuery(t, i, want.queries[i], query)
}
}
func assertQuery(t *testing.T, i int, want, got *SearchQuery) {
t.Helper()
if !reflect.DeepEqual(got.aggregateIDs, want.aggregateIDs) {
t.Errorf("wrong aggregateIDs in query %d : got: %v want: %v", i, got.aggregateIDs, want.aggregateIDs)
}
if !reflect.DeepEqual(got.aggregateTypes, want.aggregateTypes) {
t.Errorf("wrong aggregateTypes in query %d : got: %v want: %v", i, got.aggregateTypes, want.aggregateTypes)
}
if !reflect.DeepEqual(got.eventData, want.eventData) {
t.Errorf("wrong eventData in query %d : got: %v want: %v", i, got.eventData, want.eventData)
}
if got.eventSequence != want.eventSequence {
t.Errorf("wrong eventSequence in query %d : got: %v want: %v", i, got.eventSequence, want.eventSequence)
}
if !reflect.DeepEqual(got.eventTypes, want.eventTypes) {
t.Errorf("wrong eventTypes in query %d : got: %v want: %v", i, got.eventTypes, want.eventTypes)
}
}
func assertRepoQuery(t *testing.T, want, got *repository.SearchQuery) {
t.Helper()
if want == nil && got == nil {
return
}
if !reflect.DeepEqual(got.Columns, want.Columns) {
t.Errorf("wrong columns in query: got: %v want: %v", got.Columns, want.Columns)
}
if !reflect.DeepEqual(got.Desc, want.Desc) {
t.Errorf("wrong desc in query: got: %v want: %v", got.Desc, want.Desc)
}
if !reflect.DeepEqual(got.Limit, want.Limit) {
t.Errorf("wrong limit in query: got: %v want: %v", got.Limit, want.Limit)
}
if len(got.Filters) != len(want.Filters) {
t.Errorf("wrong length of filters: got: %v want: %v", len(got.Filters), len(want.Filters))
}
for filterIdx, filter := range got.Filters {
if len(got.Filters) != len(want.Filters) {
t.Errorf("wrong length of subfilters: got: %v want: %v", len(filter), len(want.Filters[filterIdx]))
}
for subFilterIdx, f := range filter {
assertFilters(t, subFilterIdx, want.Filters[filterIdx][subFilterIdx], f)
}
}
}
func assertFilters(t *testing.T, i int, want, got *repository.Filter) {
t.Helper()
if want.Field != got.Field {
t.Errorf("wrong field in filter %d : got: %v want: %v", i, got.Field, want.Field)
}
if want.Operation != got.Operation {
t.Errorf("wrong operation in filter %d : got: %v want: %v", i, got.Operation, want.Operation)
}
if !reflect.DeepEqual(want.Value, got.Value) {
t.Errorf("wrong value in filter %d : got: %v want: %v", i, got.Value, want.Value)
}
}

3
internal/eventstore/subscription.go
View File

@@ -1,9 +1,10 @@
package eventstore
import (
"sync"
v1 "github.com/caos/zitadel/internal/eventstore/v1"
"github.com/caos/zitadel/internal/eventstore/v1/models"
"sync"
)
var (

2
internal/eventstore/unique_constraint.go
View File

@@ -16,8 +16,6 @@ type UniqueConstraintAction int32
const (
UniqueConstraintAdd UniqueConstraintAction = iota
UniqueConstraintRemove
uniqueConstraintActionCount
)
func NewAddEventUniqueConstraint(