feat: user query (#3075)

* user queries

* user query

* user query

* user tests

* remove old code

* user metadata

* cleanup

* fix merge

* cleanup

* cleanup

* fixes

internal/auth/repository/eventsourcing/eventstore/user.go

@@ -7,6 +7,8 @@ import (
 	"github.com/caos/logging"
 	"github.com/golang/protobuf/ptypes"
 
+	"github.com/caos/zitadel/internal/user/model"
+
 	"github.com/caos/zitadel/internal/api/authz"
 	"github.com/caos/zitadel/internal/auth/repository/eventsourcing/view"
 	"github.com/caos/zitadel/internal/config/systemdefaults"
@@ -14,17 +16,15 @@ import (
 	"github.com/caos/zitadel/internal/errors"
 	v1 "github.com/caos/zitadel/internal/eventstore/v1"
 	"github.com/caos/zitadel/internal/eventstore/v1/models"
-	iam_model "github.com/caos/zitadel/internal/iam/repository/view/model"
-	"github.com/caos/zitadel/internal/telemetry/tracing"
-	"github.com/caos/zitadel/internal/user/model"
+	"github.com/caos/zitadel/internal/query"
 	usr_view "github.com/caos/zitadel/internal/user/repository/view"
-	usr_view_model "github.com/caos/zitadel/internal/user/repository/view/model"
 )
 
 type UserRepo struct {
 	SearchLimit     uint64
 	Eventstore      v1.Eventstore
 	View            *view.View
+	Query           *query.Queries
 	SystemDefaults  systemdefaults.SystemDefaults
 	PrefixAvatarURL string
 }
@@ -33,105 +33,6 @@ func (repo *UserRepo) Health(ctx context.Context) error {
 	return repo.Eventstore.Health(ctx)
 }
 
-func (repo *UserRepo) MyUser(ctx context.Context) (*model.UserView, error) {
-	return repo.UserByID(ctx, authz.GetCtxData(ctx).UserID)
-}
-
-func (repo *UserRepo) MyProfile(ctx context.Context) (*model.Profile, error) {
-	user, err := repo.UserByID(ctx, authz.GetCtxData(ctx).UserID)
-	if err != nil {
-		return nil, err
-	}
-	if user.HumanView == nil {
-		return nil, errors.ThrowPreconditionFailed(nil, "EVENT-H2JIT", "Errors.User.NotHuman")
-	}
-	return user.GetProfile()
-}
-
-func (repo *UserRepo) SearchMyExternalIDPs(ctx context.Context, request *model.ExternalIDPSearchRequest) (*model.ExternalIDPSearchResponse, error) {
-	err := request.EnsureLimit(repo.SearchLimit)
-	if err != nil {
-		return nil, err
-	}
-	sequence, seqErr := repo.View.GetLatestExternalIDPSequence()
-	logging.Log("EVENT-5Jsi8").OnError(seqErr).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Warn("could not read latest user sequence")
-	request.AppendUserQuery(authz.GetCtxData(ctx).UserID)
-	externalIDPS, count, err := repo.View.SearchExternalIDPs(request)
-	if err != nil {
-		return nil, err
-	}
-	result := &model.ExternalIDPSearchResponse{
-		Offset:      request.Offset,
-		Limit:       request.Limit,
-		TotalResult: count,
-		Result:      usr_view_model.ExternalIDPViewsToModel(externalIDPS),
-	}
-	if seqErr == nil {
-		result.Sequence = sequence.CurrentSequence
-		result.Timestamp = sequence.LastSuccessfulSpoolerRun
-	}
-	return result, nil
-}
-
-func (repo *UserRepo) MyEmail(ctx context.Context) (*model.Email, error) {
-	user, err := repo.UserByID(ctx, authz.GetCtxData(ctx).UserID)
-	if err != nil {
-		return nil, err
-	}
-	if user.HumanView == nil {
-		return nil, errors.ThrowPreconditionFailed(nil, "EVENT-oGRpc", "Errors.User.NotHuman")
-	}
-	return user.GetEmail()
-}
-
-func (repo *UserRepo) MyPhone(ctx context.Context) (*model.Phone, error) {
-	user, err := repo.UserByID(ctx, authz.GetCtxData(ctx).UserID)
-	if err != nil {
-		return nil, err
-	}
-	if user.HumanView == nil {
-		return nil, errors.ThrowPreconditionFailed(nil, "EVENT-DTWJb", "Errors.User.NotHuman")
-	}
-	return user.GetPhone()
-}
-
-func (repo *UserRepo) MyAddress(ctx context.Context) (*model.Address, error) {
-	user, err := repo.UserByID(ctx, authz.GetCtxData(ctx).UserID)
-	if err != nil {
-		return nil, err
-	}
-	if user.HumanView == nil {
-		return nil, errors.ThrowPreconditionFailed(nil, "EVENT-Ok9nI", "Errors.User.NotHuman")
-	}
-	return user.GetAddress()
-}
-
-func (repo *UserRepo) MyUserMFAs(ctx context.Context) ([]*model.MultiFactor, error) {
-	user, err := repo.UserByID(ctx, authz.GetCtxData(ctx).UserID)
-	if err != nil {
-		return nil, err
-	}
-	mfas := make([]*model.MultiFactor, 0)
-	if user.OTPState != model.MFAStateUnspecified {
-		mfas = append(mfas, &model.MultiFactor{Type: model.MFATypeOTP, State: user.OTPState})
-	}
-	for _, u2f := range user.U2FTokens {
-		mfas = append(mfas, &model.MultiFactor{Type: model.MFATypeU2F, State: u2f.State, Attribute: u2f.Name, ID: u2f.TokenID})
-	}
-	return mfas, nil
-}
-
-func (repo *UserRepo) GetMyPasswordless(ctx context.Context) ([]*model.WebAuthNView, error) {
-	user, err := repo.UserByID(ctx, authz.GetCtxData(ctx).UserID)
-	if err != nil {
-		return nil, err
-	}
-	if user.HumanView == nil {
-		return nil, errors.ThrowPreconditionFailed(nil, "USER-9kF98", "Errors.User.NotHuman")
-	}
-	return user.HumanView.PasswordlessTokens, nil
-}
-
 func (repo *UserRepo) UserSessionUserIDsByAgentID(ctx context.Context, agentID string) ([]string, error) {
 	userSessions, err := repo.View.UserSessionsByAgentID(agentID)
 	if err != nil {
@@ -146,53 +47,10 @@ func (repo *UserRepo) UserSessionUserIDsByAgentID(ctx context.Context, agentID s
 	return userIDs, nil
 }
 
-func (repo *UserRepo) UserByID(ctx context.Context, id string) (*model.UserView, error) {
-	user, err := repo.View.UserByID(id)
-	if err != nil {
-		return nil, err
-	}
-	events, err := repo.getUserEvents(ctx, id, user.Sequence)
-	if err != nil {
-		logging.Log("EVENT-PSoc3").WithError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Debug("error retrieving new events")
-		return usr_view_model.UserToModel(user, repo.PrefixAvatarURL), nil
-	}
-	userCopy := *user
-	for _, event := range events {
-		if err := userCopy.AppendEvent(event); err != nil {
-			return usr_view_model.UserToModel(user, repo.PrefixAvatarURL), nil
-		}
-	}
-	if userCopy.State == int32(model.UserStateDeleted) {
-		return nil, errors.ThrowNotFound(nil, "EVENT-vZ8us", "Errors.User.NotFound")
-	}
-	return usr_view_model.UserToModel(&userCopy, repo.PrefixAvatarURL), nil
-}
-
 func (repo *UserRepo) UserEventsByID(ctx context.Context, id string, sequence uint64) ([]*models.Event, error) {
 	return repo.getUserEvents(ctx, id, sequence)
 }
 
-func (repo *UserRepo) UserByLoginName(ctx context.Context, loginname string) (*model.UserView, error) {
-	user, err := repo.View.UserByLoginName(loginname)
-	if err != nil {
-		return nil, err
-	}
-	events, err := repo.getUserEvents(ctx, user.ID, user.Sequence)
-	if err != nil {
-		logging.Log("EVENT-PSoc3").WithError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Debug("error retrieving new events")
-		return usr_view_model.UserToModel(user, repo.PrefixAvatarURL), nil
-	}
-	userCopy := *user
-	for _, event := range events {
-		if err := userCopy.AppendEvent(event); err != nil {
-			return usr_view_model.UserToModel(user, repo.PrefixAvatarURL), nil
-		}
-	}
-	if userCopy.State == int32(model.UserStateDeleted) {
-		return nil, errors.ThrowNotFound(nil, "EVENT-vZ8us", "Errors.User.NotFound")
-	}
-	return usr_view_model.UserToModel(&userCopy, repo.PrefixAvatarURL), nil
-}
 func (repo *UserRepo) MyUserChanges(ctx context.Context, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*model.UserChanges, error) {
 	changes, err := repo.getUserChanges(ctx, authz.GetCtxData(ctx).UserID, lastSequence, limit, sortAscending, retention)
 	if err != nil {
@@ -201,41 +59,21 @@ func (repo *UserRepo) MyUserChanges(ctx context.Context, lastSequence uint64, li
 	for _, change := range changes.Changes {
 		change.ModifierName = change.ModifierID
 		change.ModifierLoginName = change.ModifierID
-		user, _ := repo.UserByID(ctx, change.ModifierID)
+		user, _ := repo.Query.GetUserByID(ctx, change.ModifierID)
 		if user != nil {
 			change.ModifierLoginName = user.PreferredLoginName
-			if user.HumanView != nil {
-				change.ModifierName = user.HumanView.DisplayName
-				change.ModifierAvatarURL = user.HumanView.AvatarURL
+			if user.Human != nil {
+				change.ModifierName = user.Human.DisplayName
+				change.ModifierAvatarURL = domain.AvatarURL(repo.PrefixAvatarURL, user.ResourceOwner, user.Human.AvatarKey)
 			}
-			if user.MachineView != nil {
-				change.ModifierName = user.MachineView.Name
+			if user.Machine != nil {
+				change.ModifierName = user.Machine.Name
 			}
 		}
 	}
 	return changes, nil
 }
 
-func (repo *UserRepo) SearchUsers(ctx context.Context, request *model.UserSearchRequest) (*model.UserSearchResponse, error) {
-	sequence, sequenceErr := repo.View.GetLatestUserSequence()
-	logging.Log("EVENT-Gdgsw").OnError(sequenceErr).Warn("could not read latest user sequence")
-	users, count, err := repo.View.SearchUsers(request)
-	if err != nil {
-		return nil, err
-	}
-	result := &model.UserSearchResponse{
-		Offset:      request.Offset,
-		Limit:       request.Limit,
-		TotalResult: count,
-		Result:      usr_view_model.UsersToModel(users, repo.PrefixAvatarURL),
-	}
-	if sequenceErr == nil {
-		result.Sequence = sequence.CurrentSequence
-		result.Timestamp = sequence.LastSuccessfulSpoolerRun
-	}
-	return result, nil
-}
-
 func (r *UserRepo) getUserChanges(ctx context.Context, userID string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*model.UserChanges, error) {
 	query := usr_view.ChangesQuery(userID, lastSequence, limit, sortAscending, retention)
 
@@ -287,50 +125,3 @@ func (r *UserRepo) getUserEvents(ctx context.Context, userID string, sequence ui
 	}
 	return r.Eventstore.FilterEvents(ctx, query)
 }
-
-func (repo *UserRepo) GetMyMetadataByKey(ctx context.Context, key string) (*domain.Metadata, error) {
-	ctxData := authz.GetCtxData(ctx)
-	data, err := repo.View.MetadataByKeyAndResourceOwner(ctxData.UserID, ctxData.ResourceOwner, key)
-	if err != nil {
-		return nil, err
-	}
-	return iam_model.MetadataViewToDomain(data), nil
-}
-
-func (repo *UserRepo) SearchUserMetadata(ctx context.Context, userID string) (*domain.MetadataSearchResponse, error) {
-	req := new(domain.MetadataSearchRequest)
-	return repo.searchUserMetadata(userID, "", req)
-}
-
-func (repo *UserRepo) SearchMyMetadata(ctx context.Context, req *domain.MetadataSearchRequest) (*domain.MetadataSearchResponse, error) {
-	ctxData := authz.GetCtxData(ctx)
-	err := req.EnsureLimit(repo.SearchLimit)
-	if err != nil {
-		return nil, err
-	}
-	return repo.searchUserMetadata(ctxData.UserID, ctxData.ResourceOwner, req)
-}
-
-func (repo *UserRepo) searchUserMetadata(userID, resourceOwner string, req *domain.MetadataSearchRequest) (*domain.MetadataSearchResponse, error) {
-	sequence, sequenceErr := repo.View.GetLatestUserSequence()
-	logging.Log("EVENT-N9fsd").OnError(sequenceErr).Warn("could not read latest user sequence")
-	req.AppendAggregateIDQuery(userID)
-	if resourceOwner != "" {
-		req.AppendResourceOwnerQuery(resourceOwner)
-	}
-	metadata, count, err := repo.View.SearchMetadata(req)
-	if err != nil {
-		return nil, err
-	}
-	result := &domain.MetadataSearchResponse{
-		Offset:      req.Offset,
-		Limit:       req.Limit,
-		TotalResult: count,
-		Result:      iam_model.MetadataViewsToDomain(metadata),
-	}
-	if sequenceErr == nil {
-		result.Sequence = sequence.CurrentSequence
-		result.Timestamp = sequence.LastSuccessfulSpoolerRun
-	}
-	return result, nil
-}

internal/auth/repository/eventsourcing/handler/handler.go

@@ -50,7 +50,6 @@ func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es
 			handler{view, bulkLimit, configs.cycleDuration("ExternalIDP"), errorCount, es},
 			systemDefaults),
 		newRefreshToken(handler{view, bulkLimit, configs.cycleDuration("RefreshToken"), errorCount, es}),
-		newMetadata(handler{view, bulkLimit, configs.cycleDuration("Metadata"), errorCount, es}),
 		newOrgProjectMapping(handler{view, bulkLimit, configs.cycleDuration("OrgProjectMapping"), errorCount, es}),
 	}
 }

internal/auth/repository/eventsourcing/handler/metadata.go

@@ -1,126 +0,0 @@
-package handler
-
-import (
-	"github.com/caos/logging"
-
-	caos_errs "github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/eventstore/v1"
-	es_models "github.com/caos/zitadel/internal/eventstore/v1/models"
-	"github.com/caos/zitadel/internal/eventstore/v1/query"
-	"github.com/caos/zitadel/internal/eventstore/v1/spooler"
-	iam_model "github.com/caos/zitadel/internal/iam/repository/view/model"
-	usr_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
-)
-
-type Metadata struct {
-	handler
-	subscription *v1.Subscription
-}
-
-func newMetadata(handler handler) *Metadata {
-	h := &Metadata{
-		handler: handler,
-	}
-
-	h.subscribe()
-
-	return h
-}
-
-func (m *Metadata) subscribe() {
-	m.subscription = m.es.Subscribe(m.AggregateTypes()...)
-	go func() {
-		for event := range m.subscription.Events {
-			query.ReduceEvent(m, event)
-		}
-	}()
-}
-
-const (
-	metadataTable = "auth.metadata"
-)
-
-func (m *Metadata) ViewModel() string {
-	return metadataTable
-}
-
-func (m *Metadata) Subscription() *v1.Subscription {
-	return m.subscription
-}
-
-func (_ *Metadata) AggregateTypes() []es_models.AggregateType {
-	return []es_models.AggregateType{usr_model.UserAggregate}
-}
-
-func (p *Metadata) CurrentSequence() (uint64, error) {
-	sequence, err := p.view.GetLatestMetadataSequence()
-	if err != nil {
-		return 0, err
-	}
-	return sequence.CurrentSequence, nil
-}
-
-func (m *Metadata) EventQuery() (*es_models.SearchQuery, error) {
-	sequence, err := m.view.GetLatestMetadataSequence()
-	if err != nil {
-		return nil, err
-	}
-	return es_models.NewSearchQuery().
-		AggregateTypeFilter(m.AggregateTypes()...).
-		LatestSequenceFilter(sequence.CurrentSequence), nil
-}
-
-func (m *Metadata) Reduce(event *es_models.Event) (err error) {
-	switch event.AggregateType {
-	case usr_model.UserAggregate:
-		err = m.processMetadata(event)
-	}
-	return err
-}
-
-func (m *Metadata) processMetadata(event *es_models.Event) (err error) {
-	metadata := new(iam_model.MetadataView)
-	switch event.Type {
-	case usr_model.UserMetadataSet:
-		err = metadata.SetData(event)
-		if err != nil {
-			return err
-		}
-		metadata, err = m.view.MetadataByKey(event.AggregateID, metadata.Key)
-		if err != nil && !caos_errs.IsNotFound(err) {
-			return err
-		}
-		if caos_errs.IsNotFound(err) {
-			err = nil
-			metadata = new(iam_model.MetadataView)
-			metadata.CreationDate = event.CreationDate
-		}
-		err = metadata.AppendEvent(event)
-	case usr_model.UserMetadataRemoved:
-		data := new(iam_model.MetadataView)
-		err = data.SetData(event)
-		if err != nil {
-			return err
-		}
-		return m.view.DeleteMetadata(event.AggregateID, data.Key, event)
-	case usr_model.UserMetadataRemovedAll:
-		return m.view.DeleteMetadataByAggregateID(event.AggregateID, event)
-	case usr_model.UserRemoved:
-		return m.view.DeleteMetadataByAggregateID(event.AggregateID, event)
-	default:
-		return m.view.ProcessedMetadataSequence(event)
-	}
-	if err != nil {
-		return err
-	}
-	return m.view.PutMetadata(metadata, event)
-}
-
-func (m *Metadata) OnError(event *es_models.Event, err error) error {
-	logging.LogWithFields("SPOOL-miJJs", "id", event.AggregateID).WithError(err).Warn("something went wrong in custom text handler")
-	return spooler.HandleError(event, err, m.view.GetLatestMetadataFailedEvent, m.view.ProcessedMetadataFailedEvent, m.view.ProcessedMetadataSequence, m.errorCountUntilSkip)
-}
-
-func (m *Metadata) OnSuccess() error {
-	return spooler.HandleSuccess(m.view.UpdateMetadataSpoolerRunTimestamp)
-}

internal/auth/repository/eventsourcing/view/metadata.go

@@ -1,73 +0,0 @@
-package view
-
-import (
-	"github.com/caos/zitadel/internal/domain"
-	"github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/eventstore/v1/models"
-	"github.com/caos/zitadel/internal/iam/repository/view"
-	"github.com/caos/zitadel/internal/iam/repository/view/model"
-	global_view "github.com/caos/zitadel/internal/view/repository"
-)
-
-const (
-	metadataTable = "auth.metadata"
-)
-
-func (v *View) MetadataByKey(aggregateID, key string) (*model.MetadataView, error) {
-	return view.MetadataByKey(v.Db, metadataTable, aggregateID, key)
-}
-
-func (v *View) MetadataListByAggregateID(aggregateID string) ([]*model.MetadataView, error) {
-	return view.GetMetadataList(v.Db, metadataTable, aggregateID)
-}
-
-func (v *View) MetadataByKeyAndResourceOwner(aggregateID, resourceOwner, key string) (*model.MetadataView, error) {
-	return view.MetadataByKeyAndResourceOwner(v.Db, metadataTable, aggregateID, resourceOwner, key)
-}
-
-func (v *View) SearchMetadata(request *domain.MetadataSearchRequest) ([]*model.MetadataView, uint64, error) {
-	return view.SearchMetadata(v.Db, metadataTable, request)
-}
-
-func (v *View) PutMetadata(template *model.MetadataView, event *models.Event) error {
-	err := view.PutMetadata(v.Db, metadataTable, template)
-	if err != nil {
-		return err
-	}
-	return v.ProcessedMetadataSequence(event)
-}
-
-func (v *View) DeleteMetadata(aggregateID, key string, event *models.Event) error {
-	err := view.DeleteMetadata(v.Db, metadataTable, aggregateID, key)
-	if err != nil && !errors.IsNotFound(err) {
-		return err
-	}
-	return v.ProcessedMetadataSequence(event)
-}
-
-func (v *View) DeleteMetadataByAggregateID(aggregateID string, event *models.Event) error {
-	err := view.DeleteMetadataByAggregateID(v.Db, metadataTable, aggregateID)
-	if err != nil && !errors.IsNotFound(err) {
-		return err
-	}
-	return v.ProcessedMetadataSequence(event)
-}
-func (v *View) GetLatestMetadataSequence() (*global_view.CurrentSequence, error) {
-	return v.latestSequence(metadataTable)
-}
-
-func (v *View) ProcessedMetadataSequence(event *models.Event) error {
-	return v.saveCurrentSequence(metadataTable, event)
-}
-
-func (v *View) UpdateMetadataSpoolerRunTimestamp() error {
-	return v.updateSpoolerRunSequence(metadataTable)
-}
-
-func (v *View) GetLatestMetadataFailedEvent(sequence uint64) (*global_view.FailedEvent, error) {
-	return v.latestFailedEvent(metadataTable, sequence)
-}
-
-func (v *View) ProcessedMetadataFailedEvent(failedEvent *global_view.FailedEvent) error {
-	return v.saveFailedEvent(failedEvent)
-}

internal/auth/repository/user.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/caos/zitadel/internal/domain"
 	"github.com/caos/zitadel/internal/user/model"
 )
 
@@ -12,34 +11,8 @@ type UserRepository interface {
 	myUserRepo
 
 	UserSessionUserIDsByAgentID(ctx context.Context, agentID string) ([]string, error)
-
-	UserByID(ctx context.Context, userID string) (*model.UserView, error)
-	UserByLoginName(ctx context.Context, loginName string) (*model.UserView, error)
-
-	SearchUsers(ctx context.Context, request *model.UserSearchRequest) (*model.UserSearchResponse, error)
-
-	SearchUserMetadata(ctx context.Context, userID string) (*domain.MetadataSearchResponse, error)
 }
 
 type myUserRepo interface {
-	MyUser(ctx context.Context) (*model.UserView, error)
-
-	MyProfile(ctx context.Context) (*model.Profile, error)
-
-	MyEmail(ctx context.Context) (*model.Email, error)
-
-	MyPhone(ctx context.Context) (*model.Phone, error)
-
-	MyAddress(ctx context.Context) (*model.Address, error)
-
-	SearchMyExternalIDPs(ctx context.Context, request *model.ExternalIDPSearchRequest) (*model.ExternalIDPSearchResponse, error)
-
-	MyUserMFAs(ctx context.Context) ([]*model.MultiFactor, error)
-
-	GetMyPasswordless(ctx context.Context) ([]*model.WebAuthNView, error)
-
 	MyUserChanges(ctx context.Context, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*model.UserChanges, error)
-
-	GetMyMetadataByKey(ctx context.Context, key string) (*domain.Metadata, error)
-	SearchMyMetadata(ctx context.Context, req *domain.MetadataSearchRequest) (*domain.MetadataSearchResponse, error)
 }