From 8da8fbe6ce9b1aae077d73d671d5dd34db887091 Mon Sep 17 00:00:00 2001 From: Silvan Date: Tue, 18 Apr 2023 19:29:04 +0200 Subject: [PATCH] fix(eventstore): correct creation date of events (#5683) * fix: add setup step to correct creation dates * fix(eventstore): replace now with statement ts * fix(step10): correct number * fix: handle wrong instance domain removed events --- .releaserc.js | 3 +- cmd/setup/10.go | 53 ++++++++++++++++++++++ cmd/setup/10.sql | 28 ++++++++++++ cmd/setup/config.go | 19 ++++---- cmd/setup/setup.go | 3 ++ internal/command/system_model.go | 14 ++++-- internal/eventstore/repository/sql/crdb.go | 2 +- 7 files changed, 106 insertions(+), 16 deletions(-) create mode 100644 cmd/setup/10.go create mode 100644 cmd/setup/10.sql diff --git a/.releaserc.js b/.releaserc.js index 250d6d71d..16cb8f56b 100644 --- a/.releaserc.js +++ b/.releaserc.js @@ -1,7 +1,8 @@ module.exports = { branches: [ {name: 'main'}, - {name: '1.87.x', range: '1.87.x', channel: '1.87.x'} + {name: '1.87.x', range: '1.87.x', channel: '1.87.x'}, + {name: 'next'} ], plugins: [ "@semantic-release/commit-analyzer" diff --git a/cmd/setup/10.go b/cmd/setup/10.go new file mode 100644 index 000000000..77db7c6c5 --- /dev/null +++ b/cmd/setup/10.go @@ -0,0 +1,53 @@ +package setup + +import ( + "context" + _ "embed" + + "github.com/zitadel/logging" + + "github.com/zitadel/zitadel/internal/database" +) + +var ( + //go:embed 10.sql + correctCreationDate10 string +) + +type CorrectCreationDate struct { + dbClient *database.DB +} + +func (mig *CorrectCreationDate) Execute(ctx context.Context) (err error) { + tx, err := mig.dbClient.Begin() + if err != nil { + return err + } + if mig.dbClient.Type() == "cockroach" { + if _, err := tx.Exec("SET experimental_enable_temp_tables=on"); err != nil { + return err + } + } + defer func() { + if err != nil { + logging.OnError(tx.Rollback()).Debug("rollback failed") + return + } + err = tx.Commit() + }() + for { + res, err := tx.ExecContext(ctx, correctCreationDate10) + if err != nil { + return err + } + affected, _ := res.RowsAffected() + logging.WithFields("count", affected).Info("creation dates changed") + if affected == 0 { + return nil + } + } +} + +func (mig *CorrectCreationDate) String() string { + return "10_correct_creation_date" +} diff --git a/cmd/setup/10.sql b/cmd/setup/10.sql new file mode 100644 index 000000000..ae2af741d --- /dev/null +++ b/cmd/setup/10.sql @@ -0,0 +1,28 @@ +CREATE temporary TABLE IF NOT EXISTS wrong_events ( + instance_id STRING + , event_sequence INT8 + , current_cd TIMESTAMPTZ + , next_cd TIMESTAMPTZ +); + +TRUNCATE wrong_events; + +INSERT INTO wrong_events ( + SELECT * FROM ( + SELECT + instance_id + , event_sequence + , creation_date AS current_cd + , lead(creation_date) OVER ( + PARTITION BY instance_id + ORDER BY event_sequence DESC + ) AS next_cd + FROM + eventstore.events + ) WHERE + current_cd < next_cd + ORDER BY + event_sequence DESC +); + +UPDATE eventstore.events e SET creation_date = we.next_cd FROM wrong_events we WHERE e.event_sequence = we.event_sequence and e.instance_id = we.instance_id; \ No newline at end of file diff --git a/cmd/setup/config.go b/cmd/setup/config.go index e55767604..eb55ba492 100644 --- a/cmd/setup/config.go +++ b/cmd/setup/config.go @@ -56,15 +56,16 @@ func MustNewConfig(v *viper.Viper) *Config { } type Steps struct { - s1ProjectionTable *ProjectionTable - s2AssetsTable *AssetTable - FirstInstance *FirstInstance - s4EventstoreIndexes *EventstoreIndexesNew - s5LastFailed *LastFailed - s6OwnerRemoveColumns *OwnerRemoveColumns - s7LogstoreTables *LogstoreTables - s8AuthTokens *AuthTokenIndexes - s9EventstoreIndexes2 *EventstoreIndexesNew + s1ProjectionTable *ProjectionTable + s2AssetsTable *AssetTable + FirstInstance *FirstInstance + s4EventstoreIndexes *EventstoreIndexesNew + s5LastFailed *LastFailed + s6OwnerRemoveColumns *OwnerRemoveColumns + s7LogstoreTables *LogstoreTables + s8AuthTokens *AuthTokenIndexes + s9EventstoreIndexes2 *EventstoreIndexesNew + s10EventstoreCreationDate *CorrectCreationDate } type encryptionKeyConfig struct { diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go index bc37decb5..a7deef007 100644 --- a/cmd/setup/setup.go +++ b/cmd/setup/setup.go @@ -88,6 +88,7 @@ func Setup(config *Config, steps *Steps, masterKey string) { steps.s7LogstoreTables = &LogstoreTables{dbClient: dbClient.DB, username: config.Database.Username(), dbType: config.Database.Type()} steps.s8AuthTokens = &AuthTokenIndexes{dbClient: dbClient} steps.s9EventstoreIndexes2 = New09(dbClient) + steps.s10EventstoreCreationDate = &CorrectCreationDate{dbClient: dbClient} err = projection.Create(ctx, dbClient, eventstoreClient, config.Projections, nil, nil) logging.OnError(err).Fatal("unable to start projections") @@ -123,6 +124,8 @@ func Setup(config *Config, steps *Steps, masterKey string) { logging.OnError(err).Fatal("unable to migrate step 8") err = migration.Migrate(ctx, eventstoreClient, steps.s9EventstoreIndexes2) logging.OnError(err).Fatal("unable to migrate step 9") + err = migration.Migrate(ctx, eventstoreClient, steps.s10EventstoreCreationDate) + logging.OnError(err).Fatal("unable to migrate step 10") for _, repeatableStep := range repeatableSteps { err = migration.Migrate(ctx, eventstoreClient, repeatableStep) diff --git a/internal/command/system_model.go b/internal/command/system_model.go index 9ed0684f6..40847cfab 100644 --- a/internal/command/system_model.go +++ b/internal/command/system_model.go @@ -59,12 +59,16 @@ func (wm *SystemConfigWriteModel) Reduce() error { } wm.Instances[e.Aggregate().InstanceID].Domains = append(wm.Instances[e.Aggregate().InstanceID].Domains, e.Domain) case *instance.DomainRemovedEvent: - domains := wm.Instances[e.Aggregate().InstanceID].Domains - for i, domain := range domains { + instance, ok := wm.Instances[e.Aggregate().InstanceID] + if !ok { + continue + } + + for i, domain := range instance.Domains { if domain == e.Domain { - domains[i] = domains[len(domains)-1] - domains[len(domains)-1] = "" - wm.Instances[e.Aggregate().InstanceID].Domains = domains[:len(domains)-1] + instance.Domains[i] = instance.Domains[len(instance.Domains)-1] + instance.Domains[len(instance.Domains)-1] = "" + wm.Instances[e.Aggregate().InstanceID].Domains = instance.Domains[:len(instance.Domains)-1] break } } diff --git a/internal/eventstore/repository/sql/crdb.go b/internal/eventstore/repository/sql/crdb.go index 8cbd1046d..640077625 100644 --- a/internal/eventstore/repository/sql/crdb.go +++ b/internal/eventstore/repository/sql/crdb.go @@ -67,7 +67,7 @@ const ( " $2::VARCHAR AS aggregate_type," + " $3::VARCHAR AS aggregate_id," + " $4::VARCHAR AS aggregate_version," + - " NOW() AS creation_date," + + " statement_timestamp() AS creation_date," + " $5::JSONB AS event_data," + " $6::VARCHAR AS editor_user," + " $7::VARCHAR AS editor_service," +