diff --git a/cmd/setup/28.go b/cmd/setup/28.go new file mode 100644 index 000000000..3bfcb2d61 --- /dev/null +++ b/cmd/setup/28.go @@ -0,0 +1,27 @@ +package setup + +import ( + "context" + _ "embed" + + "github.com/zitadel/zitadel/internal/database" + "github.com/zitadel/zitadel/internal/eventstore" +) + +var ( + //go:embed 28.sql + addFieldTable string +) + +type AddFieldTable struct { + dbClient *database.DB +} + +func (mig *AddFieldTable) Execute(ctx context.Context, _ eventstore.Event) error { + _, err := mig.dbClient.ExecContext(ctx, addFieldTable) + return err +} + +func (mig *AddFieldTable) String() string { + return "28_add_search_table" +} diff --git a/cmd/setup/28.sql b/cmd/setup/28.sql new file mode 100644 index 000000000..ccadf3a0a --- /dev/null +++ b/cmd/setup/28.sql @@ -0,0 +1,64 @@ +CREATE TABLE eventstore.fields ( + id TEXT NOT NULL DEFAULT gen_random_uuid() + , instance_id TEXT NOT NULL + , resource_owner TEXT NOT NULL + + , aggregate_type TEXT NOT NULL + , aggregate_id TEXT NOT NULL + + , object_type TEXT NOT NULL + , object_id TEXT NOT NULL + , object_revision INT2 + + , field_name TEXT NOT NULL + + -- all the values of fields are inserted into value column as jsonb and if we need to index something we store it to the type specific column additionally + , "value" JSONB NOT NULL + , number_value NUMERIC GENERATED ALWAYS AS (CASE WHEN should_index AND JSONB_TYPEOF("value") = 'number' THEN "value"::NUMERIC ELSE NULL END) STORED + , text_value TEXT GENERATED ALWAYS AS (CASE WHEN should_index AND JSONB_TYPEOF("value") = 'string' THEN "value" #>> '{}' ELSE NULL END) STORED + , bool_value BOOLEAN GENERATED ALWAYS AS (CASE WHEN should_index AND JSONB_TYPEOF("value") = 'boolean' THEN "value"::BOOLEAN ELSE NULL END) STORED + + -- if true the value must be unique within an instance + , value_must_be_unique BOOLEAN + -- if set to true the primitive value is indexed + , should_index BOOLEAN + + , PRIMARY KEY (instance_id, id) + -- TODO: create issue to enable the foreign key as soon as the objects table is implemented + -- , CONSTRAINT f_objects_fk FOREIGN KEY (instance_id, resource_owner, object_type, object_id, object_revision) REFERENCES eventstore.objects (instance_id, resource_owner, object_type, object_id, object_revision) ON DELETE CASCADE + + -- the constraint ensures that a primitive value is set if the value must be unique + , CONSTRAINT primitive_value_for_unique_check CHECK ( + CASE + WHEN value_must_be_unique THEN num_nonnulls(number_value, text_value, bool_value) = 1 + ELSE true + END + ) + -- the constraint ensures that a primitive value is set if the value must be indexed + , CONSTRAINT primitive_value_for_index CHECK ( + CASE + WHEN should_index THEN num_nonnulls(number_value, text_value, bool_value) = 1 + ELSE true + END + ) +); + +-- unique constraints for primitive values +CREATE UNIQUE INDEX IF NOT EXISTS f_number_unique_idx ON eventstore.fields (instance_id, field_name, number_value) WHERE value_must_be_unique; +CREATE UNIQUE INDEX IF NOT EXISTS f_text_unique_idx ON eventstore.fields (instance_id, field_name, text_value) WHERE value_must_be_unique; +CREATE UNIQUE INDEX IF NOT EXISTS f_bool_unique_idx ON eventstore.fields (instance_id, field_name, bool_value) WHERE value_must_be_unique; + +-- search index for primitive values +CREATE INDEX IF NOT EXISTS f_number_value_idx ON eventstore.fields (instance_id, object_type, field_name, number_value) + INCLUDE (resource_owner, object_id, object_revision, "value") + WHERE number_value IS NOT NULL ; +CREATE INDEX IF NOT EXISTS f_text_value_idx ON eventstore.fields (instance_id, object_type, field_name, text_value) + INCLUDE (resource_owner, object_id, object_revision, "value") + WHERE text_value IS NOT NULL ; +CREATE INDEX IF NOT EXISTS f_bool_value_idx ON eventstore.fields (instance_id, object_type, field_name, bool_value) + INCLUDE (resource_owner, object_id, object_revision, "value") + WHERE bool_value IS NOT NULL ; + +-- search index for object by id +CREATE INDEX IF NOT EXISTS f_object_idx ON eventstore.fields (instance_id, object_type, object_id, object_revision) + INCLUDE (resource_owner, field_name, "value") ; \ No newline at end of file diff --git a/cmd/setup/29.go b/cmd/setup/29.go new file mode 100644 index 000000000..1d3bfe992 --- /dev/null +++ b/cmd/setup/29.go @@ -0,0 +1,42 @@ +package setup + +import ( + "context" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/query/projection" + "github.com/zitadel/zitadel/internal/repository/instance" +) + +type FillFieldsForProjectGrant struct { + eventstore *eventstore.Eventstore +} + +func (mig *FillFieldsForProjectGrant) Execute(ctx context.Context, _ eventstore.Event) error { + instances, err := mig.eventstore.InstanceIDs( + ctx, + 0, + true, + eventstore.NewSearchQueryBuilder(eventstore.ColumnsInstanceIDs). + OrderDesc(). + AddQuery(). + AggregateTypes("instance"). + EventTypes(instance.InstanceAddedEventType). + Builder(), + ) + if err != nil { + return err + } + for _, instance := range instances { + ctx := authz.WithInstanceID(ctx, instance) + if err := projection.ProjectGrantFields.Trigger(ctx); err != nil { + return err + } + } + return nil +} + +func (mig *FillFieldsForProjectGrant) String() string { + return "29_init_fields_for_project_grant" +} diff --git a/cmd/setup/config.go b/cmd/setup/config.go index 81ec6f233..ebeaa7a91 100644 --- a/cmd/setup/config.go +++ b/cmd/setup/config.go @@ -111,6 +111,8 @@ type Steps struct { s25User11AddLowerFieldsToVerifiedEmail *User11AddLowerFieldsToVerifiedEmail s26AuthUsers3 *AuthUsers3 s27IDPTemplate6SAMLNameIDFormat *IDPTemplate6SAMLNameIDFormat + s28AddFieldTable *AddFieldTable + s29FillFieldsForProjectGrant *FillFieldsForProjectGrant } func MustNewSteps(v *viper.Viper) *Steps { diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go index 6ed0cc4dc..77395e58e 100644 --- a/cmd/setup/setup.go +++ b/cmd/setup/setup.go @@ -108,8 +108,11 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) logging.OnError(err).Fatal("unable to connect to database") config.Eventstore.Querier = old_es.NewCRDB(queryDBClient) - config.Eventstore.Pusher = new_es.NewEventstore(esPusherDBClient) + esV3 := new_es.NewEventstore(esPusherDBClient) + config.Eventstore.Pusher = esV3 + config.Eventstore.Searcher = esV3 eventstoreClient := eventstore.NewEventstore(config.Eventstore) + logging.OnError(err).Fatal("unable to start eventstore") eventstoreV4 := es_v4.NewEventstoreFromOne(es_v4_pg.New(queryDBClient, &es_v4_pg.Config{ MaxRetries: config.Eventstore.MaxRetries, @@ -153,6 +156,8 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) steps.s25User11AddLowerFieldsToVerifiedEmail = &User11AddLowerFieldsToVerifiedEmail{dbClient: esPusherDBClient} steps.s26AuthUsers3 = &AuthUsers3{dbClient: esPusherDBClient} steps.s27IDPTemplate6SAMLNameIDFormat = &IDPTemplate6SAMLNameIDFormat{dbClient: esPusherDBClient} + steps.s28AddFieldTable = &AddFieldTable{dbClient: esPusherDBClient} + steps.s29FillFieldsForProjectGrant = &FillFieldsForProjectGrant{eventstore: eventstoreClient} err = projection.Create(ctx, projectionDBClient, eventstoreClient, config.Projections, nil, nil, nil) logging.OnError(err).Fatal("unable to start projections") @@ -175,6 +180,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) steps.s14NewEventsTable, steps.s1ProjectionTable, steps.s2AssetsTable, + steps.s28AddFieldTable, steps.FirstInstance, steps.s5LastFailed, steps.s6OwnerRemoveColumns, @@ -191,6 +197,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) steps.s23CorrectGlobalUniqueConstraints, steps.s24AddActorToAuthTokens, steps.s26AuthUsers3, + steps.s29FillFieldsForProjectGrant, } { mustExecuteMigration(ctx, eventstoreClient, step, "migration failed") } diff --git a/cmd/setup/steps.yaml b/cmd/setup/steps.yaml index 03476c664..265549c37 100644 --- a/cmd/setup/steps.yaml +++ b/cmd/setup/steps.yaml @@ -54,3 +54,6 @@ CorrectCreationDate: AddEventCreatedAt: BulkAmount: 100 # ZITADEL_ADDEVENTCREATEDAT_BULKAMOUNT + +FillFields: + BatchSize: 1000 # ZITADEL_EVENTSTORE_FILLFIELDS_BULKLIMIT \ No newline at end of file diff --git a/cmd/start/start.go b/cmd/start/start.go index a7688a83c..7d2c1e61f 100644 --- a/cmd/start/start.go +++ b/cmd/start/start.go @@ -153,6 +153,7 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server } config.Eventstore.Pusher = new_es.NewEventstore(esPusherDBClient) + config.Eventstore.Searcher = new_es.NewEventstore(queryDBClient) config.Eventstore.Querier = old_es.NewCRDB(queryDBClient) eventstoreClient := eventstore.NewEventstore(config.Eventstore) eventstoreV4 := es_v4.NewEventstoreFromOne(es_v4_pg.New(queryDBClient, &es_v4_pg.Config{ diff --git a/internal/api/grpc/feature/v2/converter.go b/internal/api/grpc/feature/v2/converter.go index 259a6f8a4..15c43655a 100644 --- a/internal/api/grpc/feature/v2/converter.go +++ b/internal/api/grpc/feature/v2/converter.go @@ -109,6 +109,10 @@ func improvedPerformanceTypeToPb(typ feature.ImprovedPerformanceType) feature_pb return feature_pb.ImprovedPerformance_IMPROVED_PERFORMANCE_UNSPECIFIED case feature.ImprovedPerformanceTypeOrgByID: return feature_pb.ImprovedPerformance_IMPROVED_PERFORMANCE_ORG_BY_ID + case feature.ImprovedPerformanceTypeProjectGrant: + return feature_pb.ImprovedPerformance_IMPROVED_PERFORMANCE_PROJECT_GRANT + case feature.ImprovedPerformanceTypeProject: + return feature_pb.ImprovedPerformance_IMPROVED_PERFORMANCE_PROJECT default: return feature_pb.ImprovedPerformance(typ) } @@ -133,6 +137,10 @@ func improvedPerformanceToDomain(typ feature_pb.ImprovedPerformance) feature.Imp return feature.ImprovedPerformanceTypeUnknown case feature_pb.ImprovedPerformance_IMPROVED_PERFORMANCE_ORG_BY_ID: return feature.ImprovedPerformanceTypeOrgByID + case feature_pb.ImprovedPerformance_IMPROVED_PERFORMANCE_PROJECT_GRANT: + return feature.ImprovedPerformanceTypeProjectGrant + case feature_pb.ImprovedPerformance_IMPROVED_PERFORMANCE_PROJECT: + return feature.ImprovedPerformanceTypeProject default: return feature.ImprovedPerformanceTypeUnknown } diff --git a/internal/command/org.go b/internal/command/org.go index 570455794..e603531b3 100644 --- a/internal/command/org.go +++ b/internal/command/org.go @@ -443,6 +443,7 @@ func (c *Commands) prepareRemoveOrg(a *org.Aggregate) preparation.Validation { if a.ID == instance.DefaultOrganisationID() { return nil, zerrors.ThrowPreconditionFailed(nil, "COMMA-wG9p1", "Errors.Org.DefaultOrgNotDeletable") } + err := c.checkProjectExists(ctx, instance.ProjectID(), a.ID) // if there is no error, the ZITADEL project was found on the org to be deleted if err == nil { diff --git a/internal/command/project.go b/internal/command/project.go index be8ea242c..16ede9c7c 100644 --- a/internal/command/project.go +++ b/internal/command/project.go @@ -6,9 +6,12 @@ import ( "github.com/zitadel/logging" + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/command/preparation" "github.com/zitadel/zitadel/internal/domain" "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/feature" + "github.com/zitadel/zitadel/internal/query/projection" "github.com/zitadel/zitadel/internal/repository/project" "github.com/zitadel/zitadel/internal/telemetry/tracing" "github.com/zitadel/zitadel/internal/zerrors" @@ -165,16 +168,51 @@ func (c *Commands) getProjectByID(ctx context.Context, projectID, resourceOwner return projectWriteModelToProject(projectWriteModel), nil } +func (c *Commands) projectAggregateByID(ctx context.Context, projectID, resourceOwner string) (*eventstore.Aggregate, domain.ProjectState, error) { + result, err := c.projectState(ctx, projectID, resourceOwner) + if err != nil { + return nil, domain.ProjectStateUnspecified, zerrors.ThrowNotFound(err, "COMMA-NDQoF", "Errors.Project.NotFound") + } + if len(result) == 0 { + _ = projection.ProjectGrantFields.Trigger(ctx) + result, err = c.projectState(ctx, projectID, resourceOwner) + if err != nil || len(result) == 0 { + return nil, domain.ProjectStateUnspecified, zerrors.ThrowNotFound(err, "COMMA-U1nza", "Errors.Project.NotFound") + } + } + + var state domain.ProjectState + err = result[0].Value.Unmarshal(&state) + if err != nil { + return nil, state, zerrors.ThrowNotFound(err, "COMMA-o4n6F", "Errors.Project.NotFound") + } + return &result[0].Aggregate, state, nil +} + +func (c *Commands) projectState(ctx context.Context, projectID, resourceOwner string) ([]*eventstore.SearchResult, error) { + return c.eventstore.Search( + ctx, + map[eventstore.FieldType]any{ + eventstore.FieldTypeObjectType: project.ProjectSearchType, + eventstore.FieldTypeObjectID: projectID, + eventstore.FieldTypeObjectRevision: project.ProjectObjectRevision, + eventstore.FieldTypeFieldName: project.ProjectStateSearchField, + eventstore.FieldTypeResourceOwner: resourceOwner, + }, + ) +} + func (c *Commands) checkProjectExists(ctx context.Context, projectID, resourceOwner string) (err error) { ctx, span := tracing.NewSpan(ctx) defer func() { span.EndWithError(err) }() - projectWriteModel, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) - if err != nil { - return err + if !authz.GetFeatures(ctx).ShouldUseImprovedPerformance(feature.ImprovedPerformanceTypeProject) { + return c.checkProjectExistsOld(ctx, projectID, resourceOwner) } - if projectWriteModel.State == domain.ProjectStateUnspecified || projectWriteModel.State == domain.ProjectStateRemoved { - return zerrors.ThrowPreconditionFailed(nil, "COMMAND-EbFMN", "Errors.Project.NotFound") + + _, state, err := c.projectAggregateByID(ctx, projectID, resourceOwner) + if err != nil || !state.Valid() { + return zerrors.ThrowPreconditionFailed(err, "COMMA-VCnwD", "Errors.Project.NotFound") } return nil } @@ -184,6 +222,10 @@ func (c *Commands) ChangeProject(ctx context.Context, projectChange *domain.Proj return nil, zerrors.ThrowInvalidArgument(nil, "COMMAND-4m9vS", "Errors.Project.Invalid") } + if !authz.GetFeatures(ctx).ShouldUseImprovedPerformance(feature.ImprovedPerformanceTypeProject) { + return c.changeProjectOld(ctx, projectChange, resourceOwner) + } + existingProject, err := c.getProjectWriteModelByID(ctx, projectChange.AggregateID, resourceOwner) if err != nil { return nil, err @@ -223,6 +265,27 @@ func (c *Commands) DeactivateProject(ctx context.Context, projectID string, reso return nil, zerrors.ThrowInvalidArgument(nil, "COMMAND-88iF0", "Errors.Project.ProjectIDMissing") } + if !authz.GetFeatures(ctx).ShouldUseImprovedPerformance(feature.ImprovedPerformanceTypeProject) { + return c.deactivateProjectOld(ctx, projectID, resourceOwner) + } + + projectAgg, state, err := c.projectAggregateByID(ctx, projectID, resourceOwner) + if err != nil { + return nil, err + } + + if state == domain.ProjectStateUnspecified || state == domain.ProjectStateRemoved { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-112M9", "Errors.Project.NotFound") + } + if state != domain.ProjectStateActive { + return nil, zerrors.ThrowPreconditionFailed(nil, "COMMAND-mki55", "Errors.Project.NotActive") + } + + pushedEvents, err := c.eventstore.Push(ctx, project.NewProjectDeactivatedEvent(ctx, projectAgg)) + if err != nil { + return nil, err + } + existingProject, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) if err != nil { return nil, err @@ -234,16 +297,11 @@ func (c *Commands) DeactivateProject(ctx context.Context, projectID string, reso return nil, zerrors.ThrowPreconditionFailed(nil, "COMMAND-mki55", "Errors.Project.NotActive") } - projectAgg := ProjectAggregateFromWriteModel(&existingProject.WriteModel) - pushedEvents, err := c.eventstore.Push(ctx, project.NewProjectDeactivatedEvent(ctx, projectAgg)) - if err != nil { - return nil, err - } - err = AppendAndReduce(existingProject, pushedEvents...) - if err != nil { - return nil, err - } - return writeModelToObjectDetails(&existingProject.WriteModel), nil + return &domain.ObjectDetails{ + ResourceOwner: pushedEvents[0].Aggregate().ResourceOwner, + Sequence: pushedEvents[0].Sequence(), + EventDate: pushedEvents[0].CreatedAt(), + }, nil } func (c *Commands) ReactivateProject(ctx context.Context, projectID string, resourceOwner string) (*domain.ObjectDetails, error) { @@ -251,6 +309,23 @@ func (c *Commands) ReactivateProject(ctx context.Context, projectID string, reso return nil, zerrors.ThrowInvalidArgument(nil, "COMMAND-3ihsF", "Errors.Project.ProjectIDMissing") } + if !authz.GetFeatures(ctx).ShouldUseImprovedPerformance(feature.ImprovedPerformanceTypeProject) { + return c.reactivateProjectOld(ctx, projectID, resourceOwner) + } + + projectAgg, state, err := c.projectAggregateByID(ctx, projectID, resourceOwner) + if err != nil { + return nil, err + } + + if state == domain.ProjectStateUnspecified || state == domain.ProjectStateRemoved { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-3M9sd", "Errors.Project.NotFound") + } + + if state != domain.ProjectStateInactive { + return nil, zerrors.ThrowPreconditionFailed(nil, "COMMAND-5M9bs", "Errors.Project.NotInactive") + } + existingProject, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) if err != nil { return nil, err @@ -262,16 +337,16 @@ func (c *Commands) ReactivateProject(ctx context.Context, projectID string, reso return nil, zerrors.ThrowPreconditionFailed(nil, "COMMAND-5M9bs", "Errors.Project.NotInactive") } - projectAgg := ProjectAggregateFromWriteModel(&existingProject.WriteModel) pushedEvents, err := c.eventstore.Push(ctx, project.NewProjectReactivatedEvent(ctx, projectAgg)) if err != nil { return nil, err } - err = AppendAndReduce(existingProject, pushedEvents...) - if err != nil { - return nil, err - } - return writeModelToObjectDetails(&existingProject.WriteModel), nil + + return &domain.ObjectDetails{ + ResourceOwner: pushedEvents[0].Aggregate().ResourceOwner, + Sequence: pushedEvents[0].Sequence(), + EventDate: pushedEvents[0].CreatedAt(), + }, nil } func (c *Commands) RemoveProject(ctx context.Context, projectID, resourceOwner string, cascadingUserGrantIDs ...string) (*domain.ObjectDetails, error) { @@ -279,6 +354,10 @@ func (c *Commands) RemoveProject(ctx context.Context, projectID, resourceOwner s return nil, zerrors.ThrowInvalidArgument(nil, "COMMAND-66hM9", "Errors.Project.ProjectIDMissing") } + if !authz.GetFeatures(ctx).ShouldUseImprovedPerformance(feature.ImprovedPerformanceTypeProject) { + return c.removeProjectOld(ctx, projectID, resourceOwner) + } + existingProject, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) if err != nil { return nil, err diff --git a/internal/command/project_grant.go b/internal/command/project_grant.go index d0f2d210f..59f98e376 100644 --- a/internal/command/project_grant.go +++ b/internal/command/project_grant.go @@ -6,8 +6,11 @@ import ( "github.com/zitadel/logging" + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/domain" "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/feature" + "github.com/zitadel/zitadel/internal/repository/org" "github.com/zitadel/zitadel/internal/repository/project" "github.com/zitadel/zitadel/internal/telemetry/tracing" "github.com/zitadel/zitadel/internal/zerrors" @@ -143,10 +146,12 @@ func (c *Commands) DeactivateProjectGrant(ctx context.Context, projectID, grantI if grantID == "" || projectID == "" { return details, zerrors.ThrowInvalidArgument(nil, "PROJECT-p0s4V", "Errors.IDMissing") } + err = c.checkProjectExists(ctx, projectID, resourceOwner) if err != nil { - return details, err + return nil, err } + existingGrant, err := c.projectGrantWriteModelByID(ctx, grantID, projectID, resourceOwner) if err != nil { return details, err @@ -171,10 +176,12 @@ func (c *Commands) ReactivateProjectGrant(ctx context.Context, projectID, grantI if grantID == "" || projectID == "" { return details, zerrors.ThrowInvalidArgument(nil, "PROJECT-p0s4V", "Errors.IDMissing") } + err = c.checkProjectExists(ctx, projectID, resourceOwner) if err != nil { - return details, err + return nil, err } + existingGrant, err := c.projectGrantWriteModelByID(ctx, grantID, projectID, resourceOwner) if err != nil { return details, err @@ -198,10 +205,12 @@ func (c *Commands) RemoveProjectGrant(ctx context.Context, projectID, grantID, r if grantID == "" || projectID == "" { return details, zerrors.ThrowInvalidArgument(nil, "PROJECT-1m9fJ", "Errors.IDMissing") } + err = c.checkProjectExists(ctx, projectID, resourceOwner) if err != nil { - return details, zerrors.ThrowPreconditionFailed(err, "PROJECT-6mf9s", "Errors.Project.NotFound") + return nil, err } + existingGrant, err := c.projectGrantWriteModelByID(ctx, grantID, projectID, resourceOwner) if err != nil { return details, err @@ -247,18 +256,76 @@ func (c *Commands) projectGrantWriteModelByID(ctx context.Context, grantID, proj } func (c *Commands) checkProjectGrantPreCondition(ctx context.Context, projectGrant *domain.ProjectGrant) error { - preConditions := NewProjectGrantPreConditionReadModel(projectGrant.AggregateID, projectGrant.GrantedOrgID) - err := c.eventstore.FilterToQueryReducer(ctx, preConditions) + if !authz.GetFeatures(ctx).ShouldUseImprovedPerformance(feature.ImprovedPerformanceTypeProjectGrant) { + return c.checkProjectGrantPreConditionOld(ctx, projectGrant) + } + results, err := c.eventstore.Search( + ctx, + // project state query + map[eventstore.FieldType]any{ + eventstore.FieldTypeAggregateType: project.AggregateType, + eventstore.FieldTypeAggregateID: projectGrant.AggregateID, + eventstore.FieldTypeFieldName: project.ProjectStateSearchField, + eventstore.FieldTypeObjectType: project.ProjectSearchType, + }, + // granted org query + map[eventstore.FieldType]any{ + eventstore.FieldTypeAggregateType: org.AggregateType, + eventstore.FieldTypeAggregateID: projectGrant.GrantedOrgID, + eventstore.FieldTypeFieldName: org.OrgStateSearchField, + eventstore.FieldTypeObjectType: org.OrgSearchType, + }, + // role query + map[eventstore.FieldType]any{ + eventstore.FieldTypeAggregateType: project.AggregateType, + eventstore.FieldTypeAggregateID: projectGrant.AggregateID, + eventstore.FieldTypeFieldName: project.ProjectRoleKeySearchField, + eventstore.FieldTypeObjectType: project.ProjectRoleSearchType, + }, + ) if err != nil { return err } - if !preConditions.ProjectExists { + + var ( + existsProject bool + existsGrantedOrg bool + existingRoleKeys []string + ) + + for _, result := range results { + switch result.Object.Type { + case project.ProjectRoleSearchType: + var role string + err := result.Value.Unmarshal(&role) + if err != nil { + return err + } + existingRoleKeys = append(existingRoleKeys, role) + case org.OrgSearchType: + var state domain.OrgState + err := result.Value.Unmarshal(&state) + if err != nil { + return err + } + existsGrantedOrg = state.Valid() && state != domain.OrgStateRemoved + case project.ProjectSearchType: + var state domain.ProjectState + err := result.Value.Unmarshal(&state) + if err != nil { + return err + } + existsProject = state.Valid() && state != domain.ProjectStateRemoved + } + } + + if !existsProject { return zerrors.ThrowPreconditionFailed(err, "COMMAND-m9gsd", "Errors.Project.NotFound") } - if !preConditions.GrantedOrgExists { + if !existsGrantedOrg { return zerrors.ThrowPreconditionFailed(err, "COMMAND-3m9gg", "Errors.Org.NotFound") } - if projectGrant.HasInvalidRoles(preConditions.ExistingRoleKeys) { + if projectGrant.HasInvalidRoles(existingRoleKeys) { return zerrors.ThrowPreconditionFailed(err, "COMMAND-6m9gd", "Errors.Project.Role.NotFound") } return nil diff --git a/internal/command/project_old.go b/internal/command/project_old.go new file mode 100644 index 000000000..434df38a7 --- /dev/null +++ b/internal/command/project_old.go @@ -0,0 +1,180 @@ +package command + +import ( + "context" + + "github.com/zitadel/logging" + + "github.com/zitadel/zitadel/internal/domain" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/repository/project" + "github.com/zitadel/zitadel/internal/telemetry/tracing" + "github.com/zitadel/zitadel/internal/zerrors" +) + +func (c *Commands) checkProjectExistsOld(ctx context.Context, projectID, resourceOwner string) (err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + + projectWriteModel, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) + if err != nil { + return err + } + if projectWriteModel.State == domain.ProjectStateUnspecified || projectWriteModel.State == domain.ProjectStateRemoved { + return zerrors.ThrowPreconditionFailed(nil, "COMMAND-EbFMN", "Errors.Project.NotFound") + } + return nil +} + +func (c *Commands) changeProjectOld(ctx context.Context, projectChange *domain.Project, resourceOwner string) (*domain.Project, error) { + if !projectChange.IsValid() || projectChange.AggregateID == "" { + return nil, zerrors.ThrowInvalidArgument(nil, "COMMAND-4m9vS", "Errors.Project.Invalid") + } + + existingProject, err := c.getProjectWriteModelByID(ctx, projectChange.AggregateID, resourceOwner) + if err != nil { + return nil, err + } + if existingProject.State == domain.ProjectStateUnspecified || existingProject.State == domain.ProjectStateRemoved { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-3M9sd", "Errors.Project.NotFound") + } + + //nolint: contextcheck + projectAgg := ProjectAggregateFromWriteModel(&existingProject.WriteModel) + changedEvent, hasChanged, err := existingProject.NewChangedEvent( + ctx, + projectAgg, + projectChange.Name, + projectChange.ProjectRoleAssertion, + projectChange.ProjectRoleCheck, + projectChange.HasProjectCheck, + projectChange.PrivateLabelingSetting) + if err != nil { + return nil, err + } + if !hasChanged { + return nil, zerrors.ThrowPreconditionFailed(nil, "COMMAND-2M0fs", "Errors.NoChangesFound") + } + pushedEvents, err := c.eventstore.Push(ctx, changedEvent) + if err != nil { + return nil, err + } + err = AppendAndReduce(existingProject, pushedEvents...) + if err != nil { + return nil, err + } + return projectWriteModelToProject(existingProject), nil +} + +func (c *Commands) deactivateProjectOld(ctx context.Context, projectID string, resourceOwner string) (*domain.ObjectDetails, error) { + existingProject, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) + if err != nil { + return nil, err + } + if existingProject.State == domain.ProjectStateUnspecified || existingProject.State == domain.ProjectStateRemoved { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-112M9", "Errors.Project.NotFound") + } + if existingProject.State != domain.ProjectStateActive { + return nil, zerrors.ThrowPreconditionFailed(nil, "COMMAND-mki55", "Errors.Project.NotActive") + } + + //nolint: contextcheck + projectAgg := ProjectAggregateFromWriteModel(&existingProject.WriteModel) + pushedEvents, err := c.eventstore.Push(ctx, project.NewProjectDeactivatedEvent(ctx, projectAgg)) + if err != nil { + return nil, err + } + err = AppendAndReduce(existingProject, pushedEvents...) + if err != nil { + return nil, err + } + return writeModelToObjectDetails(&existingProject.WriteModel), nil +} + +func (c *Commands) reactivateProjectOld(ctx context.Context, projectID string, resourceOwner string) (*domain.ObjectDetails, error) { + existingProject, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) + if err != nil { + return nil, err + } + if existingProject.State == domain.ProjectStateUnspecified || existingProject.State == domain.ProjectStateRemoved { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-3M9sd", "Errors.Project.NotFound") + } + if existingProject.State != domain.ProjectStateInactive { + return nil, zerrors.ThrowPreconditionFailed(nil, "COMMAND-5M9bs", "Errors.Project.NotInactive") + } + + //nolint: contextcheck + projectAgg := ProjectAggregateFromWriteModel(&existingProject.WriteModel) + pushedEvents, err := c.eventstore.Push(ctx, project.NewProjectReactivatedEvent(ctx, projectAgg)) + if err != nil { + return nil, err + } + err = AppendAndReduce(existingProject, pushedEvents...) + if err != nil { + return nil, err + } + return writeModelToObjectDetails(&existingProject.WriteModel), nil +} + +func (c *Commands) removeProjectOld(ctx context.Context, projectID, resourceOwner string, cascadingUserGrantIDs ...string) (*domain.ObjectDetails, error) { + existingProject, err := c.getProjectWriteModelByID(ctx, projectID, resourceOwner) + if err != nil { + return nil, err + } + if existingProject.State == domain.ProjectStateUnspecified || existingProject.State == domain.ProjectStateRemoved { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-3M9sd", "Errors.Project.NotFound") + } + + samlEntityIDsAgg, err := c.getSAMLEntityIdsWriteModelByProjectID(ctx, projectID, resourceOwner) + if err != nil { + return nil, err + } + + uniqueConstraints := make([]*eventstore.UniqueConstraint, len(samlEntityIDsAgg.EntityIDs)) + for i, entityID := range samlEntityIDsAgg.EntityIDs { + uniqueConstraints[i] = project.NewRemoveSAMLConfigEntityIDUniqueConstraint(entityID.EntityID) + } + + //nolint: contextcheck + projectAgg := ProjectAggregateFromWriteModel(&existingProject.WriteModel) + events := []eventstore.Command{ + project.NewProjectRemovedEvent(ctx, projectAgg, existingProject.Name, uniqueConstraints), + } + + for _, grantID := range cascadingUserGrantIDs { + event, _, err := c.removeUserGrant(ctx, grantID, "", true) + if err != nil { + logging.WithFields("usergrantid", grantID).WithError(err).Warn("could not cascade remove user grant") + continue + } + events = append(events, event) + } + + pushedEvents, err := c.eventstore.Push(ctx, events...) + if err != nil { + return nil, err + } + err = AppendAndReduce(existingProject, pushedEvents...) + if err != nil { + return nil, err + } + return writeModelToObjectDetails(&existingProject.WriteModel), nil +} + +func (c *Commands) checkProjectGrantPreConditionOld(ctx context.Context, projectGrant *domain.ProjectGrant) error { + preConditions := NewProjectGrantPreConditionReadModel(projectGrant.AggregateID, projectGrant.GrantedOrgID) + err := c.eventstore.FilterToQueryReducer(ctx, preConditions) + if err != nil { + return err + } + if !preConditions.ProjectExists { + return zerrors.ThrowPreconditionFailed(err, "COMMAND-m9gsd", "Errors.Project.NotFound") + } + if !preConditions.GrantedOrgExists { + return zerrors.ThrowPreconditionFailed(err, "COMMAND-3m9gg", "Errors.Org.NotFound") + } + if projectGrant.HasInvalidRoles(preConditions.ExistingRoleKeys) { + return zerrors.ThrowPreconditionFailed(err, "COMMAND-6m9gd", "Errors.Project.Role.NotFound") + } + return nil +} diff --git a/internal/command/project_role.go b/internal/command/project_role.go index ec9a42624..3c8f9c725 100644 --- a/internal/command/project_role.go +++ b/internal/command/project_role.go @@ -41,7 +41,7 @@ func (c *Commands) AddProjectRole(ctx context.Context, projectRole *domain.Proje func (c *Commands) BulkAddProjectRole(ctx context.Context, projectID, resourceOwner string, projectRoles []*domain.ProjectRole) (details *domain.ObjectDetails, err error) { err = c.checkProjectExists(ctx, projectID, resourceOwner) if err != nil { - return details, err + return nil, err } roleWriteModel := NewProjectRoleWriteModel(projectID, resourceOwner) diff --git a/internal/domain/org.go b/internal/domain/org.go index 0e87d20a4..d016dde99 100644 --- a/internal/domain/org.go +++ b/internal/domain/org.go @@ -36,4 +36,10 @@ const ( OrgStateActive OrgStateInactive OrgStateRemoved + + orgStateMax ) + +func (s OrgState) Valid() bool { + return s > OrgStateUnspecified && s < orgStateMax +} diff --git a/internal/eventstore/config.go b/internal/eventstore/config.go index fd3955667..6fc3f79f2 100644 --- a/internal/eventstore/config.go +++ b/internal/eventstore/config.go @@ -8,6 +8,7 @@ type Config struct { PushTimeout time.Duration MaxRetries uint32 - Pusher Pusher - Querier Querier + Pusher Pusher + Querier Querier + Searcher Searcher } diff --git a/internal/eventstore/event.go b/internal/eventstore/event.go index 6c7f47b73..3df096f06 100644 --- a/internal/eventstore/event.go +++ b/internal/eventstore/event.go @@ -31,6 +31,8 @@ type Command interface { Payload() any // UniqueConstraints should be added for unique attributes of an event, if nil constraints will not be checked UniqueConstraints() []*UniqueConstraint + // Fields should be added for fields which should be indexed for lookup, if nil fields will not be indexed + Fields() []*FieldOperation } // Event is a stored activity diff --git a/internal/eventstore/event_base.go b/internal/eventstore/event_base.go index 5bb117b14..c2b56128a 100644 --- a/internal/eventstore/event_base.go +++ b/internal/eventstore/event_base.go @@ -123,3 +123,7 @@ func NewBaseEventForPush(ctx context.Context, aggregate *Aggregate, typ EventTyp EventType: typ, } } + +func (*BaseEvent) Fields() []*FieldOperation { + return nil +} diff --git a/internal/eventstore/eventstore.go b/internal/eventstore/eventstore.go index 34db1e9cf..e45613582 100644 --- a/internal/eventstore/eventstore.go +++ b/internal/eventstore/eventstore.go @@ -11,6 +11,7 @@ import ( "github.com/zitadel/logging" "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/zerrors" ) // Eventstore abstracts all functions needed to store valid events @@ -19,8 +20,9 @@ type Eventstore struct { PushTimeout time.Duration maxRetries int - pusher Pusher - querier Querier + pusher Pusher + querier Querier + searcher Searcher instances []string lastInstanceQuery time.Time @@ -62,8 +64,9 @@ func NewEventstore(config *Config) *Eventstore { PushTimeout: config.PushTimeout, maxRetries: int(config.MaxRetries), - pusher: config.Pusher, - querier: config.Querier, + pusher: config.Pusher, + querier: config.Querier, + searcher: config.Searcher, instancesMu: sync.Mutex{}, } @@ -127,6 +130,20 @@ func (es *Eventstore) AggregateTypes() []string { return aggregateTypes } +// FillFields implements the [Searcher] interface +func (es *Eventstore) FillFields(ctx context.Context, events ...FillFieldsEvent) error { + return es.searcher.FillFields(ctx, events...) +} + +// Search implements the [Searcher] interface +func (es *Eventstore) Search(ctx context.Context, conditions ...map[FieldType]any) ([]*SearchResult, error) { + if len(conditions) == 0 { + return nil, zerrors.ThrowInvalidArgument(nil, "V3-5Xbr1", "no search conditions") + } + + return es.searcher.Search(ctx, conditions...) +} + // Filter filters the stored events based on the searchQuery // and maps the events to the defined event structs // @@ -262,6 +279,22 @@ type Pusher interface { Push(ctx context.Context, commands ...Command) (_ []Event, err error) } +type FillFieldsEvent interface { + Event + Fields() []*FieldOperation +} + +type Searcher interface { + // Search allows to search for specific fields of objects + // The instance id is taken from the context + // The list of conditions are combined with AND + // The search fields are combined with OR + // At least one must be defined + Search(ctx context.Context, conditions ...map[FieldType]any) (result []*SearchResult, err error) + // FillFields is to insert the fields of previously stored events + FillFields(ctx context.Context, events ...FillFieldsEvent) error +} + func appendEventType(typ EventType) { i := sort.SearchStrings(eventTypes, string(typ)) if i < len(eventTypes) && eventTypes[i] == string(typ) { diff --git a/internal/eventstore/field.go b/internal/eventstore/field.go new file mode 100644 index 000000000..8ce461666 --- /dev/null +++ b/internal/eventstore/field.go @@ -0,0 +1,140 @@ +package eventstore + +// FieldOperation if the definition of the operation to be executed on the field +type FieldOperation struct { + // Set a field in the field table + // if [SearchField.UpsertConflictFields] are set the field will be updated if the conflict fields match + // if no [SearchField.UpsertConflictFields] are set the field will be inserted + Set *Field + // Remove fields using the map as `AND`ed conditions + Remove map[FieldType]any +} + +type SearchResult struct { + Aggregate Aggregate + Object Object + FieldName string + // Value represents the stored value + // use the Unmarshal method to parse the value to the desired type + Value interface { + // Unmarshal parses the value to ptr + Unmarshal(ptr any) error + } +} + +// // NumericResultValue marshals the value to the given type + +type Object struct { + // Type of the object + Type string + // ID of the object + ID string + // Revision of the object, if an object evolves the revision should be increased + // analog to current projection versioning + Revision uint8 +} + +type Field struct { + Aggregate *Aggregate + Object Object + UpsertConflictFields []FieldType + FieldName string + Value Value +} + +type Value struct { + Value any + // MustBeUnique defines if the field must be unique + // This field will replace unique constraints in the future + // If MustBeUnique is true the value must be a primitive type + MustBeUnique bool + // ShouldIndex defines if the field should be indexed + // If the field is not indexed it can not be used in search queries + // If ShouldIndex is true the value must be a primitive type + ShouldIndex bool +} + +type SearchValueType int8 + +const ( + SearchValueTypeString SearchValueType = iota + SearchValueTypeNumeric +) + +// SetSearchField sets the field based on the defined parameters +// if conflictFields are set the field will be updated if the conflict fields match +func SetField(aggregate *Aggregate, object Object, fieldName string, value *Value, conflictFields ...FieldType) *FieldOperation { + return &FieldOperation{ + Set: &Field{ + Aggregate: aggregate, + Object: object, + UpsertConflictFields: conflictFields, + FieldName: fieldName, + Value: *value, + }, + } +} + +// RemoveSearchFields removes fields using the map as `AND`ed conditions +func RemoveSearchFields(clause map[FieldType]any) *FieldOperation { + return &FieldOperation{ + Remove: clause, + } +} + +// RemoveSearchFieldsByAggregate removes fields using the aggregate as `AND`ed conditions +func RemoveSearchFieldsByAggregate(aggregate *Aggregate) *FieldOperation { + return &FieldOperation{ + Remove: map[FieldType]any{ + FieldTypeInstanceID: aggregate.InstanceID, + FieldTypeResourceOwner: aggregate.ResourceOwner, + FieldTypeAggregateType: aggregate.Type, + FieldTypeAggregateID: aggregate.ID, + }, + } +} + +// RemoveSearchFieldsByAggregateAndObject removes fields using the aggregate and object as `AND`ed conditions +func RemoveSearchFieldsByAggregateAndObject(aggregate *Aggregate, object Object) *FieldOperation { + return &FieldOperation{ + Remove: map[FieldType]any{ + FieldTypeInstanceID: aggregate.InstanceID, + FieldTypeResourceOwner: aggregate.ResourceOwner, + FieldTypeAggregateType: aggregate.Type, + FieldTypeAggregateID: aggregate.ID, + FieldTypeObjectType: object.Type, + FieldTypeObjectID: object.ID, + FieldTypeObjectRevision: object.Revision, + }, + } +} + +// RemoveSearchFieldsByAggregateAndObjectAndField removes fields using the aggregate, object and field as `AND`ed conditions +func RemoveSearchFieldsByAggregateAndObjectAndField(aggregate *Aggregate, object Object, field string) *FieldOperation { + return &FieldOperation{ + Remove: map[FieldType]any{ + FieldTypeInstanceID: aggregate.InstanceID, + FieldTypeResourceOwner: aggregate.ResourceOwner, + FieldTypeAggregateType: aggregate.Type, + FieldTypeAggregateID: aggregate.ID, + FieldTypeObjectType: object.Type, + FieldTypeObjectID: object.ID, + FieldTypeObjectRevision: object.Revision, + FieldTypeFieldName: field, + }, + } +} + +type FieldType int8 + +const ( + FieldTypeAggregateType FieldType = iota + FieldTypeAggregateID + FieldTypeInstanceID + FieldTypeResourceOwner + FieldTypeObjectType + FieldTypeObjectID + FieldTypeObjectRevision + FieldTypeFieldName + FieldTypeValue +) diff --git a/internal/eventstore/handler/v2/field_handler.go b/internal/eventstore/handler/v2/field_handler.go new file mode 100644 index 000000000..104b5a40d --- /dev/null +++ b/internal/eventstore/handler/v2/field_handler.go @@ -0,0 +1,205 @@ +package handler + +import ( + "context" + "database/sql" + "errors" + "sync" + "time" + + "github.com/jackc/pgx/v5/pgconn" + + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/telemetry/tracing" +) + +type FieldHandler struct { + Handler +} + +type fieldProjection struct { + name string +} + +// Name implements Projection. +func (f *fieldProjection) Name() string { + return f.name +} + +// Reducers implements Projection. +func (f *fieldProjection) Reducers() []AggregateReducer { + return nil +} + +var _ Projection = (*fieldProjection)(nil) + +func NewFieldHandler(config *Config, name string, eventTypes map[eventstore.AggregateType][]eventstore.EventType) *FieldHandler { + return &FieldHandler{ + Handler: Handler{ + projection: &fieldProjection{name: name}, + client: config.Client, + es: config.Eventstore, + bulkLimit: config.BulkLimit, + eventTypes: eventTypes, + requeueEvery: config.RequeueEvery, + handleActiveInstances: config.HandleActiveInstances, + now: time.Now, + maxFailureCount: config.MaxFailureCount, + retryFailedAfter: config.RetryFailedAfter, + triggeredInstancesSync: sync.Map{}, + triggerWithoutEvents: config.TriggerWithoutEvents, + txDuration: config.TransactionDuration, + }, + } +} + +func (h *FieldHandler) Trigger(ctx context.Context, opts ...TriggerOpt) (err error) { + config := new(triggerConfig) + for _, opt := range opts { + opt(config) + } + + cancel := h.lockInstance(ctx, config) + if cancel == nil { + return nil + } + defer cancel() + + for i := 0; ; i++ { + additionalIteration, err := h.processEvents(ctx, config) + h.log().OnError(err).Info("process events failed") + h.log().WithField("iteration", i).Debug("trigger iteration") + if !additionalIteration || err != nil { + return err + } + } +} + +func (h *FieldHandler) processEvents(ctx context.Context, config *triggerConfig) (additionalIteration bool, err error) { + defer func() { + pgErr := new(pgconn.PgError) + if errors.As(err, &pgErr) { + // error returned if the row is currently locked by another connection + if pgErr.Code == "55P03" { + h.log().Debug("state already locked") + err = nil + additionalIteration = false + } + } + }() + + txCtx := ctx + if h.txDuration > 0 { + var cancel, cancelTx func() + // add 100ms to store current state if iteration takes too long + txCtx, cancelTx = context.WithTimeout(ctx, h.txDuration+100*time.Millisecond) + defer cancelTx() + ctx, cancel = context.WithTimeout(ctx, h.txDuration) + defer cancel() + } + + ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx") + tx, err := h.client.BeginTx(txCtx, nil) + spanBeginTx.EndWithError(err) + if err != nil { + return false, err + } + defer func() { + if err != nil && !errors.Is(err, &executionError{}) { + rollbackErr := tx.Rollback() + h.log().OnError(rollbackErr).Debug("unable to rollback tx") + return + } + commitErr := tx.Commit() + if err == nil { + err = commitErr + } + }() + + currentState, err := h.currentState(ctx, tx, config) + if err != nil { + if errors.Is(err, errJustUpdated) { + return false, nil + } + return additionalIteration, err + } + // stop execution if currentState.eventTimestamp >= config.maxCreatedAt + if config.maxPosition != 0 && currentState.position >= config.maxPosition { + return false, nil + } + + events, additionalIteration, err := h.fetchEvents(ctx, tx, currentState) + if err != nil { + return additionalIteration, err + } + if len(events) == 0 { + err = h.setState(tx, currentState) + return additionalIteration, err + } + + err = h.es.FillFields(ctx, events...) + if err != nil { + return false, err + } + + err = h.setState(tx, currentState) + + return additionalIteration, err +} + +func (h *FieldHandler) fetchEvents(ctx context.Context, tx *sql.Tx, currentState *state) (_ []eventstore.FillFieldsEvent, additionalIteration bool, err error) { + events, err := h.es.Filter(ctx, h.eventQuery(currentState).SetTx(tx)) + if err != nil { + h.log().WithError(err).Debug("filter eventstore failed") + return nil, false, err + } + eventAmount := len(events) + + idx, offset := skipPreviouslyReducedEvents(events, currentState) + + if currentState.position == events[len(events)-1].Position() { + offset += currentState.offset + } + currentState.position = events[len(events)-1].Position() + currentState.offset = offset + currentState.aggregateID = events[len(events)-1].Aggregate().ID + currentState.aggregateType = events[len(events)-1].Aggregate().Type + currentState.sequence = events[len(events)-1].Sequence() + currentState.eventTimestamp = events[len(events)-1].CreatedAt() + + if idx+1 == len(events) { + return nil, false, nil + } + events = events[idx+1:] + + additionalIteration = eventAmount == int(h.bulkLimit) + + fillFieldsEvents := make([]eventstore.FillFieldsEvent, len(events)) + highestPosition := events[len(events)-1].Position() + for i, event := range events { + if event.Position() == highestPosition { + offset++ + } + fillFieldsEvents[i] = event.(eventstore.FillFieldsEvent) + } + + return fillFieldsEvents, additionalIteration, nil +} + +func skipPreviouslyReducedEvents(events []eventstore.Event, currentState *state) (index int, offset uint32) { + var position float64 + for i, event := range events { + if event.Position() != position { + offset = 0 + position = event.Position() + } + offset++ + if event.Position() == currentState.position && + event.Aggregate().ID == currentState.aggregateID && + event.Aggregate().Type == currentState.aggregateType && + event.Sequence() == currentState.sequence { + return i, offset + } + } + return -1, 0 +} diff --git a/internal/eventstore/handler/v2/handler.go b/internal/eventstore/handler/v2/handler.go index 0e0f8ae4d..ed16787aa 100644 --- a/internal/eventstore/handler/v2/handler.go +++ b/internal/eventstore/handler/v2/handler.go @@ -28,6 +28,7 @@ type EventStore interface { FilterToQueryReducer(ctx context.Context, reducer eventstore.QueryReducer) error Filter(ctx context.Context, queryFactory *eventstore.SearchQueryBuilder) ([]eventstore.Event, error) Push(ctx context.Context, cmds ...eventstore.Command) ([]eventstore.Event, error) + FillFields(ctx context.Context, events ...eventstore.FillFieldsEvent) error } type Config struct { @@ -542,7 +543,7 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta return []*Statement{stmt}, false, nil } - events, err := h.es.Filter(ctx, h.eventQuery(currentState)) + events, err := h.es.Filter(ctx, h.eventQuery(currentState).SetTx(tx)) if err != nil { h.log().WithError(err).Debug("filter eventstore failed") return nil, false, err @@ -554,7 +555,7 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta return nil, false, err } - idx := skipPreviouslyReduced(statements, currentState) + idx := skipPreviouslyReducedStatements(statements, currentState) if idx+1 == len(statements) { currentState.position = statements[len(statements)-1].Position currentState.offset = statements[len(statements)-1].offset @@ -576,7 +577,7 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta return statements, additionalIteration, nil } -func skipPreviouslyReduced(statements []*Statement, currentState *state) int { +func skipPreviouslyReducedStatements(statements []*Statement, currentState *state) int { for i, statement := range statements { if statement.Position == currentState.position && statement.AggregateID == currentState.aggregateID && @@ -655,12 +656,11 @@ func (h *Handler) eventQuery(currentState *state) *eventstore.SearchQueryBuilder } for aggregateType, eventTypes := range h.eventTypes { - query := builder. + builder = builder. AddQuery(). AggregateTypes(aggregateType). - EventTypes(eventTypes...) - - builder = query.Builder() + EventTypes(eventTypes...). + Builder() } return builder diff --git a/internal/eventstore/repository/event.go b/internal/eventstore/repository/event.go index d55661f15..57b85f15b 100644 --- a/internal/eventstore/repository/event.go +++ b/internal/eventstore/repository/event.go @@ -120,3 +120,7 @@ func (e *Event) Payload() any { func (e *Event) UniqueConstraints() []*eventstore.UniqueConstraint { return e.Constraints } + +func (e *Event) Fields() []*eventstore.FieldOperation { + return nil +} diff --git a/internal/eventstore/v3/field.go b/internal/eventstore/v3/field.go new file mode 100644 index 000000000..1298e59e4 --- /dev/null +++ b/internal/eventstore/v3/field.go @@ -0,0 +1,367 @@ +package eventstore + +import ( + "context" + "database/sql" + _ "embed" + "encoding/json" + "reflect" + "slices" + "strconv" + "strings" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/telemetry/tracing" + "github.com/zitadel/zitadel/internal/zerrors" +) + +type fieldValue struct { + value []byte +} + +func (value *fieldValue) Unmarshal(ptr any) error { + return json.Unmarshal(value.value, ptr) +} + +func (es *Eventstore) FillFields(ctx context.Context, events ...eventstore.FillFieldsEvent) (err error) { + ctx, span := tracing.NewSpan(ctx) + defer span.End() + + tx, err := es.client.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { + if err != nil { + _ = tx.Rollback() + return + } + err = tx.Commit() + }() + + return handleFieldFillEvents(ctx, tx, events) +} + +// Search implements the [eventstore.Search] method +func (es *Eventstore) Search(ctx context.Context, conditions ...map[eventstore.FieldType]any) (result []*eventstore.SearchResult, err error) { + ctx, span := tracing.NewSpan(ctx) + defer span.EndWithError(err) + + var builder strings.Builder + args := buildSearchStatement(ctx, &builder, conditions...) + + err = es.client.QueryContext( + ctx, + func(rows *sql.Rows) error { + for rows.Next() { + var ( + res eventstore.SearchResult + value fieldValue + ) + err = rows.Scan( + &res.Aggregate.InstanceID, + &res.Aggregate.ResourceOwner, + &res.Aggregate.Type, + &res.Aggregate.ID, + &res.Object.Type, + &res.Object.ID, + &res.Object.Revision, + &res.FieldName, + &value.value, + ) + if err != nil { + return err + } + res.Value = &value + + result = append(result, &res) + } + return nil + }, + builder.String(), + args..., + ) + if err != nil { + return nil, err + } + + return result, nil +} + +const searchQueryPrefix = `SELECT instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value FROM eventstore.fields WHERE instance_id = $1` + +func buildSearchStatement(ctx context.Context, builder *strings.Builder, conditions ...map[eventstore.FieldType]any) []any { + args := make([]any, 0, len(conditions)*4+1) + args = append(args, authz.GetInstance(ctx).InstanceID()) + + builder.WriteString(searchQueryPrefix) + + builder.WriteString(" AND ") + if len(conditions) > 1 { + builder.WriteRune('(') + } + for i, condition := range conditions { + if i > 0 { + builder.WriteString(" OR ") + } + if len(condition) > 1 { + builder.WriteRune('(') + } + args = append(args, buildSearchCondition(builder, len(args)+1, condition)...) + if len(condition) > 1 { + builder.WriteRune(')') + } + } + if len(conditions) > 1 { + builder.WriteRune(')') + } + + return args +} + +func buildSearchCondition(builder *strings.Builder, index int, conditions map[eventstore.FieldType]any) []any { + args := make([]any, 0, len(conditions)) + + orderedCondition := make([]eventstore.FieldType, 0, len(conditions)) + for field := range conditions { + orderedCondition = append(orderedCondition, field) + } + slices.Sort(orderedCondition) + + for _, field := range orderedCondition { + if len(args) > 0 { + builder.WriteString(" AND ") + } + builder.WriteString(fieldNameByType(field, conditions[field])) + builder.WriteString(" = $") + builder.WriteString(strconv.Itoa(index + len(args))) + args = append(args, conditions[field]) + } + + return args +} + +func handleFieldCommands(ctx context.Context, tx *sql.Tx, commands []eventstore.Command) error { + for _, command := range commands { + if len(command.Fields()) > 0 { + if err := handleFieldOperations(ctx, tx, command.Fields()); err != nil { + return err + } + } + } + return nil +} + +func handleFieldFillEvents(ctx context.Context, tx *sql.Tx, events []eventstore.FillFieldsEvent) error { + for _, event := range events { + if len(event.Fields()) > 0 { + if err := handleFieldOperations(ctx, tx, event.Fields()); err != nil { + return err + } + } + } + return nil +} + +func handleFieldOperations(ctx context.Context, tx *sql.Tx, operations []*eventstore.FieldOperation) error { + for _, operation := range operations { + if operation.Set != nil { + if err := handleFieldSet(ctx, tx, operation.Set); err != nil { + return err + } + continue + } + if operation.Remove != nil { + if err := handleSearchDelete(ctx, tx, operation.Remove); err != nil { + return err + } + } + } + + return nil +} + +func handleFieldSet(ctx context.Context, tx *sql.Tx, field *eventstore.Field) error { + if len(field.UpsertConflictFields) == 0 { + return handleSearchInsert(ctx, tx, field) + } + return handleSearchUpsert(ctx, tx, field) +} + +const ( + insertField = `INSERT INTO eventstore.fields (instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value, value_must_be_unique, should_index) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)` +) + +func handleSearchInsert(ctx context.Context, tx *sql.Tx, field *eventstore.Field) error { + value, err := json.Marshal(field.Value.Value) + if err != nil { + return zerrors.ThrowInvalidArgument(err, "V3-fcrW1", "unable to marshal field value") + } + _, err = tx.ExecContext( + ctx, + insertField, + + field.Aggregate.InstanceID, + field.Aggregate.ResourceOwner, + field.Aggregate.Type, + field.Aggregate.ID, + field.Object.Type, + field.Object.ID, + field.Object.Revision, + field.FieldName, + value, + field.Value.MustBeUnique, + field.Value.ShouldIndex, + ) + return err +} + +const ( + fieldsUpsertPrefix = `WITH upsert AS (UPDATE eventstore.fields SET (instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value, value_must_be_unique, should_index) = ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) WHERE ` + fieldsUpsertSuffix = ` RETURNING * ) INSERT INTO eventstore.fields (instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value, value_must_be_unique, should_index) SELECT $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 WHERE NOT EXISTS (SELECT 1 FROM upsert)` +) + +func handleSearchUpsert(ctx context.Context, tx *sql.Tx, field *eventstore.Field) error { + value, err := json.Marshal(field.Value.Value) + if err != nil { + return zerrors.ThrowInvalidArgument(err, "V3-fcrW1", "unable to marshal field value") + } + + _, err = tx.ExecContext( + ctx, + writeUpsertField(field.UpsertConflictFields), + + field.Aggregate.InstanceID, + field.Aggregate.ResourceOwner, + field.Aggregate.Type, + field.Aggregate.ID, + field.Object.Type, + field.Object.ID, + field.Object.Revision, + field.FieldName, + value, + field.Value.MustBeUnique, + field.Value.ShouldIndex, + ) + return err +} + +func writeUpsertField(fields []eventstore.FieldType) string { + var builder strings.Builder + + builder.WriteString(fieldsUpsertPrefix) + for i, fieldName := range fields { + if i > 0 { + builder.WriteString(" AND ") + } + name, index := searchFieldNameAndIndexByTypeForPush(fieldName) + + builder.WriteString(name) + builder.WriteString(" = ") + builder.WriteString(index) + } + builder.WriteString(fieldsUpsertSuffix) + + return builder.String() +} + +const removeSearch = `DELETE FROM eventstore.fields WHERE ` + +func handleSearchDelete(ctx context.Context, tx *sql.Tx, clauses map[eventstore.FieldType]any) error { + if len(clauses) == 0 { + return zerrors.ThrowInvalidArgument(nil, "V3-oqlBZ", "no conditions") + } + stmt, args := writeDeleteField(clauses) + _, err := tx.ExecContext(ctx, stmt, args...) + return err +} + +func writeDeleteField(clauses map[eventstore.FieldType]any) (string, []any) { + var ( + builder strings.Builder + args = make([]any, 0, len(clauses)) + ) + builder.WriteString(removeSearch) + + orderedCondition := make([]eventstore.FieldType, 0, len(clauses)) + for field := range clauses { + orderedCondition = append(orderedCondition, field) + } + slices.Sort(orderedCondition) + + for _, fieldName := range orderedCondition { + if len(args) > 0 { + builder.WriteString(" AND ") + } + builder.WriteString(fieldNameByType(fieldName, clauses[fieldName])) + + builder.WriteString(" = $") + builder.WriteString(strconv.Itoa(len(args) + 1)) + + args = append(args, clauses[fieldName]) + } + + return builder.String(), args +} + +func fieldNameByType(typ eventstore.FieldType, value any) string { + switch typ { + case eventstore.FieldTypeAggregateID: + return "aggregate_id" + case eventstore.FieldTypeAggregateType: + return "aggregate_type" + case eventstore.FieldTypeInstanceID: + return "instance_id" + case eventstore.FieldTypeResourceOwner: + return "resource_owner" + case eventstore.FieldTypeFieldName: + return "field_name" + case eventstore.FieldTypeObjectType: + return "object_type" + case eventstore.FieldTypeObjectID: + return "object_id" + case eventstore.FieldTypeObjectRevision: + return "object_revision" + case eventstore.FieldTypeValue: + return valueColumn(value) + } + return "" +} + +func searchFieldNameAndIndexByTypeForPush(typ eventstore.FieldType) (string, string) { + switch typ { + case eventstore.FieldTypeInstanceID: + return "instance_id", "$1" + case eventstore.FieldTypeResourceOwner: + return "resource_owner", "$2" + case eventstore.FieldTypeAggregateType: + return "aggregate_type", "$3" + case eventstore.FieldTypeAggregateID: + return "aggregate_id", "$4" + case eventstore.FieldTypeObjectType: + return "object_type", "$5" + case eventstore.FieldTypeObjectID: + return "object_id", "$6" + case eventstore.FieldTypeObjectRevision: + return "object_revision", "$7" + case eventstore.FieldTypeFieldName: + return "field_name", "$8" + case eventstore.FieldTypeValue: + return "value", "$9" + } + return "", "" +} + +func valueColumn(value any) string { + //nolint: exhaustive + switch reflect.TypeOf(value).Kind() { + case reflect.Bool: + return "bool_value" + case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Float32, reflect.Float64: + return "number_value" + case reflect.String: + return "text_value" + } + return "" +} diff --git a/internal/eventstore/v3/field_test.go b/internal/eventstore/v3/field_test.go new file mode 100644 index 000000000..0847c1b8a --- /dev/null +++ b/internal/eventstore/v3/field_test.go @@ -0,0 +1,260 @@ +package eventstore + +import ( + "context" + _ "embed" + "reflect" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/eventstore" +) + +func Test_handleSearchDelete(t *testing.T) { + type args struct { + clauses map[eventstore.FieldType]any + } + type want struct { + stmt string + args []any + } + tests := []struct { + name string + args args + want want + }{ + { + name: "1 condition", + args: args{ + clauses: map[eventstore.FieldType]any{ + eventstore.FieldTypeInstanceID: "i_id", + }, + }, + want: want{ + stmt: "DELETE FROM eventstore.fields WHERE instance_id = $1", + args: []any{"i_id"}, + }, + }, + { + name: "2 conditions", + args: args{ + clauses: map[eventstore.FieldType]any{ + eventstore.FieldTypeInstanceID: "i_id", + eventstore.FieldTypeAggregateID: "a_id", + }, + }, + want: want{ + stmt: "DELETE FROM eventstore.fields WHERE aggregate_id = $1 AND instance_id = $2", + args: []any{"a_id", "i_id"}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + stmt, args := writeDeleteField(tt.args.clauses) + if stmt != tt.want.stmt { + t.Errorf("handleSearchDelete() stmt = %q, want %q", stmt, tt.want.stmt) + } + assert.Equal(t, tt.want.args, args) + }) + } +} + +func Test_writeUpsertField(t *testing.T) { + type args struct { + fields []eventstore.FieldType + } + tests := []struct { + name string + args args + want string + }{ + { + name: "1 field", + args: args{ + fields: []eventstore.FieldType{ + eventstore.FieldTypeInstanceID, + }, + }, + want: "WITH upsert AS (UPDATE eventstore.fields SET (instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value, value_must_be_unique, should_index) = ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) WHERE instance_id = $1 RETURNING * ) INSERT INTO eventstore.fields (instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value, value_must_be_unique, should_index) SELECT $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 WHERE NOT EXISTS (SELECT 1 FROM upsert)", + }, + { + name: "2 fields", + args: args{ + fields: []eventstore.FieldType{ + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeAggregateType, + }, + }, + want: "WITH upsert AS (UPDATE eventstore.fields SET (instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value, value_must_be_unique, should_index) = ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) WHERE instance_id = $1 AND aggregate_type = $3 RETURNING * ) INSERT INTO eventstore.fields (instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value, value_must_be_unique, should_index) SELECT $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 WHERE NOT EXISTS (SELECT 1 FROM upsert)", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := writeUpsertField(tt.args.fields); got != tt.want { + t.Errorf("writeUpsertField() = %q, want %q", got, tt.want) + } + }) + } +} + +func Test_buildSearchCondition(t *testing.T) { + type args struct { + index int + conditions map[eventstore.FieldType]any + } + type want struct { + stmt string + args []any + } + tests := []struct { + name string + args args + want want + }{ + { + name: "1 condition", + args: args{ + index: 1, + conditions: map[eventstore.FieldType]any{ + eventstore.FieldTypeAggregateID: "a_id", + }, + }, + want: want{ + stmt: "aggregate_id = $1", + args: []any{"a_id"}, + }, + }, + { + name: "3 condition", + args: args{ + index: 1, + conditions: map[eventstore.FieldType]any{ + eventstore.FieldTypeAggregateID: "a_id", + eventstore.FieldTypeInstanceID: "i_id", + eventstore.FieldTypeAggregateType: "a_type", + }, + }, + want: want{ + stmt: "aggregate_type = $1 AND aggregate_id = $2 AND instance_id = $3", + args: []any{"a_type", "a_id", "i_id"}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var builder strings.Builder + + if got := buildSearchCondition(&builder, tt.args.index, tt.args.conditions); !reflect.DeepEqual(got, tt.want.args) { + t.Errorf("buildSearchCondition() = %v, want %v", got, tt.want) + } + if tt.want.stmt != builder.String() { + t.Errorf("buildSearchCondition() stmt = %q, want %q", builder.String(), tt.want.stmt) + } + }) + } +} + +func Test_buildSearchStatement(t *testing.T) { + type args struct { + index int + conditions []map[eventstore.FieldType]any + } + type want struct { + stmt string + args []any + } + tests := []struct { + name string + args args + want want + }{ + { + name: "1 condition with 1 field", + args: args{ + index: 1, + conditions: []map[eventstore.FieldType]any{ + { + eventstore.FieldTypeAggregateID: "a_id", + }, + }, + }, + want: want{ + stmt: "SELECT instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value FROM eventstore.fields WHERE instance_id = $1 AND aggregate_id = $2", + args: []any{"a_id"}, + }, + }, + { + name: "1 condition with 3 fields", + args: args{ + index: 1, + conditions: []map[eventstore.FieldType]any{ + { + eventstore.FieldTypeAggregateID: "a_id", + eventstore.FieldTypeInstanceID: "i_id", + eventstore.FieldTypeAggregateType: "a_type", + }, + }, + }, + want: want{ + stmt: "SELECT instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value FROM eventstore.fields WHERE instance_id = $1 AND (aggregate_type = $2 AND aggregate_id = $3 AND instance_id = $4)", + args: []any{"a_type", "a_id", "i_id"}, + }, + }, + { + name: "2 condition with 1 field", + args: args{ + index: 1, + conditions: []map[eventstore.FieldType]any{ + { + eventstore.FieldTypeAggregateID: "a_id", + }, + { + eventstore.FieldTypeAggregateType: "a_type", + }, + }, + }, + want: want{ + stmt: "SELECT instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value FROM eventstore.fields WHERE instance_id = $1 AND (aggregate_id = $2 OR aggregate_type = $3)", + args: []any{"a_id", "a_type"}, + }, + }, + { + name: "2 condition with 2 fields", + args: args{ + index: 1, + conditions: []map[eventstore.FieldType]any{ + { + eventstore.FieldTypeAggregateID: "a_id1", + eventstore.FieldTypeAggregateType: "a_type1", + }, + { + eventstore.FieldTypeAggregateID: "a_id2", + eventstore.FieldTypeAggregateType: "a_type2", + }, + }, + }, + want: want{ + stmt: "SELECT instance_id, resource_owner, aggregate_type, aggregate_id, object_type, object_id, object_revision, field_name, value FROM eventstore.fields WHERE instance_id = $1 AND ((aggregate_type = $2 AND aggregate_id = $3) OR (aggregate_type = $4 AND aggregate_id = $5))", + args: []any{"a_type1", "a_id1", "a_type2", "a_id2"}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var builder strings.Builder + tt.want.args = append([]any{"i_id"}, tt.want.args...) + ctx := authz.WithInstanceID(context.Background(), "i_id") + + if got := buildSearchStatement(ctx, &builder, tt.args.conditions...); !reflect.DeepEqual(got, tt.want.args) { + t.Errorf("buildSearchStatement() = %v, want %v", got, tt.want) + } + if tt.want.stmt != builder.String() { + t.Errorf("buildSearchStatement() stmt = %q, want %q", builder.String(), tt.want.stmt) + } + }) + } +} diff --git a/internal/eventstore/v3/mock_test.go b/internal/eventstore/v3/mock_test.go index 42435d9c4..8de5ae8e2 100644 --- a/internal/eventstore/v3/mock_test.go +++ b/internal/eventstore/v3/mock_test.go @@ -42,6 +42,10 @@ func (m *mockCommand) UniqueConstraints() []*eventstore.UniqueConstraint { return m.constraints } +func (e *mockCommand) Fields() []*eventstore.FieldOperation { + return nil +} + func mockEvent(aggregate *eventstore.Aggregate, sequence uint64, payload Payload) eventstore.Event { return &event{ aggregate: aggregate, diff --git a/internal/eventstore/v3/push.go b/internal/eventstore/v3/push.go index 22ea9295f..d49fc408b 100644 --- a/internal/eventstore/v3/push.go +++ b/internal/eventstore/v3/push.go @@ -41,7 +41,20 @@ func (es *Eventstore) Push(ctx context.Context, commands ...eventstore.Command) return err } - return handleUniqueConstraints(ctx, tx, commands) + if err = handleUniqueConstraints(ctx, tx, commands); err != nil { + return err + } + + // CockroachDB by default does not allow multiple modifications of the same table using ON CONFLICT + // Thats why we enable it manually + if es.client.Type() == "cockroach" { + _, err = tx.Exec("SET enable_multiple_modifications_of_table = on") + if err != nil { + return err + } + } + + return handleFieldCommands(ctx, tx, commands) }) if err != nil { diff --git a/internal/feature/feature.go b/internal/feature/feature.go index dc3e0eb59..d0f8a9f7a 100644 --- a/internal/feature/feature.go +++ b/internal/feature/feature.go @@ -42,6 +42,8 @@ type ImprovedPerformanceType int32 const ( ImprovedPerformanceTypeUnknown = iota ImprovedPerformanceTypeOrgByID + ImprovedPerformanceTypeProjectGrant + ImprovedPerformanceTypeProject ) func (f Features) ShouldUseImprovedPerformance(typ ImprovedPerformanceType) bool { diff --git a/internal/query/projection/eventstore_field.go b/internal/query/projection/eventstore_field.go new file mode 100644 index 000000000..5e0a20504 --- /dev/null +++ b/internal/query/projection/eventstore_field.go @@ -0,0 +1,19 @@ +package projection + +import ( + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/eventstore/handler/v2" + "github.com/zitadel/zitadel/internal/repository/org" + "github.com/zitadel/zitadel/internal/repository/project" +) + +func newFillProjectGrantFields(config handler.Config) *handler.FieldHandler { + return handler.NewFieldHandler( + &config, + "project_grant_fields", + map[eventstore.AggregateType][]eventstore.EventType{ + org.AggregateType: nil, + project.AggregateType: nil, + }, + ) +} diff --git a/internal/query/projection/eventstore_mock_test.go b/internal/query/projection/eventstore_mock_test.go index c1cdf434d..202664d51 100644 --- a/internal/query/projection/eventstore_mock_test.go +++ b/internal/query/projection/eventstore_mock_test.go @@ -49,3 +49,7 @@ func (m *mockEventStore) Push(ctx context.Context, cmds ...eventstore.Command) ( m.pushCounter++ return m.pushResponse[m.pushCounter-1], nil } + +func (m *mockEventStore) FillFields(ctx context.Context, events ...eventstore.FillFieldsEvent) error { + return nil +} diff --git a/internal/query/projection/projection.go b/internal/query/projection/projection.go index 30c1df687..09c3d8d86 100644 --- a/internal/query/projection/projection.go +++ b/internal/query/projection/projection.go @@ -77,6 +77,8 @@ var ( TargetProjection *handler.Handler ExecutionProjection *handler.Handler UserSchemaProjection *handler.Handler + + ProjectGrantFields *handler.FieldHandler ) type projection interface { @@ -158,6 +160,9 @@ func Create(ctx context.Context, sqlClient *database.DB, es handler.EventStore, TargetProjection = newTargetProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["targets"])) ExecutionProjection = newExecutionProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["executions"])) UserSchemaProjection = newUserSchemaProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["user_schemas"])) + + ProjectGrantFields = newFillProjectGrantFields(applyCustomConfig(projectionConfig, config.Customizations["project_grant_fields"])) + newProjectionsList() return nil } diff --git a/internal/repository/org/org.go b/internal/repository/org/org.go index af19fb18f..95c9c2f3e 100644 --- a/internal/repository/org/org.go +++ b/internal/repository/org/org.go @@ -17,6 +17,10 @@ const ( OrgDeactivatedEventType = orgEventTypePrefix + "deactivated" OrgReactivatedEventType = orgEventTypePrefix + "reactivated" OrgRemovedEventType = orgEventTypePrefix + "removed" + + OrgSearchType = "org" + OrgNameSearchField = "name" + OrgStateSearchField = "state" ) func NewAddOrgNameUniqueConstraint(orgName string) *eventstore.UniqueConstraint { @@ -46,6 +50,43 @@ func (e *OrgAddedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return []*eventstore.UniqueConstraint{NewAddOrgNameUniqueConstraint(e.Name)} } +func (e *OrgAddedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + orgSearchObject(e.Aggregate().ID), + OrgNameSearchField, + &eventstore.Value{ + Value: e.Name, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + eventstore.SetField( + e.Aggregate(), + orgSearchObject(e.Aggregate().ID), + OrgStateSearchField, + &eventstore.Value{ + Value: domain.OrgStateActive, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewOrgAddedEvent(ctx context.Context, aggregate *eventstore.Aggregate, name string) *OrgAddedEvent { return &OrgAddedEvent{ BaseEvent: *eventstore.NewBaseEventForPush( @@ -87,6 +128,28 @@ func (e *OrgChangedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { } } +func (e *OrgChangedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + orgSearchObject(e.Aggregate().ID), + OrgNameSearchField, + &eventstore.Value{ + Value: e.Name, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewOrgChangedEvent(ctx context.Context, aggregate *eventstore.Aggregate, oldName, newName string) *OrgChangedEvent { return &OrgChangedEvent{ BaseEvent: *eventstore.NewBaseEventForPush( @@ -123,6 +186,28 @@ func (e *OrgDeactivatedEvent) UniqueConstraints() []*eventstore.UniqueConstraint return nil } +func (e *OrgDeactivatedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + orgSearchObject(e.Aggregate().ID), + OrgStateSearchField, + &eventstore.Value{ + Value: domain.OrgStateInactive, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewOrgDeactivatedEvent(ctx context.Context, aggregate *eventstore.Aggregate) *OrgDeactivatedEvent { return &OrgDeactivatedEvent{ BaseEvent: *eventstore.NewBaseEventForPush( @@ -143,6 +228,28 @@ type OrgReactivatedEvent struct { eventstore.BaseEvent `json:"-"` } +func (e *OrgReactivatedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + orgSearchObject(e.Aggregate().ID), + OrgStateSearchField, + &eventstore.Value{ + Value: domain.OrgStateActive, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func (e *OrgReactivatedEvent) Payload() interface{} { return e } @@ -200,6 +307,29 @@ func (e *OrgRemovedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return constraints } +func (e *OrgRemovedEvent) Fields() []*eventstore.FieldOperation { + // TODO: project grants are currently not removed because we don't have the relationship between the granted org and the grant + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + orgSearchObject(e.Aggregate().ID), + OrgStateSearchField, + &eventstore.Value{ + Value: domain.OrgStateRemoved, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewOrgRemovedEvent(ctx context.Context, aggregate *eventstore.Aggregate, name string, usernames []string, loginMustBeDomain bool, domains []string, externalIDPs []*domain.UserIDPLink, samlEntityIDs []string) *OrgRemovedEvent { return &OrgRemovedEvent{ BaseEvent: *eventstore.NewBaseEventForPush( @@ -221,3 +351,11 @@ func OrgRemovedEventMapper(event eventstore.Event) (eventstore.Event, error) { BaseEvent: *eventstore.BaseEventFromRepo(event), }, nil } + +func orgSearchObject(id string) eventstore.Object { + return eventstore.Object{ + Type: OrgSearchType, + Revision: 1, + ID: id, + } +} diff --git a/internal/repository/project/grant.go b/internal/repository/project/grant.go index 486a62d92..a78172a03 100644 --- a/internal/repository/project/grant.go +++ b/internal/repository/project/grant.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + "github.com/zitadel/zitadel/internal/domain" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/zerrors" ) @@ -17,6 +18,13 @@ var ( GrantDeactivatedType = grantEventTypePrefix + "deactivated" GrantReactivatedType = grantEventTypePrefix + "reactivated" GrantRemovedType = grantEventTypePrefix + "removed" + + ProjectGrantSearchType = "project_grant" + ProjectGrantGrantIDSearchField = "grant_id" + ProjectGrantGrantedOrgIDSearchField = "granted_org_id" + ProjectGrantStateSearchField = "state" + ProjectGrantRoleKeySearchField = "role_key" + ProjectGrantObjectRevision = uint8(1) ) func NewAddProjectGrantUniqueConstraint(grantedOrgID, projectID string) *eventstore.UniqueConstraint { @@ -48,6 +56,76 @@ func (e *GrantAddedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return []*eventstore.UniqueConstraint{NewAddProjectGrantUniqueConstraint(e.GrantedOrgID, e.Aggregate().ID)} } +func (e *GrantAddedEvent) Fields() []*eventstore.FieldOperation { + fields := make([]*eventstore.FieldOperation, 0, len(e.RoleKeys)+3) + fields = append(fields, + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + ProjectGrantGrantIDSearchField, + &eventstore.Value{ + Value: e.GrantID, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + ProjectGrantGrantedOrgIDSearchField, + &eventstore.Value{ + Value: e.GrantedOrgID, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + ProjectGrantStateSearchField, + &eventstore.Value{ + Value: domain.ProjectGrantStateActive, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + ) + + for _, roleKey := range e.RoleKeys { + fields = append(fields, + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + ProjectGrantRoleKeySearchField, + &eventstore.Value{ + Value: roleKey, + ShouldIndex: true, + }, + ), + ) + } + + return fields +} + func NewGrantAddedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -95,6 +173,37 @@ func (e *GrantChangedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return nil } +func (e *GrantChangedEvent) Fields() []*eventstore.FieldOperation { + fields := make([]*eventstore.FieldOperation, 0, len(e.RoleKeys)+1) + + fields = append(fields, + eventstore.RemoveSearchFieldsByAggregateAndObjectAndField( + e.Aggregate(), + grantSearchObject(e.GrantID), + + ProjectGrantRoleKeySearchField, + ), + ) + + for _, roleKey := range e.RoleKeys { + fields = append(fields, + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + + ProjectGrantRoleKeySearchField, + + &eventstore.Value{ + Value: roleKey, + ShouldIndex: true, + }, + ), + ) + } + + return fields +} + func NewGrantChangedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -140,6 +249,43 @@ func (e *GrantCascadeChangedEvent) UniqueConstraints() []*eventstore.UniqueConst return nil } +func (e *GrantCascadeChangedEvent) Fields() []*eventstore.FieldOperation { + fields := make([]*eventstore.FieldOperation, 0, len(e.RoleKeys)+1) + + fields = append(fields, + eventstore.RemoveSearchFieldsByAggregateAndObjectAndField( + e.Aggregate(), + grantSearchObject(e.GrantID), + + ProjectGrantRoleKeySearchField, + ), + ) + + for _, roleKey := range e.RoleKeys { + fields = append(fields, + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + + ProjectGrantRoleKeySearchField, + &eventstore.Value{ + Value: roleKey, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + ) + } + + return fields +} + func NewGrantCascadeChangedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -184,6 +330,29 @@ func (e *GrantDeactivateEvent) UniqueConstraints() []*eventstore.UniqueConstrain return nil } +func (e *GrantDeactivateEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + + ProjectGrantStateSearchField, + &eventstore.Value{ + Value: domain.ProjectGrantStateInactive, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewGrantDeactivateEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -226,6 +395,29 @@ func (e *GrantReactivatedEvent) UniqueConstraints() []*eventstore.UniqueConstrai return nil } +func (e *GrantReactivatedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + + ProjectGrantStateSearchField, + &eventstore.Value{ + Value: domain.ProjectGrantStateActive, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewGrantReactivatedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -269,6 +461,29 @@ func (e *GrantRemovedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return []*eventstore.UniqueConstraint{NewRemoveProjectGrantUniqueConstraint(e.grantedOrgID, e.Aggregate().ID)} } +func (e *GrantRemovedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + grantSearchObject(e.GrantID), + + ProjectGrantStateSearchField, + &eventstore.Value{ + Value: domain.ProjectGrantStateRemoved, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewGrantRemovedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -298,3 +513,11 @@ func GrantRemovedEventMapper(event eventstore.Event) (eventstore.Event, error) { return e, nil } + +func grantSearchObject(id string) eventstore.Object { + return eventstore.Object{ + Type: ProjectGrantSearchType, + Revision: 1, + ID: id, + } +} diff --git a/internal/repository/project/project.go b/internal/repository/project/project.go index 977487730..6147a632e 100644 --- a/internal/repository/project/project.go +++ b/internal/repository/project/project.go @@ -16,6 +16,11 @@ const ( ProjectDeactivatedType = projectEventTypePrefix + "deactivated" ProjectReactivatedType = projectEventTypePrefix + "reactivated" ProjectRemovedType = projectEventTypePrefix + "removed" + + ProjectSearchType = "project" + ProjectObjectRevision = uint8(1) + ProjectNameSearchField = "name" + ProjectStateSearchField = "state" ) func NewAddProjectNameUniqueConstraint(projectName, resourceOwner string) *eventstore.UniqueConstraint { @@ -49,6 +54,45 @@ func (e *ProjectAddedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return []*eventstore.UniqueConstraint{NewAddProjectNameUniqueConstraint(e.Name, e.Aggregate().ResourceOwner)} } +func (e *ProjectAddedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + projectSearchObject(e.Aggregate().ID), + ProjectNameSearchField, + &eventstore.Value{ + Value: e.Name, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeObjectRevision, + eventstore.FieldTypeFieldName, + ), + eventstore.SetField( + e.Aggregate(), + projectSearchObject(e.Aggregate().ID), + ProjectStateSearchField, + &eventstore.Value{ + Value: domain.ProjectStateActive, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeObjectRevision, + eventstore.FieldTypeFieldName, + ), + } +} + func NewProjectAddedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -110,6 +154,30 @@ func (e *ProjectChangeEvent) UniqueConstraints() []*eventstore.UniqueConstraint return nil } +func (e *ProjectChangeEvent) Fields() []*eventstore.FieldOperation { + if e.Name == nil { + return nil + } + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + projectSearchObject(e.Aggregate().ID), + ProjectNameSearchField, + &eventstore.Value{ + Value: *e.Name, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewProjectChangeEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -190,6 +258,28 @@ func (e *ProjectDeactivatedEvent) UniqueConstraints() []*eventstore.UniqueConstr return nil } +func (e *ProjectDeactivatedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + projectSearchObject(e.Aggregate().ID), + ProjectStateSearchField, + &eventstore.Value{ + Value: domain.ProjectStateInactive, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewProjectDeactivatedEvent(ctx context.Context, aggregate *eventstore.Aggregate) *ProjectDeactivatedEvent { return &ProjectDeactivatedEvent{ BaseEvent: *eventstore.NewBaseEventForPush( @@ -218,6 +308,28 @@ func (e *ProjectReactivatedEvent) UniqueConstraints() []*eventstore.UniqueConstr return nil } +func (e *ProjectReactivatedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + projectSearchObject(e.Aggregate().ID), + ProjectStateSearchField, + &eventstore.Value{ + Value: domain.ProjectStateRemoved, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewProjectReactivatedEvent(ctx context.Context, aggregate *eventstore.Aggregate) *ProjectReactivatedEvent { return &ProjectReactivatedEvent{ BaseEvent: *eventstore.NewBaseEventForPush( @@ -255,6 +367,12 @@ func (e *ProjectRemovedEvent) UniqueConstraints() []*eventstore.UniqueConstraint return constraints } +func (e *ProjectRemovedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.RemoveSearchFieldsByAggregate(e.Aggregate()), + } +} + func NewProjectRemovedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -277,3 +395,11 @@ func ProjectRemovedEventMapper(event eventstore.Event) (eventstore.Event, error) BaseEvent: *eventstore.BaseEventFromRepo(event), }, nil } + +func projectSearchObject(id string) eventstore.Object { + return eventstore.Object{ + Type: ProjectSearchType, + Revision: ProjectObjectRevision, + ID: id, + } +} diff --git a/internal/repository/project/role.go b/internal/repository/project/role.go index 098b32189..2c7836da2 100644 --- a/internal/repository/project/role.go +++ b/internal/repository/project/role.go @@ -14,6 +14,12 @@ var ( RoleAddedType = roleEventTypePrefix + "added" RoleChangedType = roleEventTypePrefix + "changed" RoleRemovedType = roleEventTypePrefix + "removed" + + ProjectRoleSearchType = "project_role" + ProjectRoleRevision = uint8(1) + ProjectRoleKeySearchField = "key" + ProjectRoleDisplayNameSearchField = "display_name" + ProjectRoleGroupSearchField = "group" ) func NewAddProjectRoleUniqueConstraint(roleKey, projectID string) *eventstore.UniqueConstraint { @@ -45,6 +51,59 @@ func (e *RoleAddedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return []*eventstore.UniqueConstraint{NewAddProjectRoleUniqueConstraint(e.Key, e.Aggregate().ID)} } +func (e *RoleAddedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.SetField( + e.Aggregate(), + projectRoleSearchObject(e.Key), + ProjectRoleKeySearchField, + &eventstore.Value{ + Value: e.Key, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + eventstore.SetField( + e.Aggregate(), + projectRoleSearchObject(e.Key), + ProjectRoleDisplayNameSearchField, + &eventstore.Value{ + Value: e.DisplayName, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + eventstore.SetField( + e.Aggregate(), + projectRoleSearchObject(e.Key), + ProjectRoleGroupSearchField, + &eventstore.Value{ + Value: e.Group, + ShouldIndex: true, + }, + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + ), + } +} + func NewRoleAddedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -93,6 +152,50 @@ func (e *RoleChangedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return nil } +func (e *RoleChangedEvent) Fields() []*eventstore.FieldOperation { + operations := make([]*eventstore.FieldOperation, 0, 2) + if e.DisplayName != nil { + operations = append(operations, eventstore.SetField( + e.Aggregate(), + projectRoleSearchObject(e.Key), + ProjectRoleDisplayNameSearchField, + &eventstore.Value{ + Value: *e.DisplayName, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + )) + } + if e.Group != nil { + operations = append(operations, eventstore.SetField( + e.Aggregate(), + projectRoleSearchObject(e.Key), + ProjectRoleGroupSearchField, + &eventstore.Value{ + Value: *e.Group, + ShouldIndex: true, + }, + + eventstore.FieldTypeInstanceID, + eventstore.FieldTypeResourceOwner, + eventstore.FieldTypeAggregateType, + eventstore.FieldTypeAggregateID, + eventstore.FieldTypeObjectType, + eventstore.FieldTypeObjectID, + eventstore.FieldTypeFieldName, + )) + } + + return operations +} + func NewRoleChangedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -162,6 +265,15 @@ func (e *RoleRemovedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { return []*eventstore.UniqueConstraint{NewRemoveProjectRoleUniqueConstraint(e.Key, e.Aggregate().ID)} } +func (e *RoleRemovedEvent) Fields() []*eventstore.FieldOperation { + return []*eventstore.FieldOperation{ + eventstore.RemoveSearchFieldsByAggregateAndObject( + e.Aggregate(), + projectRoleSearchObject(e.Key), + ), + } +} + func NewRoleRemovedEvent( ctx context.Context, aggregate *eventstore.Aggregate, @@ -188,3 +300,11 @@ func RoleRemovedEventMapper(event eventstore.Event) (eventstore.Event, error) { return e, nil } + +func projectRoleSearchObject(id string) eventstore.Object { + return eventstore.Object{ + Type: ProjectRoleSearchType, + Revision: ProjectRoleRevision, + ID: id, + } +} diff --git a/proto/zitadel/feature/v2beta/feature.proto b/proto/zitadel/feature/v2beta/feature.proto index 082159b95..b000092fa 100644 --- a/proto/zitadel/feature/v2beta/feature.proto +++ b/proto/zitadel/feature/v2beta/feature.proto @@ -54,4 +54,9 @@ enum ImprovedPerformance { // Uses the eventstore to query the org by id // instead of the sql table. IMPROVED_PERFORMANCE_ORG_BY_ID = 1; + // Improves performance on write side by using + // optimized processes to query data to determine + // correctnes of data. + IMPROVED_PERFORMANCE_PROJECT_GRANT = 2; + IMPROVED_PERFORMANCE_PROJECT = 3; } \ No newline at end of file