feat: features (#1427)

* features

* features

* features

* fix json tags

* add features handler to auth

* mocks for tests

* add setup step

* fixes

* add featurelist to auth api

* grandfather state and typos

* typo

* merge new-eventstore

* fix login policy tests

* label policy in features

* audit log retention

internal/management/repository/eventsourcing/eventstore/features.go

@@ -0,0 +1,70 @@
+package eventstore
+
+import (
+	"context"
+
+	"github.com/caos/logging"
+
+	"github.com/caos/zitadel/internal/config/systemdefaults"
+	"github.com/caos/zitadel/internal/domain"
+	"github.com/caos/zitadel/internal/errors"
+	v1 "github.com/caos/zitadel/internal/eventstore/v1"
+	"github.com/caos/zitadel/internal/eventstore/v1/models"
+	features_model "github.com/caos/zitadel/internal/features/model"
+	"github.com/caos/zitadel/internal/features/repository/view/model"
+	iam_view "github.com/caos/zitadel/internal/iam/repository/view"
+	mgmt_view "github.com/caos/zitadel/internal/management/repository/eventsourcing/view"
+)
+
+type FeaturesRepo struct {
+	Eventstore v1.Eventstore
+
+	View *mgmt_view.View
+
+	SearchLimit    uint64
+	SystemDefaults systemdefaults.SystemDefaults
+}
+
+func (repo *FeaturesRepo) GetDefaultFeatures(ctx context.Context) (*features_model.FeaturesView, error) {
+	features, viewErr := repo.View.FeaturesByAggregateID(domain.IAMID)
+	if viewErr != nil && !errors.IsNotFound(viewErr) {
+		return nil, viewErr
+	}
+	if errors.IsNotFound(viewErr) {
+		features = new(model.FeaturesView)
+	}
+	events, esErr := repo.getIAMEvents(ctx, features.Sequence)
+	if errors.IsNotFound(viewErr) && len(events) == 0 {
+		return nil, errors.ThrowNotFound(nil, "EVENT-Lsoj7", "Errors.Org.NotFound")
+	}
+	if esErr != nil {
+		logging.Log("EVENT-PSoc3").WithError(esErr).Debug("error retrieving new events")
+		return model.FeaturesToModel(features), nil
+	}
+	featuresCopy := *features
+	for _, event := range events {
+		if err := featuresCopy.AppendEvent(event); err != nil {
+			return model.FeaturesToModel(&featuresCopy), nil
+		}
+	}
+	return model.FeaturesToModel(&featuresCopy), nil
+}
+
+func (repo *FeaturesRepo) GetOrgFeatures(ctx context.Context, orgID string) (*features_model.FeaturesView, error) {
+	features, err := repo.View.FeaturesByAggregateID(orgID)
+	if errors.IsNotFound(err) {
+		return repo.GetDefaultFeatures(ctx)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return model.FeaturesToModel(features), err
+}
+
+func (repo *FeaturesRepo) getIAMEvents(ctx context.Context, sequence uint64) ([]*models.Event, error) {
+	query, err := iam_view.IAMByIDQuery(domain.IAMID, sequence)
+	if err != nil {
+		return nil, err
+	}
+	return repo.Eventstore.FilterEvents(ctx, query)
+}

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

@@ -12,6 +12,7 @@ import (
 	"github.com/caos/zitadel/internal/user/repository/view"
 	"github.com/golang/protobuf/ptypes"
 	"strings"
+	"time"
 
 	"github.com/caos/logging"
 	"github.com/caos/zitadel/internal/api/authz"
@@ -93,8 +94,8 @@ func (repo *OrgRepository) SearchMyOrgDomains(ctx context.Context, request *org_
 	return result, nil
 }
 
-func (repo *OrgRepository) OrgChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*org_model.OrgChanges, error) {
-	changes, err := repo.getOrgChanges(ctx, id, lastSequence, limit, sortAscending)
+func (repo *OrgRepository) OrgChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool, auditLogRetention time.Duration) (*org_model.OrgChanges, error) {
+	changes, err := repo.getOrgChanges(ctx, id, lastSequence, limit, sortAscending, auditLogRetention)
 	if err != nil {
 		return nil, err
 	}
@@ -537,8 +538,8 @@ func (repo *OrgRepository) GetMailTexts(ctx context.Context) (*iam_model.MailTex
 	return iam_es_model.MailTextsViewToModel(texts, defaultIn), err
 }
 
-func (repo *OrgRepository) getOrgChanges(ctx context.Context, orgID string, lastSequence uint64, limit uint64, sortAscending bool) (*org_model.OrgChanges, error) {
-	query := org_view.ChangesQuery(orgID, lastSequence, limit, sortAscending)
+func (repo *OrgRepository) getOrgChanges(ctx context.Context, orgID string, lastSequence uint64, limit uint64, sortAscending bool, auditLogRetention time.Duration) (*org_model.OrgChanges, error) {
+	query := org_view.ChangesQuery(orgID, lastSequence, limit, sortAscending, auditLogRetention)
 
 	events, err := repo.Eventstore.FilterEvents(context.Background(), query)
 	if err != nil {

internal/management/repository/eventsourcing/eventstore/project.go

@@ -10,6 +10,7 @@ import (
 	iam_es_model "github.com/caos/zitadel/internal/iam/repository/eventsourcing/model"
 	iam_view "github.com/caos/zitadel/internal/iam/repository/view"
 	"strings"
+	"time"
 
 	"github.com/caos/logging"
 	"github.com/golang/protobuf/ptypes"
@@ -180,8 +181,8 @@ func (repo *ProjectRepo) SearchProjectRoles(ctx context.Context, projectID strin
 	return result, nil
 }
 
-func (repo *ProjectRepo) ProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*proj_model.ProjectChanges, error) {
-	changes, err := repo.getProjectChanges(ctx, id, lastSequence, limit, sortAscending)
+func (repo *ProjectRepo) ProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*proj_model.ProjectChanges, error) {
+	changes, err := repo.getProjectChanges(ctx, id, lastSequence, limit, sortAscending, retention)
 	if err != nil {
 		return nil, err
 	}
@@ -254,8 +255,8 @@ func (repo *ProjectRepo) SearchApplications(ctx context.Context, request *proj_m
 	return result, nil
 }
 
-func (repo *ProjectRepo) ApplicationChanges(ctx context.Context, id string, appId string, lastSequence uint64, limit uint64, sortAscending bool) (*proj_model.ApplicationChanges, error) {
-	changes, err := repo.getApplicationChanges(ctx, id, appId, lastSequence, limit, sortAscending)
+func (repo *ProjectRepo) ApplicationChanges(ctx context.Context, projectID string, appID string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*proj_model.ApplicationChanges, error) {
+	changes, err := repo.getApplicationChanges(ctx, projectID, appID, lastSequence, limit, sortAscending, retention)
 	if err != nil {
 		return nil, err
 	}
@@ -505,8 +506,8 @@ func (r *ProjectRepo) getUserEvents(ctx context.Context, userID string, sequence
 	return r.Eventstore.FilterEvents(ctx, query)
 }
 
-func (repo *ProjectRepo) getProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*proj_model.ProjectChanges, error) {
-	query := proj_view.ChangesQuery(id, lastSequence, limit, sortAscending)
+func (repo *ProjectRepo) getProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*proj_model.ProjectChanges, error) {
+	query := proj_view.ChangesQuery(id, lastSequence, limit, sortAscending, retention)
 
 	events, err := repo.Eventstore.FilterEvents(context.Background(), query)
 	if err != nil {
@@ -561,8 +562,8 @@ func (repo *ProjectRepo) getProjectEvents(ctx context.Context, id string, sequen
 	return repo.Eventstore.FilterEvents(ctx, query)
 }
 
-func (repo *ProjectRepo) getApplicationChanges(ctx context.Context, projectID string, appID string, lastSequence uint64, limit uint64, sortAscending bool) (*proj_model.ApplicationChanges, error) {
-	query := proj_view.ChangesQuery(projectID, lastSequence, limit, sortAscending)
+func (repo *ProjectRepo) getApplicationChanges(ctx context.Context, projectID string, appID string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*proj_model.ApplicationChanges, error) {
+	query := proj_view.ChangesQuery(projectID, lastSequence, limit, sortAscending, retention)
 
 	events, err := repo.Eventstore.FilterEvents(ctx, query)
 	if err != nil {

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

@@ -2,6 +2,8 @@ package eventstore
 
 import (
 	"context"
+	"time"
+
 	"github.com/caos/zitadel/internal/domain"
 	"github.com/caos/zitadel/internal/eventstore/v1"
 	"github.com/caos/zitadel/internal/eventstore/v1/models"
@@ -82,8 +84,8 @@ func (repo *UserRepo) UserIDsByDomain(ctx context.Context, domain string) ([]str
 	return repo.View.UserIDsByDomain(domain)
 }
 
-func (repo *UserRepo) UserChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*usr_model.UserChanges, error) {
-	changes, err := repo.getUserChanges(ctx, id, lastSequence, limit, sortAscending)
+func (repo *UserRepo) UserChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*usr_model.UserChanges, error) {
+	changes, err := repo.getUserChanges(ctx, id, lastSequence, limit, sortAscending, retention)
 	if err != nil {
 		return nil, err
 	}
@@ -280,8 +282,8 @@ func (repo *UserRepo) SearchUserMemberships(ctx context.Context, request *usr_mo
 	return result, nil
 }
 
-func (r *UserRepo) getUserChanges(ctx context.Context, userID string, lastSequence uint64, limit uint64, sortAscending bool) (*usr_model.UserChanges, error) {
-	query := usr_view.ChangesQuery(userID, lastSequence, limit, sortAscending)
+func (r *UserRepo) getUserChanges(ctx context.Context, userID string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*usr_model.UserChanges, error) {
+	query := usr_view.ChangesQuery(userID, lastSequence, limit, sortAscending, retention)
 
 	events, err := r.Eventstore.FilterEvents(ctx, query)
 	if err != nil {

internal/management/repository/eventsourcing/handler/features.go

@@ -0,0 +1,165 @@
+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 = "management.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) 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)
+}

internal/management/repository/eventsourcing/handler/handler.go

@@ -1,9 +1,10 @@
 package handler
 
 import (
-	"github.com/caos/zitadel/internal/eventstore/v1"
 	"time"
 
+	"github.com/caos/zitadel/internal/eventstore/v1"
+
 	"github.com/caos/zitadel/internal/config/systemdefaults"
 	"github.com/caos/zitadel/internal/config/types"
 	"github.com/caos/zitadel/internal/eventstore/v1/query"
@@ -76,6 +77,8 @@ func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es
 			handler{view, bulkLimit, configs.cycleDuration("MailTemplate"), errorCount, es}),
 		newMailText(
 			handler{view, bulkLimit, configs.cycleDuration("MailText"), errorCount, es}),
+		newFeatures(
+			handler{view, bulkLimit, configs.cycleDuration("Features"), errorCount, es}),
 	}
 }
 

internal/management/repository/eventsourcing/repository.go

@@ -27,6 +27,7 @@ type EsRepository struct {
 	eventstore.UserRepo
 	eventstore.UserGrantRepo
 	eventstore.IAMRepository
+	eventstore.FeaturesRepo
 	view *mgmt_view.View
 }
 
@@ -54,10 +55,9 @@ func Start(conf Config, systemDefaults sd.SystemDefaults, roles []string, querie
 		ProjectRepo:   eventstore.ProjectRepo{es, conf.SearchLimit, view, roles, systemDefaults.IamID},
 		UserRepo:      eventstore.UserRepo{es, conf.SearchLimit, view, systemDefaults},
 		UserGrantRepo: eventstore.UserGrantRepo{conf.SearchLimit, view},
-		IAMRepository: eventstore.IAMRepository{
-			IAMV2Query: queries,
-		},
-		view: view,
+		IAMRepository: eventstore.IAMRepository{IAMV2Query: queries},
+		FeaturesRepo:  eventstore.FeaturesRepo{es, view, conf.SearchLimit, systemDefaults},
+		view:          view,
 	}, nil
 }
 

internal/management/repository/eventsourcing/view/features.go

@@ -0,0 +1,56 @@
+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 = "management.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)
+}

internal/management/repository/features.go

@@ -0,0 +1,11 @@
+package repository
+
+import (
+	"context"
+
+	features_model "github.com/caos/zitadel/internal/features/model"
+)
+
+type FeaturesRepository interface {
+	GetOrgFeatures(ctx context.Context, id string) (*features_model.FeaturesView, error)
+}

internal/management/repository/org.go

@@ -2,6 +2,7 @@ package repository
 
 import (
 	"context"
+	"time"
 
 	iam_model "github.com/caos/zitadel/internal/iam/model"
 
@@ -11,7 +12,7 @@ import (
 type OrgRepository interface {
 	OrgByID(ctx context.Context, id string) (*org_model.OrgView, error)
 	OrgByDomainGlobal(ctx context.Context, domain string) (*org_model.OrgView, error)
-	OrgChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*org_model.OrgChanges, error)
+	OrgChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool, auditLogRetention time.Duration) (*org_model.OrgChanges, error)
 
 	SearchMyOrgDomains(ctx context.Context, request *org_model.OrgDomainSearchRequest) (*org_model.OrgDomainSearchResponse, error)
 

internal/management/repository/project.go

@@ -2,6 +2,8 @@ package repository
 
 import (
 	"context"
+	"time"
+
 	iam_model "github.com/caos/zitadel/internal/iam/model"
 
 	key_model "github.com/caos/zitadel/internal/key/model"
@@ -22,11 +24,11 @@ type ProjectRepository interface {
 	GetProjectMemberRoles(ctx context.Context) ([]string, error)
 
 	SearchProjectRoles(ctx context.Context, projectId string, request *model.ProjectRoleSearchRequest) (*model.ProjectRoleSearchResponse, error)
-	ProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*model.ProjectChanges, error)
+	ProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*model.ProjectChanges, error)
 
 	ApplicationByID(ctx context.Context, projectID, appID string) (*model.ApplicationView, error)
 	SearchApplications(ctx context.Context, request *model.ApplicationSearchRequest) (*model.ApplicationSearchResponse, error)
-	ApplicationChanges(ctx context.Context, projectID string, appID string, lastSequence uint64, limit uint64, sortAscending bool) (*model.ApplicationChanges, error)
+	ApplicationChanges(ctx context.Context, projectID string, appID string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*model.ApplicationChanges, error)
 	SearchClientKeys(ctx context.Context, request *key_model.AuthNKeySearchRequest) (*key_model.AuthNKeySearchResponse, error)
 	GetClientKey(ctx context.Context, projectID, applicationID, keyID string) (*key_model.AuthNKeyView, error)
 

internal/management/repository/repository.go

@@ -7,4 +7,5 @@ type Repository interface {
 	UserRepository
 	UserGrantRepository
 	IamRepository
+	FeaturesRepository
 }

internal/management/repository/user.go

@@ -2,6 +2,7 @@ package repository
 
 import (
 	"context"
+	"time"
 
 	key_model "github.com/caos/zitadel/internal/key/model"
 	"github.com/caos/zitadel/internal/user/model"
@@ -15,7 +16,7 @@ type UserRepository interface {
 	GetUserByLoginNameGlobal(ctx context.Context, email string) (*model.UserView, error)
 	IsUserUnique(ctx context.Context, userName, email string) (bool, error)
 
-	UserChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*model.UserChanges, error)
+	UserChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool, retention time.Duration) (*model.UserChanges, error)
 
 	ProfileByID(ctx context.Context, userID string) (*model.Profile, error)