From 4aadd290f42747daa96fb0a14b1a8ae108223bd6 Mon Sep 17 00:00:00 2001 From: adlerhurst Date: Mon, 23 Nov 2020 11:36:58 +0100 Subject: [PATCH] write model --- go.mod | 3 +- go.sum | 1 + internal/api/grpc/management/iam_converter.go | 16 +++--- internal/eventstore/v2/aggregate.go | 15 ++++++ internal/eventstore/v2/eventstore.go | 20 ++++++++ internal/eventstore/v2/write_model.go | 23 +++++---- internal/v2/business/iam/converter.go | 4 -- internal/v2/business/iam/member.go | 27 +++++----- internal/v2/business/iam/repository.go | 41 ++------------- internal/v2/repository/iam/aggregate.go | 8 ++- internal/v2/repository/iam/member.go | 28 +++++++++- internal/v2/repository/iam/members.go | 4 -- .../v2/repository/member/event_changed.go | 2 +- internal/v2/repository/member/write_model.go | 51 +++++++++++++++---- 14 files changed, 151 insertions(+), 92 deletions(-) diff --git a/go.mod b/go.mod index c897d18fb..e9884bda1 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/caos/oidc v0.12.5 github.com/census-instrumentation/opencensus-proto v0.3.0 // indirect github.com/cockroachdb/cockroach-go/v2 v2.0.8 + github.com/envoyproxy/protoc-gen-validate v0.1.0 github.com/ghodss/yaml v1.0.0 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b github.com/golang/mock v1.4.4 @@ -59,7 +60,7 @@ require ( golang.org/x/text v0.3.4 golang.org/x/tools v0.0.0-20201026223136-e84cfc6dd5ca google.golang.org/appengine v1.6.7 // indirect - google.golang.org/genproto v0.0.0-20201026171402-d4b8fe4fd877 // indirect + google.golang.org/genproto v0.0.0-20201026171402-d4b8fe4fd877 google.golang.org/grpc v1.33.1 google.golang.org/protobuf v1.25.0 gopkg.in/square/go-jose.v2 v2.5.1 diff --git a/go.sum b/go.sum index a563ae5a8..a44050b1b 100644 --- a/go.sum +++ b/go.sum @@ -112,6 +112,7 @@ github.com/denisenkom/go-mssqldb v0.0.0-20191124224453-732737034ffd/go.mod h1:xb github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= +github.com/envoyproxy/protoc-gen-validate v0.1.0 h1:EQciDnbrYxy13PgWoY8AqoxGiPrpgBZ1R8UNe3ddc+A= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 h1:Yzb9+7DPaBjB8zlTR87/ElzFsnQfuHnVUVqpZZIcV5Y= github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5/go.mod h1:a2zkGnVExMxdzMo3M0Hi/3sEU+cWnZpSni0O6/Yb/P0= diff --git a/internal/api/grpc/management/iam_converter.go b/internal/api/grpc/management/iam_converter.go index 5dbbc2c1b..255f5fd7a 100644 --- a/internal/api/grpc/management/iam_converter.go +++ b/internal/api/grpc/management/iam_converter.go @@ -20,14 +20,14 @@ func iamSetupStepFromModel(step iam_model.Step) management.IamSetupStep { return management.IamSetupStep_iam_setup_step_1 case iam_model.Step2: return management.IamSetupStep_iam_setup_step_2 - case iam_model.Step3: - return management.IamSetupStep_iam_setup_step_3 - case iam_model.Step4: - return management.IamSetupStep_iam_setup_step_4 - case iam_model.Step5: - return management.IamSetupStep_iam_setup_step_5 - case iam_model.Step6: - return management.IamSetupStep_iam_setup_step_6 + // case iam_model.Step3: + // return management.IamSetupStep_iam_setup_step_3 + // case iam_model.Step4: + // return management.IamSetupStep_iam_setup_step_4 + // case iam_model.Step5: + // return management.IamSetupStep_iam_setup_step_5 + // case iam_model.Step6: + // return management.IamSetupStep_iam_setup_step_6 default: return management.IamSetupStep_iam_setup_step_UNDEFINED diff --git a/internal/eventstore/v2/aggregate.go b/internal/eventstore/v2/aggregate.go index ed3d833a4..ab7c70d20 100644 --- a/internal/eventstore/v2/aggregate.go +++ b/internal/eventstore/v2/aggregate.go @@ -17,6 +17,21 @@ func NewAggregate( } } +func AggregateFromWriteModel( + wm *WriteModel, + typ AggregateType, + version Version, +) *Aggregate { + return &Aggregate{ + id: wm.AggregateID, + typ: typ, + resourceOwner: wm.ResourceOwner, + version: version, + previousSequence: wm.ProcessedSequence, + events: []EventPusher{}, + } +} + //Aggregate is the basic implementation of aggregater type Aggregate struct { id string `json:"-"` diff --git a/internal/eventstore/v2/eventstore.go b/internal/eventstore/v2/eventstore.go index 9304ec26b..43328cb53 100644 --- a/internal/eventstore/v2/eventstore.go +++ b/internal/eventstore/v2/eventstore.go @@ -167,6 +167,26 @@ func (es *Eventstore) LatestSequence(ctx context.Context, queryFactory *SearchQu return es.repo.LatestSequence(ctx, query) } +type queryReducer interface { + reducer + //Query returns the SearchQueryFactory for the events needed in reducer + Query() *SearchQueryFactory +} + +//FilterToQueryReducer filters the events based on the search query of the query function, +// appends all events to the reducer and calls it's reduce function +func (es *Eventstore) FilterToQueryReducer(ctx context.Context, r queryReducer) error { + events, err := es.FilterEvents(ctx, r.Query()) + if err != nil { + return err + } + if err = r.AppendEvents(events...); err != nil { + return err + } + + return r.Reduce() +} + //RegisterFilterEventMapper registers a function for mapping an eventstore event to an event func (es *Eventstore) RegisterFilterEventMapper(eventType EventType, mapper func(*repository.Event) (EventReader, error)) *Eventstore { if mapper == nil || eventType == "" { diff --git a/internal/eventstore/v2/write_model.go b/internal/eventstore/v2/write_model.go index 7d43d46a1..7c03eed2c 100644 --- a/internal/eventstore/v2/write_model.go +++ b/internal/eventstore/v2/write_model.go @@ -1,7 +1,7 @@ package eventstore -func NewReadModel() *ReadModel { - return &ReadModel{ +func NewWriteModel() *WriteModel { + return &WriteModel{ Events: []EventReader{}, } } @@ -13,6 +13,7 @@ type WriteModel struct { AggregateID string `json:"-"` ProcessedSequence uint64 `json:"-"` Events []EventReader `json:"-"` + ResourceOwner string `json:"-"` } //AppendEvents adds all the events to the read model. @@ -24,22 +25,22 @@ func (rm *WriteModel) AppendEvents(events ...EventReader) *WriteModel { //Reduce is the basic implementaion of reducer // If this function is extended the extending function should be the last step -func (rm *WriteModel) Reduce() error { - if len(rm.Events) == 0 { +func (wm *WriteModel) Reduce() error { + if len(wm.Events) == 0 { return nil } - if rm.AggregateID == "" { - rm.AggregateID = rm.Events[0].AggregateID() + if wm.AggregateID == "" { + wm.AggregateID = wm.Events[0].AggregateID() } - if rm.ResourceOwner == "" { - rm.ResourceOwner = rm.Events[0].ResourceOwner() + if wm.ResourceOwner == "" { + wm.ResourceOwner = wm.Events[0].ResourceOwner() } - rm.ProcessedSequence = rm.Events[len(rm.Events)-1].Sequence() + wm.ProcessedSequence = wm.Events[len(wm.Events)-1].Sequence() // all events processed and not needed anymore - rm.Events = nil - rm.Events = []EventReader{} + wm.Events = nil + wm.Events = []EventReader{} return nil } diff --git a/internal/v2/business/iam/converter.go b/internal/v2/business/iam/converter.go index 8d6a27bfb..1cbc36924 100644 --- a/internal/v2/business/iam/converter.go +++ b/internal/v2/business/iam/converter.go @@ -115,7 +115,3 @@ func readModelToMember(readModel *member.ReadModel) *model.IAMMember { UserID: readModel.UserID, } } - -func writeModelFromMember(member *model.IAMMember) *member.WriteModel { - return &member.WriteModel{} -} diff --git a/internal/v2/business/iam/member.go b/internal/v2/business/iam/member.go index 8b7aff98a..85fd0ca60 100644 --- a/internal/v2/business/iam/member.go +++ b/internal/v2/business/iam/member.go @@ -43,38 +43,37 @@ func (r *Repository) AddIAMMember(ctx context.Context, member *iam_model.IAMMemb return readModelToMember(addedMember), nil } +//ChangeIAMMember updates an existing member +//TODO: refactor to ChangeMember func (r *Repository) ChangeIAMMember(ctx context.Context, member *iam_model.IAMMember) (*iam_model.IAMMember, error) { if !member.IsValid() { return nil, caos_errs.ThrowPreconditionFailed(nil, "IAM-LiaZi", "Errors.IAM.MemberInvalid") } - iam, err := r.iamByID(ctx, member.AggregateID) - if err != nil { - return nil, err - } - existingMember, err := r.memberWriteModelByID(ctx, member.AggregateID, member.UserID) if err != nil { return nil, err } - iamAgg := iam_repo.AggregateFromReadModel(iam). - PushMemberChanged(ctx, existingMember, nil) + changedMember := *existingMember + changedMember.Roles = member.Roles - events, err := r.eventstore.PushAggregates(ctx, iamAgg) + iam := iam_repo.AggregateFromWriteModel(&existingMember.WriteModel.WriteModel). + PushMemberChanged(ctx, existingMember, &changedMember) + + events, err := r.eventstore.PushAggregates(ctx, iam) if err != nil { return nil, err } - if err = iam.AppendAndReduce(events...); err != nil { + if err = existingMember.AppendEvents(events...); err != nil { + return nil, err + } + if err = existingMember.Reduce(); err != nil { return nil, err } - _, addedMember := iam.Members.MemberByUserID(member.UserID) - if member == nil { - return nil, errors.ThrowInternal(nil, "IAM-E5nTQ", "member not saved") - } - return readModelToMember(addedMember), nil + return nil, nil } func (r *Repository) RemoveIAMMember(ctx context.Context, member *iam_model.IAMMember) error { diff --git a/internal/v2/business/iam/repository.go b/internal/v2/business/iam/repository.go index f2f954331..a21dd9bfc 100644 --- a/internal/v2/business/iam/repository.go +++ b/internal/v2/business/iam/repository.go @@ -8,7 +8,6 @@ import ( 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" - "github.com/caos/zitadel/internal/v2/repository/member" ) type Repository struct { @@ -53,47 +52,15 @@ func (r *Repository) memberWriteModelByID(ctx context.Context, iamID, userID str ctx, span := tracing.NewSpan(ctx) defer func() { span.EndWithError(err) }() - query := eventstore.NewSearchQueryFactory(eventstore.ColumnsEvent, iam_repo.AggregateType).AggregateIDs(iamID) - - writeModel := new(memberWriteModel) - err = r.eventstore.FilterToReducer(ctx, query, writeModel) + writeModel := iam_repo.PrepareMemberWriteModel(iamID, userID) + err = r.eventstore.FilterToQueryReducer(ctx, writeModel) if err != nil { return nil, err } - if writeModel.isDeleted { + if writeModel.IsRemoved { return nil, errors.ThrowNotFound(nil, "IAM-D8JxR", "Errors.NotFound") } - return &writeModel.MemberWriteModel, nil -} - -type memberWriteModel struct { - iam_repo.MemberWriteModel - - userID string - isDeleted bool -} - -func (wm *memberWriteModel) AppendEvents(events ...eventstore.EventReader) error { - for _, event := range events { - switch e := event.(type) { - case *member.AddedEvent: - if e.UserID == wm.userID { - wm.isDeleted = false - wm.MemberWriteModel.AppendEvents(event) - } - case *member.ChangedEvent: - if e.UserID == wm.userID { - wm.MemberWriteModel.AppendEvents(event) - } - case *member.RemovedEvent: - if e.UserID == wm.userID { - wm.isDeleted = true - wm.MemberWriteModel = iam_repo.MemberWriteModel{} - } - } - } - - return nil + return writeModel, nil } diff --git a/internal/v2/repository/iam/aggregate.go b/internal/v2/repository/iam/aggregate.go index f44978c0f..4c8fb2470 100644 --- a/internal/v2/repository/iam/aggregate.go +++ b/internal/v2/repository/iam/aggregate.go @@ -21,8 +21,6 @@ type Aggregate struct { SetUpStarted Step SetUpDone Step - - Members MembersAggregate } func NewAggregate( @@ -42,6 +40,12 @@ func NewAggregate( } } +func AggregateFromWriteModel(wm *eventstore.WriteModel) *Aggregate { + return &Aggregate{ + Aggregate: *eventstore.AggregateFromWriteModel(wm, AggregateType, AggregateVersion), + } +} + func AggregateFromReadModel(rm *ReadModel) *Aggregate { return &Aggregate{ Aggregate: *eventstore.NewAggregate(rm.AggregateID, AggregateType, rm.ResourceOwner, AggregateVersion, rm.ProcessedSequence), diff --git a/internal/v2/repository/iam/member.go b/internal/v2/repository/iam/member.go index ad4fce148..f01c974b0 100644 --- a/internal/v2/repository/iam/member.go +++ b/internal/v2/repository/iam/member.go @@ -31,6 +31,32 @@ func (rm *MemberReadModel) AppendEvents(events ...eventstore.EventReader) (err e return nil } +type MemberWriteModel struct { + member.WriteModel +} + +func PrepareMemberWriteModel(iamID, userID string) *MemberWriteModel { + return &MemberWriteModel{ + WriteModel: *member.PrepareWriteModel(userID, AggregateType, iamID), + } +} + +func (wm *MemberWriteModel) AppendEvents(events ...eventstore.EventReader) (err error) { + for _, event := range events { + switch e := event.(type) { + case *MemberAddedEvent: + wm.WriteModel.AppendEvents(&e.AddedEvent) + case *MemberChangedEvent: + wm.WriteModel.AppendEvents(&e.ChangedEvent) + case *MemberRemovedEvent: + wm.WriteModel.AppendEvents(&e.RemovedEvent) + default: + wm.WriteModel.AppendEvents(e) + } + } + return nil +} + type MemberAddedEvent struct { member.AddedEvent } @@ -63,7 +89,7 @@ func NewMemberAddedEvent( func NewMemberChangedEvent( ctx context.Context, current, - changed *MemberAggregate, + changed *MemberWriteModel, ) (*MemberChangedEvent, error) { m, err := member.NewChangedEvent( diff --git a/internal/v2/repository/iam/members.go b/internal/v2/repository/iam/members.go index 2cc912f22..fe9d6c6e0 100644 --- a/internal/v2/repository/iam/members.go +++ b/internal/v2/repository/iam/members.go @@ -6,10 +6,6 @@ import ( "github.com/caos/zitadel/internal/v2/repository/members" ) -type MembersAggregate struct { - members.Aggregate -} - type MembersReadModel struct { members.ReadModel } diff --git a/internal/v2/repository/member/event_changed.go b/internal/v2/repository/member/event_changed.go index c38b0bfb3..23c299bbe 100644 --- a/internal/v2/repository/member/event_changed.go +++ b/internal/v2/repository/member/event_changed.go @@ -34,7 +34,7 @@ func (e *ChangedEvent) Data() interface{} { func NewChangedEvent( base *eventstore.BaseEvent, current, - changed *Aggregate, + changed *WriteModel, ) (*ChangedEvent, error) { change := &ChangedEvent{ diff --git a/internal/v2/repository/member/write_model.go b/internal/v2/repository/member/write_model.go index f17c91f5d..0c6c291f5 100644 --- a/internal/v2/repository/member/write_model.go +++ b/internal/v2/repository/member/write_model.go @@ -8,21 +8,54 @@ import "github.com/caos/zitadel/internal/eventstore/v2" type WriteModel struct { eventstore.WriteModel - UserID string - Roles []string + UserID string + Roles []string + IsRemoved bool + + userID string + aggregateType eventstore.AggregateType + aggregateID string +} + +func PrepareWriteModel( + userID string, + aggregateType eventstore.AggregateType, + aggregateID string, +) *WriteModel { + + return &WriteModel{ + WriteModel: *eventstore.NewWriteModel(), + userID: userID, + aggregateType: aggregateType, + aggregateID: aggregateID, + } } //Reduce extends eventstore.ReadModel -func (rm *WriteModel) Reduce() error { - for _, event := range rm.Events { +func (wm *WriteModel) Reduce() error { + for _, event := range wm.Events { switch e := event.(type) { case *AddedEvent: - rm.UserID = e.UserID - rm.Roles = e.Roles + if e.UserID != wm.userID { + continue + } + wm.UserID = e.UserID + wm.Roles = e.Roles case *ChangedEvent: - rm.UserID = e.UserID - rm.Roles = e.Roles + if e.UserID != wm.userID { + continue + } + wm.UserID = e.UserID + wm.Roles = e.Roles + case *RemovedEvent: + wm.Roles = nil + wm.IsRemoved = true } } - return rm.ReadModel.Reduce() + return wm.WriteModel.Reduce() +} + +func (wm *WriteModel) Query() *eventstore.SearchQueryFactory { + return eventstore.NewSearchQueryFactory(eventstore.ColumnsEvent, wm.aggregateType). + AggregateIDs(wm.aggregateID) }