write model

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

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=

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:
+	// case iam_model.Step3:
-		return management.IamSetupStep_iam_setup_step_3
+	// 	return management.IamSetupStep_iam_setup_step_3
-	case iam_model.Step4:
+	// case iam_model.Step4:
-		return management.IamSetupStep_iam_setup_step_4
+	// 	return management.IamSetupStep_iam_setup_step_4
-	case iam_model.Step5:
+	// case iam_model.Step5:
-		return management.IamSetupStep_iam_setup_step_5
+	// 	return management.IamSetupStep_iam_setup_step_5
-	case iam_model.Step6:
+	// case iam_model.Step6:
-		return management.IamSetupStep_iam_setup_step_6
+	// 	return management.IamSetupStep_iam_setup_step_6
 
 	default:
 		return management.IamSetupStep_iam_setup_step_UNDEFINED

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:"-"`

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 == "" {

internal/eventstore/v2/write_model.go

@@ -1,7 +1,7 @@
 package eventstore
 
-func NewReadModel() *ReadModel {
+func NewWriteModel() *WriteModel {
-	return &ReadModel{
+	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 {
+func (wm *WriteModel) Reduce() error {
-	if len(rm.Events) == 0 {
+	if len(wm.Events) == 0 {
 		return nil
 	}
 
-	if rm.AggregateID == "" {
+	if wm.AggregateID == "" {
-		rm.AggregateID = rm.Events[0].AggregateID()
+		wm.AggregateID = wm.Events[0].AggregateID()
 	}
-	if rm.ResourceOwner == "" {
+	if wm.ResourceOwner == "" {
-		rm.ResourceOwner = rm.Events[0].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
+	wm.Events = nil
-	rm.Events = []EventReader{}
+	wm.Events = []EventReader{}
 	return nil
 }

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{}
-}

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).
+	changedMember := *existingMember
-		PushMemberChanged(ctx, existingMember, nil)
+	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)
+	return nil, nil
-	if member == nil {
-		return nil, errors.ThrowInternal(nil, "IAM-E5nTQ", "member not saved")
-	}
-	return readModelToMember(addedMember), nil
 }
 
 func (r *Repository) RemoveIAMMember(ctx context.Context, member *iam_model.IAMMember) error {

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 := iam_repo.PrepareMemberWriteModel(iamID, userID)
-
+	err = r.eventstore.FilterToQueryReducer(ctx, writeModel)
-	writeModel := new(memberWriteModel)
-	err = r.eventstore.FilterToReducer(ctx, query, 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
+	return writeModel, 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
 }

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),

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(

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
 }

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{

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
+	UserID    string
-	Roles  []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 {
+func (wm *WriteModel) Reduce() error {
-	for _, event := range rm.Events {
+	for _, event := range wm.Events {
 		switch e := event.(type) {
 		case *AddedEvent:
-			rm.UserID = e.UserID
+			if e.UserID != wm.userID {
-			rm.Roles = e.Roles
+				continue
+			}
+			wm.UserID = e.UserID
+			wm.Roles = e.Roles
 		case *ChangedEvent:
-			rm.UserID = e.UserID
+			if e.UserID != wm.userID {
-			rm.Roles = e.Roles
+				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)
 }