fix(setup): steps 10 and 11 (#5987)

* fix(step10): count amount of wrong_events manually

* fix(step11): create index for better performance

cmd/setup/10.go

@@ -19,6 +19,10 @@ var (
 	correctCreationDate10FillTable string
 	//go:embed 10/10_update.sql
 	correctCreationDate10Update string
+	//go:embed 10/10_count_wrong_events.sql
+	correctCreationDate10CountWrongEvents string
+	//go:embed 10/10_empty_table.sql
+	correctCreationDate10Truncate string
 )
 
 type CorrectCreationDate struct {
@@ -31,7 +35,7 @@ func (mig *CorrectCreationDate) Execute(ctx context.Context) (err error) {
 	defer cancel()
 
 	for {
-		var affected int64
+		affected := int64(0)
 		err = crdb.ExecuteTx(ctx, mig.dbClient.DB, nil, func(tx *sql.Tx) error {
 			if mig.dbClient.Type() == "cockroach" {
 				if _, err := tx.Exec("SET experimental_enable_temp_tables=on"); err != nil {
@@ -43,16 +47,24 @@ func (mig *CorrectCreationDate) Execute(ctx context.Context) (err error) {
 				return err
 			}
 
+			_, err = tx.ExecContext(ctx, correctCreationDate10Truncate)
+			if err != nil {
+				return err
+			}
 			_, err = tx.ExecContext(ctx, correctCreationDate10FillTable)
 			if err != nil {
 				return err
 			}
 
-			res, err := tx.ExecContext(ctx, correctCreationDate10Update)
+			res := tx.QueryRowContext(ctx, correctCreationDate10CountWrongEvents)
+			if err := res.Scan(&affected); err != nil || affected == 0 {
+				return err
+			}
+
+			_, err = tx.ExecContext(ctx, correctCreationDate10Update)
 			if err != nil {
 				return err
 			}
-			affected, _ = res.RowsAffected()
 			logging.WithFields("count", affected).Info("creation dates changed")
 			return nil
 		})

cmd/setup/10/10_count_wrong_events.sql

@@ -0,0 +1 @@
+SELECT COUNT(*) FROM wrong_events

cmd/setup/10/10_empty_table.sql

@@ -0,0 +1 @@
+TRUNCATE wrong_events

cmd/setup/10/10_fill_table.sql

@@ -1,5 +1,3 @@
-TRUNCATE wrong_events;
-
 INSERT INTO wrong_events (
     SELECT * FROM (
         SELECT

cmd/setup/11.go

@@ -3,7 +3,7 @@ package setup
 import (
 	"context"
 	"database/sql"
-	_ "embed"
+	"embed"
 	"time"
 
 	"github.com/cockroachdb/cockroach-go/v2/crdb"
@@ -21,11 +21,18 @@ var (
 	fillCreatedAt string
 	//go:embed 11/11_set_column.sql
 	setCreatedAtDetails string
+	//go:embed 11/postgres/create_index.sql
+	//go:embed 11/cockroach/create_index.sql
+	createdAtIndexCreateStmt embed.FS
+	//go:embed 11/postgres/drop_index.sql
+	//go:embed 11/cockroach/drop_index.sql
+	createdAtIndexDropStmt embed.FS
 )
 
 type AddEventCreatedAt struct {
-	step10   *CorrectCreationDate
+	BulkAmount int
-	dbClient *database.DB
+	step10     *CorrectCreationDate
+	dbClient   *database.DB
 }
 
 func (mig *AddEventCreatedAt) Execute(ctx context.Context) error {
@@ -39,10 +46,19 @@ func (mig *AddEventCreatedAt) Execute(ctx context.Context) error {
 		return err
 	}
 
+	createIndex, err := readStmt(createdAtIndexCreateStmt, "11", mig.dbClient.Type(), "create_index.sql")
+	if err != nil {
+		return err
+	}
+	_, err = mig.dbClient.ExecContext(ctx, createIndex)
+	if err != nil {
+		return err
+	}
+
 	for {
 		var count int
 		err = crdb.ExecuteTx(ctx, mig.dbClient.DB, nil, func(tx *sql.Tx) error {
-			rows, err := tx.Query(fetchCreatedAt)
+			rows, err := tx.Query(fetchCreatedAt, mig.BulkAmount)
 			if err != nil {
 				return err
 			}
@@ -80,7 +96,7 @@ func (mig *AddEventCreatedAt) Execute(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		logging.WithFields("count", count).Debug("creation dates set")
+		logging.WithFields("count", count).Info("creation dates set")
 		if count < 20 {
 			break
 		}
@@ -88,6 +104,16 @@ func (mig *AddEventCreatedAt) Execute(ctx context.Context) error {
 
 	logging.Info("set details")
 	_, err = mig.dbClient.ExecContext(ctx, setCreatedAtDetails)
+	if err != nil {
+		return err
+	}
+
+	dropIndex, err := readStmt(createdAtIndexDropStmt, "11", mig.dbClient.Type(), "drop_index.sql")
+	if err != nil {
+		return err
+	}
+	_, err = mig.dbClient.ExecContext(ctx, dropIndex)
+
 	return err
 }
 

cmd/setup/11/11_fetch_events.sql

@@ -1 +1,12 @@
-SELECT id, creation_date FROM eventstore.events WHERE created_at IS NULL ORDER BY event_sequence DESC, instance_id LIMIT 20 FOR UPDATE
+SELECT 
+    id
+    , creation_date 
+FROM 
+    eventstore.events 
+WHERE 
+    created_at IS NULL 
+ORDER BY 
+    event_sequence DESC
+    , instance_id 
+LIMIT $1 
+FOR UPDATE

cmd/setup/11/cockroach/create_index.sql

@@ -0,0 +1,8 @@
+CREATE INDEX IF NOT EXISTS ca_fill_idx ON eventstore.events (
+    event_sequence DESC
+    , instance_id
+) STORING (
+    id
+    , creation_date
+    , created_at
+) WHERE created_at IS NULL; 

cmd/setup/11/cockroach/drop_index.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS eventstore.events@ca_fill_idx;

cmd/setup/11/postgres/create_index.sql

@@ -0,0 +1,4 @@
+CREATE INDEX IF NOT EXISTS ca_fill_idx ON eventstore.events (
+    event_sequence DESC
+    , instance_id
+) WHERE created_at IS NULL; 

cmd/setup/11/postgres/drop_index.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS eventstore.ca_fill_idx;

cmd/setup/config.go

@@ -66,7 +66,7 @@ type Steps struct {
 	s8AuthTokens         *AuthTokenIndexes
 	s9EventstoreIndexes2 *EventstoreIndexesNew
 	CorrectCreationDate  *CorrectCreationDate
-	s11AddEventCreatedAt *AddEventCreatedAt
+	AddEventCreatedAt    *AddEventCreatedAt
 }
 
 type encryptionKeyConfig struct {

cmd/setup/setup.go

@@ -91,7 +91,8 @@ func Setup(config *Config, steps *Steps, masterKey string) {
 	steps.s8AuthTokens = &AuthTokenIndexes{dbClient: dbClient}
 	steps.s9EventstoreIndexes2 = New09(dbClient)
 	steps.CorrectCreationDate.dbClient = dbClient
-	steps.s11AddEventCreatedAt = &AddEventCreatedAt{dbClient: dbClient, step10: steps.CorrectCreationDate}
+	steps.AddEventCreatedAt.dbClient = dbClient
+	steps.AddEventCreatedAt.step10 = steps.CorrectCreationDate
 
 	err = projection.Create(ctx, dbClient, eventstoreClient, config.Projections, nil, nil)
 	logging.OnError(err).Fatal("unable to start projections")
@@ -129,7 +130,7 @@ func Setup(config *Config, steps *Steps, masterKey string) {
 	logging.OnError(err).Fatal("unable to migrate step 9")
 	err = migration.Migrate(ctx, eventstoreClient, steps.CorrectCreationDate)
 	logging.OnError(err).Fatal("unable to migrate step 10")
-	err = migration.Migrate(ctx, eventstoreClient, steps.s11AddEventCreatedAt)
+	err = migration.Migrate(ctx, eventstoreClient, steps.AddEventCreatedAt)
 	logging.OnError(err).Fatal("unable to migrate step 11")
 
 	for _, repeatableStep := range repeatableSteps {

cmd/setup/steps.yaml

@@ -30,5 +30,9 @@ FirstInstance:
       MachineKey:
         ExpirationDate:
         Type:
+
 CorrectCreationDate:
   FailAfter: 5m
+
+AddEventCreatedAt:
+  BulkAmount: 100