From f8028f07d0a0814d139993e3df5d9efaf9111924 Mon Sep 17 00:00:00 2001 From: adlerhurst Date: Mon, 23 Nov 2020 19:31:12 +0100 Subject: [PATCH] event data search query --- .../eventstore/v2/repository/search_query.go | 4 +++ internal/eventstore/v2/repository/sql/crdb.go | 4 +++ .../eventstore/v2/repository/sql/query.go | 5 +++ internal/eventstore/v2/search_query.go | 14 ++++++++ internal/v2/business/iam/converter.go | 18 ++++++++++ internal/v2/business/iam/member.go | 36 +++++++++++++------ internal/v2/repository/iam/aggregate.go | 12 ++++--- internal/v2/repository/iam/member.go | 28 ++++++++++++--- .../v2/repository/member/event_changed.go | 30 ++++++++++------ internal/v2/repository/member/write_model.go | 3 ++ internal/v2/repository/org/member.go | 31 ++++++++++++---- 11 files changed, 148 insertions(+), 37 deletions(-) diff --git a/internal/eventstore/v2/repository/search_query.go b/internal/eventstore/v2/repository/search_query.go index 194d2467f6..927eb37c9a 100644 --- a/internal/eventstore/v2/repository/search_query.go +++ b/internal/eventstore/v2/repository/search_query.go @@ -48,6 +48,8 @@ const ( OperationLess //OperationIn checks if a stored value matches one of the passed value list OperationIn + //OperationJSONContains checks if a stored value matches the given json + OperationJSONContains operationCount ) @@ -70,6 +72,8 @@ const ( FieldEditorUser //FieldEventType represents the event type field FieldEventType + //FieldEventData represents the event data field + FieldEventData fieldCount ) diff --git a/internal/eventstore/v2/repository/sql/crdb.go b/internal/eventstore/v2/repository/sql/crdb.go index aac3fcc193..869accd282 100644 --- a/internal/eventstore/v2/repository/sql/crdb.go +++ b/internal/eventstore/v2/repository/sql/crdb.go @@ -252,6 +252,8 @@ func (db *CRDB) columnName(col repository.Field) string { return "editor_user" case repository.FieldEventType: return "event_type" + case repository.FieldEventData: + return "event_data" default: return "" } @@ -272,6 +274,8 @@ func (db *CRDB) operation(operation repository.Operation) string { return ">" case repository.OperationLess: return "<" + case repository.OperationJSONContains: + return "@>" } return "" } diff --git a/internal/eventstore/v2/repository/sql/query.go b/internal/eventstore/v2/repository/sql/query.go index 8d21fd52a6..319fa29ee9 100644 --- a/internal/eventstore/v2/repository/sql/query.go +++ b/internal/eventstore/v2/repository/sql/query.go @@ -3,6 +3,7 @@ package sql import ( "context" "database/sql" + "encoding/json" "errors" "fmt" "strings" @@ -137,6 +138,10 @@ func prepareCondition(criteria querier, filters []*repository.Filter) (clause st 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) diff --git a/internal/eventstore/v2/search_query.go b/internal/eventstore/v2/search_query.go index 4d9cdd9e4e..a203d04be8 100644 --- a/internal/eventstore/v2/search_query.go +++ b/internal/eventstore/v2/search_query.go @@ -15,6 +15,7 @@ type SearchQueryFactory struct { aggregateIDs []string eventSequence uint64 eventTypes []EventType + eventData map[string]interface{} resourceOwner string } @@ -83,6 +84,11 @@ func (factory *SearchQueryFactory) OrderAsc() *SearchQueryFactory { return factory } +func (factory *SearchQueryFactory) EventData(query map[string]interface{}) *SearchQueryFactory { + factory.eventData = query + return factory +} + func (factory *SearchQueryFactory) build() (*repository.SearchQuery, error) { if factory == nil || len(factory.aggregateTypes) < 1 || @@ -98,6 +104,7 @@ func (factory *SearchQueryFactory) build() (*repository.SearchQuery, error) { factory.eventSequenceFilter, factory.eventTypeFilter, factory.resourceOwnerFilter, + factory.eventDataFilter, } { if filter := f(); filter != nil { if err := filter.Validate(); err != nil { @@ -159,3 +166,10 @@ func (factory *SearchQueryFactory) resourceOwnerFilter() *repository.Filter { } return repository.NewFilter(repository.FieldResourceOwner, factory.resourceOwner, repository.OperationEquals) } + +func (factory *SearchQueryFactory) eventDataFilter() *repository.Filter { + if len(factory.eventData) == 0 { + return nil + } + return repository.NewFilter(repository.FieldEventData, factory.eventData, repository.OperationJSONContains) +} diff --git a/internal/v2/business/iam/converter.go b/internal/v2/business/iam/converter.go index 1cbc36924d..a406981d08 100644 --- a/internal/v2/business/iam/converter.go +++ b/internal/v2/business/iam/converter.go @@ -108,6 +108,16 @@ func readModelToObjectRoot(readModel eventstore.ReadModel) models.ObjectRoot { } } +func writeModelToObjectRoot(readModel eventstore.WriteModel) models.ObjectRoot { + return models.ObjectRoot{ + AggregateID: readModel.AggregateID, + // ChangeDate: readModel.ChangeDate, + // CreationDate: readModel.CreationDate, + ResourceOwner: readModel.ResourceOwner, + Sequence: readModel.ProcessedSequence, + } +} + func readModelToMember(readModel *member.ReadModel) *model.IAMMember { return &model.IAMMember{ ObjectRoot: readModelToObjectRoot(readModel.ReadModel), @@ -115,3 +125,11 @@ func readModelToMember(readModel *member.ReadModel) *model.IAMMember { UserID: readModel.UserID, } } + +func writeModelToMember(writeModel *iam.MemberWriteModel) *model.IAMMember { + return &model.IAMMember{ + ObjectRoot: writeModelToObjectRoot(writeModel.WriteModel.WriteModel), + Roles: writeModel.Roles, + UserID: writeModel.UserID, + } +} diff --git a/internal/v2/business/iam/member.go b/internal/v2/business/iam/member.go index 85fd0ca600..e316c30dcb 100644 --- a/internal/v2/business/iam/member.go +++ b/internal/v2/business/iam/member.go @@ -5,7 +5,9 @@ import ( "github.com/caos/zitadel/internal/errors" caos_errs "github.com/caos/zitadel/internal/errors" + "github.com/caos/zitadel/internal/eventstore/v2" iam_model "github.com/caos/zitadel/internal/iam/model" + "github.com/caos/zitadel/internal/tracing" iam_repo "github.com/caos/zitadel/internal/v2/repository/iam" ) @@ -55,25 +57,20 @@ func (r *Repository) ChangeIAMMember(ctx context.Context, member *iam_model.IAMM return nil, err } - changedMember := *existingMember - changedMember.Roles = member.Roles - iam := iam_repo.AggregateFromWriteModel(&existingMember.WriteModel.WriteModel). - PushMemberChanged(ctx, existingMember, &changedMember) + PushMemberChangedFromExisting(ctx, existingMember, member.Roles...) - events, err := r.eventstore.PushAggregates(ctx, iam) + _, err = r.eventstore.PushAggregates(ctx, iam) if err != nil { return nil, err } - if err = existingMember.AppendEvents(events...); err != nil { - return nil, err - } - if err = existingMember.Reduce(); err != nil { + updatedMember, err := r.MemberByID(ctx, member.AggregateID, member.UserID) + if err != nil { return nil, err } - return nil, nil + return readModelToMember(&updatedMember.ReadModel), nil } func (r *Repository) RemoveIAMMember(ctx context.Context, member *iam_model.IAMMember) error { @@ -97,3 +94,22 @@ func (r *Repository) RemoveIAMMember(ctx context.Context, member *iam_model.IAMM return iam.AppendAndReduce(events...) } + +func (r *Repository) MemberByID(ctx context.Context, iamID, userID string) (member *iam_repo.MemberReadModel, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + + query := eventstore.NewSearchQueryFactory(eventstore.ColumnsEvent, iam_repo.AggregateType). + AggregateIDs(iamID). + EventData(map[string]interface{}{ + "userId": userID, + }) + + member = new(iam_repo.MemberReadModel) + err = r.eventstore.FilterToReducer(ctx, query, member) + if err != nil { + return nil, err + } + + return member, nil +} diff --git a/internal/v2/repository/iam/aggregate.go b/internal/v2/repository/iam/aggregate.go index 4c8fb24703..fc7fd590a2 100644 --- a/internal/v2/repository/iam/aggregate.go +++ b/internal/v2/repository/iam/aggregate.go @@ -3,7 +3,6 @@ package iam import ( "context" - "github.com/caos/logging" "github.com/caos/zitadel/internal/eventstore/v2" ) @@ -59,13 +58,16 @@ func (a *Aggregate) PushMemberAdded(ctx context.Context, userID string, roles .. return a } -func (a *Aggregate) PushMemberChanged(ctx context.Context, current, changed *MemberWriteModel) *Aggregate { - e, err := NewMemberChangedEvent(ctx, current, changed) +func (a *Aggregate) PushMemberChanged(ctx context.Context, changed *MemberWriteModel) *Aggregate { + a.Aggregate = *a.PushEvents(NewMemberChangedEvent(ctx, changed.UserID, changed.Roles...)) + return a +} + +func (a *Aggregate) PushMemberChangedFromExisting(ctx context.Context, current *MemberWriteModel, roles ...string) *Aggregate { + e, err := MemberChangedEventFromExisting(ctx, current, roles...) if err != nil { - logging.Log("IAM-KH21C").OnError(err).Warn("unable to push member changed") return a } - a.Aggregate = *a.PushEvents(e) return a } diff --git a/internal/v2/repository/iam/member.go b/internal/v2/repository/iam/member.go index f01c974b0f..1014f2488d 100644 --- a/internal/v2/repository/iam/member.go +++ b/internal/v2/repository/iam/member.go @@ -86,19 +86,19 @@ func NewMemberAddedEvent( } } -func NewMemberChangedEvent( +func MemberChangedEventFromExisting( ctx context.Context, - current, - changed *MemberWriteModel, + current *MemberWriteModel, + roles ...string, ) (*MemberChangedEvent, error) { - m, err := member.NewChangedEvent( + m, err := member.ChangeEventFromExisting( eventstore.NewBaseEventForPush( ctx, MemberChangedEventType, ), ¤t.WriteModel, - &changed.WriteModel, + roles..., ) if err != nil { return nil, err @@ -109,6 +109,24 @@ func NewMemberChangedEvent( }, nil } +func NewMemberChangedEvent( + ctx context.Context, + userID string, + roles ...string, +) *MemberChangedEvent { + + return &MemberChangedEvent{ + ChangedEvent: *member.NewChangedEvent( + eventstore.NewBaseEventForPush( + ctx, + MemberChangedEventType, + ), + userID, + roles..., + ), + } +} + func NewMemberRemovedEvent( ctx context.Context, userID string, diff --git a/internal/v2/repository/member/event_changed.go b/internal/v2/repository/member/event_changed.go index 23c299bbe9..cbf2295e2a 100644 --- a/internal/v2/repository/member/event_changed.go +++ b/internal/v2/repository/member/event_changed.go @@ -31,25 +31,21 @@ func (e *ChangedEvent) Data() interface{} { return e } -func NewChangedEvent( +func ChangeEventFromExisting( base *eventstore.BaseEvent, - current, - changed *WriteModel, + current *WriteModel, + roles ...string, ) (*ChangedEvent, error) { change := &ChangedEvent{ BaseEvent: *base, - } - - if current.UserID != changed.UserID { - change.UserID = changed.UserID - change.hasChanged = true + UserID: current.UserID, } sort.Strings(current.Roles) - sort.Strings(changed.Roles) - if !reflect.DeepEqual(current.Roles, changed.Roles) { - change.Roles = changed.Roles + sort.Strings(roles) + if !reflect.DeepEqual(current.Roles, roles) { + change.Roles = roles change.hasChanged = true } @@ -58,7 +54,19 @@ func NewChangedEvent( } return change, nil +} +func NewChangedEvent( + base *eventstore.BaseEvent, + userID string, + roles ...string, +) *ChangedEvent { + + return &ChangedEvent{ + BaseEvent: *base, + UserID: userID, + Roles: roles, + } } func ChangedEventMapper(event *repository.Event) (eventstore.EventReader, error) { diff --git a/internal/v2/repository/member/write_model.go b/internal/v2/repository/member/write_model.go index 0c6c291f5b..cabdc3ab33 100644 --- a/internal/v2/repository/member/write_model.go +++ b/internal/v2/repository/member/write_model.go @@ -48,6 +48,9 @@ func (wm *WriteModel) Reduce() error { wm.UserID = e.UserID wm.Roles = e.Roles case *RemovedEvent: + if e.UserID != wm.userID { + continue + } wm.Roles = nil wm.IsRemoved = true } diff --git a/internal/v2/repository/org/member.go b/internal/v2/repository/org/member.go index beaaf603c7..0eaf44cc8c 100644 --- a/internal/v2/repository/org/member.go +++ b/internal/v2/repository/org/member.go @@ -92,28 +92,47 @@ func NewMemberAddedEvent( } } -func NewMemberChangedEvent( +func MemberChangedEventFromExisting( ctx context.Context, - current, - changed *MemberWriteModel, + current *MemberWriteModel, + roles ...string, ) (*MemberChangedEvent, error) { - event, err := member.NewChangedEvent( + m, err := member.ChangeEventFromExisting( eventstore.NewBaseEventForPush( ctx, MemberChangedEventType, ), ¤t.WriteModel, - &changed.WriteModel, + roles..., ) if err != nil { return nil, err } + return &MemberChangedEvent{ - ChangedEvent: *event, + ChangedEvent: *m, }, nil } +func NewMemberChangedEvent( + ctx context.Context, + userID string, + roles ...string, +) *MemberChangedEvent { + + return &MemberChangedEvent{ + ChangedEvent: *member.NewChangedEvent( + eventstore.NewBaseEventForPush( + ctx, + MemberChangedEventType, + ), + userID, + roles..., + ), + } +} + func NewMemberRemovedEvent( ctx context.Context, userID string,