mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-13 03:24:26 +00:00
b5564572bc
This implementation increases parallel write capabilities of the eventstore. Please have a look at the technical advisories: [05](https://zitadel.com/docs/support/advisory/a10005) and [06](https://zitadel.com/docs/support/advisory/a10006). The implementation of eventstore.push is rewritten and stored events are migrated to a new table `eventstore.events2`. If you are using cockroach: make sure that the database user of ZITADEL has `VIEWACTIVITY` grant. This is used to query events.
89 lines
2.4 KiB
Go
89 lines
2.4 KiB
Go
package setup
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"embed"
|
|
"time"
|
|
|
|
"github.com/cockroachdb/cockroach-go/v2/crdb"
|
|
"github.com/zitadel/logging"
|
|
|
|
"github.com/zitadel/zitadel/internal/database"
|
|
)
|
|
|
|
var (
|
|
//go:embed 10/10_create_temp_table.sql
|
|
correctCreationDate10CreateTable string
|
|
//go:embed 10/10_fill_table.sql
|
|
correctCreationDate10FillTable string
|
|
//go:embed 10/cockroach/10_update.sql
|
|
//go:embed 10/postgres/10_update.sql
|
|
correctCreationDate10Update embed.FS
|
|
//go:embed 10/10_count_wrong_events.sql
|
|
correctCreationDate10CountWrongEvents string
|
|
//go:embed 10/10_empty_table.sql
|
|
correctCreationDate10Truncate string
|
|
)
|
|
|
|
type CorrectCreationDate struct {
|
|
dbClient *database.DB
|
|
FailAfter time.Duration
|
|
}
|
|
|
|
func (mig *CorrectCreationDate) Execute(ctx context.Context) (err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, mig.FailAfter)
|
|
defer cancel()
|
|
|
|
for i := 0; ; i++ {
|
|
logging.WithFields("mig", mig.String(), "iteration", i).Debug("start iteration")
|
|
var affected int64
|
|
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 {
|
|
return err
|
|
}
|
|
}
|
|
_, err := tx.ExecContext(ctx, correctCreationDate10CreateTable)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
logging.WithFields("mig", mig.String(), "iteration", i).Debug("temp table created")
|
|
|
|
_, err = tx.ExecContext(ctx, correctCreationDate10Truncate)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(ctx, correctCreationDate10FillTable)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
logging.WithFields("mig", mig.String(), "iteration", i).Debug("temp table filled")
|
|
|
|
res := tx.QueryRowContext(ctx, correctCreationDate10CountWrongEvents)
|
|
if err := res.Scan(&affected); err != nil || affected == 0 {
|
|
return err
|
|
}
|
|
|
|
updateStmt, err := readStmt(correctCreationDate10Update, "10", mig.dbClient.Type(), "10_update.sql")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(ctx, updateStmt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
logging.WithFields("mig", mig.String(), "iteration", i, "count", affected).Debug("creation dates updated")
|
|
return nil
|
|
})
|
|
logging.WithFields("mig", mig.String(), "iteration", i).Debug("end iteration")
|
|
if affected == 0 || err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
func (mig *CorrectCreationDate) String() string {
|
|
return "10_correct_creation_date"
|
|
}
|