TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-12 01:47:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: features query (#2610)

Browse Source
This commit is contained in:
Fabi
2021-11-21 20:22:25 +01:00
committed by GitHub
parent 56e10ecf30
commit 76346cb070
46 changed files with 604 additions and 1619 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
internal/authz/repository/eventsourcing/eventstore/token_verifier.go
View File

@@ -15,10 +15,10 @@ import (
v1 "github.com/caos/zitadel/internal/eventstore/v1"
"github.com/caos/zitadel/internal/eventstore/v1/models"
es_sdk "github.com/caos/zitadel/internal/eventstore/v1/sdk"
features_view_model "github.com/caos/zitadel/internal/features/repository/view/model"
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"
"github.com/caos/zitadel/internal/query"
"github.com/caos/zitadel/internal/telemetry/tracing"
usr_model "github.com/caos/zitadel/internal/user/model"
usr_view "github.com/caos/zitadel/internal/user/repository/view"
@@ -30,6 +30,7 @@ type TokenVerifierRepo struct {
IAMID string
Eventstore v1.Eventstore
View *view.View
Query *query.Queries
}
func (repo *TokenVerifierRepo) TokenByID(ctx context.Context, tokenID, userID string) (*usr_model.TokenView, error) {
@@ -110,17 +111,14 @@ func (repo *TokenVerifierRepo) ProjectIDAndOriginsByClientID(ctx context.Context
}
func (repo *TokenVerifierRepo) CheckOrgFeatures(ctx context.Context, orgID string, requiredFeatures ...string) error {
features, err := repo.View.FeaturesByAggregateID(orgID)
if caos_errs.IsNotFound(err) {
return repo.checkDefaultFeatures(ctx, requiredFeatures...)
}
features, err := repo.Query.FeaturesByOrgID(ctx, orgID)
if err != nil {
return err
}
return checkFeatures(features, requiredFeatures...)
}
func checkFeatures(features *features_view_model.FeaturesView, requiredFeatures ...string) error {
func checkFeatures(features *query.Features, requiredFeatures ...string) error {
for _, requiredFeature := range requiredFeatures {
if strings.HasPrefix(requiredFeature, domain.FeatureLoginPolicy) {
if err := checkLoginPolicyFeatures(features, requiredFeature); err != nil {
@@ -187,7 +185,7 @@ func checkFeatures(features *features_view_model.FeaturesView, requiredFeatures
return nil
}
func checkLoginPolicyFeatures(features *features_view_model.FeaturesView, requiredFeature string) error {
func checkLoginPolicyFeatures(features *query.Features, requiredFeature string) error {
switch requiredFeature {
case domain.FeatureLoginPolicyFactors:
if !features.LoginPolicyFactors {
@@ -221,7 +219,7 @@ func checkLoginPolicyFeatures(features *features_view_model.FeaturesView, requir
return nil
}
func checkLabelPolicyFeatures(features *features_view_model.FeaturesView, requiredFeature string) error {
func checkLabelPolicyFeatures(features *query.Features, requiredFeature string) error {
switch requiredFeature {
case domain.FeatureLabelPolicyPrivateLabel:
if !features.LabelPolicyPrivateLabel {
@@ -280,28 +278,11 @@ func (u *TokenVerifierRepo) getIAMByID(ctx context.Context) (*iam_model.IAM, err
}
func (repo *TokenVerifierRepo) checkDefaultFeatures(ctx context.Context, requiredFeatures ...string) error {
features, viewErr := repo.View.FeaturesByAggregateID(domain.IAMID)
if viewErr != nil && !caos_errs.IsNotFound(viewErr) {
return viewErr
features, err := repo.Query.DefaultFeatures(ctx)
if err != nil {
return err
}
if caos_errs.IsNotFound(viewErr) {
features = new(features_view_model.FeaturesView)
}
events, esErr := repo.getIAMEvents(ctx, features.Sequence)
if caos_errs.IsNotFound(viewErr) && len(events) == 0 {
return checkFeatures(features, requiredFeatures...)
}
if esErr != nil {
logging.Log("EVENT-PSoc3").WithError(esErr).Debug("error retrieving new events")
return esErr
}
featuresCopy := *features
for _, event := range events {
if err := featuresCopy.AppendEvent(event); err != nil {
return checkFeatures(features, requiredFeatures...)
}
}
return checkFeatures(&featuresCopy, requiredFeatures...)
return checkFeatures(features, requiredFeatures...)
}
func (repo *TokenVerifierRepo) getIAMEvents(ctx context.Context, sequence uint64) ([]*models.Event, error) {

169
internal/authz/repository/eventsourcing/handler/features.go
View File

@@ -1,169 +0,0 @@
package handler
import (
"context"
"github.com/caos/logging"
"github.com/caos/zitadel/internal/domain"
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"
"github.com/caos/zitadel/internal/features/repository/view/model"
"github.com/caos/zitadel/internal/iam/repository/eventsourcing"
iam_es_model "github.com/caos/zitadel/internal/iam/repository/eventsourcing/model"
org_es_model "github.com/caos/zitadel/internal/org/repository/eventsourcing/model"
iam_repo "github.com/caos/zitadel/internal/repository/iam"
org_repo "github.com/caos/zitadel/internal/repository/org"
)
const (
featuresTable = "authz.features"
)
type Features struct {
handler
subscription *v1.Subscription
}
func newFeatures(handler handler) *Features {
h := &Features{
handler: handler,
}
h.subscribe()
return h
}
func (p *Features) subscribe() {
p.subscription = p.es.Subscribe(p.AggregateTypes()...)
go func() {
for event := range p.subscription.Events {
query.ReduceEvent(p, event)
}
}()
}
func (p *Features) ViewModel() string {
return featuresTable
}
func (p *Features) Subscription() *v1.Subscription {
return p.subscription
}
func (p *Features) AggregateTypes() []es_models.AggregateType {
return []es_models.AggregateType{iam_es_model.IAMAggregate, org_es_model.OrgAggregate}
}
func (p *Features) EventQuery() (*es_models.SearchQuery, error) {
sequence, err := p.view.GetLatestFeaturesSequence()
if err != nil {
return nil, err
}
return es_models.NewSearchQuery().
AggregateTypeFilter(p.AggregateTypes()...).
LatestSequenceFilter(sequence.CurrentSequence), nil
}
func (p *Features) CurrentSequence() (uint64, error) {
sequence, err := p.view.GetLatestFeaturesSequence()
if err != nil {
return 0, err
}
return sequence.CurrentSequence, nil
}
func (p *Features) Reduce(event *es_models.Event) (err error) {
switch event.AggregateType {
case org_es_model.OrgAggregate, iam_es_model.IAMAggregate:
err = p.processFeatures(event)
}
return err
}
func (p *Features) processFeatures(event *es_models.Event) (err error) {
features := new(model.FeaturesView)
switch string(event.Type) {
case string(org_es_model.OrgAdded):
features, err = p.getDefaultFeatures()
if err != nil {
return err
}
features.AggregateID = event.AggregateID
features.Default = true
case string(iam_repo.FeaturesSetEventType):
defaultFeatures, err := p.view.AllDefaultFeatures()
if err != nil {
return err
}
for _, features := range defaultFeatures {
err = features.AppendEvent(event)
if err != nil {
return err
}
}
return p.view.PutFeaturesList(defaultFeatures, event)
case string(org_repo.FeaturesSetEventType):
features, err = p.view.FeaturesByAggregateID(event.AggregateID)
if err != nil {
return err
}
err = features.AppendEvent(event)
case string(org_repo.FeaturesRemovedEventType):
features, err = p.getDefaultFeatures()
if err != nil {
return err
}
features.AggregateID = event.AggregateID
features.Default = true
default:
return p.view.ProcessedFeaturesSequence(event)
}
if err != nil {
return err
}
return p.view.PutFeatures(features, event)
}
func (p *Features) OnError(event *es_models.Event, err error) error {
logging.LogWithFields("SPOOL-Wj8sf", "id", event.AggregateID).WithError(err).Warn("something went wrong in login features handler")
return spooler.HandleError(event, err, p.view.GetLatestFeaturesFailedEvent, p.view.ProcessedFeaturesFailedEvent, p.view.ProcessedFeaturesSequence, p.errorCountUntilSkip)
}
func (p *Features) OnSuccess() error {
return spooler.HandleSuccess(p.view.UpdateFeaturesSpoolerRunTimestamp)
}
func (p *Features) getDefaultFeatures() (*model.FeaturesView, error) {
features, featuresErr := p.view.FeaturesByAggregateID(domain.IAMID)
if featuresErr != nil && !caos_errs.IsNotFound(featuresErr) {
return nil, featuresErr
}
if features == nil {
features = &model.FeaturesView{}
}
events, err := p.getIAMEvents(features.Sequence)
if err != nil {
return features, featuresErr
}
featuresCopy := *features
for _, event := range events {
if err := featuresCopy.AppendEvent(event); err != nil {
return features, nil
}
}
return &featuresCopy, nil
}
func (p *Features) getIAMEvents(sequence uint64) ([]*es_models.Event, error) {
query, err := eventsourcing.IAMByIDQuery(domain.IAMID, sequence)
if err != nil {
return nil, err
}
return p.es.FilterEvents(context.Background(), query)
}

2
internal/authz/repository/eventsourcing/handler/handler.go
View File

@@ -38,8 +38,6 @@ func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es
handler{view, bulkLimit, configs.cycleDuration("UserMemberships"), errorCount, es}),
newApplication(
handler{view, bulkLimit, configs.cycleDuration("Application"), errorCount, es}),
newFeatures(
handler{view, bulkLimit, configs.cycleDuration("Features"), errorCount, es}),
}
}

1
internal/authz/repository/eventsourcing/repository.go
View File

@@ -72,6 +72,7 @@ func Start(conf Config, authZ authz.Config, systemDefaults sd.SystemDefaults, qu
Eventstore: es,
IAMID: systemDefaults.IamID,
View: view,
Query: queries,
},
}, nil
}

56
internal/authz/repository/eventsourcing/view/features.go
View File

@@ -1,56 +0,0 @@
package view
import (
"github.com/caos/zitadel/internal/eventstore/v1/models"
"github.com/caos/zitadel/internal/features/repository/view"
"github.com/caos/zitadel/internal/features/repository/view/model"
global_view "github.com/caos/zitadel/internal/view/repository"
)
const (
featuresTable = "authz.features"
)
func (v *View) AllDefaultFeatures() ([]*model.FeaturesView, error) {
return view.GetDefaultFeatures(v.Db, featuresTable)
}
func (v *View) FeaturesByAggregateID(aggregateID string) (*model.FeaturesView, error) {
return view.GetFeaturesByAggregateID(v.Db, featuresTable, aggregateID)
}
func (v *View) PutFeatures(features *model.FeaturesView, event *models.Event) error {
err := view.PutFeatures(v.Db, featuresTable, features)
if err != nil {
return err
}
return v.ProcessedFeaturesSequence(event)
}
func (v *View) PutFeaturesList(features []*model.FeaturesView, event *models.Event) error {
err := view.PutFeaturesList(v.Db, featuresTable, features...)
if err != nil {
return err
}
return v.ProcessedFeaturesSequence(event)
}
func (v *View) GetLatestFeaturesSequence() (*global_view.CurrentSequence, error) {
return v.latestSequence(featuresTable)
}
func (v *View) ProcessedFeaturesSequence(event *models.Event) error {
return v.saveCurrentSequence(featuresTable, event)
}
func (v *View) UpdateFeaturesSpoolerRunTimestamp() error {
return v.updateSpoolerRunSequence(featuresTable)
}
func (v *View) GetLatestFeaturesFailedEvent(sequence uint64) (*global_view.FailedEvent, error) {
return v.latestFailedEvent(featuresTable, sequence)
}
func (v *View) ProcessedFeaturesFailedEvent(failedEvent *global_view.FailedEvent) error {
return v.saveFailedEvent(failedEvent)
}