feat: Iam projection (#3074)

* feat: implement projection for iam and clean up code

* feat: add migration

* fix: remove unused tests

* fix: handler

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

@@ -8,6 +8,7 @@ import (
 	"github.com/caos/zitadel/internal/config/types"
 	v1 "github.com/caos/zitadel/internal/eventstore/v1"
 	"github.com/caos/zitadel/internal/eventstore/v1/query"
+	query2 "github.com/caos/zitadel/internal/query"
 )
 
 type Configs map[string]*Config
@@ -29,11 +30,11 @@ func (h *handler) Eventstore() v1.Eventstore {
 	return h.es
 }
 
-func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es v1.Eventstore, systemDefaults sd.SystemDefaults) []query.Handler {
+func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es v1.Eventstore, systemDefaults sd.SystemDefaults, queries *query2.Queries) []query.Handler {
 	return []query.Handler{
 		newUser(
 			handler{view, bulkLimit, configs.cycleDuration("User"), errorCount, es},
-			systemDefaults.IamID),
+			systemDefaults.IamID, queries),
 		newUserSession(
 			handler{view, bulkLimit, configs.cycleDuration("UserSession"), errorCount, es}),
 		newToken(
@@ -45,10 +46,10 @@ func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es
 			handler{view, bulkLimit, configs.cycleDuration("IDPConfig"), errorCount, es}),
 		newIDPProvider(
 			handler{view, bulkLimit, configs.cycleDuration("IDPProvider"), errorCount, es},
-			systemDefaults),
+			systemDefaults, queries),
 		newExternalIDP(
 			handler{view, bulkLimit, configs.cycleDuration("ExternalIDP"), errorCount, es},
-			systemDefaults),
+			systemDefaults, queries),
 		newRefreshToken(handler{view, bulkLimit, configs.cycleDuration("RefreshToken"), errorCount, es}),
 		newOrgProjectMapping(handler{view, bulkLimit, configs.cycleDuration("OrgProjectMapping"), errorCount, es}),
 	}

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

@@ -2,13 +2,10 @@ package handler
 
 import (
 	"context"
+
 	"github.com/caos/zitadel/internal/domain"
-	"github.com/caos/zitadel/internal/errors"
 	"github.com/caos/zitadel/internal/eventstore/v1"
-	es_sdk "github.com/caos/zitadel/internal/eventstore/v1/sdk"
-	iam_view "github.com/caos/zitadel/internal/iam/repository/view"
-	org_model "github.com/caos/zitadel/internal/org/model"
-	"github.com/caos/zitadel/internal/org/repository/view"
+	query2 "github.com/caos/zitadel/internal/query"
 
 	"github.com/caos/logging"
 	"github.com/caos/zitadel/internal/config/systemdefaults"
@@ -30,15 +27,18 @@ type IDPProvider struct {
 	handler
 	systemDefaults systemdefaults.SystemDefaults
 	subscription   *v1.Subscription
+	queries        *query2.Queries
 }
 
 func newIDPProvider(
 	h handler,
 	defaults systemdefaults.SystemDefaults,
+	queries *query2.Queries,
 ) *IDPProvider {
 	idpProvider := &IDPProvider{
 		handler:        h,
 		systemDefaults: defaults,
+		queries:        queries,
 	}
 
 	idpProvider.subscribe()
@@ -120,7 +120,7 @@ func (i *IDPProvider) processIdpProvider(event *es_models.Event) (err error) {
 		if err != nil {
 			return err
 		}
-		config := new(iam_model.IDPConfig)
+		config := new(query2.IDP)
 		if event.AggregateID == i.systemDefaults.IamID {
 			config, err = i.getDefaultIDPConfig(context.TODO(), esConfig.IDPConfigID)
 		} else {
@@ -145,7 +145,7 @@ func (i *IDPProvider) processIdpProvider(event *es_models.Event) (err error) {
 }
 
 func (i *IDPProvider) fillData(provider *iam_view_model.IDPProviderView) (err error) {
-	var config *iam_model.IDPConfig
+	var config *query2.IDP
 	if provider.IDPProviderType == int32(iam_model.IDPProviderTypeSystem) {
 		config, err = i.getDefaultIDPConfig(context.Background(), provider.IDPConfigID)
 	} else {
@@ -158,10 +158,14 @@ func (i *IDPProvider) fillData(provider *iam_view_model.IDPProviderView) (err er
 	return nil
 }
 
-func (i *IDPProvider) fillConfigData(provider *iam_view_model.IDPProviderView, config *iam_model.IDPConfig) {
+func (i *IDPProvider) fillConfigData(provider *iam_view_model.IDPProviderView, config *query2.IDP) {
 	provider.Name = config.Name
 	provider.StylingType = int32(config.StylingType)
-	provider.IDPConfigType = int32(config.Type)
+	if config.OIDCIDP != nil {
+		provider.IDPConfigType = int32(domain.IDPConfigTypeOIDC)
+	} else if config.JWTIDP != nil {
+		provider.IDPConfigType = int32(domain.IDPConfigTypeJWT)
+	}
 	provider.IDPState = int32(config.State)
 }
 
@@ -174,63 +178,10 @@ func (i *IDPProvider) OnSuccess() error {
 	return spooler.HandleSuccess(i.view.UpdateIDPProviderSpoolerRunTimestamp)
 }
 
-func (i *IDPProvider) getOrgIDPConfig(ctx context.Context, aggregateID, idpConfigID string) (*iam_model.IDPConfig, error) {
-	existing, err := i.getOrgByID(ctx, aggregateID)
-	if err != nil {
-		return nil, err
-	}
-	if _, i := existing.GetIDP(idpConfigID); i != nil {
-		return i, nil
-	}
-	return nil, errors.ThrowNotFound(nil, "EVENT-2m9fS", "Errors.IDP.NotExisting")
+func (i *IDPProvider) getOrgIDPConfig(ctx context.Context, aggregateID, idpConfigID string) (*query2.IDP, error) {
+	return i.queries.IDPByIDAndResourceOwner(ctx, idpConfigID, aggregateID)
 }
 
-func (i *IDPProvider) getOrgByID(ctx context.Context, orgID string) (*org_model.Org, error) {
-	query, err := view.OrgByIDQuery(orgID, 0)
-	if err != nil {
-		return nil, err
-	}
-
-	esOrg := &org_es_model.Org{
-		ObjectRoot: es_models.ObjectRoot{
-			AggregateID: orgID,
-		},
-	}
-	err = es_sdk.Filter(ctx, i.Eventstore().FilterEvents, esOrg.AppendEvents, query)
-	if err != nil && !errors.IsNotFound(err) {
-		return nil, err
-	}
-	if esOrg.Sequence == 0 {
-		return nil, errors.ThrowNotFound(nil, "EVENT-6m0fS", "Errors.Org.NotFound")
-	}
-
-	return org_es_model.OrgToModel(esOrg), nil
-}
-
-func (u *IDPProvider) getIAMByID(ctx context.Context) (*iam_model.IAM, error) {
-	query, err := iam_view.IAMByIDQuery(domain.IAMID, 0)
-	if err != nil {
-		return nil, err
-	}
-	iam := &model.IAM{
-		ObjectRoot: es_models.ObjectRoot{
-			AggregateID: domain.IAMID,
-		},
-	}
-	err = es_sdk.Filter(ctx, u.Eventstore().FilterEvents, iam.AppendEvents, query)
-	if err != nil && errors.IsNotFound(err) && iam.Sequence == 0 {
-		return nil, err
-	}
-	return model.IAMToModel(iam), nil
-}
-
-func (u *IDPProvider) getDefaultIDPConfig(ctx context.Context, idpConfigID string) (*iam_model.IDPConfig, error) {
-	existing, err := u.getIAMByID(ctx)
-	if err != nil {
-		return nil, err
-	}
-	if _, existingIDP := existing.GetIDP(idpConfigID); existingIDP != nil {
-		return existingIDP, nil
-	}
-	return nil, errors.ThrowNotFound(nil, "EVENT-49O0f", "Errors.IDP.NotExisting")
+func (u *IDPProvider) getDefaultIDPConfig(ctx context.Context, idpConfigID string) (*query2.IDP, error) {
+	return u.queries.IDPByIDAndResourceOwner(ctx, idpConfigID, domain.IAMID)
 }

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

@@ -2,20 +2,18 @@ package handler
 
 import (
 	"context"
+
 	"github.com/caos/logging"
-	"github.com/caos/zitadel/internal/domain"
 	"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"
 	es_sdk "github.com/caos/zitadel/internal/eventstore/v1/sdk"
 	"github.com/caos/zitadel/internal/eventstore/v1/spooler"
-	iam_model "github.com/caos/zitadel/internal/iam/model"
-	"github.com/caos/zitadel/internal/iam/repository/eventsourcing/model"
-	iam_view "github.com/caos/zitadel/internal/iam/repository/view"
 	org_model "github.com/caos/zitadel/internal/org/model"
 	org_es_model "github.com/caos/zitadel/internal/org/repository/eventsourcing/model"
 	"github.com/caos/zitadel/internal/org/repository/view"
+	query2 "github.com/caos/zitadel/internal/query"
 	user_repo "github.com/caos/zitadel/internal/repository/user"
 	es_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
 	view_model "github.com/caos/zitadel/internal/user/repository/view/model"
@@ -29,15 +27,18 @@ type User struct {
 	handler
 	iamID        string
 	subscription *v1.Subscription
+	queries      *query2.Queries
 }
 
 func newUser(
 	handler handler,
 	iamID string,
+	queries *query2.Queries,
 ) *User {
 	h := &User{
 		handler: handler,
 		iamID:   iamID,
+		queries: queries,
 	}
 
 	h.subscribe()
@@ -178,7 +179,7 @@ func (u *User) fillLoginNames(user *view_model.UserView) (err error) {
 	if err != nil {
 		return err
 	}
-	policy := org.OrgIamPolicy
+	policy := new(query2.OrgIAMPolicy)
 	if policy == nil {
 		policy, err = u.getDefaultOrgIAMPolicy(context.Background())
 		if err != nil {
@@ -210,7 +211,7 @@ func (u *User) fillLoginNamesOnOrgUsers(event *es_models.Event) error {
 	if err != nil {
 		return err
 	}
-	policy := org.OrgIamPolicy
+	policy := new(query2.OrgIAMPolicy)
 	if policy == nil {
 		policy, err = u.getDefaultOrgIAMPolicy(context.Background())
 		if err != nil {
@@ -232,7 +233,7 @@ func (u *User) fillPreferredLoginNamesOnOrgUsers(event *es_models.Event) error {
 	if err != nil {
 		return err
 	}
-	policy := org.OrgIamPolicy
+	policy := new(query2.OrgIAMPolicy)
 	if policy == nil {
 		policy, err = u.getDefaultOrgIAMPolicy(context.Background())
 		if err != nil {
@@ -283,30 +284,6 @@ func (u *User) getOrgByID(ctx context.Context, orgID string) (*org_model.Org, er
 	return org_es_model.OrgToModel(esOrg), nil
 }
 
-func (u *User) getIAMByID(ctx context.Context) (*iam_model.IAM, error) {
-	query, err := iam_view.IAMByIDQuery(domain.IAMID, 0)
-	if err != nil {
-		return nil, err
-	}
-	iam := &model.IAM{
-		ObjectRoot: es_models.ObjectRoot{
-			AggregateID: domain.IAMID,
-		},
-	}
-	err = es_sdk.Filter(ctx, u.Eventstore().FilterEvents, iam.AppendEvents, query)
-	if err != nil && errors.IsNotFound(err) && iam.Sequence == 0 {
-		return nil, err
-	}
-	return model.IAMToModel(iam), nil
-}
-
-func (u *User) getDefaultOrgIAMPolicy(ctx context.Context) (*iam_model.OrgIAMPolicy, error) {
-	existingIAM, err := u.getIAMByID(ctx)
-	if err != nil {
-		return nil, err
-	}
-	if existingIAM.DefaultOrgIAMPolicy == nil {
-		return nil, errors.ThrowNotFound(nil, "EVENT-3m9fs", "Errors.IAM.OrgIAMPolicy.NotExisting")
-	}
-	return existingIAM.DefaultOrgIAMPolicy, nil
+func (u *User) getDefaultOrgIAMPolicy(ctx context.Context) (*query2.OrgIAMPolicy, error) {
+	return u.queries.DefaultOrgIAMPolicy(ctx)
 }

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

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"context"
+
 	"github.com/caos/logging"
 	"github.com/caos/zitadel/internal/config/systemdefaults"
 	"github.com/caos/zitadel/internal/domain"
@@ -9,15 +10,12 @@ import (
 	"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"
-	es_sdk "github.com/caos/zitadel/internal/eventstore/v1/sdk"
 	"github.com/caos/zitadel/internal/eventstore/v1/spooler"
 	iam_model "github.com/caos/zitadel/internal/iam/model"
 	iam_es_model "github.com/caos/zitadel/internal/iam/repository/eventsourcing/model"
-	iam_view "github.com/caos/zitadel/internal/iam/repository/view"
 	iam_view_model "github.com/caos/zitadel/internal/iam/repository/view/model"
-	org_model "github.com/caos/zitadel/internal/org/model"
 	org_es_model "github.com/caos/zitadel/internal/org/repository/eventsourcing/model"
-	"github.com/caos/zitadel/internal/org/repository/view"
+	query2 "github.com/caos/zitadel/internal/query"
 	"github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
 	usr_view_model "github.com/caos/zitadel/internal/user/repository/view/model"
 )
@@ -30,15 +28,18 @@ type ExternalIDP struct {
 	handler
 	systemDefaults systemdefaults.SystemDefaults
 	subscription   *v1.Subscription
+	queries        *query2.Queries
 }
 
 func newExternalIDP(
 	handler handler,
 	defaults systemdefaults.SystemDefaults,
+	queries *query2.Queries,
 ) *ExternalIDP {
 	h := &ExternalIDP{
 		handler:        handler,
 		systemDefaults: defaults,
+		queries:        queries,
 	}
 
 	h.subscribe()
@@ -125,7 +126,7 @@ func (i *ExternalIDP) processIdpConfig(event *es_models.Event) (err error) {
 	switch event.Type {
 	case iam_es_model.IDPConfigChanged, org_es_model.IDPConfigChanged:
 		configView := new(iam_view_model.IDPConfigView)
-		config := new(iam_model.IDPConfig)
+		config := new(query2.IDP)
 		if event.Type == iam_es_model.IDPConfigChanged {
 			configView.AppendEvent(iam_model.IDPProviderTypeSystem, event)
 		} else {
@@ -165,7 +166,7 @@ func (i *ExternalIDP) fillData(externalIDP *usr_view_model.ExternalIDPView) erro
 	return nil
 }
 
-func (i *ExternalIDP) fillConfigData(externalIDP *usr_view_model.ExternalIDPView, config *iam_model.IDPConfig) {
+func (i *ExternalIDP) fillConfigData(externalIDP *usr_view_model.ExternalIDPView, config *query2.IDP) {
 	externalIDP.IDPName = config.Name
 }
 
@@ -178,63 +179,10 @@ func (i *ExternalIDP) OnSuccess() error {
 	return spooler.HandleSuccess(i.view.UpdateExternalIDPSpoolerRunTimestamp)
 }
 
-func (i *ExternalIDP) getOrgIDPConfig(ctx context.Context, aggregateID, idpConfigID string) (*iam_model.IDPConfig, error) {
-	existing, err := i.getOrgByID(ctx, aggregateID)
-	if err != nil {
-		return nil, err
-	}
-	if _, i := existing.GetIDP(idpConfigID); i != nil {
-		return i, nil
-	}
-	return nil, caos_errs.ThrowNotFound(nil, "EVENT-2m9fS", "Errors.Org.IdpNotExisting")
+func (i *ExternalIDP) getOrgIDPConfig(ctx context.Context, aggregateID, idpConfigID string) (*query2.IDP, error) {
+	return i.queries.IDPByIDAndResourceOwner(ctx, idpConfigID, aggregateID)
 }
 
-func (i *ExternalIDP) getOrgByID(ctx context.Context, orgID string) (*org_model.Org, error) {
-	query, err := view.OrgByIDQuery(orgID, 0)
-	if err != nil {
-		return nil, err
-	}
-
-	esOrg := &org_es_model.Org{
-		ObjectRoot: es_models.ObjectRoot{
-			AggregateID: orgID,
-		},
-	}
-	err = es_sdk.Filter(ctx, i.Eventstore().FilterEvents, esOrg.AppendEvents, query)
-	if err != nil && !caos_errs.IsNotFound(err) {
-		return nil, err
-	}
-	if esOrg.Sequence == 0 {
-		return nil, caos_errs.ThrowNotFound(nil, "EVENT-6m0fS", "Errors.Org.NotFound")
-	}
-
-	return org_es_model.OrgToModel(esOrg), nil
-}
-
-func (u *ExternalIDP) getIAMByID(ctx context.Context) (*iam_model.IAM, error) {
-	query, err := iam_view.IAMByIDQuery(domain.IAMID, 0)
-	if err != nil {
-		return nil, err
-	}
-	iam := &iam_es_model.IAM{
-		ObjectRoot: es_models.ObjectRoot{
-			AggregateID: domain.IAMID,
-		},
-	}
-	err = es_sdk.Filter(ctx, u.Eventstore().FilterEvents, iam.AppendEvents, query)
-	if err != nil && caos_errs.IsNotFound(err) && iam.Sequence == 0 {
-		return nil, err
-	}
-	return iam_es_model.IAMToModel(iam), nil
-}
-
-func (u *ExternalIDP) getDefaultIDPConfig(ctx context.Context, idpConfigID string) (*iam_model.IDPConfig, error) {
-	existing, err := u.getIAMByID(ctx)
-	if err != nil {
-		return nil, err
-	}
-	if _, existingIDP := existing.GetIDP(idpConfigID); existingIDP != nil {
-		return existingIDP, nil
-	}
-	return nil, caos_errs.ThrowNotFound(nil, "EVENT-mmk5d", "Errors.IAM.IdpNotExisting")
+func (i *ExternalIDP) getDefaultIDPConfig(ctx context.Context, idpConfigID string) (*query2.IDP, error) {
+	return i.queries.IDPByIDAndResourceOwner(ctx, idpConfigID, domain.IAMID)
 }

internal/auth/repository/eventsourcing/repository.go

@@ -79,7 +79,7 @@ func Start(conf Config, authZ authz.Config, systemDefaults sd.SystemDefaults, co
 	statikLoginFS, err := fs.NewWithNamespace("login")
 	logging.Log("CONFI-20opp").OnError(err).Panic("unable to start login statik dir")
 
-	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, systemDefaults)
+	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, systemDefaults, queries)
 
 	userRepo := eventstore.UserRepo{
 		SearchLimit:     conf.SearchLimit,

internal/auth/repository/eventsourcing/spooler/spooler.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 
 	v1 "github.com/caos/zitadel/internal/eventstore/v1"
+	"github.com/caos/zitadel/internal/query"
 
 	"github.com/caos/zitadel/internal/auth/repository/eventsourcing/handler"
 	"github.com/caos/zitadel/internal/auth/repository/eventsourcing/view"
@@ -18,12 +19,12 @@ type SpoolerConfig struct {
 	Handlers              handler.Configs
 }
 
-func StartSpooler(c SpoolerConfig, es v1.Eventstore, view *view.View, client *sql.DB, systemDefaults sd.SystemDefaults) *spooler.Spooler {
+func StartSpooler(c SpoolerConfig, es v1.Eventstore, view *view.View, client *sql.DB, systemDefaults sd.SystemDefaults, queries *query.Queries) *spooler.Spooler {
 	spoolerConfig := spooler.Config{
 		Eventstore:        es,
 		Locker:            &locker{dbClient: client},
 		ConcurrentWorkers: c.ConcurrentWorkers,
-		ViewHandlers:      handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, es, systemDefaults),
+		ViewHandlers:      handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, es, systemDefaults, queries),
 	}
 	spool := spoolerConfig.New()
 	spool.Start()