fix(setup): make step 39 repeatable (#9085)

# Which Problems Are Solved

When downgrading zitadel and upgrading it again, it might be that orgs
deleted in this period still have stale entries in the fields table.

# How the Problems Are Solved

- Make the cleanup repeatable
- Scope the query by instance so that an index is used.

(cherry picked from commit da706a8b30dca5ab4d7d96ae4225dcc9d55d85a5)

cmd/setup/39.go

@@ -3,9 +3,12 @@ package setup
 import (
 	"context"
 	_ "embed"
+	"fmt"
+
+	"github.com/zitadel/logging"
 
-	"github.com/zitadel/zitadel/internal/database"
 	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/repository/instance"
 )
 
 var (
@@ -14,14 +17,35 @@ var (
 )
 
 type DeleteStaleOrgFields struct {
-	dbClient *database.DB
+	eventstore *eventstore.Eventstore
 }
 
 func (mig *DeleteStaleOrgFields) Execute(ctx context.Context, _ eventstore.Event) error {
-	_, err := mig.dbClient.ExecContext(ctx, deleteStaleOrgFields)
-	return err
+	instances, err := mig.eventstore.InstanceIDs(
+		ctx,
+		eventstore.NewSearchQueryBuilder(eventstore.ColumnsInstanceIDs).
+			OrderDesc().
+			AddQuery().
+			AggregateTypes("instance").
+			EventTypes(instance.InstanceAddedEventType).
+			Builder(),
+	)
+	if err != nil {
+		return err
+	}
+	for i, instance := range instances {
+		logging.WithFields("instance_id", instance, "migration", mig.String(), "progress", fmt.Sprintf("%d/%d", i+1, len(instances))).Info("execute delete query")
+		if _, err := mig.eventstore.Client().ExecContext(ctx, deleteStaleOrgFields, instance); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
-func (mig *DeleteStaleOrgFields) String() string {
-	return "39_delete_stale_org_fields"
+func (*DeleteStaleOrgFields) Check(map[string]any) bool {
+	return true
+}
+
+func (*DeleteStaleOrgFields) String() string {
+	return "repeatable_delete_stale_org_fields"
 }

cmd/setup/39.sql

@@ -3,6 +3,7 @@ WHERE aggregate_type = 'org'
 AND aggregate_id IN (
 	SELECT aggregate_id
 	FROM eventstore.events2
-	WHERE aggregate_type = 'org'
+	WHERE instance_id = $1
+	AND aggregate_type = 'org'
 	AND event_type = 'org.removed'
 );

cmd/setup/config.go

@@ -127,7 +127,6 @@ type Steps struct {
 	s37Apps7OIDConfigsBackChannelLogoutURI  *Apps7OIDConfigsBackChannelLogoutURI
 	s38BackChannelLogoutNotificationStart   *BackChannelLogoutNotificationStart
 	s40InitPushFunc                         *InitPushFunc
-	s39DeleteStaleOrgFields                 *DeleteStaleOrgFields
 }
 
 func MustNewSteps(v *viper.Viper) *Steps {

cmd/setup/setup.go

@@ -169,7 +169,6 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 	steps.s36FillV2Milestones = &FillV3Milestones{dbClient: queryDBClient, eventstore: eventstoreClient}
 	steps.s37Apps7OIDConfigsBackChannelLogoutURI = &Apps7OIDConfigsBackChannelLogoutURI{dbClient: esPusherDBClient}
 	steps.s38BackChannelLogoutNotificationStart = &BackChannelLogoutNotificationStart{dbClient: esPusherDBClient, esClient: eventstoreClient}
-	steps.s39DeleteStaleOrgFields = &DeleteStaleOrgFields{dbClient: esPusherDBClient}
 	steps.s40InitPushFunc = &InitPushFunc{dbClient: esPusherDBClient}
 
 	err = projection.Create(ctx, projectionDBClient, eventstoreClient, config.Projections, nil, nil, nil)
@@ -187,6 +186,9 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 			es:      eventstoreClient,
 			Version: build.Version(),
 		},
+		&DeleteStaleOrgFields{
+			eventstore: eventstoreClient,
+		},
 		&FillFieldsForInstanceDomains{
 			eventstore: eventstoreClient,
 		},
@@ -238,7 +240,6 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 		steps.s32AddAuthSessionID,
 		steps.s33SMSConfigs3TwilioAddVerifyServiceSid,
 		steps.s37Apps7OIDConfigsBackChannelLogoutURI,
-		steps.s39DeleteStaleOrgFields,
 	} {
 		mustExecuteMigration(ctx, eventstoreClient, step, "migration failed")
 	}