fix: claim verified domain from usernames (#603)

* fix: return orgDomain validationType

* added missing translations for orgDomain activity

* claim org domain

* show message if domain token was requested

* fix tests

* fix tests

Co-authored-by: Max Peintner <max@caos.ch>

internal/admin/repository/eventsourcing/eventstore/org.go

@@ -2,11 +2,13 @@ package eventstore
 
 import (
 	"context"
+
 	"github.com/caos/logging"
 
 	admin_model "github.com/caos/zitadel/internal/admin/model"
 	admin_view "github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
 	"github.com/caos/zitadel/internal/eventstore"
+	es_models "github.com/caos/zitadel/internal/eventstore/models"
 	"github.com/caos/zitadel/internal/eventstore/sdk"
 	org_model "github.com/caos/zitadel/internal/org/model"
 	org_es "github.com/caos/zitadel/internal/org/repository/eventsourcing"
@@ -40,7 +42,14 @@ func (repo *OrgRepo) SetUpOrg(ctx context.Context, setUp *admin_model.SetupOrg)
 	if err != nil {
 		return nil, err
 	}
-	org, aggregates, err := repo.OrgEventstore.PrepareCreateOrg(ctx, setUp.Org)
+	users := func(ctx context.Context, domain string) ([]*es_models.Aggregate, error) {
+		userIDs, err := repo.View.UserIDsByDomain(domain)
+		if err != nil {
+			return nil, err
+		}
+		return repo.UserEventstore.PrepareDomainClaimed(ctx, userIDs)
+	}
+	org, aggregates, err := repo.OrgEventstore.PrepareCreateOrg(ctx, setUp.Org, users)
 	if err != nil {
 		return nil, err
 	}

internal/admin/repository/eventsourcing/handler/handler.go

@@ -5,7 +5,9 @@ import (
 
 	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
 	"github.com/caos/zitadel/internal/config/types"
+	"github.com/caos/zitadel/internal/eventstore"
 	"github.com/caos/zitadel/internal/eventstore/query"
+	org_event "github.com/caos/zitadel/internal/org/repository/eventsourcing"
 	usr_event "github.com/caos/zitadel/internal/user/repository/eventsourcing"
 )
 
@@ -24,12 +26,14 @@ type handler struct {
 
 type EventstoreRepos struct {
 	UserEvents *usr_event.UserEventstore
+	OrgEvents  *org_event.OrgEventstore
 }
 
-func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, repos EventstoreRepos) []query.Handler {
+func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, eventstore eventstore.Eventstore, repos EventstoreRepos) []query.Handler {
 	return []query.Handler{
 		&Org{handler: handler{view, bulkLimit, configs.cycleDuration("Org"), errorCount}},
 		&IamMember{handler: handler{view, bulkLimit, configs.cycleDuration("IamMember"), errorCount}, userEvents: repos.UserEvents},
+		&User{handler: handler{view, bulkLimit, configs.cycleDuration("User"), errorCount}, eventstore: eventstore, orgEvents: repos.OrgEvents},
 	}
 }
 

internal/admin/repository/eventsourcing/handler/user.go

@@ -0,0 +1,177 @@
+package handler
+
+import (
+	"context"
+
+	"github.com/caos/logging"
+
+	"github.com/caos/zitadel/internal/eventstore"
+	"github.com/caos/zitadel/internal/eventstore/models"
+	es_models "github.com/caos/zitadel/internal/eventstore/models"
+	"github.com/caos/zitadel/internal/eventstore/spooler"
+	org_model "github.com/caos/zitadel/internal/org/model"
+	org_events "github.com/caos/zitadel/internal/org/repository/eventsourcing"
+	org_es_model "github.com/caos/zitadel/internal/org/repository/eventsourcing/model"
+	es_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
+	view_model "github.com/caos/zitadel/internal/user/repository/view/model"
+)
+
+type User struct {
+	handler
+	eventstore eventstore.Eventstore
+	orgEvents  *org_events.OrgEventstore
+}
+
+const (
+	userTable = "adminapi.users"
+)
+
+func (u *User) ViewModel() string {
+	return userTable
+}
+
+func (u *User) EventQuery() (*models.SearchQuery, error) {
+	sequence, err := u.view.GetLatestUserSequence()
+	if err != nil {
+		return nil, err
+	}
+	return es_models.NewSearchQuery().
+		AggregateTypeFilter(es_model.UserAggregate, org_es_model.OrgAggregate).
+		LatestSequenceFilter(sequence.CurrentSequence), nil
+}
+
+func (u *User) Reduce(event *models.Event) (err error) {
+	switch event.AggregateType {
+	case es_model.UserAggregate:
+		return u.ProcessUser(event)
+	case org_es_model.OrgAggregate:
+		return u.ProcessOrg(event)
+	default:
+		return nil
+	}
+}
+
+func (u *User) ProcessUser(event *models.Event) (err error) {
+	user := new(view_model.UserView)
+	switch event.Type {
+	case es_model.UserAdded,
+		es_model.UserRegistered:
+		err = user.AppendEvent(event)
+		if err != nil {
+			return err
+		}
+		err = u.fillLoginNames(user)
+	case es_model.UserProfileChanged,
+		es_model.UserEmailChanged,
+		es_model.UserEmailVerified,
+		es_model.UserPhoneChanged,
+		es_model.UserPhoneVerified,
+		es_model.UserPhoneRemoved,
+		es_model.UserAddressChanged,
+		es_model.UserDeactivated,
+		es_model.UserReactivated,
+		es_model.UserLocked,
+		es_model.UserUnlocked,
+		es_model.MfaOtpAdded,
+		es_model.MfaOtpVerified,
+		es_model.MfaOtpRemoved:
+		user, err = u.view.UserByID(event.AggregateID)
+		if err != nil {
+			return err
+		}
+		err = user.AppendEvent(event)
+	case es_model.DomainClaimed:
+		user, err = u.view.UserByID(event.AggregateID)
+		if err != nil {
+			return err
+		}
+		err = user.AppendEvent(event)
+		if err != nil {
+			return err
+		}
+		err = u.fillLoginNames(user)
+	case es_model.UserRemoved:
+		err = u.view.DeleteUser(event.AggregateID, event.Sequence)
+	default:
+		return u.view.ProcessedUserSequence(event.Sequence)
+	}
+	if err != nil {
+		return err
+	}
+	return u.view.PutUser(user, user.Sequence)
+}
+
+func (u *User) ProcessOrg(event *models.Event) (err error) {
+	switch event.Type {
+	case org_es_model.OrgDomainVerified,
+		org_es_model.OrgDomainRemoved,
+		org_es_model.OrgIamPolicyAdded,
+		org_es_model.OrgIamPolicyChanged,
+		org_es_model.OrgIamPolicyRemoved:
+		return u.fillLoginNamesOnOrgUsers(event)
+	case org_es_model.OrgDomainPrimarySet:
+		return u.fillPreferredLoginNamesOnOrgUsers(event)
+	default:
+		return u.view.ProcessedUserSequence(event.Sequence)
+	}
+}
+
+func (u *User) fillLoginNamesOnOrgUsers(event *models.Event) error {
+	org, err := u.orgEvents.OrgByID(context.Background(), org_model.NewOrg(event.ResourceOwner))
+	if err != nil {
+		return err
+	}
+	policy, err := u.orgEvents.GetOrgIamPolicy(context.Background(), event.ResourceOwner)
+	if err != nil {
+		return err
+	}
+	users, err := u.view.UsersByOrgID(event.AggregateID)
+	if err != nil {
+		return err
+	}
+	for _, user := range users {
+		user.SetLoginNames(policy, org.Domains)
+	}
+	return u.view.PutUsers(users, event.Sequence)
+}
+
+func (u *User) fillPreferredLoginNamesOnOrgUsers(event *models.Event) error {
+	org, err := u.orgEvents.OrgByID(context.Background(), org_model.NewOrg(event.ResourceOwner))
+	if err != nil {
+		return err
+	}
+	policy, err := u.orgEvents.GetOrgIamPolicy(context.Background(), event.ResourceOwner)
+	if err != nil {
+		return err
+	}
+	if !policy.UserLoginMustBeDomain {
+		return nil
+	}
+	users, err := u.view.UsersByOrgID(event.AggregateID)
+	if err != nil {
+		return err
+	}
+	for _, user := range users {
+		user.PreferredLoginName = user.GenerateLoginName(org.GetPrimaryDomain().Domain, policy.UserLoginMustBeDomain)
+	}
+	return u.view.PutUsers(users, event.Sequence)
+}
+
+func (u *User) fillLoginNames(user *view_model.UserView) (err error) {
+	org, err := u.orgEvents.OrgByID(context.Background(), org_model.NewOrg(user.ResourceOwner))
+	if err != nil {
+		return err
+	}
+	policy, err := u.orgEvents.GetOrgIamPolicy(context.Background(), user.ResourceOwner)
+	if err != nil {
+		return err
+	}
+	user.SetLoginNames(policy, org.Domains)
+	user.PreferredLoginName = user.GenerateLoginName(org.GetPrimaryDomain().Domain, policy.UserLoginMustBeDomain)
+	return nil
+}
+
+func (u *User) OnError(event *models.Event, err error) error {
+	logging.LogWithFields("SPOOL-is8wa", "id", event.AggregateID).WithError(err).Warn("something went wrong in user handler")
+	return spooler.HandleError(event, err, u.view.GetLatestUserFailedEvent, u.view.ProcessedUserFailedEvent, u.view.ProcessedUserSequence, u.errorCountUntilSkip)
+}

internal/admin/repository/eventsourcing/repository.go

@@ -86,7 +86,7 @@ func Start(ctx context.Context, conf Config, systemDefaults sd.SystemDefaults, r
 	err = setup.StartSetup(systemDefaults, eventstoreRepos).Execute(ctx)
 	logging.Log("SERVE-djs3R").OnError(err).Panic("failed to execute setup")
 
-	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, handler.EventstoreRepos{UserEvents: user})
+	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, handler.EventstoreRepos{UserEvents: user, OrgEvents: org})
 
 	return &EsRepository{
 		spooler: spool,

internal/admin/repository/eventsourcing/setup/setup.go

@@ -211,7 +211,7 @@ func (setUp *initializer) org(ctx context.Context, org types.Org) (*org_model.Or
 		Name:    org.Name,
 		Domains: []*org_model.OrgDomain{{Domain: org.Domain}},
 	}
-	return setUp.repos.OrgEvents.CreateOrg(ctx, createOrg)
+	return setUp.repos.OrgEvents.CreateOrg(ctx, createOrg, nil)
 }
 
 func (setUp *initializer) iamorgpolicy(ctx context.Context, org *org_model.Org) (*org_model.OrgIamPolicy, error) {

internal/admin/repository/eventsourcing/spooler/spooler.go

@@ -21,7 +21,7 @@ func StartSpooler(c SpoolerConfig, es eventstore.Eventstore, view *view.View, sq
 		Eventstore:        es,
 		Locker:            &locker{dbClient: sql},
 		ConcurrentWorkers: c.ConcurrentWorkers,
-		ViewHandlers:      handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, repos),
+		ViewHandlers:      handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, es, repos),
 	}
 	spool := spoolerConfig.New()
 	spool.Start()

internal/admin/repository/eventsourcing/view/user.go

@@ -0,0 +1,83 @@
+package view
+
+import (
+	usr_model "github.com/caos/zitadel/internal/user/model"
+	"github.com/caos/zitadel/internal/user/repository/view"
+	"github.com/caos/zitadel/internal/user/repository/view/model"
+	"github.com/caos/zitadel/internal/view/repository"
+)
+
+const (
+	userTable = "adminapi.users"
+)
+
+func (v *View) UserByID(userID string) (*model.UserView, error) {
+	return view.UserByID(v.Db, userTable, userID)
+}
+
+func (v *View) SearchUsers(request *usr_model.UserSearchRequest) ([]*model.UserView, uint64, error) {
+	return view.SearchUsers(v.Db, userTable, request)
+}
+
+func (v *View) GetGlobalUserByEmail(email string) (*model.UserView, error) {
+	return view.GetGlobalUserByEmail(v.Db, userTable, email)
+}
+
+func (v *View) UsersByOrgID(orgID string) ([]*model.UserView, error) {
+	return view.UsersByOrgID(v.Db, userTable, orgID)
+}
+
+func (v *View) UserIDsByDomain(domain string) ([]string, error) {
+	return view.UserIDsByDomain(v.Db, userTable, domain)
+}
+
+func (v *View) IsUserUnique(userName, email string) (bool, error) {
+	return view.IsUserUnique(v.Db, userTable, userName, email)
+}
+
+func (v *View) UserMfas(userID string) ([]*usr_model.MultiFactor, error) {
+	return view.UserMfas(v.Db, userTable, userID)
+}
+
+func (v *View) PutUsers(user []*model.UserView, sequence uint64) error {
+	err := view.PutUsers(v.Db, userTable, user...)
+	if err != nil {
+		return err
+	}
+	return v.ProcessedUserSequence(sequence)
+}
+
+func (v *View) PutUser(user *model.UserView, sequence uint64) error {
+	err := view.PutUser(v.Db, userTable, user)
+	if err != nil {
+		return err
+	}
+	if sequence != 0 {
+		return v.ProcessedUserSequence(sequence)
+	}
+	return nil
+}
+
+func (v *View) DeleteUser(userID string, eventSequence uint64) error {
+	err := view.DeleteUser(v.Db, userTable, userID)
+	if err != nil {
+		return nil
+	}
+	return v.ProcessedUserSequence(eventSequence)
+}
+
+func (v *View) GetLatestUserSequence() (*repository.CurrentSequence, error) {
+	return v.latestSequence(userTable)
+}
+
+func (v *View) ProcessedUserSequence(eventSequence uint64) error {
+	return v.saveCurrentSequence(userTable, eventSequence)
+}
+
+func (v *View) GetLatestUserFailedEvent(sequence uint64) (*repository.FailedEvent, error) {
+	return v.latestFailedEvent(userTable, sequence)
+}
+
+func (v *View) ProcessedUserFailedEvent(failedEvent *repository.FailedEvent) error {
+	return v.saveFailedEvent(failedEvent)
+}